446 lines
17 KiB
Python
446 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import random
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime, time, timedelta, timezone
|
||
from typing import Any, Optional
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import httpx
|
||
from celery import current_app, shared_task
|
||
from sqlalchemy import select, update
|
||
|
||
from app.core.config import get_settings
|
||
from app.db.models.push_preference import PushPreference
|
||
from app.db.models.push_send_log import PushSendLog
|
||
from app.db.models.push_token import PushToken
|
||
from app.db.session import AsyncSessionLocal
|
||
from app.features.personalized_reco.content_repository.types import normalize_locale
|
||
from app.features.user_profile_scoring.scoring import build_user_profile_from_questionnaire
|
||
from app.features.user_profile_scoring.types import QuestionnaireAnswersV1_2, UserProfileV1_2
|
||
|
||
|
||
def _ensure_tz(tz_name: Optional[str]) -> ZoneInfo:
|
||
raw = (tz_name or "").strip()
|
||
if not raw:
|
||
return ZoneInfo("UTC")
|
||
try:
|
||
return ZoneInfo(raw)
|
||
except Exception:
|
||
return ZoneInfo("UTC")
|
||
|
||
|
||
def _pick_reco_locale(pref_locale: Optional[str]) -> str:
|
||
"""
|
||
将客户端 locale（如 zh-CN/en/zh-TW）映射到推荐系统 locale（en/tc）。
|
||
"""
|
||
|
||
raw = (pref_locale or "").strip().lower()
|
||
if raw.startswith("zh"):
|
||
return "tc"
|
||
# 其他语言在当前版本统一回退到 en（与现有 reco/ legal 链路一致）
|
||
return "en"
|
||
|
||
|
||
def _pick_title(locale: str) -> str:
|
||
# 需求：tc 语言使用繁体标题
|
||
return "每日推薦" if str(locale) == "tc" else "Daily Reminder"
|
||
|
||
|
||
async def _send_expo_push(*, to: str, title: str, body: str, data: Optional[dict[str, Any]] = None) -> dict[str, Any]:
|
||
settings = get_settings()
|
||
url = "https://exp.host/--/api/v2/push/send"
|
||
|
||
headers: dict[str, str] = {"Content-Type": "application/json"}
|
||
if settings.expo_access_token:
|
||
headers["Authorization"] = f"Bearer {settings.expo_access_token}"
|
||
|
||
payload: dict[str, Any] = {"to": to, "title": title, "body": body}
|
||
if data:
|
||
payload["data"] = data
|
||
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
res = await client.post(url, headers=headers, json=payload)
|
||
res.raise_for_status()
|
||
return res.json()
|
||
|
||
|
||
def _uniform_jitter_times(*, start: datetime, end: datetime, n: int) -> list[datetime]:
|
||
"""
|
||
将窗口均匀切分为 n 个区间，并在每段内随机取一个时间点（抖动）。
|
||
"""
|
||
|
||
if n <= 0:
|
||
return []
|
||
total = (end - start).total_seconds()
|
||
if total <= 0:
|
||
return []
|
||
|
||
out: list[datetime] = []
|
||
for i in range(n):
|
||
seg_start = start + timedelta(seconds=total * i / n)
|
||
seg_end = start + timedelta(seconds=total * (i + 1) / n)
|
||
seg = (seg_end - seg_start).total_seconds()
|
||
if seg <= 0:
|
||
out.append(seg_start)
|
||
continue
|
||
jitter = random.random() * seg
|
||
out.append(seg_start + timedelta(seconds=jitter))
|
||
return out
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _ScheduleTarget:
|
||
local_date: date
|
||
start_local: datetime
|
||
end_local: datetime
|
||
|
||
|
||
def _pick_schedule_target(*, now_utc: datetime, tz: ZoneInfo) -> _ScheduleTarget:
|
||
"""
|
||
为某个用户选择“要生成计划的本地日期”：
|
||
- 若当前本地时间 < 09:00：生成“今天”
|
||
- 否则：生成“明天”
|
||
|
||
目的：确保生成的时间点尽量都在未来，避免任务 ETA 立刻触发导致体验异常。
|
||
"""
|
||
|
||
now_local = now_utc.astimezone(tz)
|
||
today = now_local.date()
|
||
day = today if now_local.time() < time(9, 0) else (today + timedelta(days=1))
|
||
start_local = datetime.combine(day, time(9, 0), tzinfo=tz)
|
||
end_local = datetime.combine(day + timedelta(days=1), time(0, 0), tzinfo=tz) # 24:00
|
||
return _ScheduleTarget(local_date=day, start_local=start_local, end_local=end_local)
|
||
|
||
|
||
async def _generate_schedule_once(*, now_utc: datetime, max_users: int = 5000) -> dict[str, int]:
|
||
"""
|
||
为所有开启每日提醒的用户生成当天/明天的推送计划，并投递 ETA 发送任务。
|
||
|
||
幂等：
|
||
- `push_send_log` 唯一键（client_user_id + local_date + slot_index）保证同一天不会重复排程
|
||
- 即使重复运行，最多只会补齐缺失 slot
|
||
"""
|
||
|
||
created = 0
|
||
scheduled = 0
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
q = (
|
||
select(PushPreference)
|
||
.where(PushPreference.enabled == True, PushPreference.times_per_day > 0) # noqa: E712
|
||
.limit(int(max_users))
|
||
)
|
||
rows = await session.execute(q)
|
||
prefs = list(rows.scalars().all())
|
||
|
||
for pref in prefs:
|
||
tz = _ensure_tz(pref.timezone)
|
||
target = _pick_schedule_target(now_utc=now_utc, tz=tz)
|
||
|
||
n = int(pref.times_per_day or 0)
|
||
n = max(0, min(5, n))
|
||
if n <= 0:
|
||
continue
|
||
|
||
times_local = _uniform_jitter_times(start=target.start_local, end=target.end_local, n=n)
|
||
|
||
for idx, dt_local in enumerate(times_local, start=1):
|
||
dt_utc = dt_local.astimezone(timezone.utc)
|
||
|
||
# 先尝试插入日志（幂等）
|
||
log = PushSendLog(
|
||
client_user_id=pref.client_user_id,
|
||
local_date=target.local_date,
|
||
slot_index=int(idx),
|
||
scheduled_at=dt_utc.replace(tzinfo=None), # DB 存 naive（约定为 UTC）
|
||
status="scheduled",
|
||
)
|
||
session.add(log)
|
||
try:
|
||
await session.commit()
|
||
except Exception:
|
||
await session.rollback()
|
||
# 可能已存在（唯一键冲突），跳过
|
||
continue
|
||
|
||
created += 1
|
||
|
||
# 投递 ETA 发送任务
|
||
try:
|
||
current_app.send_task(
|
||
"tasks.push.send_scheduled",
|
||
kwargs={
|
||
"client_user_id": pref.client_user_id,
|
||
"local_date": target.local_date.isoformat(),
|
||
"slot_index": int(idx),
|
||
},
|
||
eta=dt_utc,
|
||
)
|
||
scheduled += 1
|
||
except Exception as e:
|
||
# 关键：如果投递失败（例如 broker 短暂不可用），不要让 log 永远卡在 scheduled
|
||
log.status = "failed"
|
||
log.error = f"enqueue_failed:{type(e).__name__}"
|
||
try:
|
||
await session.commit()
|
||
except Exception:
|
||
await session.rollback()
|
||
|
||
return {"created": created, "scheduled": scheduled}
|
||
|
||
|
||
@shared_task(name="tasks.push.generate_daily_schedule")
|
||
def generate_daily_schedule(*, max_users: int = 5000) -> dict[str, int]:
|
||
"""
|
||
生成每日推送计划（入口任务）。
|
||
|
||
说明：
|
||
- 建议由 celery beat 定时触发（见 app/worker.py）
|
||
- 入参尽量小，避免 Redis 队列膨胀
|
||
"""
|
||
|
||
now_utc = datetime.now(timezone.utc)
|
||
return asyncio.run(_generate_schedule_once(now_utc=now_utc, max_users=int(max_users)))
|
||
|
||
|
||
async def _send_once_async(*, client_user_id: str, local_date: date, slot_index: int) -> dict[str, Any]:
|
||
async with AsyncSessionLocal() as session:
|
||
# 1) 查 log，避免重复发送
|
||
qlog = select(PushSendLog).where(
|
||
PushSendLog.client_user_id == client_user_id,
|
||
PushSendLog.local_date == local_date,
|
||
PushSendLog.slot_index == int(slot_index),
|
||
)
|
||
rlog = await session.execute(qlog)
|
||
log = rlog.scalar_one_or_none()
|
||
if log is None:
|
||
return {"status": "noop", "reason": "no_log"}
|
||
if str(log.status) == "sent":
|
||
return {"status": "noop", "reason": "already_sent"}
|
||
if str(log.status) not in ("scheduled", "sending"):
|
||
# 例如 failed/skipped：不再重复尝试
|
||
return {"status": "noop", "reason": f"not_retryable:{log.status}"}
|
||
|
||
# 原子抢占：避免重复发送
|
||
# - scheduled：正常抢占 scheduled -> sending
|
||
# - sending：如果长时间卡在 sending（进程崩溃/网络异常等），允许“超时接管”继续执行
|
||
now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None)
|
||
steal_cutoff = now_utc_naive - timedelta(minutes=10)
|
||
res = await session.execute(
|
||
update(PushSendLog)
|
||
.where(
|
||
PushSendLog.id == log.id,
|
||
(
|
||
(PushSendLog.status == "scheduled")
|
||
| (
|
||
(PushSendLog.status == "sending")
|
||
& (PushSendLog.sent_at.is_(None))
|
||
& (PushSendLog.scheduled_at <= steal_cutoff)
|
||
)
|
||
),
|
||
)
|
||
.values(status="sending", error=None)
|
||
)
|
||
await session.commit()
|
||
if (res.rowcount or 0) <= 0:
|
||
return {"status": "noop", "reason": "already_in_progress_or_processed"}
|
||
log.status = "sending"
|
||
|
||
# 2) 当前偏好检查（用户可能中途关闭/改次数）
|
||
qpref = select(PushPreference).where(PushPreference.client_user_id == client_user_id)
|
||
rpref = await session.execute(qpref)
|
||
pref = rpref.scalar_one_or_none()
|
||
if pref is None or (not bool(pref.enabled)) or int(pref.times_per_day or 0) < int(slot_index):
|
||
log.status = "skipped"
|
||
log.error = "disabled_or_reduced"
|
||
await session.commit()
|
||
return {"status": "skipped"}
|
||
|
||
# 3) 找 token
|
||
qtok = (
|
||
select(PushToken)
|
||
.where(PushToken.client_user_id == client_user_id, PushToken.is_active == True) # noqa: E712
|
||
.order_by(PushToken.last_seen_at.desc())
|
||
.limit(1)
|
||
)
|
||
rtok = await session.execute(qtok)
|
||
token = rtok.scalar_one_or_none()
|
||
if token is None:
|
||
log.status = "failed"
|
||
log.error = "no_active_token"
|
||
await session.commit()
|
||
return {"status": "failed", "reason": "no_active_token"}
|
||
|
||
try:
|
||
# 4) 生成文案（复用推荐模块 push 场景）
|
||
reco_locale = str(normalize_locale(_pick_reco_locale(pref.locale)))
|
||
title = _pick_title(reco_locale)
|
||
|
||
if pref.user_profile_json:
|
||
user_profile = UserProfileV1_2.model_validate(pref.user_profile_json)
|
||
else:
|
||
# 无画像：用“全跳过”的默认画像（降个性化/降风险）
|
||
user_profile = UserProfileV1_2.model_validate(
|
||
build_user_profile_from_questionnaire(QuestionnaireAnswersV1_2()).model_dump()
|
||
)
|
||
|
||
# 关键：这里不能调用 tasks.reco.generate（内部会 asyncio.run），否则会嵌套事件循环崩溃。
|
||
from app.tasks.reco import run_reco_payload_async
|
||
|
||
# 去重：同一用户同一天尽量不重复推送相同 content
|
||
used_ids: list[int] = []
|
||
try:
|
||
qused = (
|
||
select(PushSendLog.content_id)
|
||
.where(
|
||
PushSendLog.client_user_id == client_user_id,
|
||
PushSendLog.local_date == local_date,
|
||
PushSendLog.content_id.is_not(None),
|
||
PushSendLog.id != log.id,
|
||
)
|
||
.order_by(PushSendLog.slot_index.asc())
|
||
)
|
||
rused = await session.execute(qused)
|
||
used_ids = [int(x) for x in rused.scalars().all() if x is not None]
|
||
except Exception:
|
||
used_ids = []
|
||
|
||
body = ""
|
||
picked_content_id: int | None = None
|
||
try:
|
||
reco_payload = await run_reco_payload_async(
|
||
scene="push",
|
||
user_profile=user_profile,
|
||
k=3,
|
||
locale=reco_locale,
|
||
already_recommended_ids=used_ids,
|
||
)
|
||
items = (reco_payload or {}).get("items") or []
|
||
if items and isinstance(items, list):
|
||
for it in items:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
cid = it.get("content_id")
|
||
txt = str(it.get("text") or "").strip()
|
||
if not txt:
|
||
continue
|
||
if cid is not None:
|
||
try:
|
||
cid_i = int(cid)
|
||
except Exception:
|
||
cid_i = None
|
||
else:
|
||
cid_i = None
|
||
if cid_i is not None and cid_i in used_ids:
|
||
continue
|
||
picked_content_id = cid_i
|
||
body = txt
|
||
break
|
||
except Exception:
|
||
body = ""
|
||
|
||
if not body:
|
||
# tc 语言兜底文案使用繁体
|
||
body = "給自己一句溫柔的話。" if reco_locale == "tc" else "给自己一句温柔的话。"
|
||
|
||
# 5) 发送
|
||
expo_res = await _send_expo_push(
|
||
to=str(token.push_token),
|
||
title=title,
|
||
body=body,
|
||
data={"client_user_id": client_user_id, "scene": "push"},
|
||
)
|
||
|
||
# 6) 解析 Expo 回执，必要时停用 token
|
||
try:
|
||
data_list = (expo_res or {}).get("data") or []
|
||
if data_list and isinstance(data_list, list):
|
||
first = data_list[0] or {}
|
||
if first.get("status") == "error":
|
||
details = first.get("details") or {}
|
||
err = str(details.get("error") or first.get("message") or "expo_error")
|
||
log.status = "failed"
|
||
log.error = err
|
||
if "DeviceNotRegistered" in err:
|
||
token.is_active = False
|
||
await session.commit()
|
||
return {"status": "failed", "expo": expo_res}
|
||
except Exception:
|
||
# 忽略解析异常，继续按成功处理
|
||
pass
|
||
|
||
log.status = "sent"
|
||
log.sent_at = datetime.now(timezone.utc).replace(tzinfo=None)
|
||
log.error = None
|
||
log.title = title
|
||
log.body = body
|
||
log.content_id = picked_content_id
|
||
await session.commit()
|
||
return {"status": "sent", "expo": expo_res}
|
||
except Exception as e:
|
||
# 兜底：任何未预期异常都不要让状态卡在 sending
|
||
log.status = "failed"
|
||
log.error = f"unexpected:{type(e).__name__}"
|
||
await session.commit()
|
||
return {"status": "failed", "error": str(e)}
|
||
|
||
|
||
@shared_task(name="tasks.push.send_scheduled")
|
||
def send_scheduled(*, client_user_id: str, local_date: str, slot_index: int) -> dict[str, Any]:
|
||
"""
|
||
ETA 发送任务：发送某用户某天第 slot 条推送。
|
||
"""
|
||
|
||
d = date.fromisoformat(str(local_date))
|
||
return asyncio.run(_send_once_async(client_user_id=str(client_user_id), local_date=d, slot_index=int(slot_index)))
|
||
|
||
|
||
@shared_task(name="tasks.push.requeue_overdue")
|
||
def requeue_overdue(*, grace_seconds: int = 300, limit: int = 200) -> dict[str, Any]:
|
||
"""
|
||
补偿任务：扫描“已到时间但仍处于 scheduled”的记录并重新投递发送任务。
|
||
|
||
目的：
|
||
- 覆盖 broker 短暂不可用、worker 重启、ETA 任务丢失等导致的“scheduled 卡住”
|
||
- 与 send_scheduled 内部的原子状态抢占配合，避免重复发送
|
||
"""
|
||
|
||
now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None)
|
||
cutoff = now_utc_naive - timedelta(seconds=int(grace_seconds))
|
||
|
||
async def _run() -> dict[str, Any]:
|
||
requeued = 0
|
||
async with AsyncSessionLocal() as session:
|
||
q = (
|
||
select(PushSendLog)
|
||
.where(
|
||
PushSendLog.status.in_(("scheduled", "sending")),
|
||
PushSendLog.sent_at.is_(None),
|
||
PushSendLog.scheduled_at <= cutoff,
|
||
)
|
||
.order_by(PushSendLog.scheduled_at.asc())
|
||
.limit(int(limit))
|
||
)
|
||
rows = await session.execute(q)
|
||
logs = list(rows.scalars().all())
|
||
for log in logs:
|
||
try:
|
||
current_app.send_task(
|
||
"tasks.push.send_scheduled",
|
||
kwargs={
|
||
"client_user_id": str(log.client_user_id),
|
||
"local_date": str(log.local_date),
|
||
"slot_index": int(log.slot_index),
|
||
},
|
||
)
|
||
requeued += 1
|
||
except Exception:
|
||
# 忽略单条投递失败，交给下一轮补偿
|
||
continue
|
||
return {"status": "ok", "requeued": requeued, "cutoff": cutoff.isoformat()}
|
||
|
||
return asyncio.run(_run())
|
||
|