diff --git a/client/src/constants/env.ts b/client/src/constants/env.ts index b069af3..fc73f15 100644 --- a/client/src/constants/env.ts +++ b/client/src/constants/env.ts @@ -46,7 +46,7 @@ function getApiBaseUrl(env: AppRuntimeEnv): string { return getOptionalEnv('EXPO_PUBLIC_API_BASE_URL_PROD', getOptionalEnv('EXPO_PUBLIC_API_BASE_URL_LOCAL', 'https://api.damer.fun')); } -export const API_BASE_URL = getApiBaseUrl(APP_ENV); +export const API_BASE_URL = getApiBaseUrl(APPpai qa /** * 调试：打印环境变量注入结果（仅开发环境） diff --git a/server/app/tasks/push.py b/server/app/tasks/push.py index 136ca66..a9872dd 100644 --- a/server/app/tasks/push.py +++ b/server/app/tasks/push.py @@ -218,17 +218,27 @@ async def _send_once_async(*, client_user_id: str, local_date: date, slot_index: return {"status": "noop", "reason": "no_log"} if str(log.status) == "sent": return {"status": "noop", "reason": "already_sent"} - if str(log.status) != "scheduled": - # 例如 failed/skipped/sending：不再重复尝试 - return {"status": "noop", "reason": f"not_scheduled:{log.status}"} + if str(log.status) not in ("scheduled", "sending"): + # 例如 failed/skipped：不再重复尝试 + return {"status": "noop", "reason": f"not_retryable:{log.status}"} - # 原子抢占：避免同一条 scheduled 被重复发送（例如 ETA 任务丢失后被补偿重投递） - # 只有从 scheduled -> sending 抢占成功的任务才能继续执行。 + # 原子抢占：避免重复发送 + # - 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 == "scheduled") + | ( + (PushSendLog.status == "sending") + & (PushSendLog.sent_at.is_(None)) + & (PushSendLog.scheduled_at <= steal_cutoff) + ) + ), ) .values(status="sending", error=None) ) @@ -262,71 +272,72 @@ async def _send_once_async(*, client_user_id: str, local_date: date, slot_index: 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: + # 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 + body = "" + try: + reco_payload = await run_reco_payload_async(scene="push", user_profile=user_profile, k=1, locale=reco_locale) + 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 = "给自己一句温柔的话。" + if not body: + body = "给自己一句温柔的话。" - # 5) 发送 - try: + # 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 + await session.commit() + return {"status": "sent", "expo": expo_res} except Exception as e: + # 兜底：任何未预期异常都不要让状态卡在 sending log.status = "failed" - log.error = f"send_failed:{type(e).__name__}" + log.error = f"unexpected:{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]: @@ -356,7 +367,11 @@ def requeue_overdue(*, grace_seconds: int = 300, limit: int = 200) -> dict[str, async with AsyncSessionLocal() as session: q = ( select(PushSendLog) - .where(PushSendLog.status == "scheduled", PushSendLog.scheduled_at <= cutoff) + .where( + PushSendLog.status.in_(("scheduled", "sending")), + PushSendLog.sent_at.is_(None), + PushSendLog.scheduled_at <= cutoff, + ) .order_by(PushSendLog.scheduled_at.asc()) .limit(int(limit)) ) diff --git a/server/app/tasks/reco.py b/server/app/tasks/reco.py index 0e34fc2..2c2fcfa 100644 --- a/server/app/tasks/reco.py +++ b/server/app/tasks/reco.py @@ -53,6 +53,45 @@ async def _run_reco_async( ) +async def run_reco_payload_async( + *, + scene: Scene, + user_profile: UserProfileV1_2, + already_recommended_ids: Optional[list[Any]] = None, + touched_or_viewed_ids: Optional[list[Any]] = None, + k: Optional[int] = None, + now: Optional[datetime] = None, + locale: Optional[str] = None, +) -> dict[str, Any]: + """ + 在“已有事件循环”内运行推荐并返回 payload。 + + 用途： + - 供 Push 等 async 任务内部调用，避免 `asyncio.run()` 嵌套导致 RuntimeError + - 也便于未来在 API/任务间复用 + """ + + effective_now = _ensure_now(now) + effective_locale = _ensure_locale(locale) + + # k 默认按场景（与 generate 保持一致） + if k is None: + k_i = 30 if scene == "feed" else 1 + else: + k_i = int(k) + + result = await _run_reco_async( + scene=scene, + user_profile=user_profile, + already_recommended_ids=list(already_recommended_ids or []), + touched_or_viewed_ids=list(touched_or_viewed_ids or []), + k=int(k_i), + now=effective_now, + locale=effective_locale, + ) + return result.model_dump() + + def _run_reco_sync( *, scene: Scene, diff --git a/server/celerybeat-schedule-shm b/server/celerybeat-schedule-shm new file mode 100644 index 0000000..a285d7b Binary files /dev/null and b/server/celerybeat-schedule-shm differ diff --git a/server/celerybeat-schedule-wal b/server/celerybeat-schedule-wal new file mode 100644 index 0000000..967e4cc Binary files /dev/null and b/server/celerybeat-schedule-wal differ