Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ceef642854 | |||
| 1ec376880a | |||
| f29b28769e | |||
| d872755bb8 | |||
| ceceeb3458 | |||
| 07cf806805 | |||
| 31f61f6aad |
@@ -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
|
||||||
@@ -5,6 +5,7 @@
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
from decimal import Decimal, InvalidOperation
|
from decimal import Decimal, InvalidOperation
|
||||||
from typing import Any
|
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.admin.repositories.queries import _as_utc, offset_paginate
|
||||||
from app.integrations import jd_union, meituan
|
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_activity import CpsActivity
|
||||||
from app.models.cps_group import CpsGroup
|
from app.models.cps_group import CpsGroup
|
||||||
from app.models.cps_link import CpsClick, CpsLink
|
from app.models.cps_link import CpsClick, CpsLink
|
||||||
from app.models.cps_order import CpsOrder
|
from app.models.cps_order import CpsOrder
|
||||||
from app.models.cps_wx_user import CpsWxUser
|
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)为佣金真正到账
|
# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
|
||||||
_INVALID_STATUS = {"4", "5"}
|
_INVALID_STATUS = {"4", "5"}
|
||||||
@@ -378,18 +381,35 @@ def effective_commission_cents(order: CpsOrder) -> int:
|
|||||||
def reconcile_orders(
|
def reconcile_orders(
|
||||||
db: Session, *, start_time: int, end_time: int,
|
db: Session, *, start_time: int, end_time: int,
|
||||||
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
|
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
|
||||||
|
audit_context: dict[str, Any] | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
|
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
|
||||||
|
|
||||||
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
|
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
|
||||||
"""
|
"""
|
||||||
fetched = inserted = updated = pages = 0
|
fetched = inserted = updated = pages = api_requests = 0
|
||||||
page = 1
|
page = 1
|
||||||
while page <= max_pages:
|
while page <= max_pages:
|
||||||
resp = meituan.query_order(
|
api_requests += 1
|
||||||
sid=sid, start_time=start_time, end_time=end_time,
|
try:
|
||||||
query_time_type=query_time_type, page=page, limit=100,
|
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 []
|
rows = ((resp.get("data") or {}).get("dataList")) or []
|
||||||
if not rows:
|
if not rows:
|
||||||
break
|
break
|
||||||
@@ -419,30 +439,58 @@ def reconcile_orders(
|
|||||||
break
|
break
|
||||||
page += 1
|
page += 1
|
||||||
db.commit()
|
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(
|
def reconcile_jd_orders(
|
||||||
db: Session, *, start_time: datetime, end_time: datetime,
|
db: Session, *, start_time: datetime, end_time: datetime,
|
||||||
query_time_type: int = 3, max_pages: int = 100,
|
query_time_type: int = 3, max_pages: int = 100,
|
||||||
|
audit_context: dict[str, Any] | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""调京东 order.row.query 拉单 → 按订单行 upsert。
|
"""调京东 order.row.query 拉单 → 按订单行 upsert。
|
||||||
|
|
||||||
京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
|
京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
|
||||||
"""
|
"""
|
||||||
fetched = inserted = updated = pages = 0
|
fetched = inserted = updated = pages = api_requests = windows = 0
|
||||||
cur = start_time
|
cur = start_time
|
||||||
while cur < end_time:
|
while cur < end_time:
|
||||||
win_end = min(cur + timedelta(hours=1), end_time)
|
win_end = min(cur + timedelta(hours=1), end_time)
|
||||||
|
windows += 1
|
||||||
page = 1
|
page = 1
|
||||||
while page <= max_pages:
|
while page <= max_pages:
|
||||||
resp = jd_union.query_order_rows(
|
api_requests += 1
|
||||||
start_time=cur,
|
try:
|
||||||
end_time=win_end,
|
resp = jd_union.query_order_rows(
|
||||||
query_time_type=query_time_type,
|
start_time=cur,
|
||||||
page_index=page,
|
end_time=win_end,
|
||||||
page_size=200,
|
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 []
|
rows = resp.get("rows") or []
|
||||||
has_more = bool(resp.get("has_more"))
|
has_more = bool(resp.get("has_more"))
|
||||||
if not rows:
|
if not rows:
|
||||||
@@ -469,7 +517,14 @@ def reconcile_jd_orders(
|
|||||||
page += 1
|
page += 1
|
||||||
cur = win_end
|
cur = win_end
|
||||||
db.commit()
|
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(
|
def list_orders(
|
||||||
|
|||||||
@@ -192,7 +192,7 @@ def withdraw_info(
|
|||||||
wechat_bound=bool(u and u.wechat_openid),
|
wechat_bound=bool(u and u.wechat_openid),
|
||||||
wechat_nickname=u.wechat_nickname if u else None,
|
wechat_nickname=u.wechat_nickname if u else None,
|
||||||
wechat_avatar_url=u.wechat_avatar_url 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)],
|
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:
|
def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
|
||||||
auth = crud_wallet.sync_transfer_auth(db, user.id)
|
auth = crud_wallet.sync_transfer_auth(db, user.id)
|
||||||
state = auth.state if auth else "none"
|
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(
|
@router.post(
|
||||||
|
|||||||
@@ -9,15 +9,17 @@ import asyncio
|
|||||||
import contextlib
|
import contextlib
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import socket
|
||||||
import time
|
import time
|
||||||
from collections.abc import Iterator
|
from collections.abc import Callable, Iterator
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
from app.admin.repositories import cps as cps_repo
|
from app.admin.repositories import cps as cps_repo
|
||||||
from app.core.config import settings
|
from app.core.config import settings
|
||||||
from app.core.rewards import CN_TZ
|
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")
|
logger = logging.getLogger("shagua.cps_reconcile")
|
||||||
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
|
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
|
||||||
@@ -27,6 +29,21 @@ def _cn_now() -> datetime:
|
|||||||
return datetime.now(CN_TZ)
|
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:
|
def _touch_lock() -> None:
|
||||||
with contextlib.suppress(FileNotFoundError):
|
with contextlib.suppress(FileNotFoundError):
|
||||||
os.utime(_LOCK_PATH, None)
|
os.utime(_LOCK_PATH, None)
|
||||||
@@ -67,52 +84,240 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
|
|||||||
|
|
||||||
|
|
||||||
def _empty_result() -> dict:
|
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)
|
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))
|
lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
|
||||||
start = end - timedelta(days=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 = {
|
result = {
|
||||||
|
"run_id": run_id,
|
||||||
|
"trigger": trigger,
|
||||||
"window_start": start.isoformat(),
|
"window_start": start.isoformat(),
|
||||||
"window_end": end.isoformat(),
|
"window_end": end.isoformat(),
|
||||||
"meituan": None,
|
"meituan": None,
|
||||||
"jd": None,
|
"jd": None,
|
||||||
"errors": {},
|
"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:
|
if settings.mt_cps_configured:
|
||||||
try:
|
def reconcile_meituan() -> dict:
|
||||||
with SessionLocal() as db:
|
with SessionLocal() as db:
|
||||||
result["meituan"] = cps_repo.reconcile_orders(
|
return cps_repo.reconcile_orders(
|
||||||
db,
|
db,
|
||||||
start_time=int(start.timestamp()),
|
start_time=int(start.timestamp()),
|
||||||
end_time=int(end.timestamp()),
|
end_time=int(end.timestamp()),
|
||||||
query_time_type=2,
|
query_time_type=2,
|
||||||
|
audit_context=common,
|
||||||
)
|
)
|
||||||
except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台
|
|
||||||
result["errors"]["meituan"] = str(exc)
|
result["meituan"], error = _run_platform(
|
||||||
logger.exception("CPS auto reconcile failed platform=meituan")
|
platform="meituan",
|
||||||
|
query_time_type=2,
|
||||||
|
common=common,
|
||||||
|
reconcile=reconcile_meituan,
|
||||||
|
)
|
||||||
|
if error is not None:
|
||||||
|
result["errors"]["meituan"] = error
|
||||||
else:
|
else:
|
||||||
result["meituan"] = {**_empty_result(), "skipped": "not_configured"}
|
result["meituan"] = _skip_platform("meituan", common)
|
||||||
|
|
||||||
_touch_lock()
|
_touch_lock()
|
||||||
if settings.jd_union_configured:
|
if settings.jd_union_configured:
|
||||||
try:
|
def reconcile_jd() -> dict:
|
||||||
with SessionLocal() as db:
|
with SessionLocal() as db:
|
||||||
result["jd"] = cps_repo.reconcile_jd_orders(
|
return cps_repo.reconcile_jd_orders(
|
||||||
db,
|
db,
|
||||||
start_time=start,
|
start_time=start,
|
||||||
end_time=end,
|
end_time=end,
|
||||||
query_time_type=3,
|
query_time_type=3,
|
||||||
|
audit_context=common,
|
||||||
)
|
)
|
||||||
except Exception as exc: # noqa: BLE001 - 单平台失败不能阻断另一平台
|
|
||||||
result["errors"]["jd"] = str(exc)
|
result["jd"], error = _run_platform(
|
||||||
logger.exception("CPS auto reconcile failed platform=jd")
|
platform="jd",
|
||||||
|
query_time_type=3,
|
||||||
|
common=common,
|
||||||
|
reconcile=reconcile_jd,
|
||||||
|
)
|
||||||
|
if error is not None:
|
||||||
|
result["errors"]["jd"] = error
|
||||||
else:
|
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
|
return result
|
||||||
|
|
||||||
@@ -127,42 +332,90 @@ async def _run_loop() -> None:
|
|||||||
lock_stale_after = max(interval * 3, 1800)
|
lock_stale_after = max(interval * 3, 1800)
|
||||||
with _single_instance_lock(lock_stale_after) as lock_acquired:
|
with _single_instance_lock(lock_stale_after) as lock_acquired:
|
||||||
if not 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
|
return
|
||||||
await _run_locked_loop(interval, run_hour)
|
await _run_locked_loop(interval, run_hour)
|
||||||
|
|
||||||
|
|
||||||
async def _run_locked_loop(interval: int, run_hour: int) -> None:
|
async def _run_locked_loop(interval: int, run_hour: int) -> None:
|
||||||
|
worker_started_at = _cn_now()
|
||||||
logger.info(
|
logger.info(
|
||||||
"CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
|
"CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
|
||||||
run_hour,
|
run_hour,
|
||||||
interval,
|
interval,
|
||||||
settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
|
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
|
last_run: date | None = None
|
||||||
try:
|
try:
|
||||||
while True:
|
while True:
|
||||||
|
run_id: str | None = None
|
||||||
try:
|
try:
|
||||||
_touch_lock()
|
_touch_lock()
|
||||||
now = _cn_now()
|
now = _cn_now()
|
||||||
if _should_run(last_run, now, run_hour):
|
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()
|
last_run = now.date()
|
||||||
logger.info("CPS auto reconcile done date=%s result=%s", last_run, result)
|
|
||||||
except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出
|
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)
|
await asyncio.sleep(interval)
|
||||||
except asyncio.CancelledError:
|
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
|
raise
|
||||||
|
|
||||||
|
|
||||||
def start_cps_reconcile_worker() -> asyncio.Task | None:
|
def start_cps_reconcile_worker() -> asyncio.Task | None:
|
||||||
if not settings.CPS_AUTO_RECONCILE_ENABLED:
|
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
|
return None
|
||||||
if not settings.mt_cps_configured and not settings.jd_union_configured:
|
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 None
|
||||||
return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
|
return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
|
||||||
|
|
||||||
|
|||||||
+51
-2
@@ -22,13 +22,13 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
from contextvars import ContextVar
|
from contextvars import ContextVar
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from logging.handlers import RotatingFileHandler
|
from logging.handlers import RotatingFileHandler
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
|
# 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
|
||||||
# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
|
# 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
|
||||||
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
|
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
|
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
|
_CONFIGURED = False
|
||||||
|
|
||||||
|
|
||||||
@@ -126,7 +175,7 @@ def setup_logging(debug: bool = False) -> None:
|
|||||||
Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
|
Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
|
||||||
)
|
)
|
||||||
Path(log_file).parent.mkdir(parents=True, exist_ok=True)
|
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",
|
log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
|
||||||
)
|
)
|
||||||
file_handler.setFormatter(JsonFormatter(service))
|
file_handler.setFormatter(JsonFormatter(service))
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -30,6 +30,7 @@ from app.core.config import settings
|
|||||||
from app.db.session import SessionLocal
|
from app.db.session import SessionLocal
|
||||||
from app.models.price_report import PriceReport
|
from app.models.price_report import PriceReport
|
||||||
from app.models.user import User
|
from app.models.user import User
|
||||||
|
from app.repositories.user import _gen_unique_username
|
||||||
|
|
||||||
if hasattr(sys.stdout, "reconfigure"):
|
if hasattr(sys.stdout, "reconfigure"):
|
||||||
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
|
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文/¥
|
||||||
@@ -84,10 +85,11 @@ def seed(db) -> list[PriceReport]:
|
|||||||
# 1) 建 3 个 mock 用户
|
# 1) 建 3 个 mock 用户
|
||||||
users: dict[str, User] = {}
|
users: dict[str, User] = {}
|
||||||
for i, (phone, nickname) in enumerate(
|
for i, (phone, nickname) in enumerate(
|
||||||
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
|
zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
|
||||||
):
|
):
|
||||||
u = User(
|
u = User(
|
||||||
phone=phone,
|
phone=phone,
|
||||||
|
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
|
||||||
nickname=nickname,
|
nickname=nickname,
|
||||||
register_channel="sms",
|
register_channel="sms",
|
||||||
status="active",
|
status="active",
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ import argparse
|
|||||||
import sys
|
import sys
|
||||||
import uuid
|
import uuid
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
from datetime import datetime, timedelta, timezone
|
from datetime import UTC, datetime, timedelta
|
||||||
|
|
||||||
from sqlalchemy import delete, select
|
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.ad_reward import AdRewardRecord
|
||||||
from app.models.user import User
|
from app.models.user import User
|
||||||
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
|
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
|
||||||
|
from app.repositories.user import _gen_unique_username
|
||||||
|
|
||||||
# Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
|
# Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
|
||||||
if hasattr(sys.stdout, "reconfigure"):
|
if hasattr(sys.stdout, "reconfigure"):
|
||||||
@@ -51,7 +52,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r
|
|||||||
|
|
||||||
def _naive_utc_now() -> datetime:
|
def _naive_utc_now() -> datetime:
|
||||||
"""与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
|
"""与 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:
|
def _mock_transfer_no() -> str:
|
||||||
@@ -96,6 +97,7 @@ def seed(db) -> list[WithdrawOrder]:
|
|||||||
for phone, nickname in users_spec:
|
for phone, nickname in users_spec:
|
||||||
u = User(
|
u = User(
|
||||||
phone=phone,
|
phone=phone,
|
||||||
|
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
|
||||||
nickname=nickname,
|
nickname=nickname,
|
||||||
register_channel="sms",
|
register_channel="sms",
|
||||||
status="active",
|
status="active",
|
||||||
|
|||||||
@@ -1,11 +1,15 @@
|
|||||||
"""美团、京东 CPS 每日自动对账 worker。"""
|
"""美团、京东 CPS 每日自动对账 worker。"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
from app.core import cps_reconcile_worker as worker
|
from app.core import cps_reconcile_worker as worker
|
||||||
from app.core.config import settings
|
from app.core.config import settings
|
||||||
from app.core.rewards import CN_TZ
|
from app.core.rewards import CN_TZ
|
||||||
|
from app.db.session import SessionLocal
|
||||||
from app.integrations.meituan import MeituanCpsError
|
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):
|
def fake_meituan(db, **kwargs):
|
||||||
calls["meituan"] = 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):
|
def fake_jd(db, **kwargs):
|
||||||
calls["jd"] = 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_orders", fake_meituan)
|
||||||
monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
|
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["meituan"]["start_time"] == int((now - timedelta(days=3)).timestamp())
|
||||||
assert calls["jd"]["start_time"] == now - timedelta(days=3)
|
assert calls["jd"]["start_time"] == now - timedelta(days=3)
|
||||||
assert calls["jd"]["end_time"] == now
|
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:
|
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 result["errors"]["meituan"] == "temporary failure"
|
||||||
assert jd_called is True
|
assert jd_called is True
|
||||||
assert result["jd"]["fetched"] == 1
|
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:
|
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
|
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:
|
def test_start_worker_respects_auto_switch(monkeypatch) -> None:
|
||||||
monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False)
|
monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False)
|
||||||
assert worker.start_cps_reconcile_worker() is None
|
assert worker.start_cps_reconcile_worker() is None
|
||||||
|
|||||||
@@ -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
@@ -9,7 +9,7 @@ from sqlalchemy import select
|
|||||||
|
|
||||||
from app.db.session import SessionLocal
|
from app.db.session import SessionLocal
|
||||||
from app.models.user import User
|
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
|
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))
|
r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
|
||||||
assert r.json()["status"] == "rejected"
|
assert r.json()["status"] == "rejected"
|
||||||
assert r.json()["fail_reason"] == "测试拒绝"
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user