andros
cfdf3aa20d
Extract netloc parsing into _author_from_post so tests can import and verify it independently. _get_author_nick uses it as fallback instead of duplicating the urlparse logic.
147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
import logging
|
|
import re
|
|
import time
|
|
from urllib.parse import urlparse
|
|
|
|
import httpx
|
|
import jwt
|
|
import requests
|
|
from django.conf import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
APNS_HOST_PRODUCTION = "https://api.push.apple.com"
|
|
APNS_HOST_SANDBOX = "https://api.sandbox.push.apple.com"
|
|
TOKEN_TTL = 3300 # 55 minutes (APNs tokens valid for 1 hour)
|
|
NAME_CACHE_TTL = 3600 # 1 hour
|
|
|
|
_token_cache: dict = {"token": None, "generated_at": 0}
|
|
_name_cache: dict[str, tuple[str, float]] = {}
|
|
|
|
|
|
def _load_private_key() -> str:
|
|
with open(settings.APNS_PRIVATE_KEY_PATH) as f:
|
|
return f.read()
|
|
|
|
|
|
def _get_token() -> str:
|
|
now = time.time()
|
|
if _token_cache["token"] and now - _token_cache["generated_at"] < TOKEN_TTL:
|
|
return _token_cache["token"]
|
|
|
|
private_key = _load_private_key()
|
|
token = jwt.encode(
|
|
{"iss": settings.APNS_TEAM_ID, "iat": int(now)},
|
|
private_key,
|
|
algorithm="ES256",
|
|
headers={"kid": settings.APNS_KEY_ID},
|
|
)
|
|
_token_cache["token"] = token
|
|
_token_cache["generated_at"] = now
|
|
return token
|
|
|
|
|
|
def _author_from_post(post_url: str) -> str:
|
|
feed_url = post_url.split("#")[0]
|
|
return urlparse(feed_url).netloc or feed_url
|
|
|
|
|
|
def _get_author_nick(feed_url: str) -> str:
|
|
now = time.time()
|
|
if feed_url in _name_cache:
|
|
nick, ts = _name_cache[feed_url]
|
|
if now - ts < NAME_CACHE_TTL:
|
|
return nick
|
|
|
|
fallback = _author_from_post(feed_url)
|
|
try:
|
|
response = requests.get(
|
|
f"{settings.RELAY_BASE_URL.rstrip('/')}/feed-content/",
|
|
params={"feed": feed_url},
|
|
timeout=5,
|
|
)
|
|
if response.ok:
|
|
content = response.json().get("data", {}).get("content", "")
|
|
nick_match = re.search(r"^#\+NICK:\s*(.+)$", content, re.MULTILINE)
|
|
title_match = re.search(r"^#\+TITLE:\s*(.+)$", content, re.MULTILINE)
|
|
nick_val = nick_match.group(1).strip() if nick_match else ""
|
|
title_val = title_match.group(1).strip() if title_match else ""
|
|
nick = nick_val or title_val or fallback
|
|
_name_cache[feed_url] = (nick, now)
|
|
return nick
|
|
except Exception:
|
|
logger.warning("Could not fetch nick for %s", feed_url)
|
|
|
|
return fallback
|
|
|
|
|
|
def _build_payload(notification: dict) -> dict:
|
|
ntype = notification.get("type", "")
|
|
post_url = notification.get("post", "")
|
|
author_feed = post_url.split("#")[0]
|
|
author = _get_author_nick(author_feed)
|
|
|
|
bodies: dict[str, str] = {
|
|
"mention": f"{author} mentioned you",
|
|
"reply": f"{author} replied to your post",
|
|
"reaction": f"{author} reacted with {notification.get('emoji', '')}",
|
|
"boost": f"{author} boosted your post",
|
|
}
|
|
body = bodies.get(ntype, "New notification")
|
|
|
|
target_urls: dict[str, str] = {
|
|
"mention": post_url,
|
|
"reply": notification.get("parent", post_url),
|
|
"reaction": notification.get("parent", post_url),
|
|
"boost": notification.get("boosted", post_url),
|
|
}
|
|
target_url = target_urls.get(ntype, post_url)
|
|
|
|
return {
|
|
"aps": {
|
|
"alert": {"title": "Org Social", "body": body},
|
|
"sound": "default",
|
|
},
|
|
"notification": notification,
|
|
"post_url": post_url,
|
|
"target_url": target_url,
|
|
}
|
|
|
|
|
|
def send_push(device_token: str, notification: dict, sandbox: bool = False) -> bool:
|
|
host = APNS_HOST_SANDBOX if sandbox else APNS_HOST_PRODUCTION
|
|
url = f"{host}/3/device/{device_token}"
|
|
|
|
try:
|
|
token = _get_token()
|
|
headers = {
|
|
"authorization": f"bearer {token}",
|
|
"apns-topic": settings.APNS_BUNDLE_ID,
|
|
"apns-push-type": "alert",
|
|
"apns-priority": "10",
|
|
}
|
|
with httpx.Client(http2=True, timeout=10.0) as client:
|
|
response = client.post(url, json=_build_payload(notification), headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
return True
|
|
|
|
logger.warning(
|
|
"APNs %s for token %s...: %s",
|
|
response.status_code,
|
|
device_token[:10],
|
|
response.text,
|
|
)
|
|
|
|
if response.status_code == 410:
|
|
from app.subscriptions.models import Device
|
|
|
|
Device.objects.filter(device_token=device_token).delete()
|
|
logger.info("Removed stale device token %s...", device_token[:10])
|
|
|
|
return False
|
|
|
|
except Exception:
|
|
logger.exception("Failed to send push to %s...", device_token[:10])
|
|
return False
|