From eef5210c994c196bcb37672085c0128990bdaeb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=95=E6=96=B0=E9=9B=A8?= Date: Wed, 11 Feb 2026 13:50:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=E6=9B=B4=E6=96=B0=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1push?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client/src/constants/env.ts | 2 +- server/app/tasks/push.py | 129 ++++++++++++++++++--------------- server/app/tasks/reco.py | 39 ++++++++++ server/celerybeat-schedule-shm | Bin 0 -> 32768 bytes server/celerybeat-schedule-wal | Bin 0 -> 238992 bytes 5 files changed, 112 insertions(+), 58 deletions(-) create mode 100644 server/celerybeat-schedule-shm create mode 100644 server/celerybeat-schedule-wal 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 0000000000000000000000000000000000000000..a285d7b33d522575ca6b79176a80401bb108e9b9 GIT binary patch literal 32768 zcmeI)IWEOf6bIm+dFGj)d7eGE%K z`tIAe_&ok9wMAF`_pjHlfi(mO5FkK+009C72oNAZfB*pk1PBlyK!5-N0t5&UAV7cs z0RjXF5FkK+009C72oNAZfB*pk1PBlyK!5-N0t5&UAV7cs0RjXF{A+=lV&h6fn$nWC zbfhcY=}mtIGnA2HAIwCiGLyM1WGO4DWIdbN&TjT|_^;Ir|2G6`Q=i5(r#0>AOesC- z%RtH*&S=Innd!`CK8sn-YSyxmt?Xnk2dQo{5(EekAV7cs0RjXF5FkK+009C72oNAZ sfB*pk1PBlyK!5-N0t5&UAV7cs0RjXF5FkK+009C72oNAZ;CBVy0WB#leE?WIR5;lzp7*f3puU-JNK;*xkdPS@l>Ai~RT~OD}?l0MGHoIYG z76=mgBSfeptz~RakJEZrThBYiKA$aT`yTkPJzk}H=bu+mlzaBn%30-F|J1(P-fwny z$lxuB*#_9me({0BZg%H4pB?5sGGBhbdFv}SDJ`!}Qj|Ky{888YhdtX5uc$R%SEsa3 zopz0UsLcL&$JYMJ_@*yhP%9tw)Gko``~16V7x)jEU&9Xs5I_I{1Q0*~0R#|000Eo8 zW;Nw$*mj*rXsMV`H+b&6KJ)i0Md>%1^~T<`8g1zk36TnEq9qcF#rs;+a8yLnabf7y zVG$RpzCecvX@*|AEce7>`Cfg-j^SV5q=nROHDK-}VC^KB*-3D4CzTPoYc19z_cyq? z!&v0=9x1C<&Yk7kvrRtJk&eZ+SVA>)Zh z(}~_#xLY2?fN_RaptQ`8VKejp&5T0X!6)+^oB&&6t%L;qjv!a#Rzn zk2DnS7KunO92cQPy4R2+i_Vk(W%uT*s;H=ZelertxVl92ALAM_fcUc(w*6E^^n+)Yi&V(3?9yv!6?#Np&~P4hC0A z)>twT49c&lvc95xPGD}J)djg5kfPY_69WsMK009ILKmY**5I_I{1d2g`UVvV}j6QM| z!|Ch=v{CE@+JD#i{YP$?zLj2}7%no50RaRMKmY**5I_I{1Q0*~f#C#fk$`RU1igSZ zTrcp%71zIfxb?_Rs~0fm7BJ@+7;Y3ij{pJ)AbG+#&Am#t7LXhKKmY**5I_I{1Q0*~0R&Euz(DOPPlH}JIP3q1%=+*2 zS^uL`@~lEIKC{9@N@SKPXi3UB_9Jq+0&}+ixq5?bd4}9q1dn}%wd}xx3El(U$1MaP zmmQdsXW4;fy|ybA3X7mB!pTHLt_E@(r35kLR|1Q0*~0R#|0U=#)D1xoK< z@#noCPcL9xWuyc3`FerXqqrWyzx?IXFTCTu`^WSGqj(kLtq?!}0R#|0009ILKmY** z5IAW9wn)IXd4gVG^>FJEbY3)Xm8u@zG}sH6wE*i8oV1URH%0&f1Q0*~0R#|00D<8I z=mqEnG{jO?EwkVS9n#r7!$Cv3#G;>IJg3fd5c294TW!009ILKmY**5I_Kd zk`$m9pckMQSi1KC)l`u43yi22=yNyC4hH4g4Qecz2nIKJ?ySmPMfCDHfw_TJBOrh8 zHh5I_I{1Q0*~0R#{ja{{(Vz_$4* zdx1WAxS5qdt?vu?t~UOmbJfw;tzN*KTfl#)=H4;K!hjJ#009ILKmY**5J12w0t2;c zJq>!DNNA~;P~{EJ4>Ny@QIvk8S#Qj&Y2PIhA{Ej^OC%JF_qC|usEDNF!qBV3A}&&W zfesPU483+)?uo_nz50wD!@s^s3#r{|z}!i|+DS08li=V^DkE~&TC7LzZ*X&mvB>8= zQdX^;JIl9cn|!7t9gAzRglg#CE}?}qEhPt_SFTMaL@d#nl$-5xbBVbrAFm3g6TPu; zw>*de^WWCHmtVceXmgp{=Fhe=y)JiX);P>J4yTd{E!1HiS(~p-(OuJwXk)a|ywBWk zbWH7+@7ohv6HKe}_OY>a>lvziqbBySO%mF+co!qRur_Gp^8R7Ax>^E4C!2Go2 zr{fiCdSfXeKdn>O3tTvU!BZP|3NO8YQ!b$#2LcEnfB*srAb+<7`8Y4gmxZKmY**5I_I{ z1WuX2z{NMoyZ_qY-T$fPzipO3xchI04P)l+zeiEZzdqf%`~STm@BR4#A}S=kpQSJVbM4f*hNB#Rt~dT)DB$$x|jY>$P2}P*?<25l$u|^0xeW zF1xpSb+y1l+7tYTUQcSQalu)=^BqG6ZT!a$oUc-zjPWaps+vt?a6~~RvHPg^t zvvcDctUFFqFL3bf9nOK`Ag6 z=Lzu0+wtwotoz-MIkglRB62uk6EIXSaK)JE1ui&a$+h?2_VpmWz!6ELW;WFi=p%Xw7g{#@(ja{_Y% ztwuop-Yj?A);fhxxS#vhfyd6;O)oIYS3uql0R#|0009ILKmY**5I_KdlOkY?1ZOPWO{*ti#ICFFW`h~ zV5&SLwpi7`6U9{xoYUl+7q!ObVSa&;%`foIXJ4Mvb+C1Rwih^{_>cMz_}}s${L?Q1 zyaNIVAb!hG!$bf91Q0*~0R#|00D+Qh z+@~|Yz)An$*Z~3vAb=Mo4B7gt_2q1s}0tg_000Ib%QGxuuz~%ybf$&r7UOqhi!&>GS7^9~H z!$trB1Q0*~0R#|0plAiMy?~rwpq%*yoKOucRFn(fx2l0>i>n%#r0G;ik*t$Xq>6jd)!E>G8SuUTWO zq8AuhFYx5qzy9V&H$DCky}&43_IL{f5I_I{1Q0*~0R#|0009I}x;M4-5I_I{1Q0*~ff5rq)(d#)1)R_ej8{CLtG4dRA1tn3V7#Jy z;XZ3@)${@*>ji3df9-phT=-H8y+DawqB&dy5I_I{1Q0*~0R#|00D&|!R009ILKmY**5I~@41Y|Ez>1nuK5eY37 z6Y2)foip6o!Sx%>dSh=|jka`&gh+)n(Gm&8;(aY@I4UCPxG?nUu!xIPU!X&TG()dl zmV08ce6K!Z$MCOj(n4yt8ZdVfuyzv6>?An2lgfzPwHE7<`y1TcVJz}_kCas_=g#u& z*(RUqNXOz@ETI~@w@YXtO-soE=#^`e2@y+lCgoS;dlb5}&GvS*ShxPix_zNQ64#BC^4=H863tyKbhoR{o*#=KjEE&$K{OpSx*xaAdte#g-}S zE=s9e=mm=A;>kb|KmY**5I_I{1Q0*~0R#{jLLh%Hu(iNm;G7CkUfZYqp7{laAm=Fr z5I_I{1Q0*~0R#{zKp@)-3|zdiiunbcPz_90lt(98cfJ3sxT=A(6lHs#HMSb&7Z_PD zaM$u1Z@7QUE5D!@DBv>3OAtT+0R#|0009ILKmY**5IE%m`FnwF1@;1uRVM%O=X;D_ z(hHpO50V`sfB*srAb!_MH~tN$mI9@_9>KoP z-m?C3@zZb93lz=8lYtl9fwgP*BkN@)q&mF&w&0v0k zA;@_O0R#|0009ILKmY**3J}Ql0&+cq8s-;pLN)L?MREVgss@e}S2Zw2QCx3YV;jf( z0wbGWVCw(gI%D%|Pp_vJDBv>3OAtT+0R#|0009ILKmY**5IE%m`FjDqz+T|`1CjIp z`->a4(+iyP50V`sfB*srAb&^a2jQBy)HOAb