Compare commits

...

7 Commits

Author SHA1 Message Date
guke ceef642854 fix(scripts): seed 脚本补 username 并新增反馈 mock 脚本
User.username 现为 NOT NULL+unique,原 seed 脚本未设 username 会 IntegrityError,补 _gen_unique_username(仿真实注册生成)。顺带 zip(strict=True)、timezone.utc→datetime.UTC(与代码库统一)。新增 seed_mock_feedback.py:造待审核/已采纳/未采纳三态用户反馈 + 可加载纯色 PNG 截图,供后台反馈工单页联调。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-24 09:55:47 +08:00
guke 1ec376880a fix(logging): Windows 日志轮转改用 copytruncate 避免 WinError 32 卡死
stdlib RotatingFileHandler 靠 rename 活动文件轮转;Windows 上只要有第二个句柄(admin 第二进程、残留 --reload worker、IDE 索引、杀软)开着它,rename 就 WinError 32、轮转永久卡死。SafeRotatingFileHandler 在 Windows 改走 copytruncate(拷备份→自有句柄就地清空,不 rename 活动文件),POSIX 保持 stdlib 原子轮转不变。含 test_logging_rotation.py:monkeypatch os.name=nt,任意 CI 覆盖该分支。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-24 09:55:39 +08:00
guke f29b28769e Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/withdrawal-auth-gate 2026-07-24 09:49:44 +08:00
guke d872755bb8 test(withdraw): 测试健壮性 polish(openid 取自 User + 补 status_code 断言) 2026-07-23 18:28:51 +08:00
marco ceceeb3458 fix(db): 合并 3 个 alembic 迁移 head(发车 0.4.4 前置,no-op merge 节点) 2026-07-23 17:55:35 +08:00
guke 07cf806805 fix(withdraw): 免确认授权已开启判定加 authorization_id 非空,与打款一致
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-23 16:48:55 +08:00
linkeyu 31f61f6aad 增强:CPS 每日自动对账增加可回查日志 (#163)
## 背景

PR #162 已合并。本 PR 为其后续日志增强,方便人工按一次任务完整回查美团、京东每日自动对账。

## 日志内容

- 每次运行生成唯一 `run_id`,记录 `scheduled` / `startup_catchup` 触发来源
- 记录北京时间计划日期、近 3 天回拉窗口、环境、数据库方言、主机名和 PID
- 分平台记录开始、成功、跳过、失败及执行耗时
- 成功结果记录 fetched / inserted / updated / pages / api_requests,京东额外记录小时窗口数
- 京东上游失败记录具体小时窗口、页码和请求序号;美团记录失败页码
- 最终汇总记录 success / partial_success / failed、失败平台、是否需要人工补跑和下次执行时间
- 日志使用现有 `extra` 结构化字段写入 JSON 日志,便于 SLS/人工检索

## 兼容性

- 手动对账接口、事务和返回模型不变
- 不记录密钥、Token、完整上游响应或订单明细
- 仓储层仅增加可选审计上下文和请求计数;不改变拉取及 upsert 逻辑

## 验证

- `pytest tests/test_cps_reconcile_worker.py tests/test_cps_admin.py tests/test_admin_read.py tests/test_observe.py -q`:44 passed
- 相关文件 Ruff 检查通过(忽略文件原有 UP017 提示)
- `git diff --check` 通过

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #163
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-23 11:57:37 +08:00
11 changed files with 959 additions and 49 deletions
@@ -0,0 +1,26 @@
"""merge notification/comparison_user_idx/monitoring_audit heads
Revision ID: 8e04cc13a211
Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
Create Date: 2026-07-23 15:37:26.967540
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '8e04cc13a211'
down_revision: Union[str, Sequence[str], None] = ('comparison_user_created_idx', 'monitoring_audit_rbac', 'notification_table')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
+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(
+3 -2
View File
@@ -192,7 +192,7 @@ def withdraw_info(
wechat_bound=bool(u and u.wechat_openid),
wechat_nickname=u.wechat_nickname if u else None,
wechat_avatar_url=u.wechat_avatar_url if u else None,
transfer_auth_enabled=bool(auth and auth.state == "active"),
transfer_auth_enabled=bool(auth and auth.state == "active" and auth.authorization_id),
tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
)
@@ -335,7 +335,8 @@ def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOu
def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
auth = crud_wallet.sync_transfer_auth(db, user.id)
state = auth.state if auth else "none"
return TransferAuthStatusOut(state=state, enabled=(state == "active"))
enabled = bool(auth and auth.state == "active" and auth.authorization_id)
return TransferAuthStatusOut(state=state, enabled=enabled)
@router.post(
+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")
+51 -2
View File
@@ -22,13 +22,13 @@ from __future__ import annotations
import json
import logging
import os
import shutil
import sys
from contextvars import ContextVar
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
@@ -91,6 +91,55 @@ class TextFormatter(logging.Formatter):
return f"{base} trace={tid}" if tid else base
class SafeRotatingFileHandler(RotatingFileHandler):
"""Windows 下不会被外部句柄卡死的 RotatingFileHandler。
stdlib 轮转靠 rename 活动文件(app-server.log .1);Windows 只要有别的句柄(IDE 索引
app.admin.main 第二进程残留 --reload worker杀软扫描)开着它, rename WinError 32,
轮转永久卡死文件停在 maxBytes之后每条日志被丢这里 Windows 改用 copytruncate:把活动
文件拷进备份再通过自己的句柄原地清空, 从不 rename 活动文件, 故外部句柄开着也能转
POSIX(生产 Linux)rename 打开中的文件本就合法, 保留 stdlib 的原子轮转不变
代价:copytruncate 拷贝清空极窄窗口内并发写可能丢几行(仅跨进程;同进程 emit
handler 锁串行, 无此问题)对本地开发日志可接受
"""
def doRollover(self) -> None:
if os.name != "nt":
super().doRollover()
return
if self.stream is None:
self.stream = self._open()
else:
self.stream.flush()
try:
self._copytruncate_backups()
except OSError:
# 备份腾挪是尽力而为:任一备份被占用也绝不能挡住下面的清空, 否则活动文件继续涨、
# 轮转又卡死——那就白改了。
pass
# 通过自己独占的句柄原地清空:不涉及 rename, 外部只读句柄不受影响。
self.stream.seek(0)
self.stream.truncate()
self.stream.flush()
def _copytruncate_backups(self) -> None:
"""把 .N-1→.N 逐级腾挪, 再把活动文件拷到 .1(不动活动文件本身)。"""
if self.backupCount <= 0:
return
for i in range(self.backupCount - 1, 0, -1):
sfn = self.rotation_filename(f"{self.baseFilename}.{i}")
dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}")
if os.path.exists(sfn):
if os.path.exists(dfn):
os.remove(dfn)
os.replace(sfn, dfn)
dfn = self.rotation_filename(f"{self.baseFilename}.1")
if os.path.exists(dfn):
os.remove(dfn)
shutil.copyfile(self.baseFilename, dfn)
_CONFIGURED = False
@@ -126,7 +175,7 @@ def setup_logging(debug: bool = False) -> None:
Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
)
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
file_handler = SafeRotatingFileHandler(
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
)
file_handler.setFormatter(JsonFormatter(service))
+277
View File
@@ -0,0 +1,277 @@
"""一次性 mock:造几条不同状态的用户反馈,供运营后台「反馈工单」页联调验收。
覆盖:
- 三个状态 tab(待审核 pending / 已采纳 adopted / 未采纳 rejected),重点铺待审核;
- 两种反馈类型 source(普通反馈 profile /我的页入口比价反馈 comparison / 比价结果页入口),
比价反馈带问题场景scene(找错商品/优惠不对/比价太慢);
- 提交端环境快照(app_version / device_model / rom_name / android_version)新端反馈才有,
另留 1~2 条历史反馈(env NULLcontact 有值)旧数据展示;
- 截图: data/media/feedback/ 生成真实可加载的纯色 PNG(手写字节,无需 Pillow),
让审核抽屉的图能真加载出来( seed_mock_price_reports 同法)
- 已采纳条带 reward_coins + admin_reply + review_note;未采纳条带 reject_reason + admin_reply
幂等:每次运行先按固定 mock 手机号清掉旧 mock 用户/反馈 + mock 截图再重建仅清理用 --clean-only
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py --clean-only
看图:前端反馈工单(http://localhost:3001 反馈)图经 NEXT_PUBLIC_MEDIA_BASE
(本地 = http://localhost:8770) App 后端 /media 加载改过 .env.local 后需重启 next dev,
App 后端(:8770)要在跑
"""
from __future__ import annotations
import argparse
import struct
import sys
import zlib
from datetime import UTC, datetime, timedelta
from pathlib import Path
from sqlalchemy import delete, select
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.feedback import Feedback
from app.models.user import User
from app.repositories.user import _gen_unique_username
# Windows GBK 控制台下也能正常打印中文/¥(避免 UnicodeEncodeError)
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
# 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。
# 刻意与 seed_mock_withdraws / seed_mock_price_reports 的号段错开,互不干扰。
MOCK_PHONES = [
"13255550001",
"13255550002",
"13255550003",
"13255550004",
"13255550005",
]
_FEEDBACK_DIR = Path(settings.MEDIA_ROOT) / "feedback"
_MOCK_IMG_GLOB = "mock_fb_*.png" # 本脚本生成的图前缀,清理时按此删
def _naive_utc_now() -> datetime:
"""与 func.now() 在 SQLite 的口径一致:naive UTC。反馈 created_at 走 server_default=func.now(),
这里显式造数据也用 naive UTC,和真实反馈行同源,前端展示口径一致"""
return datetime.now(UTC).replace(tzinfo=None)
def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
"""生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
def _chunk(typ: bytes, data: bytes) -> bytes:
body = typ + data
return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8bit/通道, color type 2 = RGB
row = b"\x00" + bytes(rgb) * width # 每行前缀 filter byte 0
idat = zlib.compress(row * height, 9)
return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
"""写一张 mock 截图到 media/feedback/,返回其相对 URL(/media/feedback/<name>)。"""
_FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
(_FEEDBACK_DIR / name).write_bytes(_solid_png(320, 320, rgb))
return f"{settings.MEDIA_URL_PREFIX}/feedback/{name}"
def clean(db) -> int:
uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
if uids:
db.execute(delete(Feedback).where(Feedback.user_id.in_(uids)))
db.execute(delete(User).where(User.id.in_(uids)))
db.commit()
# 删 mock 截图文件
if _FEEDBACK_DIR.exists():
for f in _FEEDBACK_DIR.glob(_MOCK_IMG_GLOB):
f.unlink(missing_ok=True)
return len(uids)
def seed(db) -> list[Feedback]:
now = _naive_utc_now()
def ago(**kw) -> datetime:
return now - timedelta(**kw)
# 1) 建 5 个 mock 用户(U5 昵称留空测 "-" 展示)
users_spec = [
("13255550001", "反馈小达人"),
("13255550002", "比价挑刺王"),
("13255550003", "热心用户阿明"),
("13255550004", "老用户张姐"),
("13255550005", None),
]
users: dict[str, User] = {}
for phone, nickname in users_spec:
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
created_at=ago(days=15),
last_login_at=ago(hours=1),
)
db.add(u)
users[phone] = u
db.flush() # 拿 user.id
# 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
palette = [
(24, 144, 255), # 蓝
(82, 196, 26), # 绿
(250, 173, 20), # 橙
(245, 34, 45), # 红
]
imgs = [_write_mock_image(f"mock_fb_{i}.png", palette[i]) for i in range(len(palette))]
# 3) 造反馈记录
def fb(
phone: str,
content: str,
*,
source: str = "profile",
scene: str | None = None,
images: list[str] | None = None,
contact: str = "",
status: str = "pending",
reject_reason: str | None = None,
reward_coins: int | None = None,
review_note: str | None = None,
admin_reply: str | None = None,
app_version: str | None = None,
device_model: str | None = None,
rom_name: str | None = None,
android_version: str | None = None,
created: datetime,
) -> Feedback:
return Feedback(
user_id=users[phone].id,
content=content,
contact=contact, # 列 NOT NULL:新端存空串,历史数据有值
source=source,
scene=scene,
images=images,
status=status,
reject_reason=reject_reason,
reward_coins=reward_coins,
review_note=review_note,
admin_reply=admin_reply,
app_version=app_version,
device_model=device_model,
rom_name=rom_name,
android_version=android_version,
reviewed_at=(created + timedelta(hours=2)) if status != "pending" else None,
created_at=created,
)
feedbacks = [
# ===== 待审核 pending(默认 tab,重点铺量)=====
# 普通反馈 · 新端(带环境快照)· 无图
fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
source="profile",
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
created=ago(minutes=6)),
# 比价反馈 · scene=优惠不对 · 新端 · 2 图
fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(minutes=22)),
# 比价反馈 · scene=找错商品 · 新端 · 无图
fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
source="comparison", scene="找错商品",
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
created=ago(hours=1)),
# 普通反馈 · 新端 · 1 图(表扬 + 小问题)
fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
source="profile", images=[imgs[2]],
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
created=ago(hours=3)),
# 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
source="comparison", scene="比价太慢", contact="微信 zhangjie_66",
created=ago(days=1, hours=2)),
# 普通反馈 · 历史数据(env 全 NULL、contact 有值)· 无昵称用户
fb("13255550005", "希望能增加支付宝提现,微信零钱用不太习惯。",
source="profile", contact="QQ 100200300",
created=ago(days=1, hours=8)),
# ===== 已采纳 adopted(发金币 + 回复)=====
fb("13255550003", "建议在比价结果页加个「一键复制口令」,分享给家人更方便。",
source="profile", images=[imgs[3]],
status="adopted", reward_coins=2000,
review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
created=ago(days=2)),
# ===== 未采纳 rejected(带原因 + 回复)=====
fb("13255550002", "你们算的价格不准,我看到的更便宜。",
source="comparison", scene="价格不准",
status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(days=3)),
]
db.add_all(feedbacks)
db.commit()
for f in feedbacks:
db.refresh(f)
return feedbacks
def main() -> None:
parser = argparse.ArgumentParser(description="造用户反馈 mock 数据(运营后台反馈工单页联调用)")
parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
args = parser.parse_args()
db = SessionLocal()
try:
removed = clean(db)
if removed:
print(f"🧹 已清理旧 mock:{removed} 个用户及其反馈 + mock 截图")
if args.clean_only:
print("✅ 仅清理,已完成。")
return
feedbacks = seed(db)
status_label = {"pending": "待审核", "adopted": "已采纳", "rejected": "未采纳"}
source_label = {"profile": "普通反馈", "comparison": "比价反馈"}
by_status: dict[str, list[Feedback]] = {}
for f in feedbacks:
by_status.setdefault(f.status, []).append(f)
print(f"\n✅ 已生成 {len(feedbacks)} 条反馈(截图落 {_FEEDBACK_DIR}),分布:")
for st in ("pending", "adopted", "rejected"):
lst = by_status.get(st, [])
print(f" {status_label[st]:<4} {len(lst)}")
print("\n 明细(#id | 状态 | 类型/场景 | 图 | 内容):")
uid2phone = dict(
db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()
)
for f in feedbacks:
src = source_label.get(f.source, f.source)
scene = f"·{f.scene}" if f.scene else ""
nimg = len(f.images or [])
snippet = f.content[:20] + ("" if len(f.content) > 20 else "")
print(
f" #{f.id} [{status_label.get(f.status, f.status)}] "
f"{src}{scene} {nimg}{uid2phone.get(f.user_id, '?')} {snippet}"
)
print(
"\n👉 打开 http://localhost:3001 → 反馈 查看(默认「待审核」tab)。"
"\n 图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
" ② App 后端(:8770)是否在跑(它托管 /media)。"
)
finally:
db.close()
if __name__ == "__main__":
main()
+3 -1
View File
@@ -30,6 +30,7 @@ from app.core.config import settings
from app.db.session import SessionLocal
from app.models.price_report import PriceReport
from app.models.user import User
from app.repositories.user import _gen_unique_username
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
@@ -84,10 +85,11 @@ def seed(db) -> list[PriceReport]:
# 1) 建 3 个 mock 用户
users: dict[str, User] = {}
for i, (phone, nickname) in enumerate(
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
):
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
+4 -2
View File
@@ -21,7 +21,7 @@ import argparse
import sys
import uuid
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, select
@@ -31,6 +31,7 @@ from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
from app.repositories.user import _gen_unique_username
# Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
if hasattr(sys.stdout, "reconfigure"):
@@ -51,7 +52,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r
def _naive_utc_now() -> datetime:
"""与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
return datetime.now(timezone.utc).replace(tzinfo=None)
return datetime.now(UTC).replace(tzinfo=None)
def _mock_transfer_no() -> str:
@@ -96,6 +97,7 @@ def seed(db) -> list[WithdrawOrder]:
for phone, nickname in users_spec:
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
+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
+73
View File
@@ -0,0 +1,73 @@
"""SafeRotatingFileHandler:Windows 轮转不被外部句柄卡死(WinError 32)。
stdlib RotatingFileHandler rename 活动文件轮转;Windows 上只要有第二个句柄(admin
第二进程残留 --reload workerIDE 索引杀软)开着它, rename WinError 32轮转永久
卡死Safe 版在 Windows 改走 copytruncate(拷贝就地清空, 从不 rename 活动文件)
这些用例用 monkeypatch os.name 强制成 "nt", 使 copytruncate 分支在任何 OS CI 上都被
覆盖;在真实 Windows 上则天然命中
"""
from __future__ import annotations
import logging
from pathlib import Path
from app.core.logging import SafeRotatingFileHandler
def _emit(handler: logging.Handler, msg: str) -> None:
handler.emit(logging.LogRecord("t", logging.INFO, __file__, 0, msg, (), None))
def test_rollover_survives_second_open_handle(tmp_path: Path, monkeypatch) -> None:
"""第二个句柄开着活动文件时轮转:不抛异常, 且确实转了(生成 .1、活动文件就地清空)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
# maxBytes 放大, 避免 emit 期间自动轮转干扰;本用例手动触发 doRollover。
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=3, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for i in range(20):
_emit(handler, f"line-{i:03d}")
handler.flush()
before = log_file.stat().st_size
assert before > 0
# 正是 Windows 上 rename 失败的条件:另一个句柄开着活动文件。
with open(log_file, "a", encoding="utf-8"):
handler.doRollover() # 不应抛 PermissionError / WinError 32
backup = tmp_path / "app-server.log.1"
assert backup.exists()
assert backup.stat().st_size == before # 轮转前内容完整进了备份
assert log_file.stat().st_size == 0 # 活动文件就地清空(不是 rename)
# 句柄没被 rename 破坏, 仍能继续写。
_emit(handler, "after-rollover")
handler.flush()
assert log_file.stat().st_size > 0
finally:
handler.close()
def test_backups_shift_and_capped(tmp_path: Path, monkeypatch) -> None:
"""多次轮转:.1/.2 逐级腾挪, 超过 backupCount 的丢弃(不出现 .3)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=2, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for _ in range(4):
_emit(handler, "x" * 50)
handler.doRollover()
assert (tmp_path / "app-server.log.1").exists()
assert (tmp_path / "app-server.log.2").exists()
assert not (tmp_path / "app-server.log.3").exists()
finally:
handler.close()
+60 -1
View File
@@ -9,7 +9,7 @@ from sqlalchemy import select
from app.db.session import SessionLocal
from app.models.user import User
from app.models.wallet import CoinAccount, WithdrawOrder
from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
from app.repositories import wallet as crud_wallet
@@ -426,3 +426,62 @@ def test_withdraw_reject_refunds(client, monkeypatch) -> None:
r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
assert r.json()["status"] == "rejected"
assert r.json()["fail_reason"] == "测试拒绝"
# ===== §fix 免确认授权 enabled 判定收严:active+authorization_id 非空才算已授权 =====
def _seed_transfer_auth(phone: str, state: str, authorization_id: str | None) -> None:
"""直接在库里写/改该用户的免确认授权记录(绕过微信,用于判定测试)。"""
db = SessionLocal()
try:
user = db.execute(select(User).where(User.phone == phone)).scalar_one()
auth = db.get(WechatTransferAuthorization, user.id)
if auth is None:
auth = WechatTransferAuthorization(
user_id=user.id, openid=user.wechat_openid or "openid_test_abc",
out_authorization_no=f"oan_{user.id}",
)
db.add(auth)
auth.state = state
auth.authorization_id = authorization_id
db.commit()
finally:
db.close()
def test_withdraw_info_auth_enabled_requires_authorization_id(client, monkeypatch) -> None:
_patch_userinfo(monkeypatch)
token = _login(client, "13800002051")
# 触发建号 + 绑定微信(withdraw-info 读 openid)
client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
client.post("/api/v1/wallet/bind-wechat", json={"code": "c1"}, headers=_auth(token))
# active 但无 authorization_id → 视为未授权
_seed_transfer_auth("13800002051", "active", None)
r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["transfer_auth_enabled"] is False
# active 且有 authorization_id → 已授权
_seed_transfer_auth("13800002051", "active", "wx_auth_123")
r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["transfer_auth_enabled"] is True
def test_transfer_auth_status_requires_authorization_id(client, monkeypatch) -> None:
_patch_userinfo(monkeypatch)
token = _login(client, "13800002052")
client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
client.post("/api/v1/wallet/bind-wechat", json={"code": "c2"}, headers=_auth(token))
_seed_transfer_auth("13800002052", "active", None)
r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["state"] == "active"
assert r.json()["enabled"] is False
_seed_transfer_auth("13800002052", "active", "wx_auth_456")
r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
assert r.status_code == 200, r.text
assert r.json()["enabled"] is True