From b7cfcf74955f97733758241b150abb1fed58fb13 Mon Sep 17 00:00:00 2001 From: linkeyu Date: Thu, 23 Jul 2026 10:51:20 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E7=BE=8E=E5=9B=A2?= =?UTF-8?q?=E5=92=8C=E4=BA=AC=E4=B8=9C=20CPS=20=E6=AF=8F=E6=97=A5=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=AF=B9=E8=B4=A6=20(#162)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 需求 - 保留现有后台手动对账逻辑不变 - 每天北京时间 05:00 自动刷新 CPS 对账 - 当前仅处理美团和京东 - 每次按更新时间回拉近 3 天，覆盖延迟更新和订单状态变化 ## 实现 - 新增进程内 CPS 自动对账 worker，并接入应用生命周期 - 美团使用更新时间查询类型 2，京东使用更新时间查询类型 3 - 复用现有仓储层对账及 order_id 幂等更新逻辑 - 美团和京东独立会话、独立异常处理，单个平台失败不阻塞另一平台 - 增加单实例锁、开关、执行小时、回拉天数和轮询间隔配置 - 服务在 05:00 后重启时会补跑当天任务 ## 验证 - `pytest tests/test_cps_reconcile_worker.py tests/test_cps_admin.py tests/test_admin_read.py -q`：27 passed - 相关文件 Ruff 检查通过 - `git diff --check` 通过 --------- Co-authored-by: unknown <798648091@qq.com> Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/162 Co-authored-by: linkeyu Co-committed-by: linkeyu --- .env.example | 7 ++ app/core/config.py | 7 ++ app/core/cps_reconcile_worker.py | 175 +++++++++++++++++++++++++++++ app/main.py | 6 + tests/test_cps_reconcile_worker.py | 84 ++++++++++++++ 5 files changed, 279 insertions(+) create mode 100644 app/core/cps_reconcile_worker.py create mode 100644 tests/test_cps_reconcile_worker.py diff --git a/.env.example b/.env.example index c3ba1d8..a7f92f8 100644 --- a/.env.example +++ b/.env.example @@ -113,6 +113,13 @@ JD_UNION_APP_SECRET= JD_UNION_SITE_ID= JD_UNION_AUTH_KEY= +# 美团 + 京东订单每天北京时间 05:00 自动对账；按更新时间回拉近 3 天，重叠防漏单并刷新状态。 +# 手动对账按钮不受该开关影响。通常保持开启；临时停自动任务时设为 false。 +CPS_AUTO_RECONCILE_ENABLED=true +CPS_AUTO_RECONCILE_RUN_HOUR=5 +CPS_AUTO_RECONCILE_LOOKBACK_DAYS=3 +CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC=60 + # ===== Pricebot 上游 (领券/比价业务透传目标) ===== # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。 diff --git a/app/core/config.py b/app/core/config.py index 2f13587..afae3ee 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -188,6 +188,13 @@ class Settings(BaseSettings): """京东联盟订单查询凭证齐全。""" return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET) + # 美团 + 京东 CPS 订单自动对账：进程内 worker 每天北京时间 05:00 后跑一轮。 + # 按更新时间回拉近 N 天（重叠窗口防漏单并刷新状态），order_id 幂等更新；手动接口不受影响。 + CPS_AUTO_RECONCILE_ENABLED: bool = True + CPS_AUTO_RECONCILE_RUN_HOUR: int = 5 + CPS_AUTO_RECONCILE_LOOKBACK_DAYS: int = 3 + CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC: int = 60 + # ===== 微信服务号(网页授权) ===== # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。 # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 diff --git a/app/core/cps_reconcile_worker.py b/app/core/cps_reconcile_worker.py new file mode 100644 index 0000000..9c0e6d2 --- /dev/null +++ b/app/core/cps_reconcile_worker.py @@ -0,0 +1,175 @@ +"""美团、京东 CPS 订单每日自动对账任务。 + +每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`（默认 05:00）后执行一次，按更新时间 +回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import time +from collections.abc import Iterator +from datetime import date, datetime, timedelta +from pathlib import Path + +from app.admin.repositories import cps as cps_repo +from app.core.config import settings +from app.core.rewards import CN_TZ +from app.db.session import SessionLocal + +logger = logging.getLogger("shagua.cps_reconcile") +_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock" + + +def _cn_now() -> datetime: + return datetime.now(CN_TZ) + + +def _touch_lock() -> None: + with contextlib.suppress(FileNotFoundError): + os.utime(_LOCK_PATH, None) + + +@contextlib.contextmanager +def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]: + """同机多进程保护：同一时间只允许一个 CPS 自动对账 worker 运行。""" + _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) + fd: int | None = None + try: + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + try: + age = time.time() - _LOCK_PATH.stat().st_mtime + except FileNotFoundError: + age = stale_after_sec + 1 + if age > stale_after_sec: + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + fd = None + + if fd is None: + yield False + return + + os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii")) + yield True + finally: + if fd is not None: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + + +def _empty_result() -> dict: + return {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0} + + +def _reconcile_once(now: datetime | None = None) -> dict: + """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。""" + end = (now or _cn_now()).astimezone(CN_TZ) + lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS)) + start = end - timedelta(days=lookback_days) + result = { + "window_start": start.isoformat(), + "window_end": end.isoformat(), + "meituan": None, + "jd": None, + "errors": {}, + } + + if settings.mt_cps_configured: + try: + with SessionLocal() as db: + result["meituan"] = cps_repo.reconcile_orders( + db, + start_time=int(start.timestamp()), + end_time=int(end.timestamp()), + query_time_type=2, + ) + except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台 + result["errors"]["meituan"] = str(exc) + logger.exception("CPS auto reconcile failed platform=meituan") + else: + result["meituan"] = {**_empty_result(), "skipped": "not_configured"} + + _touch_lock() + if settings.jd_union_configured: + try: + with SessionLocal() as db: + result["jd"] = cps_repo.reconcile_jd_orders( + db, + start_time=start, + end_time=end, + query_time_type=3, + ) + except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台 + result["errors"]["jd"] = str(exc) + logger.exception("CPS auto reconcile failed platform=jd") + else: + result["jd"] = {**_empty_result(), "skipped": "not_configured"} + + return result + + +def _should_run(last_run: date | None, now: datetime, run_hour: int) -> bool: + return last_run != now.date() and now.hour >= run_hour + + +async def _run_loop() -> None: + interval = max(30, int(settings.CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC)) + run_hour = min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23) + lock_stale_after = max(interval * 3, 1800) + with _single_instance_lock(lock_stale_after) as lock_acquired: + if not lock_acquired: + logger.warning("CPS auto reconcile skipped: another worker owns lock") + return + await _run_locked_loop(interval, run_hour) + + +async def _run_locked_loop(interval: int, run_hour: int) -> None: + logger.info( + "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s", + run_hour, + interval, + settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS, + ) + last_run: date | None = None + try: + while True: + try: + _touch_lock() + now = _cn_now() + if _should_run(last_run, now, run_hour): + result = await asyncio.to_thread(_reconcile_once, now) + last_run = now.date() + logger.info("CPS auto reconcile done date=%s result=%s", last_run, result) + except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出 + logger.exception("CPS auto reconcile unexpected error") + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("CPS auto reconcile worker stopped") + raise + + +def start_cps_reconcile_worker() -> asyncio.Task | None: + if not settings.CPS_AUTO_RECONCILE_ENABLED: + logger.info("CPS auto reconcile disabled") + return None + if not settings.mt_cps_configured and not settings.jd_union_configured: + logger.warning("CPS auto reconcile not started: Meituan and JD credentials are missing") + return None + return asyncio.create_task(_run_loop(), name="cps-auto-reconcile") + + +async def stop_cps_reconcile_worker(task: asyncio.Task | None) -> None: + if task is None: + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task diff --git a/app/main.py b/app/main.py index 0a1fcd1..dcff43c 100644 --- a/app/main.py +++ b/app/main.py @@ -43,6 +43,10 @@ from app.api.v1.user import router as user_router from app.api.v1.wallet import router as wallet_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings +from app.core.cps_reconcile_worker import ( + start_cps_reconcile_worker, + stop_cps_reconcile_worker, +) from app.core.daily_exchange_worker import ( start_daily_exchange_worker, stop_daily_exchange_worker, @@ -89,6 +93,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: except Exception: # noqa: BLE001 logger.exception("reverse_geocoder 预热失败(城市反查将在首个请求时懒加载)") reconcile_task = start_withdraw_reconcile_worker() + cps_reconcile_task = start_cps_reconcile_worker() heartbeat_task = start_heartbeat_monitor() daily_exchange_task = start_daily_exchange_worker() observe_task = start_observe_worker() @@ -98,6 +103,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: finally: await stop_heartbeat_monitor(heartbeat_task) await stop_withdraw_reconcile_worker(reconcile_task) + await stop_cps_reconcile_worker(cps_reconcile_task) await stop_daily_exchange_worker(daily_exchange_task) await stop_observe_worker(observe_task) await stop_inactivity_reset_worker(inactivity_task) diff --git a/tests/test_cps_reconcile_worker.py b/tests/test_cps_reconcile_worker.py new file mode 100644 index 0000000..953533a --- /dev/null +++ b/tests/test_cps_reconcile_worker.py @@ -0,0 +1,84 @@ +"""美团、京东 CPS 每日自动对账 worker。""" +from __future__ import annotations + +from datetime import date, datetime, timedelta + +from app.core import cps_reconcile_worker as worker +from app.core.config import settings +from app.core.rewards import CN_TZ +from app.integrations.meituan import MeituanCpsError + + +def _configure_platforms(monkeypatch) -> None: + monkeypatch.setattr(settings, "MT_CPS_APP_KEY", "mt-key") + monkeypatch.setattr(settings, "MT_CPS_APP_SECRET", "mt-secret") + monkeypatch.setattr(settings, "JD_UNION_APP_KEY", "jd-key") + monkeypatch.setattr(settings, "JD_UNION_APP_SECRET", "jd-secret") + monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_LOOKBACK_DAYS", 3) + + +def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None: + _configure_platforms(monkeypatch) + calls: dict[str, dict] = {} + + def fake_meituan(db, **kwargs): + calls["meituan"] = kwargs + return {"fetched": 2, "inserted": 1, "updated": 1, "pages": 1} + + def fake_jd(db, **kwargs): + calls["jd"] = kwargs + return {"fetched": 3, "inserted": 2, "updated": 1, "pages": 2} + + monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fake_meituan) + monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd) + monkeypatch.setattr(worker, "_touch_lock", lambda: None) + now = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ) + + result = worker._reconcile_once(now) + + assert result["errors"] == {} + assert result["meituan"]["fetched"] == 2 + assert result["jd"]["fetched"] == 3 + assert calls["meituan"]["query_time_type"] == 2 + assert calls["jd"]["query_time_type"] == 3 + assert calls["meituan"]["end_time"] == int(now.timestamp()) + assert calls["meituan"]["start_time"] == int((now - timedelta(days=3)).timestamp()) + assert calls["jd"]["start_time"] == now - timedelta(days=3) + assert calls["jd"]["end_time"] == now + + +def test_meituan_failure_does_not_block_jd(monkeypatch) -> None: + _configure_platforms(monkeypatch) + jd_called = False + + def fail_meituan(db, **kwargs): + raise MeituanCpsError("temporary failure") + + def fake_jd(db, **kwargs): + nonlocal jd_called + jd_called = True + return {"fetched": 1, "inserted": 1, "updated": 0, "pages": 1} + + monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fail_meituan) + monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd) + monkeypatch.setattr(worker, "_touch_lock", lambda: None) + + result = worker._reconcile_once(datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)) + + assert result["errors"]["meituan"] == "temporary failure" + assert jd_called is True + assert result["jd"]["fetched"] == 1 + + +def test_should_run_once_after_five_am() -> None: + before = datetime(2026, 7, 22, 4, 59, tzinfo=CN_TZ) + at_five = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ) + + assert worker._should_run(None, before, 5) is False + assert worker._should_run(None, at_five, 5) is True + assert worker._should_run(date(2026, 7, 22), at_five, 5) is False + + +def test_start_worker_respects_auto_switch(monkeypatch) -> None: + monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False) + assert worker.start_cps_reconcile_worker() is None