feat(compare): limit daily starts to 100

app/admin/routers/feedback.py

@@ -12,10 +12,6 @@ from app.admin.repositories import mutations, queries
 from app.admin.schemas.common import CursorPage, OkResponse
 from app.admin.schemas.feedback import (
     FeedbackApproveRequest,
-    FeedbackBulkApproveRequest,
-    FeedbackBulkItemResult,
-    FeedbackBulkRejectRequest,
-    FeedbackBulkResult,
     FeedbackOut,
     FeedbackRejectRequest,
     FeedbackSummary,
@@ -37,123 +33,6 @@ def _ensure_pending(fb: Feedback) -> None:
         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="反馈工单列表")
 def list_feedbacks(
     db: AdminDb,
@@ -194,50 +73,6 @@ def feedback_summary(db: AdminDb) -> FeedbackSummary:
     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="标记反馈已处理")
 def handle_feedback(
     feedback_id: int,
@@ -258,7 +93,53 @@ def approve_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> 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="拒绝采纳反馈")
@@ -269,4 +150,41 @@ def reject_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> 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

app/admin/routers/price_report.py

@@ -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.schemas.common import CursorPage, OkResponse
 from app.admin.schemas.price_report import (
-    PriceReportBulkItemResult,
-    PriceReportBulkRejectRequest,
-    PriceReportBulkRequest,
-    PriceReportBulkResult,
     PriceReportOut,
     PriceReportRejectRequest,
     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="上报更低价列表(筛选+分页)")
 def list_price_reports(
     db: AdminDb,
@@ -117,50 +60,6 @@ def price_report_summary(db: AdminDb) -> PriceReportSummary:
     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="通过上报(发固定金币)")
 def approve_price_report(
     report_id: int,
@@ -168,7 +67,27 @@ def approve_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> 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()
 
 
@@ -180,5 +99,16 @@ def reject_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> 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()

app/admin/schemas/feedback.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 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
 
@@ -55,54 +55,6 @@ class FeedbackRejectRequest(BaseModel):
     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):
     """审核台顶部各状态计数(pending 含历史 new 态)。"""
 

app/admin/schemas/price_report.py

@@ -56,42 +56,6 @@ class PriceReportRejectRequest(BaseModel):
         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):
     """审核台顶部各状态计数。"""
 

app/api/v1/compare_record.py

@@ -20,6 +20,8 @@ from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.schemas.compare_record import (
+    CompareStartReserveIn,
+    CompareStartReserveOut,
     CompareStatsOut,
     ComparisonRecordCreatedOut,
     ComparisonRecordDetailOut,
@@ -35,6 +37,41 @@ logger = logging.getLogger("shagua.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(
     "/record",
     response_model=ComparisonRecordCreatedOut,

app/repositories/comparison.py

@@ -5,7 +5,7 @@
 """
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, defer
@@ -14,8 +14,19 @@ from app.core.rewards import CN_TZ
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.comparison import ComparisonRecord
 from app.models.savings import SavingsRecord
+from app.models.user import User
 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:
     """元(float)→ 分(int)。None 透传。"""
@@ -243,6 +254,78 @@ def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | 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(
     db: Session,
     *,

app/schemas/compare_record.py

@@ -13,7 +13,6 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
-
 # ===== 上报请求 =====
 
 class ComparisonItemIn(BaseModel):
@@ -198,6 +197,20 @@ class ComparisonRecordCreatedOut(BaseModel):
     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):
     """「我的」页省钱战绩卡(比价口径)聚合。"""
 

tests/test_admin_write.py

@@ -14,7 +14,6 @@ from app.core.security import hash_password
 from app.db.session import SessionLocal
 from app.models.admin import AdminAuditLog
 from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
 from app.models.user import User
 from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder
 from app.repositories import user as user_repo
@@ -83,26 +82,6 @@ def _seed_feedback(phone: str) -> int:
         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:
@@ -418,100 +397,6 @@ def test_feedback_review_stores_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) =====
 
 def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:

tests/test_compare_daily_limit.py

@@ -0,0 +1,120 @@
+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