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.push_payload import build_home_push_data 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 # 受限抖动：在每段的 [25%, 75%] 区间内取点 # 这样相邻两段的最小间隔为 50% 段长，能显著减少“随机挤在一起”。 mid = seg_start + timedelta(seconds=seg * 0.5) jitter = (random.random() - 0.5) * (seg * 0.5) # [-0.25*seg, +0.25*seg] out.append(mid + timedelta(seconds=jitter)) # 保序（理论上天然有序，这里再保险） out.sort() 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 # 去重：用户推送过的内容尽量不再推送 # 说明： # - 依赖 push_send_log.content_id（需先完成对应 DB 迁移） # - 为避免历史过长导致 already_recommended_ids 过大，这里取“最近若干条已推送内容”近似全量去重 used_ids: list[int] = [] try: qused = ( select(PushSendLog.content_id) .where( PushSendLog.client_user_id == client_user_id, PushSendLog.content_id.is_not(None), PushSendLog.id != log.id, ) # 优先排除最近发送过的内容 .order_by(PushSendLog.local_date.desc(), PushSendLog.slot_index.desc()) .limit(5000) ) 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=build_home_push_data( client_user_id=client_user_id, body=body, scene="push", content_id=picked_content_id, ), ) # 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())