Compare commits

..

9 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
linkeyu b7cfcf7495 功能:美团和京东 CPS 每日自动对账 (#162)
## 需求

- 保留现有后台手动对账逻辑不变
- 每天北京时间 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: #162
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-23 10:51:20 +08:00
linkeyu 77f772f47c 性能:比价和领券分位数改为 PostgreSQL 聚合 (#161)
## 背景
- 比价记录页和领券记录页前端已经只拉当前页,并直接展示后端 summary。
- 原后端仍会将筛选区间内的全部耗时值取回 Python 计算分位数,生产数据量增大后会放大数据库读取和应用内存开销。

## 修改内容
- 比价记录:成功耗时 P5/P50/P95/P99、平均耗时以及中途退出耗时 P5/P50/P95 改用 PostgreSQL percentile_cont/AVG 聚合。
- 领券记录:发起数、完成数、平均耗时和完成耗时 P5/P50/P95/P99 合并为一条 PostgreSQL 聚合查询。
- 日期、用户、环境、状态、店铺和商品等筛选条件继续与列表共用,统计口径不变。
- SQLite 不支持 percentile_cont,仅在本地和测试环境回退读取耗时单列;不加载完整业务记录。
- API 字段与前端展示保持不变,无需前端改动。

## 验证
- 比价/领券及关联广告收益、点位、按券统计测试:26 passed。
- 本次涉及文件 ruff 检查通过。
- PostgreSQL SQL 编译测试确认使用 ordered-set percentile_cont 聚合。
- 全量测试:497 passed;8 个现有失败集中在邀请奖励、提现档位和代理转发等无关模块。

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #161
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>
2026-07-23 10:35:41 +08:00
18 changed files with 1436 additions and 69 deletions
+7
View File
@@ -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)。
@@ -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
+79 -18
View File
@@ -1,7 +1,7 @@
"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)
数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。生产 PostgreSQL
使用 percentile_cont 聚合耗时分位SQLite 本地/测试环境回退读取耗时单列计算
- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
@@ -45,6 +45,75 @@ def _percentile(sorted_vals: list[int], q: float) -> int | None:
return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)
def _round_duration_ms(value) -> int | None:
"""将数据库聚合结果按既有 Python round 口径转为整数毫秒。"""
if value is None:
return None
return int(round(value))
def _coupon_summary_aggregate_stmt(conditions: list):
"""PostgreSQL 汇总卡聚合语句;计数、均值与四个分位一次返回。"""
completed = CouponSession.status == "completed"
completed_elapsed = completed & CouponSession.elapsed_ms.is_not(None)
return select(
func.count(CouponSession.id),
func.sum(case((completed, 1), else_=0)),
func.avg(CouponSession.elapsed_ms).filter(completed_elapsed),
*(
func.percentile_cont(q)
.within_group(CouponSession.elapsed_ms)
.filter(completed_elapsed)
for q in (0.05, 0.5, 0.95, 0.99)
),
).where(*conditions)
def _coupon_summary_aggregates(db: Session, conditions: list) -> dict:
"""汇总卡基础指标;生产 PG 全部在数据库内完成,SQLite 仅作测试回退。"""
if db.bind is not None and db.bind.dialect.name == "postgresql":
row = db.execute(_coupon_summary_aggregate_stmt(conditions)).one()
return {
"started_count": int(row[0] or 0),
"completed_count": int(row[1] or 0),
"avg_elapsed_ms": _round_duration_ms(row[2]),
"p5_ms": _round_duration_ms(row[3]),
"p50_ms": _round_duration_ms(row[4]),
"p95_ms": _round_duration_ms(row[5]),
"p99_ms": _round_duration_ms(row[6]),
}
counts = db.execute(
select(
func.count(CouponSession.id),
func.sum(case((CouponSession.status == "completed", 1), else_=0)),
).where(*conditions)
).one()
# SQLite 没有 percentile_cont;本地/测试只回退读取耗时单列,不加载完整记录。
completed_elapsed = list(
db.execute(
select(CouponSession.elapsed_ms)
.where(
*conditions,
CouponSession.status == "completed",
CouponSession.elapsed_ms.is_not(None),
)
.order_by(CouponSession.elapsed_ms)
).scalars()
)
return {
"started_count": int(counts[0] or 0),
"completed_count": int(counts[1] or 0),
"avg_elapsed_ms": _round_duration_ms(
sum(completed_elapsed) / len(completed_elapsed)
) if completed_elapsed else None,
"p5_ms": _percentile(completed_elapsed, 5),
"p50_ms": _percentile(completed_elapsed, 50),
"p95_ms": _percentile(completed_elapsed, 95),
"p99_ms": _percentile(completed_elapsed, 99),
}
def _avg(vals: list[int]) -> int | None:
return round(sum(vals) / len(vals)) if vals else None
@@ -215,30 +284,22 @@ def coupon_data_report(
if not user_ids:
return _empty_result()
stmt = select(CouponSession).where(
conditions = [
CouponSession.started_date >= d_from,
CouponSession.started_date <= d_to,
)
]
if app_env is not None:
stmt = stmt.where(CouponSession.app_env == app_env)
conditions.append(CouponSession.app_env == app_env)
if statuses:
stmt = stmt.where(CouponSession.status.in_(statuses))
conditions.append(CouponSession.status.in_(statuses))
if user_ids is not None:
stmt = stmt.where(CouponSession.user_id.in_(user_ids))
conditions.append(CouponSession.user_id.in_(user_ids))
stmt = select(CouponSession).where(*conditions)
rows = list(db.execute(stmt).scalars())
# ── 汇总卡 ──
completed_elapsed = sorted(
r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
)
summary = {
"started_count": len(rows),
"completed_count": sum(1 for r in rows if r.status == "completed"),
"avg_elapsed_ms": _avg(completed_elapsed),
"p5_ms": _percentile(completed_elapsed, 5),
"p50_ms": _percentile(completed_elapsed, 50),
"p95_ms": _percentile(completed_elapsed, 95),
"p99_ms": _percentile(completed_elapsed, 99),
**_coupon_summary_aggregates(db, conditions),
**_success_rates(rows),
}
@@ -323,7 +384,7 @@ def coupon_data_report(
"summary": summary,
"daily": daily,
"hourly": hourly,
"total": len(rows),
"total": summary["started_count"],
"items": items,
}
+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(
+73 -25
View File
@@ -6,6 +6,7 @@
from __future__ import annotations
from datetime import date, datetime, time, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal
from zoneinfo import ZoneInfo
from sqlalchemy import Select, asc, case, desc, func, or_, select
@@ -297,6 +298,58 @@ def _comparison_percentile(sorted_values: list[int], q: float) -> int | None:
return int(value + 0.5)
def _round_duration_ms(value) -> int | None:
"""将数据库聚合结果按既有口径四舍五入为整数毫秒。"""
if value is None:
return None
return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
"""PostgreSQL 耗时聚合语句;每种状态只返回一行。"""
return select(
func.avg(ComparisonRecord.total_ms),
*(
func.percentile_cont(q).within_group(ComparisonRecord.total_ms)
for q in quantiles
),
).where(
*conditions,
ComparisonRecord.status == status,
ComparisonRecord.total_ms.is_not(None),
)
def _comparison_duration_aggregates(
db: Session,
*,
conditions: list,
status: str,
quantiles: tuple[float, ...],
) -> list[int | None]:
"""返回平均值和各分位数;生产 PG 在数据库内聚合,SQLite 仅作测试回退。"""
if db.bind is not None and db.bind.dialect.name == "postgresql":
row = db.execute(
_comparison_duration_aggregate_stmt(conditions, status, quantiles)
).one()
return [_round_duration_ms(value) for value in row]
# SQLite 没有 percentile_cont;本地/测试只回退读取耗时单列,不加载完整记录。
values = list(
db.execute(
select(ComparisonRecord.total_ms)
.where(
*conditions,
ComparisonRecord.status == status,
ComparisonRecord.total_ms.is_not(None),
)
.order_by(ComparisonRecord.total_ms)
).scalars()
)
average = _round_duration_ms(sum(values) / len(values)) if values else None
return [average, *(_comparison_percentile(values, q) for q in quantiles)]
def comparison_records_summary(
db: Session,
*,
@@ -332,20 +385,18 @@ def comparison_records_summary(
success = int(row[2] or 0)
lower_price = int(row[4] or 0)
cancelled = int(row[5] or 0)
success_durations = sorted(db.execute(
select(ComparisonRecord.total_ms).where(
*conditions,
ComparisonRecord.status == "success",
ComparisonRecord.total_ms.is_not(None),
)
).scalars().all())
cancelled_durations = sorted(db.execute(
select(ComparisonRecord.total_ms).where(
*conditions,
ComparisonRecord.status == "cancelled",
ComparisonRecord.total_ms.is_not(None),
)
).scalars().all())
success_duration_stats = _comparison_duration_aggregates(
db,
conditions=conditions,
status="success",
quantiles=(0.05, 0.5, 0.95, 0.99),
)
cancelled_duration_stats = _comparison_duration_aggregates(
db,
conditions=conditions,
status="cancelled",
quantiles=(0.05, 0.5, 0.95),
)
success_rate_denominator = started - cancelled
return {
"started": started,
@@ -354,19 +405,16 @@ def comparison_records_summary(
"success_rate": success / success_rate_denominator if success_rate_denominator else None,
"avg_token_cost": float(row[3]) if row[3] is not None else None,
"lower_price_rate": lower_price / success if success else None,
"avg_duration_ms": (
int(sum(success_durations) / len(success_durations) + 0.5)
if success_durations else None
),
"p5_duration_ms": _comparison_percentile(success_durations, 0.05),
"p50_duration_ms": _comparison_percentile(success_durations, 0.5),
"p95_duration_ms": _comparison_percentile(success_durations, 0.95),
"p99_duration_ms": _comparison_percentile(success_durations, 0.99),
"avg_duration_ms": success_duration_stats[0],
"p5_duration_ms": success_duration_stats[1],
"p50_duration_ms": success_duration_stats[2],
"p95_duration_ms": success_duration_stats[3],
"p99_duration_ms": success_duration_stats[4],
"cancelled": cancelled,
"cancelled_rate": cancelled / started if started else None,
"cancelled_p5_ms": _comparison_percentile(cancelled_durations, 0.05),
"cancelled_p50_ms": _comparison_percentile(cancelled_durations, 0.5),
"cancelled_p95_ms": _comparison_percentile(cancelled_durations, 0.95),
"cancelled_p5_ms": cancelled_duration_stats[1],
"cancelled_p50_ms": cancelled_duration_stats[2],
"cancelled_p95_ms": cancelled_duration_stats[3],
}
+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(
+7
View File
@@ -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 移动应用,用于微信支付);这是【已认证服务号】。
+428
View File
@@ -0,0 +1,428 @@
"""美团、京东 CPS 订单每日自动对账任务。
每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`(默认 05:00)后执行一次,按更新时间
回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
import os
import socket
import time
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, engine
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 _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)
@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,
"api_requests": 0,
}
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:
def reconcile_meituan() -> dict:
with SessionLocal() as db:
return cps_repo.reconcile_orders(
db,
start_time=int(start.timestamp()),
end_time=int(end.timestamp()),
query_time_type=2,
audit_context=common,
)
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"] = _skip_platform("meituan", common)
_touch_lock()
if settings.jd_union_configured:
def reconcile_jd() -> dict:
with SessionLocal() as db:
return cps_repo.reconcile_jd_orders(
db,
start_time=start,
end_time=end,
query_time_type=3,
audit_context=common,
)
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"] = _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
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 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):
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()
except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出
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",
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",
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",
extra={
"event": "cps_reconcile.worker_skipped",
"skip_reason": "all_platform_credentials_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
+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))
+6
View File
@@ -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)
+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 全 NULL、contact 有值)测「旧数据」展示;
- 截图:在 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",
+16
View File
@@ -4,12 +4,28 @@ from __future__ import annotations
from datetime import UTC, date, datetime
import pytest
from sqlalchemy.dialects import postgresql
from app.admin.repositories import queries
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
stmt = queries._comparison_duration_aggregate_stmt(
[], "success", (0.05, 0.5, 0.95, 0.99)
)
sql = str(
stmt.compile(
dialect=postgresql.dialect(),
compile_kwargs={"literal_binds": True},
)
)
assert sql.count("percentile_cont") == 4
assert "comparison_record.status = 'success'" in sql
def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
db = SessionLocal()
try:
+55 -2
View File
@@ -7,8 +7,12 @@ from __future__ import annotations
from datetime import UTC, date, datetime
from sqlalchemy import delete, func, select
from sqlalchemy.dialects import postgresql
from app.admin.repositories.coupon_data import coupon_data_report
from app.admin.repositories.coupon_data import (
_coupon_summary_aggregate_stmt,
coupon_data_report,
)
from app.db.session import SessionLocal
from app.models.coupon_state import CouponSession
from app.repositories.coupon_state import (
@@ -19,7 +23,12 @@ from app.repositories.coupon_state import (
def _agg_session(
trace: str, platforms, platform_success, *, status: str = "completed"
trace: str,
platforms,
platform_success,
*,
status: str = "completed",
elapsed_ms: int | None = None,
) -> CouponSession:
"""构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
return CouponSession(
@@ -31,9 +40,53 @@ def _agg_session(
platform_success=platform_success,
started_at=datetime(2020, 1, 2, tzinfo=UTC),
started_date=date(2020, 1, 2),
elapsed_ms=elapsed_ms,
)
def test_postgresql_coupon_summary_uses_ordered_set_aggregates() -> None:
sql = str(
_coupon_summary_aggregate_stmt([]).compile(
dialect=postgresql.dialect(),
compile_kwargs={"literal_binds": True},
)
)
assert sql.count("percentile_cont") == 4
assert "FILTER (WHERE coupon_session.status = 'completed'" in sql
def test_coupon_duration_summary_uses_only_completed_rows() -> None:
db = SessionLocal()
try:
db.add_all([
_agg_session("duration-a", [], [], elapsed_ms=1000),
_agg_session("duration-b", [], [], elapsed_ms=3000),
_agg_session(
"duration-failed", [], [], status="failed", elapsed_ms=100_000
),
])
db.flush()
summary = coupon_data_report(
db,
date_from="2020-01-02",
date_to="2020-01-02",
app_env="prod",
)["summary"]
assert summary["started_count"] == 3
assert summary["completed_count"] == 2
assert summary["avg_elapsed_ms"] == 2000
assert summary["p5_ms"] == 1100
assert summary["p50_ms"] == 2000
assert summary["p95_ms"] == 2900
assert summary["p99_ms"] == 2980
finally:
db.rollback()
db.close()
def _make_session(db, trace_id: str, **kw) -> CouponSession:
row = CouponSession(
trace_id=trace_id,
+197
View File
@@ -0,0 +1,197 @@
"""美团、京东 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
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,
"api_requests": 1,
}
def fake_jd(db, **kwargs):
calls["jd"] = kwargs
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)
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
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:
_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
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:
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_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 worker、IDE 索引、杀软)开着它, 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