WonderableAI/shaguabijia-app-server
11
0
Fork 0
Code Issues Pull Requests 12 Actions Packages Projects Releases Wiki Activity

feat(cps): add reconciliation audit logs

Browse Source
This commit is contained in:
unknown
2026-07-23 11:35:55 +08:00
parent b7cfcf7495
commit 2004e8c910
3 changed files with 462 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/admin/repositories/cps.py
+71 -16
View File
@@ -5,6 +5,7 @@
"""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from typing import Any
@@ -15,12 +16,14 @@ from sqlalchemy.orm import Session
from app.admin.repositories.queries import _as_utc, offset_paginate
from app.integrations import jd_union, meituan
from app.repositories import cps_link as cps_link_repo
from app.models.cps_activity import CpsActivity
from app.models.cps_group import CpsGroup
from app.models.cps_link import CpsClick, CpsLink
from app.models.cps_order import CpsOrder
from app.models.cps_wx_user import CpsWxUser
from app.repositories import cps_link as cps_link_repo
logger = logging.getLogger("shagua.cps_reconcile")
# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
_INVALID_STATUS = {"4", "5"}
@@ -378,18 +381,35 @@ def effective_commission_cents(order: CpsOrder) -> int:
def reconcile_orders(
db: Session, *, start_time: int, end_time: int,
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
audit_context: dict[str, Any] | None = None,
) -> dict:
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
"""
fetched = inserted = updated = pages = 0
fetched = inserted = updated = pages = api_requests = 0
page = 1
while page <= max_pages:
resp = meituan.query_order(
sid=sid, start_time=start_time, end_time=end_time,
query_time_type=query_time_type, page=page, limit=100,
)
api_requests += 1
try:
resp = meituan.query_order(
sid=sid, start_time=start_time, end_time=end_time,
query_time_type=query_time_type, page=page, limit=100,
)
except Exception: # noqa: BLE001 - 记录失败页后保持原异常类型继续抛出
if audit_context is not None:
logger.exception(
"CPS reconcile upstream request failed platform=meituan page=%s",
page,
extra={
**audit_context,
"event": "cps_reconcile.request_failed",
"platform": "meituan",
"failed_page": page,
"api_request_number": api_requests,
},
)
raise
rows = ((resp.get("data") or {}).get("dataList")) or []
if not rows:
break
@@ -419,30 +439,58 @@ def reconcile_orders(
break
page += 1
db.commit()
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
return {
"fetched": fetched,
"inserted": inserted,
"updated": updated,
"pages": pages,
"api_requests": api_requests,
}
def reconcile_jd_orders(
db: Session, *, start_time: datetime, end_time: datetime,
query_time_type: int = 3, max_pages: int = 100,
audit_context: dict[str, Any] | None = None,
) -> dict:
"""调京东 order.row.query 拉单 → 按订单行 upsert。
京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
"""
fetched = inserted = updated = pages = 0
fetched = inserted = updated = pages = api_requests = windows = 0
cur = start_time
while cur < end_time:
win_end = min(cur + timedelta(hours=1), end_time)
windows += 1
page = 1
while page <= max_pages:
resp = jd_union.query_order_rows(
start_time=cur,
end_time=win_end,
query_time_type=query_time_type,
page_index=page,
page_size=200,
)
api_requests += 1
try:
resp = jd_union.query_order_rows(
start_time=cur,
end_time=win_end,
query_time_type=query_time_type,
page_index=page,
page_size=200,
)
except Exception: # noqa: BLE001 - 记录失败窗口后保持原异常类型继续抛出
if audit_context is not None:
logger.exception(
"CPS reconcile upstream request failed platform=jd window=%s..%s page=%s",
cur.isoformat(),
win_end.isoformat(),
page,
extra={
**audit_context,
"event": "cps_reconcile.request_failed",
"platform": "jd",
"failed_window_start": cur.isoformat(),
"failed_window_end": win_end.isoformat(),
"failed_page": page,
"api_request_number": api_requests,
},
)
raise
rows = resp.get("rows") or []
has_more = bool(resp.get("has_more"))
if not rows:
@@ -469,7 +517,14 @@ def reconcile_jd_orders(
page += 1
cur = win_end
db.commit()
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
return {
"fetched": fetched,
"inserted": inserted,
"updated": updated,
"pages": pages,
"api_requests": api_requests,
"windows": windows,
}
def list_orders(
app/core/cps_reconcile_worker.py
+276 -23
View File
@@ -9,15 +9,17 @@ import asyncio
import contextlib
import logging
import os
import socket
import time
from collections.abc import Iterator
from collections.abc import Callable, Iterator
from datetime import date, datetime, timedelta
from pathlib import Path
from uuid import uuid4
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
from app.db.session import SessionLocal, engine
logger = logging.getLogger("shagua.cps_reconcile")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
@@ -27,6 +29,21 @@ def _cn_now() -> datetime:
return datetime.now(CN_TZ)
def _new_run_id(now: datetime) -> str:
return f"{now:%Y%m%d-%H%M%S}-{uuid4().hex[:8]}"
def _next_run_at(now: datetime, run_hour: int) -> datetime:
scheduled = now.replace(hour=run_hour, minute=0, second=0, microsecond=0)
return scheduled if now < scheduled else scheduled + timedelta(days=1)
def _trigger_for_run(worker_started_at: datetime, now: datetime, run_hour: int) -> str:
if worker_started_at.date() == now.date() and worker_started_at.hour >= run_hour:
return "startup_catchup"
return "scheduled"
def _touch_lock() -> None:
with contextlib.suppress(FileNotFoundError):
os.utime(_LOCK_PATH, None)
@@ -67,52 +84,240 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
def _empty_result() -> dict:
return {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0}
return {
"fetched": 0,
"inserted": 0,
"updated": 0,
"pages": 0,
"api_requests": 0,
}
def _reconcile_once(now: datetime | None = None) -> dict:
def _platform_log_fields(platform: str, result: dict) -> dict:
fields = {
"platform": platform,
"platform_status": result["status"],
"duration_ms": result["duration_ms"],
}
for key in ("fetched", "inserted", "updated", "pages", "api_requests", "windows"):
if key in result:
fields[key] = result[key]
return fields
def _run_platform(
*,
platform: str,
query_time_type: int,
common: dict,
reconcile: Callable[[], dict],
) -> tuple[dict, str | None]:
started = time.perf_counter()
logger.info(
"CPS auto reconcile platform started run_id=%s platform=%s",
common["run_id"],
platform,
extra={
**common,
"event": "cps_reconcile.platform_started",
"platform": platform,
"query_time_type": query_time_type,
},
)
try:
raw_result = reconcile()
except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台
result = {
"status": "failed",
"duration_ms": round((time.perf_counter() - started) * 1000, 3),
"error_type": type(exc).__name__,
"error_summary": str(exc)[:500],
}
logger.exception(
"CPS auto reconcile platform failed run_id=%s platform=%s error_type=%s",
common["run_id"],
platform,
type(exc).__name__,
extra={
**common,
"event": "cps_reconcile.platform_failed",
**_platform_log_fields(platform, result),
"error_type": type(exc).__name__,
"error_summary": str(exc)[:500],
},
)
return result, str(exc)
result = {
**raw_result,
"status": "success",
"duration_ms": round((time.perf_counter() - started) * 1000, 3),
}
logger.info(
"CPS auto reconcile platform completed run_id=%s platform=%s fetched=%s inserted=%s updated=%s",
common["run_id"],
platform,
result.get("fetched", 0),
result.get("inserted", 0),
result.get("updated", 0),
extra={
**common,
"event": "cps_reconcile.platform_completed",
**_platform_log_fields(platform, result),
},
)
return result, None
def _skip_platform(platform: str, common: dict) -> dict:
result = {
**_empty_result(),
"status": "skipped",
"skipped": "not_configured",
"duration_ms": 0.0,
}
logger.warning(
"CPS auto reconcile platform skipped run_id=%s platform=%s reason=not_configured",
common["run_id"],
platform,
extra={
**common,
"event": "cps_reconcile.platform_skipped",
**_platform_log_fields(platform, result),
"skip_reason": "not_configured",
},
)
return result
def _reconcile_once(
now: datetime | None = None,
*,
run_id: str | None = None,
trigger: str = "scheduled",
) -> dict:
"""独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
end = (now or _cn_now()).astimezone(CN_TZ)
run_id = run_id or _new_run_id(end)
lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
start = end - timedelta(days=lookback_days)
task_started = time.perf_counter()
common = {
"run_id": run_id,
"trigger": trigger,
"scheduled_date": end.date().isoformat(),
"app_env": settings.APP_ENV,
"db_dialect": engine.dialect.name,
"hostname": socket.gethostname(),
"pid": os.getpid(),
"window_start": start.isoformat(),
"window_end": end.isoformat(),
"lookback_days": lookback_days,
}
result = {
"run_id": run_id,
"trigger": trigger,
"window_start": start.isoformat(),
"window_end": end.isoformat(),
"meituan": None,
"jd": None,
"errors": {},
}
logger.info(
"CPS auto reconcile started run_id=%s trigger=%s window=%s..%s",
run_id,
trigger,
result["window_start"],
result["window_end"],
extra={**common, "event": "cps_reconcile.started"},
)
if settings.mt_cps_configured:
try:
def reconcile_meituan() -> dict:
with SessionLocal() as db:
result["meituan"] = cps_repo.reconcile_orders(
return cps_repo.reconcile_orders(
db,
start_time=int(start.timestamp()),
end_time=int(end.timestamp()),
query_time_type=2,
audit_context=common,
)
except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台
result["errors"]["meituan"] = str(exc)
logger.exception("CPS auto reconcile failed platform=meituan")
result["meituan"], error = _run_platform(
platform="meituan",
query_time_type=2,
common=common,
reconcile=reconcile_meituan,
)
if error is not None:
result["errors"]["meituan"] = error
else:
result["meituan"] = {**_empty_result(), "skipped": "not_configured"}
result["meituan"] = _skip_platform("meituan", common)
_touch_lock()
if settings.jd_union_configured:
try:
def reconcile_jd() -> dict:
with SessionLocal() as db:
result["jd"] = cps_repo.reconcile_jd_orders(
return cps_repo.reconcile_jd_orders(
db,
start_time=start,
end_time=end,
query_time_type=3,
audit_context=common,
)
except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台
result["errors"]["jd"] = str(exc)
logger.exception("CPS auto reconcile failed platform=jd")
result["jd"], error = _run_platform(
platform="jd",
query_time_type=3,
common=common,
reconcile=reconcile_jd,
)
if error is not None:
result["errors"]["jd"] = error
else:
result["jd"] = {**_empty_result(), "skipped": "not_configured"}
result["jd"] = _skip_platform("jd", common)
successes = sum(
platform_result.get("status") == "success"
for platform_result in (result["meituan"], result["jd"])
)
if result["errors"]:
task_status = "partial_success" if successes else "failed"
else:
task_status = "success"
completed_at = _cn_now()
duration_ms = round((time.perf_counter() - task_started) * 1000, 3)
next_run_at = _next_run_at(
completed_at,
min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23),
).isoformat()
result.update({
"status": task_status,
"duration_ms": duration_ms,
"manual_retry_required": bool(result["errors"]),
"next_run_at": next_run_at,
})
completion_fields = {
**common,
"event": "cps_reconcile.completed",
"task_status": task_status,
"duration_ms": duration_ms,
"manual_retry_required": result["manual_retry_required"],
"failed_platforms": sorted(result["errors"]),
"next_run_at": next_run_at,
"meituan_result": result["meituan"],
"jd_result": result["jd"],
}
log_method = logger.info if task_status == "success" else logger.warning
log_method(
"CPS auto reconcile completed run_id=%s status=%s duration_ms=%s failed_platforms=%s next_run_at=%s",
run_id,
task_status,
duration_ms,
",".join(sorted(result["errors"])) or "-",
next_run_at,
extra=completion_fields,
)
return result
@@ -127,42 +332,90 @@ async def _run_loop() -> None:
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")
logger.warning(
"CPS auto reconcile worker skipped: another worker owns lock",
extra={
"event": "cps_reconcile.worker_skipped",
"skip_reason": "lock_not_acquired",
"pid": os.getpid(),
},
)
return
await _run_locked_loop(interval, run_hour)
async def _run_locked_loop(interval: int, run_hour: int) -> None:
worker_started_at = _cn_now()
logger.info(
"CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
run_hour,
interval,
settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
extra={
"event": "cps_reconcile.worker_started",
"pid": os.getpid(),
"hostname": socket.gethostname(),
"app_env": settings.APP_ENV,
"db_dialect": engine.dialect.name,
"run_hour": run_hour,
"interval_sec": interval,
"lookback_days": settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
"next_run_at": _next_run_at(worker_started_at, run_hour).isoformat(),
},
)
last_run: date | None = None
try:
while True:
run_id: str | None = None
try:
_touch_lock()
now = _cn_now()
if _should_run(last_run, now, run_hour):
result = await asyncio.to_thread(_reconcile_once, now)
run_id = _new_run_id(now)
trigger = _trigger_for_run(worker_started_at, now, run_hour)
await asyncio.to_thread(
_reconcile_once,
now,
run_id=run_id,
trigger=trigger,
)
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")
logger.exception(
"CPS auto reconcile unexpected worker error",
extra={
"event": "cps_reconcile.unexpected_failed",
"run_id": run_id or "unassigned",
"pid": os.getpid(),
},
)
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("CPS auto reconcile worker stopped")
logger.info(
"CPS auto reconcile worker stopped",
extra={"event": "cps_reconcile.worker_stopped", "pid": os.getpid()},
)
raise
def start_cps_reconcile_worker() -> asyncio.Task | None:
if not settings.CPS_AUTO_RECONCILE_ENABLED:
logger.info("CPS auto reconcile disabled")
logger.info(
"CPS auto reconcile disabled",
extra={
"event": "cps_reconcile.worker_skipped",
"skip_reason": "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")
logger.warning(
"CPS auto reconcile not started: Meituan and JD credentials are missing",
extra={
"event": "cps_reconcile.worker_skipped",
"skip_reason": "all_platform_credentials_missing",
},
)
return None
return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
tests/test_cps_reconcile_worker.py
+115 -2
View File
@@ -1,11 +1,15 @@
"""美团、京东 CPS 每日自动对账 worker。"""
from __future__ import annotations
import logging
from datetime import date, datetime, timedelta
import pytest
from app.core import cps_reconcile_worker as worker
from app.core.config import settings
from app.core.rewards import CN_TZ
from app.db.session import SessionLocal
from app.integrations.meituan import MeituanCpsError
@@ -23,11 +27,24 @@ def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None
def fake_meituan(db, **kwargs):
calls["meituan"] = kwargs
return {"fetched": 2, "inserted": 1, "updated": 1, "pages": 1}
return {
"fetched": 2,
"inserted": 1,
"updated": 1,
"pages": 1,
"api_requests": 1,
}
def fake_jd(db, **kwargs):
calls["jd"] = kwargs
return {"fetched": 3, "inserted": 2, "updated": 1, "pages": 2}
return {
"fetched": 3,
"inserted": 2,
"updated": 1,
"pages": 2,
"api_requests": 72,
"windows": 72,
}
monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fake_meituan)
monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
@@ -45,6 +62,10 @@ def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None
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
assert calls["meituan"]["audit_context"]["run_id"] == result["run_id"]
assert calls["jd"]["audit_context"]["run_id"] == result["run_id"]
assert result["status"] == "success"
assert result["manual_retry_required"] is False
def test_meituan_failure_does_not_block_jd(monkeypatch) -> None:
@@ -68,6 +89,9 @@ def test_meituan_failure_does_not_block_jd(monkeypatch) -> None:
assert result["errors"]["meituan"] == "temporary failure"
assert jd_called is True
assert result["jd"]["fetched"] == 1
assert result["meituan"]["status"] == "failed"
assert result["status"] == "partial_success"
assert result["manual_retry_required"] is True
def test_should_run_once_after_five_am() -> None:
@@ -79,6 +103,95 @@ def test_should_run_once_after_five_am() -> None:
assert worker._should_run(date(2026, 7, 22), at_five, 5) is False
def test_trigger_and_next_run_are_beijing_time() -> None:
before_five = datetime(2026, 7, 22, 4, 30, tzinfo=CN_TZ)
after_five = datetime(2026, 7, 22, 6, 0, tzinfo=CN_TZ)
assert worker._trigger_for_run(before_five, after_five, 5) == "scheduled"
assert worker._trigger_for_run(after_five, after_five, 5) == "startup_catchup"
assert worker._next_run_at(before_five, 5) == datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
assert worker._next_run_at(after_five, 5) == datetime(2026, 7, 23, 5, 0, tzinfo=CN_TZ)
def test_reconcile_emits_structured_audit_logs(monkeypatch, caplog) -> None:
_configure_platforms(monkeypatch)
monkeypatch.setattr(
worker.cps_repo,
"reconcile_orders",
lambda db, **kwargs: {
"fetched": 2,
"inserted": 1,
"updated": 1,
"pages": 1,
"api_requests": 1,
},
)
monkeypatch.setattr(
worker.cps_repo,
"reconcile_jd_orders",
lambda db, **kwargs: {
"fetched": 3,
"inserted": 2,
"updated": 1,
"pages": 1,
"api_requests": 72,
"windows": 72,
},
)
monkeypatch.setattr(worker, "_touch_lock", lambda: None)
monkeypatch.setattr(
worker,
"_cn_now",
lambda: datetime(2026, 7, 22, 5, 1, tzinfo=CN_TZ),
)
with caplog.at_level(logging.INFO, logger=worker.logger.name):
result = worker._reconcile_once(
datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ),
run_id="audit-run-1",
trigger="scheduled",
)
records = {record.event: record for record in caplog.records if hasattr(record, "event")}
assert records["cps_reconcile.started"].run_id == "audit-run-1"
assert records["cps_reconcile.started"].lookback_days == 3
assert records["cps_reconcile.started"].scheduled_date == "2026-07-22"
assert records["cps_reconcile.started"].hostname
assert records["cps_reconcile.completed"].task_status == "success"
assert records["cps_reconcile.completed"].manual_retry_required is False
assert records["cps_reconcile.completed"].meituan_result["fetched"] == 2
assert records["cps_reconcile.completed"].jd_result["api_requests"] == 72
assert result["next_run_at"] == "2026-07-23T05:00:00+08:00"
def test_jd_request_failure_logs_exact_window(monkeypatch, caplog) -> None:
start = datetime(2026, 7, 20, 5, 0, tzinfo=CN_TZ)
def fail_query(**kwargs):
raise RuntimeError("upstream timeout")
monkeypatch.setattr(worker.cps_repo.jd_union, "query_order_rows", fail_query)
with SessionLocal() as db, caplog.at_level(logging.ERROR, logger=worker.logger.name):
with pytest.raises(RuntimeError, match="upstream timeout"):
worker.cps_repo.reconcile_jd_orders(
db,
start_time=start,
end_time=start + timedelta(hours=2),
audit_context={"run_id": "audit-run-2", "trigger": "scheduled"},
)
record = next(
item
for item in caplog.records
if getattr(item, "event", "") == "cps_reconcile.request_failed"
)
assert record.run_id == "audit-run-2"
assert record.platform == "jd"
assert record.failed_window_start == "2026-07-20T05:00:00+08:00"
assert record.failed_window_end == "2026-07-20T06:00:00+08:00"
assert record.failed_page == 1
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