Compare commits

..

1 Commits

Author SHA1 Message Date
unknown 49e271927a 修复:领券单券成功率按任务独立统计 2026-07-23 21:55:02 +08:00
18 changed files with 134 additions and 1248 deletions
+4 -3
View File
@@ -30,17 +30,18 @@ from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder from app.models.wallet import CoinTransaction, WithdrawOrder
_BEIJING = timezone(timedelta(hours=8)) _BEIJING = timezone(timedelta(hours=8))
REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward", "signin_boost") REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景); # 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未 # coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。 # 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
# reward_video/ad_reward 及历史 signin_boost 均归看视频桶,不再混进领券奖励或常规任务 # ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)
COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward") COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward") COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
# 常规任务必须按明确来源相加;不能从全部正向流水反减排除项,否则新增广告/运营 # 常规任务必须按明确来源相加;不能从全部正向流水反减排除项,否则新增广告/运营
# biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。 # biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。
REGULAR_TASK_EXACT_BIZ_TYPES = ( REGULAR_TASK_EXACT_BIZ_TYPES = (
"signin", "signin",
"signin_boost",
"price_report_reward", "price_report_reward",
"feedback_reward", "feedback_reward",
) )
+85 -167
View File
@@ -12,10 +12,6 @@ from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.feedback import ( from app.admin.schemas.feedback import (
FeedbackApproveRequest, FeedbackApproveRequest,
FeedbackBulkApproveRequest,
FeedbackBulkItemResult,
FeedbackBulkRejectRequest,
FeedbackBulkResult,
FeedbackOut, FeedbackOut,
FeedbackRejectRequest, FeedbackRejectRequest,
FeedbackSummary, FeedbackSummary,
@@ -37,123 +33,6 @@ def _ensure_pending(fb: Feedback) -> None:
raise HTTPException(status_code=400, detail="反馈已审核") raise HTTPException(status_code=400, detail="反馈已审核")
def _approve_feedback(
db: AdminDb,
admin: AdminUser,
feedback_id: int,
payload: FeedbackApproveRequest | FeedbackBulkApproveRequest,
ip: str,
*,
bulk: bool = False,
) -> FeedbackOut:
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
db,
fb,
status="adopted",
reward_coins=payload.reward_coins,
review_note=payload.note,
admin_reply=payload.reply,
reviewed_by_admin_id=admin.id,
commit=False,
)
wallet_repo.grant_coins(
db,
fb.user_id,
payload.reward_coins,
biz_type="feedback_reward",
ref_id=str(fb.id),
remark="意见反馈被采纳",
)
detail = {
"before": before,
"after": "adopted",
"reward_coins": payload.reward_coins,
"note": payload.note,
"reply": payload.reply,
}
if bulk:
detail["bulk"] = True
write_audit(
db,
admin,
action="feedback.approve",
target_type="feedback",
target_id=feedback_id,
detail=detail,
ip=ip,
commit=False,
)
db.commit()
db.refresh(fb)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reward(db, fb)
return out
def _reject_feedback(
db: AdminDb,
admin: AdminUser,
feedback_id: int,
payload: FeedbackRejectRequest | FeedbackBulkRejectRequest,
ip: str,
*,
bulk: bool = False,
) -> FeedbackOut:
fb = db.get(Feedback, feedback_id, with_for_update=True)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
db,
fb,
status="rejected",
reject_reason=payload.reason,
review_note=payload.note,
admin_reply=payload.reply,
reviewed_by_admin_id=admin.id,
commit=False,
)
detail = {
"before": before,
"after": "rejected",
"reason": payload.reason,
"note": payload.note,
"reply": payload.reply,
}
if bulk:
detail["bulk"] = True
write_audit(
db,
admin,
action="feedback.reject",
target_type="feedback",
target_id=feedback_id,
detail=detail,
ip=ip,
commit=False,
)
db.commit()
db.refresh(fb)
out = FeedbackOut.model_validate(fb)
notification_events.notify_feedback_reply(db, fb)
return out
def _bulk_result(items: list[FeedbackBulkItemResult]) -> FeedbackBulkResult:
success = sum(1 for item in items if item.ok)
return FeedbackBulkResult(
total=len(items), success=success, failed=len(items) - success, items=items,
)
@router.get("", response_model=CursorPage[FeedbackOut], summary="反馈工单列表") @router.get("", response_model=CursorPage[FeedbackOut], summary="反馈工单列表")
def list_feedbacks( def list_feedbacks(
db: AdminDb, db: AdminDb,
@@ -194,50 +73,6 @@ def feedback_summary(db: AdminDb) -> FeedbackSummary:
return FeedbackSummary.model_validate(queries.feedback_summary(db)) return FeedbackSummary.model_validate(queries.feedback_summary(db))
@router.post("/bulk/approve", response_model=FeedbackBulkResult, summary="批量采纳反馈并发金币")
def bulk_approve_feedbacks(
body: FeedbackBulkApproveRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
try:
out = _approve_feedback(db, admin, feedback_id, body, ip, bulk=True)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
@router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
def bulk_reject_feedbacks(
body: FeedbackBulkRejectRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> FeedbackBulkResult:
results: list[FeedbackBulkItemResult] = []
ip = get_client_ip(request)
for feedback_id in body.ids:
try:
out = _reject_feedback(db, admin, feedback_id, body, ip, bulk=True)
results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
except HTTPException as exc:
db.rollback()
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
return _bulk_result(results)
@router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理") @router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
def handle_feedback( def handle_feedback(
feedback_id: int, feedback_id: int,
@@ -258,7 +93,53 @@ def approve_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> FeedbackOut: ) -> FeedbackOut:
return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request)) fb = db.get(Feedback, feedback_id)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
db,
fb,
status="adopted",
reward_coins=payload.reward_coins,
review_note=payload.note,
admin_reply=payload.reply,
reviewed_by_admin_id=admin.id,
commit=False,
)
wallet_repo.grant_coins(
db,
fb.user_id,
payload.reward_coins,
biz_type="feedback_reward",
ref_id=str(fb.id),
remark="意见反馈被采纳",
)
write_audit(
db,
admin,
action="feedback.approve",
target_type="feedback",
target_id=feedback_id,
detail={
"before": before,
"after": "adopted",
"reward_coins": payload.reward_coins,
"note": payload.note,
"reply": payload.reply,
},
ip=get_client_ip(request),
commit=False,
)
db.commit()
db.refresh(fb)
out = FeedbackOut.model_validate(fb)
# PRD #10 反馈奖励:采纳发金币后通知用户(站内 + push,必带官方留言)。
# 业务已 commit,通知失败只 log 不影响审核结果。
notification_events.notify_feedback_reward(db, fb)
return out
@router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈") @router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -269,4 +150,41 @@ def reject_feedback(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> FeedbackOut: ) -> FeedbackOut:
return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request)) fb = db.get(Feedback, feedback_id)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
_ensure_pending(fb)
before = fb.status
mutations.review_feedback(
db,
fb,
status="rejected",
reject_reason=payload.reason,
review_note=payload.note,
admin_reply=payload.reply,
reviewed_by_admin_id=admin.id,
commit=False,
)
write_audit(
db,
admin,
action="feedback.reject",
target_type="feedback",
target_id=feedback_id,
detail={
"before": before,
"after": "rejected",
"reason": payload.reason,
"note": payload.note,
"reply": payload.reply,
},
ip=get_client_ip(request),
commit=False,
)
db.commit()
db.refresh(fb)
out = FeedbackOut.model_validate(fb)
# PRD #9 官方回复:未采纳也回复了用户(原因/留言用户端可见),通知去反馈历史页查看。
notification_events.notify_feedback_reply(db, fb)
return out
+33 -103
View File
@@ -17,10 +17,6 @@ from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_ro
from app.admin.repositories import mutations, queries from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.price_report import ( from app.admin.schemas.price_report import (
PriceReportBulkItemResult,
PriceReportBulkRejectRequest,
PriceReportBulkRequest,
PriceReportBulkResult,
PriceReportOut, PriceReportOut,
PriceReportRejectRequest, PriceReportRejectRequest,
PriceReportSummary, PriceReportSummary,
@@ -38,59 +34,6 @@ router = APIRouter(
) )
def _approve_price_report(
db: AdminDb, admin: AdminUser, report_id: int, ip: str, *, bulk: bool = False
) -> PriceReport:
rep = db.get(PriceReport, report_id, with_for_update=True)
if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending":
raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
coins = PRICE_REPORT_REWARD_COINS
mutations.review_price_report(db, rep, status="approved", reward_coins=coins, commit=False)
wallet_repo.grant_coins(
db, rep.user_id, coins,
biz_type="price_report_reward", ref_id=str(rep.id), remark="上报更低价审核通过",
)
detail = {"reward_coins": coins, "user_id": rep.user_id}
if bulk:
detail["bulk"] = True
write_audit(
db, admin, action="price_report.approve", target_type="price_report", target_id=report_id,
detail=detail, ip=ip, commit=False,
)
db.commit()
notification_events.notify_report_approved(db, rep)
return rep
def _reject_price_report(
db: AdminDb, admin: AdminUser, report_id: int, reason: str, ip: str, *, bulk: bool = False
) -> PriceReport:
rep = db.get(PriceReport, report_id, with_for_update=True)
if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending":
raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
mutations.review_price_report(db, rep, status="rejected", reject_reason=reason, commit=False)
detail = {"reason": reason, "user_id": rep.user_id}
if bulk:
detail["bulk"] = True
write_audit(
db, admin, action="price_report.reject", target_type="price_report", target_id=report_id,
detail=detail, ip=ip, commit=False,
)
db.commit()
return rep
def _bulk_result(items: list[PriceReportBulkItemResult]) -> PriceReportBulkResult:
success = sum(1 for item in items if item.ok)
return PriceReportBulkResult(
total=len(items), success=success, failed=len(items) - success, items=items,
)
@router.get("", response_model=CursorPage[PriceReportOut], summary="上报更低价列表(筛选+分页)") @router.get("", response_model=CursorPage[PriceReportOut], summary="上报更低价列表(筛选+分页)")
def list_price_reports( def list_price_reports(
db: AdminDb, db: AdminDb,
@@ -117,50 +60,6 @@ def price_report_summary(db: AdminDb) -> PriceReportSummary:
return PriceReportSummary.model_validate(queries.price_report_summary(db)) return PriceReportSummary.model_validate(queries.price_report_summary(db))
@router.post("/bulk/approve", response_model=PriceReportBulkResult, summary="批量通过上报(发固定金币)")
def bulk_approve_price_reports(
body: PriceReportBulkRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> PriceReportBulkResult:
results: list[PriceReportBulkItemResult] = []
ip = get_client_ip(request)
for report_id in body.ids:
try:
rep = _approve_price_report(db, admin, report_id, ip, bulk=True)
results.append(PriceReportBulkItemResult(id=report_id, ok=True, status=rep.status))
except HTTPException as exc:
db.rollback()
results.append(PriceReportBulkItemResult(id=report_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
results.append(PriceReportBulkItemResult(id=report_id, ok=False, error="系统异常"))
return _bulk_result(results)
@router.post("/bulk/reject", response_model=PriceReportBulkResult, summary="批量拒绝上报")
def bulk_reject_price_reports(
body: PriceReportBulkRejectRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> PriceReportBulkResult:
results: list[PriceReportBulkItemResult] = []
ip = get_client_ip(request)
for report_id in body.ids:
try:
rep = _reject_price_report(db, admin, report_id, body.reason, ip, bulk=True)
results.append(PriceReportBulkItemResult(id=report_id, ok=True, status=rep.status))
except HTTPException as exc:
db.rollback()
results.append(PriceReportBulkItemResult(id=report_id, ok=False, error=str(exc.detail)))
except Exception: # noqa: BLE001 - 单笔失败不打断整批
db.rollback()
results.append(PriceReportBulkItemResult(id=report_id, ok=False, error="系统异常"))
return _bulk_result(results)
@router.post("/{report_id}/approve", response_model=OkResponse, summary="通过上报(发固定金币)") @router.post("/{report_id}/approve", response_model=OkResponse, summary="通过上报(发固定金币)")
def approve_price_report( def approve_price_report(
report_id: int, report_id: int,
@@ -168,7 +67,27 @@ def approve_price_report(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> OkResponse: ) -> OkResponse:
_approve_price_report(db, admin, report_id, get_client_ip(request)) # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖,
# 锁住该行串行化,第二个请求拿锁后看到 approved → 走 400。SQLite 下 FOR UPDATE 为 no-op。
rep = db.get(PriceReport, report_id, with_for_update=True)
if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending":
raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
coins = PRICE_REPORT_REWARD_COINS
# 改状态 + 发金币 + 审计同一事务(commit=False),最后一起 commit:改了就有痕、发了就留账
mutations.review_price_report(db, rep, status="approved", reward_coins=coins, commit=False)
wallet_repo.grant_coins(
db, rep.user_id, coins,
biz_type="price_report_reward", ref_id=str(rep.id), remark="上报更低价审核通过",
)
write_audit(
db, admin, action="price_report.approve", target_type="price_report", target_id=report_id,
detail={"reward_coins": coins, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
)
db.commit()
# PRD #11 爆料审核通过:发金币后通知用户(站内 + push)。业务已 commit,通知失败只 log。
notification_events.notify_report_approved(db, rep)
return OkResponse() return OkResponse()
@@ -180,5 +99,16 @@ def reject_price_report(
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
) -> OkResponse: ) -> OkResponse:
_reject_price_report(db, admin, report_id, body.reason, get_client_ip(request)) rep = db.get(PriceReport, report_id, with_for_update=True) # 行锁,同 approve(防并发重复审核)
if rep is None:
raise HTTPException(status_code=404, detail="上报记录不存在")
if rep.status != "pending":
raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
reason = body.reason.strip()
mutations.review_price_report(db, rep, status="rejected", reject_reason=reason, commit=False)
write_audit(
db, admin, action="price_report.reject", target_type="price_report", target_id=report_id,
detail={"reason": reason, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse() return OkResponse()
+1 -49
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
from datetime import datetime from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator from pydantic import BaseModel, ConfigDict, Field
from app.core.rewards import FEEDBACK_REWARD_MAX_COINS from app.core.rewards import FEEDBACK_REWARD_MAX_COINS
@@ -55,54 +55,6 @@ class FeedbackRejectRequest(BaseModel):
reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见") reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
class FeedbackBulkRequest(BaseModel):
ids: list[int] = Field(min_length=1, max_length=50, description="待审核反馈 ID 列表")
@field_validator("ids")
@classmethod
def _ids_must_be_unique(cls, ids: list[int]) -> list[int]:
if len(ids) != len(set(ids)):
raise ValueError("反馈 ID 不能重复")
return ids
class FeedbackBulkApproveRequest(FeedbackBulkRequest):
reward_coins: int = Field(
ge=1,
le=FEEDBACK_REWARD_MAX_COINS,
description="每条采纳反馈发放的金币数",
)
note: str | None = Field(default=None, max_length=256, description="采纳要点/审核备注(内部)")
reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
class FeedbackBulkRejectRequest(FeedbackBulkRequest):
reason: str = Field(min_length=1, max_length=256, description="批量未采纳原因,用户端可见")
note: str | None = Field(default=None, max_length=256, description="运营内部审核备注")
reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
@field_validator("reason")
@classmethod
def _reason_not_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("未采纳原因不能为空")
return value.strip()
class FeedbackBulkItemResult(BaseModel):
id: int
ok: bool
status: str | None = None
error: str | None = None
class FeedbackBulkResult(BaseModel):
total: int
success: int
failed: int
items: list[FeedbackBulkItemResult]
class FeedbackSummary(BaseModel): class FeedbackSummary(BaseModel):
"""审核台顶部各状态计数(pending 含历史 new 态)。""" """审核台顶部各状态计数(pending 含历史 new 态)。"""
-36
View File
@@ -56,42 +56,6 @@ class PriceReportRejectRequest(BaseModel):
return v.strip() return v.strip()
class PriceReportBulkRequest(BaseModel):
ids: list[int] = Field(min_length=1, max_length=50, description="待审核上报 ID 列表")
@field_validator("ids")
@classmethod
def _ids_must_be_unique(cls, ids: list[int]) -> list[int]:
if len(ids) != len(set(ids)):
raise ValueError("上报 ID 不能重复")
return ids
class PriceReportBulkRejectRequest(PriceReportBulkRequest):
reason: str = Field(min_length=1, max_length=256, description="批量拒绝理由,用户端记录页会看到")
@field_validator("reason")
@classmethod
def _reason_not_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("拒绝理由不能为空")
return value.strip()
class PriceReportBulkItemResult(BaseModel):
id: int
ok: bool
status: str | None = None
error: str | None = None
class PriceReportBulkResult(BaseModel):
total: int
success: int
failed: int
items: list[PriceReportBulkItemResult]
class PriceReportSummary(BaseModel): class PriceReportSummary(BaseModel):
"""审核台顶部各状态计数。""" """审核台顶部各状态计数。"""
-37
View File
@@ -20,8 +20,6 @@ from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord from app.models.comparison import ComparisonRecord
from app.repositories import comparison as crud_compare from app.repositories import comparison as crud_compare
from app.schemas.compare_record import ( from app.schemas.compare_record import (
CompareStartReserveIn,
CompareStartReserveOut,
CompareStatsOut, CompareStatsOut,
ComparisonRecordCreatedOut, ComparisonRecordCreatedOut,
ComparisonRecordDetailOut, ComparisonRecordDetailOut,
@@ -37,41 +35,6 @@ logger = logging.getLogger("shagua.compare_record")
router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"]) router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
@router.post(
"/start",
response_model=CompareStartReserveOut,
summary="预占一次当日比价发起次数(每人每天最多100次)",
)
def reserve_compare_start(
payload: CompareStartReserveIn,
user: CurrentUser,
db: DbSession,
) -> CompareStartReserveOut:
try:
_, used = crud_compare.reserve_daily_start(
db,
user_id=user.id,
trace_id=payload.trace_id,
business_type=payload.business_type,
device_id=payload.device_id,
)
except crud_compare.DailyCompareStartLimitExceeded:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="今日已比价超过100次,请明天再试",
) from None
except crud_compare.ComparisonTraceOwnershipError:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="比价任务标识冲突,请重新发起",
) from None
return CompareStartReserveOut(
limit=crud_compare.DAILY_COMPARE_START_LIMIT,
used=used,
remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
)
@router.post( @router.post(
"/record", "/record",
response_model=ComparisonRecordCreatedOut, response_model=ComparisonRecordCreatedOut,
+2 -3
View File
@@ -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" and auth.authorization_id), transfer_auth_enabled=bool(auth and auth.state == "active"),
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,8 +335,7 @@ 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"
enabled = bool(auth and auth.state == "active" and auth.authorization_id) return TransferAuthStatusOut(state=state, enabled=(state == "active"))
return TransferAuthStatusOut(state=state, enabled=enabled)
@router.post( @router.post(
+2 -51
View File
@@ -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,55 +91,6 @@ 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
@@ -175,7 +126,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 = SafeRotatingFileHandler( file_handler = RotatingFileHandler(
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))
+1 -84
View File
@@ -5,7 +5,7 @@
""" """
from __future__ import annotations from __future__ import annotations
from datetime import datetime, timedelta from datetime import datetime
from sqlalchemy import func, or_, select from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, defer from sqlalchemy.orm import Session, defer
@@ -14,19 +14,8 @@ from app.core.rewards import CN_TZ
from app.models.ad_feed_reward import AdFeedRewardRecord from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.comparison import ComparisonRecord from app.models.comparison import ComparisonRecord
from app.models.savings import SavingsRecord from app.models.savings import SavingsRecord
from app.models.user import User
from app.schemas.compare_record import ComparisonRecordIn from app.schemas.compare_record import ComparisonRecordIn
DAILY_COMPARE_START_LIMIT = 100
class DailyCompareStartLimitExceeded(Exception):
"""The authenticated user has consumed today's comparison-start quota."""
class ComparisonTraceOwnershipError(Exception):
"""A trace id already belongs to a different authenticated user."""
def _yuan_to_cents(yuan: float | None) -> int | None: def _yuan_to_cents(yuan: float | None) -> int | None:
"""元(float)→ 分(int)。None 透传。""" """元(float)→ 分(int)。None 透传。"""
@@ -254,78 +243,6 @@ def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
).scalar_one_or_none() ).scalar_one_or_none()
def reserve_daily_start(
db: Session,
*,
user_id: int,
trace_id: str,
business_type: str = "food",
device_id: str | None = None,
now: datetime | None = None,
) -> tuple[ComparisonRecord, int]:
"""Atomically reserve one of a user's 100 Beijing-day comparison starts.
``trace_id`` makes client retries idempotent. Locking the user row serializes
concurrent starts for one account, so parallel requests cannot both consume
the final available slot. The reservation is the existing ``running``
comparison row; later result reporting updates that same row.
"""
db.execute(select(User.id).where(User.id == user_id).with_for_update()).scalar_one()
existing = _get_by_trace(db, trace_id)
if existing is not None:
if existing.user_id not in (None, user_id):
raise ComparisonTraceOwnershipError
if existing.user_id is None:
existing.user_id = user_id
if existing.device_id is None and device_id:
existing.device_id = device_id
db.commit()
db.refresh(existing)
existing_at = existing.created_at
if existing_at.tzinfo is not None:
existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= day_start,
ComparisonRecord.created_at < day_end,
)
) or 0
return existing, int(used)
current = now or datetime.now(CN_TZ)
if current.tzinfo is not None:
current = current.astimezone(CN_TZ).replace(tzinfo=None)
day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + timedelta(days=1)
used = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.user_id == user_id,
ComparisonRecord.created_at >= day_start,
ComparisonRecord.created_at < day_end,
)
) or 0
if used >= DAILY_COMPARE_START_LIMIT:
raise DailyCompareStartLimitExceeded
rec = ComparisonRecord(
trace_id=trace_id,
user_id=user_id,
business_type=business_type or "food",
device_id=device_id,
status="running",
created_at=current,
)
db.add(rec)
db.commit()
db.refresh(rec)
return rec, int(used) + 1
def harvest_running( def harvest_running(
db: Session, db: Session,
*, *,
+1 -14
View File
@@ -13,6 +13,7 @@ from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator from pydantic import BaseModel, ConfigDict, Field, field_validator
# ===== 上报请求 ===== # ===== 上报请求 =====
class ComparisonItemIn(BaseModel): class ComparisonItemIn(BaseModel):
@@ -197,20 +198,6 @@ class ComparisonRecordCreatedOut(BaseModel):
id: int = Field(..., description="写入(或已存在)的记录 id") id: int = Field(..., description="写入(或已存在)的记录 id")
class CompareStartReserveIn(BaseModel):
"""Reserve one authenticated comparison start before the agent begins."""
trace_id: str = Field(..., min_length=1, max_length=64)
business_type: str = Field(default="food", min_length=1, max_length=16)
device_id: str | None = Field(default=None, max_length=64)
class CompareStartReserveOut(BaseModel):
limit: int
used: int
remaining: int
class CompareStatsOut(BaseModel): class CompareStatsOut(BaseModel):
"""「我的」页省钱战绩卡(比价口径)聚合。""" """「我的」页省钱战绩卡(比价口径)聚合。"""
-277
View File
@@ -1,277 +0,0 @@
"""一次性 mock:造几条不同状态的用户反馈,供运营后台「反馈工单」页联调验收。
覆盖:
- 三个状态 tab(待审核 pending / 已采纳 adopted / 未采纳 rejected),重点铺待审核;
- 两种反馈类型 source(普通反馈 profile /我的页入口比价反馈 comparison / 比价结果页入口),
比价反馈带问题场景scene(找错商品/优惠不对/比价太慢);
- 提交端环境快照(app_version / device_model / rom_name / android_version)新端反馈才有,
另留 1~2 条历史反馈(env NULLcontact 有值)旧数据展示;
- 截图: data/media/feedback/ 生成真实可加载的纯色 PNG(手写字节,无需 Pillow),
让审核抽屉的图能真加载出来( seed_mock_price_reports 同法)
- 已采纳条带 reward_coins + admin_reply + review_note;未采纳条带 reject_reason + admin_reply
幂等:每次运行先按固定 mock 手机号清掉旧 mock 用户/反馈 + mock 截图再重建仅清理用 --clean-only
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py
.venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py --clean-only
看图:前端反馈工单(http://localhost:3001 反馈)图经 NEXT_PUBLIC_MEDIA_BASE
(本地 = http://localhost:8770) App 后端 /media 加载改过 .env.local 后需重启 next dev,
App 后端(:8770)要在跑
"""
from __future__ import annotations
import argparse
import struct
import sys
import zlib
from datetime import UTC, datetime, timedelta
from pathlib import Path
from sqlalchemy import delete, select
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.feedback import Feedback
from app.models.user import User
from app.repositories.user import _gen_unique_username
# Windows GBK 控制台下也能正常打印中文/¥(避免 UnicodeEncodeError)
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
# 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。
# 刻意与 seed_mock_withdraws / seed_mock_price_reports 的号段错开,互不干扰。
MOCK_PHONES = [
"13255550001",
"13255550002",
"13255550003",
"13255550004",
"13255550005",
]
_FEEDBACK_DIR = Path(settings.MEDIA_ROOT) / "feedback"
_MOCK_IMG_GLOB = "mock_fb_*.png" # 本脚本生成的图前缀,清理时按此删
def _naive_utc_now() -> datetime:
"""与 func.now() 在 SQLite 的口径一致:naive UTC。反馈 created_at 走 server_default=func.now(),
这里显式造数据也用 naive UTC,和真实反馈行同源,前端展示口径一致"""
return datetime.now(UTC).replace(tzinfo=None)
def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
"""生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
def _chunk(typ: bytes, data: bytes) -> bytes:
body = typ + data
return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8bit/通道, color type 2 = RGB
row = b"\x00" + bytes(rgb) * width # 每行前缀 filter byte 0
idat = zlib.compress(row * height, 9)
return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
"""写一张 mock 截图到 media/feedback/,返回其相对 URL(/media/feedback/<name>)。"""
_FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
(_FEEDBACK_DIR / name).write_bytes(_solid_png(320, 320, rgb))
return f"{settings.MEDIA_URL_PREFIX}/feedback/{name}"
def clean(db) -> int:
uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
if uids:
db.execute(delete(Feedback).where(Feedback.user_id.in_(uids)))
db.execute(delete(User).where(User.id.in_(uids)))
db.commit()
# 删 mock 截图文件
if _FEEDBACK_DIR.exists():
for f in _FEEDBACK_DIR.glob(_MOCK_IMG_GLOB):
f.unlink(missing_ok=True)
return len(uids)
def seed(db) -> list[Feedback]:
now = _naive_utc_now()
def ago(**kw) -> datetime:
return now - timedelta(**kw)
# 1) 建 5 个 mock 用户(U5 昵称留空测 "-" 展示)
users_spec = [
("13255550001", "反馈小达人"),
("13255550002", "比价挑刺王"),
("13255550003", "热心用户阿明"),
("13255550004", "老用户张姐"),
("13255550005", None),
]
users: dict[str, User] = {}
for phone, nickname in users_spec:
u = User(
phone=phone,
username=_gen_unique_username(db), # 列 NOT NULL + unique,仿真实注册生成
nickname=nickname,
register_channel="sms",
status="active",
created_at=ago(days=15),
last_login_at=ago(hours=1),
)
db.add(u)
users[phone] = u
db.flush() # 拿 user.id
# 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
palette = [
(24, 144, 255), # 蓝
(82, 196, 26), # 绿
(250, 173, 20), # 橙
(245, 34, 45), # 红
]
imgs = [_write_mock_image(f"mock_fb_{i}.png", palette[i]) for i in range(len(palette))]
# 3) 造反馈记录
def fb(
phone: str,
content: str,
*,
source: str = "profile",
scene: str | None = None,
images: list[str] | None = None,
contact: str = "",
status: str = "pending",
reject_reason: str | None = None,
reward_coins: int | None = None,
review_note: str | None = None,
admin_reply: str | None = None,
app_version: str | None = None,
device_model: str | None = None,
rom_name: str | None = None,
android_version: str | None = None,
created: datetime,
) -> Feedback:
return Feedback(
user_id=users[phone].id,
content=content,
contact=contact, # 列 NOT NULL:新端存空串,历史数据有值
source=source,
scene=scene,
images=images,
status=status,
reject_reason=reject_reason,
reward_coins=reward_coins,
review_note=review_note,
admin_reply=admin_reply,
app_version=app_version,
device_model=device_model,
rom_name=rom_name,
android_version=android_version,
reviewed_at=(created + timedelta(hours=2)) if status != "pending" else None,
created_at=created,
)
feedbacks = [
# ===== 待审核 pending(默认 tab,重点铺量)=====
# 普通反馈 · 新端(带环境快照)· 无图
fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
source="profile",
app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
created=ago(minutes=6)),
# 比价反馈 · scene=优惠不对 · 新端 · 2 图
fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(minutes=22)),
# 比价反馈 · scene=找错商品 · 新端 · 无图
fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
source="comparison", scene="找错商品",
app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
created=ago(hours=1)),
# 普通反馈 · 新端 · 1 图(表扬 + 小问题)
fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
source="profile", images=[imgs[2]],
app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
created=ago(hours=3)),
# 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
source="comparison", scene="比价太慢", contact="微信 zhangjie_66",
created=ago(days=1, hours=2)),
# 普通反馈 · 历史数据(env 全 NULL、contact 有值)· 无昵称用户
fb("13255550005", "希望能增加支付宝提现,微信零钱用不太习惯。",
source="profile", contact="QQ 100200300",
created=ago(days=1, hours=8)),
# ===== 已采纳 adopted(发金币 + 回复)=====
fb("13255550003", "建议在比价结果页加个「一键复制口令」,分享给家人更方便。",
source="profile", images=[imgs[3]],
status="adopted", reward_coins=2000,
review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
created=ago(days=2)),
# ===== 未采纳 rejected(带原因 + 回复)=====
fb("13255550002", "你们算的价格不准,我看到的更便宜。",
source="comparison", scene="价格不准",
status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
created=ago(days=3)),
]
db.add_all(feedbacks)
db.commit()
for f in feedbacks:
db.refresh(f)
return feedbacks
def main() -> None:
parser = argparse.ArgumentParser(description="造用户反馈 mock 数据(运营后台反馈工单页联调用)")
parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
args = parser.parse_args()
db = SessionLocal()
try:
removed = clean(db)
if removed:
print(f"🧹 已清理旧 mock:{removed} 个用户及其反馈 + mock 截图")
if args.clean_only:
print("✅ 仅清理,已完成。")
return
feedbacks = seed(db)
status_label = {"pending": "待审核", "adopted": "已采纳", "rejected": "未采纳"}
source_label = {"profile": "普通反馈", "comparison": "比价反馈"}
by_status: dict[str, list[Feedback]] = {}
for f in feedbacks:
by_status.setdefault(f.status, []).append(f)
print(f"\n✅ 已生成 {len(feedbacks)} 条反馈(截图落 {_FEEDBACK_DIR}),分布:")
for st in ("pending", "adopted", "rejected"):
lst = by_status.get(st, [])
print(f" {status_label[st]:<4} {len(lst)}")
print("\n 明细(#id | 状态 | 类型/场景 | 图 | 内容):")
uid2phone = dict(
db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()
)
for f in feedbacks:
src = source_label.get(f.source, f.source)
scene = f"·{f.scene}" if f.scene else ""
nimg = len(f.images or [])
snippet = f.content[:20] + ("" if len(f.content) > 20 else "")
print(
f" #{f.id} [{status_label.get(f.status, f.status)}] "
f"{src}{scene} {nimg}{uid2phone.get(f.user_id, '?')} {snippet}"
)
print(
"\n👉 打开 http://localhost:3001 → 反馈 查看(默认「待审核」tab)。"
"\n 图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
" ② App 后端(:8770)是否在跑(它托管 /media)。"
)
finally:
db.close()
if __name__ == "__main__":
main()
+1 -3
View File
@@ -30,7 +30,6 @@ 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 控制台输出中文/¥
@@ -85,11 +84,10 @@ 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, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True) zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
): ):
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",
+2 -4
View File
@@ -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 UTC, datetime, timedelta from datetime import datetime, timedelta, timezone
from sqlalchemy import delete, select from sqlalchemy import delete, select
@@ -31,7 +31,6 @@ 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"):
@@ -52,7 +51,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(UTC).replace(tzinfo=None) return datetime.now(timezone.utc).replace(tzinfo=None)
def _mock_transfer_no() -> str: def _mock_transfer_no() -> str:
@@ -97,7 +96,6 @@ 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 -49
View File
@@ -468,13 +468,13 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
d = "2021-06-18" d = "2021-06-18"
included = { included = {
"signin": 100, "signin": 100,
"signin_boost": 200,
"task_enable_notification": 300, "task_enable_notification": 300,
"task_other": 400, "task_other": 400,
"price_report_reward": 500, "price_report_reward": 500,
"feedback_reward": 600, "feedback_reward": 600,
} }
excluded = { excluded = {
"signin_boost": 200,
"feed_ad_reward_coupon": 700, "feed_ad_reward_coupon": 700,
"feed_ad_reward_comparison": 800, "feed_ad_reward_comparison": 800,
"feed_ad_reward": 900, "feed_ad_reward": 900,
@@ -514,51 +514,3 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
coins = response.json()["period"]["coins"] coins = response.json()["period"]["coins"]
assert coins["regular_task_coin_total"] == sum(included.values()) assert coins["regular_task_coin_total"] == sum(included.values())
assert coins["task_coin_total"] == 700 assert coins["task_coin_total"] == 700
assert coins["reward_video_coin_total"] == 1200
def test_period_signin_boost_moves_to_reward_video_without_double_count(
admin_client: TestClient, admin_token: str
) -> None:
"""历史签到膨胀归看视频桶,不再进常规任务;本期发放总额不变且不重复计算。"""
from datetime import datetime
from app.models.wallet import CoinTransaction
d = "2021-06-19"
rows = [
("signin_boost", 200),
("reward_video", 100),
("signin", 50),
]
db = SessionLocal()
try:
uid = user_repo.upsert_user_for_login(
db, phone="13800008805", register_channel="sms"
).id
balance = 0
for index, (biz_type, amount) in enumerate(rows, start=1):
balance += amount
db.add(CoinTransaction(
user_id=uid,
amount=amount,
balance_after=balance,
biz_type=biz_type,
ref_id=f"signin-boost-route-{index}",
created_at=datetime(2021, 6, 19, 12, 0, index),
))
db.commit()
finally:
db.close()
response = admin_client.get(
"/admin/api/stats/overview",
params={"date_from": d, "date_to": d},
headers=_auth(admin_token),
)
assert response.status_code == 200, response.text
coins = response.json()["period"]["coins"]
assert coins["granted_total"] == 350
assert coins["reward_video_coin_total"] == 300
assert coins["regular_task_coin_total"] == 50
assert coins["signin_boost_coin_total"] == 200
-115
View File
@@ -14,7 +14,6 @@ from app.core.security import hash_password
from app.db.session import SessionLocal from app.db.session import SessionLocal
from app.models.admin import AdminAuditLog from app.models.admin import AdminAuditLog
from app.models.feedback import Feedback from app.models.feedback import Feedback
from app.models.price_report import PriceReport
from app.models.user import User from app.models.user import User
from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder
from app.repositories import user as user_repo from app.repositories import user as user_repo
@@ -83,26 +82,6 @@ def _seed_feedback(phone: str) -> int:
db.close() db.close()
def _seed_price_report(phone: str) -> int:
uid = _seed_user(phone)
db = SessionLocal()
try:
report = PriceReport(
user_id=uid,
store_name="测试门店",
reported_platform_id="eleme",
reported_platform_name="饿了么",
reported_price_cents=2990,
images=[],
status="pending",
)
db.add(report)
db.commit()
return report.id
finally:
db.close()
# ===== 调金币 ===== # ===== 调金币 =====
def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None: def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -418,100 +397,6 @@ def test_feedback_review_stores_admin_reply(
assert r.json()["admin_reply"] == "已收到,后续跟进" assert r.json()["admin_reply"] == "已收到,后续跟进"
def test_bulk_approve_feedbacks_returns_per_item_results(
admin_client: TestClient, operator_token: str
) -> None:
first_id = _seed_feedback("13900000031")
second_id = _seed_feedback("13900000032")
r = admin_client.post(
"/admin/api/feedbacks/bulk/approve",
json={"ids": [first_id, second_id, 999999], "reward_coins": 600, "note": "批量采纳"},
headers=_auth(operator_token),
)
assert r.status_code == 200, r.text
payload = r.json()
assert payload["total"] == 3 and payload["success"] == 2 and payload["failed"] == 1
assert payload["items"][-1] == {"id": 999999, "ok": False, "status": None, "error": "反馈不存在"}
db = SessionLocal()
try:
for feedback_id in (first_id, second_id):
feedback = db.get(Feedback, feedback_id)
assert feedback is not None and feedback.status == "adopted"
assert db.get(CoinAccount, feedback.user_id).coin_balance == 600
log = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "feedback.approve",
AdminAuditLog.target_id == str(feedback_id),
)
).scalar_one()
assert log.detail["bulk"] is True
finally:
db.close()
def test_bulk_approve_price_reports_returns_per_item_results(
admin_client: TestClient, operator_token: str
) -> None:
first_id = _seed_price_report("13900000041")
second_id = _seed_price_report("13900000042")
r = admin_client.post(
"/admin/api/price-reports/bulk/approve",
json={"ids": [first_id, second_id, 999999]},
headers=_auth(operator_token),
)
assert r.status_code == 200, r.text
payload = r.json()
assert payload["total"] == 3 and payload["success"] == 2 and payload["failed"] == 1
assert payload["items"][-1] == {"id": 999999, "ok": False, "status": None, "error": "上报记录不存在"}
db = SessionLocal()
try:
for report_id in (first_id, second_id):
report = db.get(PriceReport, report_id)
assert report is not None and report.status == "approved"
assert report.reward_coins == 1000
assert db.get(CoinAccount, report.user_id).coin_balance == 1000
log = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "price_report.approve",
AdminAuditLog.target_id == str(report_id),
)
).scalar_one()
assert log.detail["bulk"] is True
finally:
db.close()
def test_bulk_reject_review_requests_apply_shared_reason(
admin_client: TestClient, operator_token: str
) -> None:
feedback_id = _seed_feedback("13900000051")
report_id = _seed_price_report("13900000052")
feedback_response = admin_client.post(
"/admin/api/feedbacks/bulk/reject",
json={"ids": [feedback_id], "reason": "信息不足", "reply": "请补充完整截图"},
headers=_auth(operator_token),
)
report_response = admin_client.post(
"/admin/api/price-reports/bulk/reject",
json={"ids": [report_id], "reason": "截图无法核实"},
headers=_auth(operator_token),
)
assert feedback_response.status_code == 200, feedback_response.text
assert report_response.status_code == 200, report_response.text
assert feedback_response.json()["success"] == 1
assert report_response.json()["success"] == 1
db = SessionLocal()
try:
feedback = db.get(Feedback, feedback_id)
report = db.get(PriceReport, report_id)
assert feedback is not None and feedback.status == "rejected"
assert feedback.reject_reason == "信息不足" and feedback.admin_reply == "请补充完整截图"
assert report is not None and report.status == "rejected"
assert report.reject_reason == "截图无法核实"
finally:
db.close()
# ===== admin 账号管理(super_admin) ===== # ===== admin 账号管理(super_admin) =====
def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None: def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:
-120
View File
@@ -1,120 +0,0 @@
from __future__ import annotations
import time
from datetime import datetime, timedelta
from sqlalchemy import func, select
from app.core.rewards import CN_TZ
from app.core.security import decode_token
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord
def _login(client) -> tuple[str, int]:
phone = f"137{int(time.time() * 1000) % 100000000:08d}"
sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
assert sent.status_code == 200, sent.text
logged_in = client.post(
"/api/v1/auth/sms/login",
json={"phone": phone, "code": "123456"},
)
assert logged_in.status_code == 200, logged_in.text
token = logged_in.json()["access_token"]
return token, int(decode_token(token, expected_type="access")["sub"])
def _headers(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def test_compare_start_requires_login(client) -> None:
response = client.post(
"/api/v1/compare/start",
json={"trace_id": "quota-no-auth", "business_type": "food"},
)
assert response.status_code == 401
def test_compare_start_is_idempotent_by_trace_id(client) -> None:
token, user_id = _login(client)
payload = {
"trace_id": f"quota-idempotent-{user_id}",
"business_type": "ecom",
"device_id": "quota-device",
}
first = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
retry = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
assert first.status_code == 200, first.text
assert first.json() == {"limit": 100, "used": 1, "remaining": 99}
assert retry.status_code == 200, retry.text
assert retry.json() == first.json()
with SessionLocal() as db:
count = db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.trace_id == payload["trace_id"]
)
)
record = db.execute(
select(ComparisonRecord).where(
ComparisonRecord.trace_id == payload["trace_id"]
)
).scalar_one()
assert count == 1
assert record.user_id == user_id
assert record.status == "running"
assert record.business_type == "ecom"
assert record.device_id == "quota-device"
def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
token, user_id = _login(client)
now = datetime.now(CN_TZ).replace(tzinfo=None)
with SessionLocal() as db:
db.add_all(
[
ComparisonRecord(
user_id=user_id,
trace_id=f"quota-full-{user_id}-{index}",
status="failed",
created_at=now,
)
for index in range(99)
]
)
db.add(
ComparisonRecord(
user_id=user_id,
trace_id=f"quota-yesterday-{user_id}",
status="success",
created_at=now - timedelta(days=1),
)
)
db.commit()
final_allowed_trace = f"quota-final-allowed-{user_id}"
allowed = client.post(
"/api/v1/compare/start",
json={"trace_id": final_allowed_trace, "business_type": "food"},
headers=_headers(token),
)
assert allowed.status_code == 200, allowed.text
assert allowed.json() == {"limit": 100, "used": 100, "remaining": 0}
rejected_trace = f"quota-rejected-{user_id}"
response = client.post(
"/api/v1/compare/start",
json={"trace_id": rejected_trace, "business_type": "food"},
headers=_headers(token),
)
assert response.status_code == 429
assert response.json()["detail"] == "今日已比价超过100次,请明天再试"
with SessionLocal() as db:
assert db.scalar(
select(func.count(ComparisonRecord.id)).where(
ComparisonRecord.trace_id == rejected_trace
)
) == 0
-73
View File
@@ -1,73 +0,0 @@
"""SafeRotatingFileHandler:Windows 轮转不被外部句柄卡死(WinError 32)。
stdlib RotatingFileHandler rename 活动文件轮转;Windows 上只要有第二个句柄(admin
第二进程残留 --reload workerIDE 索引杀软)开着它, rename WinError 32轮转永久
卡死Safe 版在 Windows 改走 copytruncate(拷贝就地清空, 从不 rename 活动文件)
这些用例用 monkeypatch os.name 强制成 "nt", 使 copytruncate 分支在任何 OS CI 上都被
覆盖;在真实 Windows 上则天然命中
"""
from __future__ import annotations
import logging
from pathlib import Path
from app.core.logging import SafeRotatingFileHandler
def _emit(handler: logging.Handler, msg: str) -> None:
handler.emit(logging.LogRecord("t", logging.INFO, __file__, 0, msg, (), None))
def test_rollover_survives_second_open_handle(tmp_path: Path, monkeypatch) -> None:
"""第二个句柄开着活动文件时轮转:不抛异常, 且确实转了(生成 .1、活动文件就地清空)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
# maxBytes 放大, 避免 emit 期间自动轮转干扰;本用例手动触发 doRollover。
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=3, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for i in range(20):
_emit(handler, f"line-{i:03d}")
handler.flush()
before = log_file.stat().st_size
assert before > 0
# 正是 Windows 上 rename 失败的条件:另一个句柄开着活动文件。
with open(log_file, "a", encoding="utf-8"):
handler.doRollover() # 不应抛 PermissionError / WinError 32
backup = tmp_path / "app-server.log.1"
assert backup.exists()
assert backup.stat().st_size == before # 轮转前内容完整进了备份
assert log_file.stat().st_size == 0 # 活动文件就地清空(不是 rename)
# 句柄没被 rename 破坏, 仍能继续写。
_emit(handler, "after-rollover")
handler.flush()
assert log_file.stat().st_size > 0
finally:
handler.close()
def test_backups_shift_and_capped(tmp_path: Path, monkeypatch) -> None:
"""多次轮转:.1/.2 逐级腾挪, 超过 backupCount 的丢弃(不出现 .3)。"""
monkeypatch.setattr("app.core.logging.os.name", "nt")
log_file = tmp_path / "app-server.log"
handler = SafeRotatingFileHandler(
str(log_file), maxBytes=10**9, backupCount=2, encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(message)s"))
try:
for _ in range(4):
_emit(handler, "x" * 50)
handler.doRollover()
assert (tmp_path / "app-server.log.1").exists()
assert (tmp_path / "app-server.log.2").exists()
assert not (tmp_path / "app-server.log.3").exists()
finally:
handler.close()
+1 -60
View File
@@ -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, WechatTransferAuthorization from app.models.wallet import CoinAccount, WithdrawOrder
from app.repositories import wallet as crud_wallet from app.repositories import wallet as crud_wallet
@@ -426,62 +426,3 @@ 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