313 lines
11 KiB
Python
313 lines
11 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
|
||
|
||
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:
|
||
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 发送任务
|
||
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
|
||
|
||
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"}
|
||
|
||
# 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"}
|
||
|
||
# 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()
|
||
)
|
||
|
||
# 直接复用 reco 的 Celery 任务实现（同步函数）
|
||
from app.tasks.reco import generate as reco_generate
|
||
|
||
reco_payload = reco_generate(scene="push", user_profile=user_profile.model_dump(), k=1, locale=reco_locale)
|
||
body = ""
|
||
try:
|
||
items = (reco_payload or {}).get("items") or []
|
||
if items and isinstance(items, list):
|
||
body = str(items[0].get("text") or "").strip()
|
||
except Exception:
|
||
body = ""
|
||
|
||
if not body:
|
||
body = "给自己一句温柔的话。"
|
||
|
||
# 5) 发送
|
||
try:
|
||
expo_res = await _send_expo_push(
|
||
to=str(token.push_token),
|
||
title=title,
|
||
body=body,
|
||
data={"client_user_id": client_user_id, "scene": "push"},
|
||
)
|
||
except Exception as e:
|
||
log.status = "failed"
|
||
log.error = f"send_failed:{type(e).__name__}"
|
||
await session.commit()
|
||
return {"status": "failed", "error": str(e)}
|
||
|
||
# 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
|
||
await session.commit()
|
||
return {"status": "sent", "expo": expo_res}
|
||
|
||
|
||
@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)))
|
||
|