Compare commits

...

3 Commits

Author SHA1 Message Date
xianze fcfac7ad07 feat(report): 上报更低价后端——price_report 表 + 提交/列表接口
- 新表 price_report:关联比价记录,存原最低价快照(反查 best_*)/上报平台+价/截图URL/审核状态(pending/approved/rejected)/奖励字段
- POST /api/v1/report (multipart):反查比价记录填原最低价,校验上报价 < 原最低价(D),截图复用 core.media 落 /media/price_report/,提交一律 pending
- GET /api/v1/report/records?status= :按状态筛选 + 四态 counts,供「上报记录」页
- core.media 加 save_report_image;main 挂 report 路由;models 登记 PriceReport
- alembic 迁移仅建 price_report(autogenerate 误检的「删 order_record」已手动剔除——那是降级为审计的历史表)

奖励发放(通过→钱包+1000金币)是人工审核后台动作,本期不含。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-05 10:22:21 +08:00
marco b50495bebe feat: 短信接入极光真实发送 + 新增运营 admin 后台子应用
短信(SMS_MOCK 切 mock/real):
- integrations/sms.py 重写: real 模式走极光短信 REST /v1/messages 自定义验证码(本服务 secrets
  生成 6 位码 + 进程内存 + 本地校验一次性/防爆破), 鉴权复用极光一键登录 JG_APP_KEY/MASTER_SECRET
  (同一极光应用, 上线只需 SMS_MOCK=false); mock 仍"任意6位通过"不动其余测试
- 防刷四层: 单号冷却 + 单号每日上限 + 单IP rate_limit(/sms/send 10/min、/sms/login 20/min)
  + 单码失败次数作废; SmsError 带 status_code 映射 429/503/400
- config 增 SMS_SEND_ENDPOINT/SIGN_ID/TEMPLATE_ID/CODE_LENGTH/DAILY_LIMIT/MAX_VERIFY_ATTEMPTS;
  test_auth 加 real 模式单测; sms.md/后端技术实现/待办账本同步

admin 后台(app/admin/ 独立子应用, uvicorn app.admin.main:admin_app :8771):
- 复用主仓 models/repositories/integrations + 同库, 鉴权完全隔离(ADMIN_JWT_SECRET≠JWT_SECRET_KEY
  + payload typ=admin + bcrypt 密码 + 可选 IP 白名单); 主 app 不 import 本包, admin 崩不影响主进程
- 路由: 登录 / 账号管理(RBAC: super_admin·finance·operator) / 用户列表+360详情+封禁+手动调币 /
  钱包流水 / 提现重试对账 / 反馈工单 / 数据大盘; 全写操作落 admin_audit_log(涉钱与业务写同事务)
- 涉钱逻辑(调微信/退款/对账)复用 app.repositories.wallet 不重写
- 新增 models/admin.py(AdminUser/AdminAuditLog) + admin_tables 迁移 + create_admin.py +
  deploy/shaguabijia-admin.service; 依赖加 bcrypt

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-04 03:02:41 +08:00
ouzhou 90c0d65a16 chore(alembic): merge comparison_milestone 和 savings_report_fields 两个 head (#11)
Co-authored-by: OuYingJun1024 <1034284404@qq.com>
Reviewed-on: #11
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>
2026-06-03 09:48:28 +08:00
67 changed files with 3087 additions and 104 deletions
@@ -0,0 +1,65 @@
"""新增 price_report 上报更低价表
Revision ID: 9258bddde4ea
Revises: ad60a1b2c3d4
Create Date: 2026-06-05 10:15:51.508598
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '9258bddde4ea'
down_revision: Union[str, Sequence[str], None] = 'ad60a1b2c3d4'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('price_report',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('comparison_record_id', sa.Integer(), nullable=True),
sa.Column('store_name', sa.String(length=128), nullable=True),
sa.Column('dish_summary', sa.String(length=256), nullable=True),
sa.Column('original_platform_id', sa.String(length=32), nullable=True),
sa.Column('original_platform_name', sa.String(length=32), nullable=True),
sa.Column('original_price_cents', sa.Integer(), nullable=True),
sa.Column('reported_platform_id', sa.String(length=32), nullable=False),
sa.Column('reported_platform_name', sa.String(length=32), nullable=False),
sa.Column('reported_price_cents', sa.Integer(), nullable=False),
sa.Column('images', sa.JSON(), nullable=False),
sa.Column('status', sa.String(length=16), nullable=False),
sa.Column('reject_reason', sa.String(length=256), nullable=True),
sa.Column('reward_coins', sa.Integer(), nullable=True),
sa.Column('reviewed_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['comparison_record_id'], ['comparison_record.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_price_report_comparison_record_id'), ['comparison_record_id'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_created_at'), ['created_at'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_status'), ['status'], unique=False)
batch_op.create_index(batch_op.f('ix_price_report_user_id'), ['user_id'], unique=False)
# 注:autogenerate 另检测到「删除 order_record 表」——那是已降级为审计的历史表
# (model 已移除、库表保留),与本次无关,手动剔除,避免误删他人审计数据。
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# (对称:upgrade 未删 order_record,downgrade 也不重建它)
with op.batch_alter_table('price_report', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_price_report_user_id'))
batch_op.drop_index(batch_op.f('ix_price_report_status'))
batch_op.drop_index(batch_op.f('ix_price_report_created_at'))
batch_op.drop_index(batch_op.f('ix_price_report_comparison_record_id'))
op.drop_table('price_report')
# ### end Alembic commands ###
+65
View File
@@ -0,0 +1,65 @@
"""admin_user + admin_audit_log tables (运营 admin 后台:账号 + 操作审计)
Revision ID: ad60a1b2c3d4
Revises: 8ac524a8ea02
Create Date: 2026-06-03 10:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ad60a1b2c3d4'
down_revision: Union[str, Sequence[str], None] = '8ac524a8ea02'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'admin_user',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('username', sa.String(length=64), nullable=False),
sa.Column('password_hash', sa.String(length=255), nullable=False),
sa.Column('role', sa.String(length=20), nullable=False, server_default='operator'),
sa.Column('status', sa.String(length=20), nullable=False, server_default='active'),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.Column('last_login_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_user_username'), ['username'], unique=True)
op.create_table(
'admin_audit_log',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('admin_id', sa.Integer(), nullable=False),
sa.Column('admin_username', sa.String(length=64), nullable=False),
sa.Column('action', sa.String(length=64), nullable=False),
sa.Column('target_type', sa.String(length=32), nullable=False),
sa.Column('target_id', sa.String(length=64), nullable=True),
sa.Column('detail', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
sa.Column('ip', sa.String(length=64), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['admin_id'], ['admin_user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_admin_audit_log_admin_id'), ['admin_id'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_action'), ['action'], unique=False)
batch_op.create_index(batch_op.f('ix_admin_audit_log_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('admin_audit_log', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_audit_log_created_at'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_action'))
batch_op.drop_index(batch_op.f('ix_admin_audit_log_admin_id'))
op.drop_table('admin_audit_log')
with op.batch_alter_table('admin_user', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_admin_user_username'))
op.drop_table('admin_user')
+6
View File
@@ -0,0 +1,6 @@
"""运营 Admin 后台子应用。
独立 FastAPI app(app.admin.main:admin_app),独立进程/端口运行,复用 App 的
models/repositories/integrations + 同一个 DB,但鉴权完全隔离(独立 JWT secret)。
现有 app.main:app 不 import 本包,admin 崩溃不影响 App 主进程。
"""
+38
View File
@@ -0,0 +1,38 @@
"""审计写入门面。
每个 admin 写操作调一次 write_audit,把"谁(admin)在哪个 IP 对什么(target)做了什么
(action)+ 前后值(detail)"落进 admin_audit_log。
⚠️ 涉钱/涉状态的写操作:传 commit=False,和业务写操作放同一事务一起 commit,
保证"改了就有痕、有痕就改了"原子(见 plan 风险点 1)。轻量操作可 commit=True。
"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.admin.repositories import audit_log as audit_repo
from app.models.admin import AdminUser
def write_audit(
db: Session,
admin: AdminUser,
*,
action: str,
target_type: str,
target_id: str | int | None = None,
detail: dict | None = None,
ip: str | None = None,
commit: bool = True,
) -> None:
audit_repo.add_audit_log(
db,
admin_id=admin.id,
admin_username=admin.username,
action=action,
target_type=target_type,
target_id=str(target_id) if target_id is not None else None,
detail=detail,
ip=ip,
commit=commit,
)
+84
View File
@@ -0,0 +1,84 @@
"""Admin API 共享依赖:DB session、当前 admin、角色守卫、客户端 IP。
仿 app/api/deps.py 的 get_current_user,但验的是独立的 admin JWT、查的是 admin_user 表。
"""
from __future__ import annotations
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from app.admin.repositories import admin_user as admin_repo
from app.admin.security import AdminTokenError, decode_admin_token
from app.db.session import get_db
from app.models.admin import AdminUser
_bearer = HTTPBearer(auto_error=False, scheme_name="AdminBearer")
def get_current_admin(
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
db: Annotated[Session, Depends(get_db)],
) -> AdminUser:
"""从 Authorization: Bearer <admin_token> 解出当前管理员。失败统一 401。"""
if credentials is None or credentials.scheme.lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="missing bearer token",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = decode_admin_token(credentials.credentials)
except AdminTokenError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=str(e),
headers={"WWW-Authenticate": "Bearer"},
) from e
admin = admin_repo.get_by_id(db, int(payload["sub"]))
if admin is None or admin.status != "active":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="admin not found or disabled",
headers={"WWW-Authenticate": "Bearer"},
)
return admin
CurrentAdmin = Annotated[AdminUser, Depends(get_current_admin)]
AdminDb = Annotated[Session, Depends(get_db)]
def require_role(*roles: str):
"""角色守卫依赖工厂。super_admin 恒通过(全权)。
用法:在路由签名加 `_: Annotated[AdminUser, Depends(require_role("finance"))]`,
或 `dependencies=[Depends(require_role("finance"))]`(不需要拿 admin 时)。
"""
allowed = set(roles)
def _checker(admin: CurrentAdmin) -> AdminUser:
if admin.role != "super_admin" and admin.role not in allowed:
need = sorted(allowed | {"super_admin"})
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"role '{admin.role}' not allowed (need one of {need})",
)
return admin
return _checker
def get_client_ip(request: Request) -> str:
"""取客户端 IP(审计日志用)。生产经 nginx 反代,优先 X-Forwarded-For 第一段;否则直连 IP。
⚠️ XFF 可被客户端伪造。nginx 必须用 `proxy_set_header X-Forwarded-For $remote_addr`
覆盖客户端传入值(见 M4 部署),否则审计里的 IP 可被伪造。审计 IP 仅作记录、不参与鉴权。
"""
xff = request.headers.get("x-forwarded-for")
if xff:
return xff.split(",")[0].strip()
return request.client.host if request.client else ""
+80
View File
@@ -0,0 +1,80 @@
"""Admin 后台 FastAPI app(独立进程)。
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
现有 app.main:app 不 import 本模块,两进程互不影响。
"""
from __future__ import annotations
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.admin.routers.admins import router as admins_router
from app.admin.routers.audit import router as audit_router
from app.admin.routers.auth import router as auth_router
from app.admin.routers.dashboard import router as dashboard_router
from app.admin.routers.feedback import router as feedback_router
from app.admin.routers.users import router as users_router
from app.admin.routers.wallet import router as wallet_router
from app.admin.routers.withdraw import router as withdraw_router
from app.core.config import settings
from app.core.logging import setup_logging
setup_logging(debug=settings.APP_DEBUG)
logger = logging.getLogger("shagua.admin")
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
logger.info(
"admin app started env=%s db=%s",
settings.APP_ENV,
settings.DATABASE_URL.split("://", 1)[0],
)
yield
logger.info("admin app shutting down")
admin_app = FastAPI(
title=f"{settings.APP_NAME} · Admin",
version="0.1.0",
docs_url="/admin/docs" if not settings.is_prod else None,
redoc_url=None,
lifespan=lifespan,
)
# admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
_dev_origins = [
"http://localhost:3001",
"http://127.0.0.1:3001",
"http://localhost:3000",
"http://127.0.0.1:3000",
]
_origins = settings.cors_origins_list or ([] if settings.is_prod else _dev_origins)
if _origins:
admin_app.add_middleware(
CORSMiddleware,
allow_origins=_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@admin_app.get("/admin/api/health", tags=["meta"])
def health() -> dict[str, str]:
return {"status": "ok", "service": "admin"}
admin_app.include_router(auth_router)
admin_app.include_router(dashboard_router)
admin_app.include_router(users_router)
admin_app.include_router(wallet_router)
admin_app.include_router(withdraw_router)
admin_app.include_router(feedback_router)
admin_app.include_router(admins_router)
admin_app.include_router(audit_router)
+1
View File
@@ -0,0 +1 @@
"""admin 专用数据访问层(跨用户查询 + admin 账号 + 审计 + 大盘聚合)。"""
+64
View File
@@ -0,0 +1,64 @@
"""admin_user 表 CRUD。"""
from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.security import hash_password
from app.models.admin import AdminUser
def get_by_id(db: Session, admin_id: int) -> AdminUser | None:
return db.get(AdminUser, admin_id)
def get_by_username(db: Session, username: str) -> AdminUser | None:
stmt = select(AdminUser).where(AdminUser.username == username)
return db.execute(stmt).scalar_one_or_none()
def create_admin(
db: Session, *, username: str, password: str, role: str = "operator"
) -> AdminUser:
admin = AdminUser(
username=username,
password_hash=hash_password(password),
role=role,
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
def update_last_login(db: Session, admin: AdminUser) -> None:
admin.last_login_at = datetime.now(timezone.utc)
db.commit()
def list_admins(db: Session) -> list[AdminUser]:
stmt = select(AdminUser).order_by(AdminUser.id)
return list(db.execute(stmt).scalars().all())
def set_role(db: Session, admin: AdminUser, *, role: str) -> AdminUser:
admin.role = role
db.commit()
db.refresh(admin)
return admin
def set_status(db: Session, admin: AdminUser, *, status: str) -> AdminUser:
admin.status = status
db.commit()
db.refresh(admin)
return admin
def set_password(db: Session, admin: AdminUser, *, password: str) -> AdminUser:
admin.password_hash = hash_password(password)
db.commit()
db.refresh(admin)
return admin
+70
View File
@@ -0,0 +1,70 @@
"""admin_audit_log 写入 + 查询。
审计日志只增不改不删——任何写操作经 app.admin.audit.write_audit 落一条。
"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.admin import AdminAuditLog
def add_audit_log(
db: Session,
*,
admin_id: int,
admin_username: str,
action: str,
target_type: str,
target_id: str | None = None,
detail: dict | None = None,
ip: str | None = None,
commit: bool = True,
) -> AdminAuditLog:
"""插一条审计。commit=False 时只 flush,让调用方把审计和业务写操作放同一事务。"""
log = AdminAuditLog(
admin_id=admin_id,
admin_username=admin_username,
action=action,
target_type=target_type,
target_id=target_id,
detail=detail,
ip=ip,
)
db.add(log)
if commit:
db.commit()
db.refresh(log)
else:
db.flush()
return log
def list_audit_logs(
db: Session,
*,
action: str | None = None,
target_type: str | None = None,
admin_id: int | None = None,
limit: int = 50,
cursor: int | None = None,
) -> tuple[list[AdminAuditLog], int | None]:
"""游标分页(id 倒序),与现有 list_* 约定一致。返回 (rows, next_cursor)。"""
stmt = select(AdminAuditLog)
if action:
stmt = stmt.where(AdminAuditLog.action == action)
if target_type:
stmt = stmt.where(AdminAuditLog.target_type == target_type)
if admin_id is not None:
stmt = stmt.where(AdminAuditLog.admin_id == admin_id)
if cursor is not None:
stmt = stmt.where(AdminAuditLog.id < cursor)
stmt = stmt.order_by(AdminAuditLog.id.desc())
rows = list(db.execute(stmt.limit(limit + 1)).scalars().all())
has_more = len(rows) > limit
items = rows[:limit]
# next_cursor 必须是"本页返回的最后一条"的 id(下一页查 id < 它),不能用 rows[limit]——
# rows[limit] 是探测下一页用的第 limit+1 条,它既不在本页也不在下页 → 每页边界丢一条。
next_cursor = items[-1].id if has_more else None
return items, next_cursor
+36
View File
@@ -0,0 +1,36 @@
"""admin 写操作 repo(状态改写)。
涉钱的金币/提现复用 app.repositories.wallet(grant_coins / refresh_withdraw_status /
reconcile_pending_withdraws),不在这里重写——重写涉钱逻辑就是给自己埋雷。
set_user_status / update_feedback_status 支持 commit=False,让 router 把"业务写 + 审计写"
放进同一事务一起 commit(原子:改了就有审计、有审计就真改了,见 plan 风险点 1)。
"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.models.feedback import Feedback
from app.models.user import User
def set_user_status(db: Session, user: User, *, status: str, commit: bool = True) -> User:
user.status = status
if commit:
db.commit()
db.refresh(user)
else:
db.flush()
return user
def update_feedback_status(
db: Session, feedback: Feedback, *, status: str, commit: bool = True
) -> Feedback:
feedback.status = status
if commit:
db.commit()
db.refresh(feedback)
else:
db.flush()
return feedback
+153
View File
@@ -0,0 +1,153 @@
"""admin 跨用户查询(去掉现有 repo 的 user_id 强制过滤)+ 通用游标分页 helper + 用户概览。
现有 app/repositories/ 的 list_* 都强绑单个 user_id(C 端只看自己);admin 要看全量、按条件筛,
所以在这里另起一套。游标约定与现有一致:id 倒序,cursor=上页最后一条 id,返回 (items, next_cursor)。
"""
from __future__ import annotations
from sqlalchemy import Select, func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
def cursor_paginate(
db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
) -> tuple[list, int | None]:
"""通用游标分页(id 倒序)。stmt 不要预先带 order_by/limit。
多取 1 条探测有没有下一页;next_cursor 取本页最后一条的 id(下一页查 id < 它),
绝不用第 limit+1 条的 id——那条既不在本页也不在下页,会每页边界丢一条(见 audit_log 同款修复)。
"""
if cursor is not None:
stmt = stmt.where(id_col < cursor)
stmt = stmt.order_by(id_col.desc()).limit(limit + 1)
rows = list(db.execute(stmt).scalars().all())
has_more = len(rows) > limit
items = rows[:limit]
next_cursor = items[-1].id if has_more else None
return items, next_cursor
def list_users(
db: Session,
*,
phone: str | None = None,
register_channel: str | None = None,
status: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[User], int | None]:
stmt = select(User)
if phone:
stmt = stmt.where(User.phone.like(f"{phone}%")) # 前缀匹配
if register_channel:
stmt = stmt.where(User.register_channel == register_channel)
if status:
stmt = stmt.where(User.status == status)
return cursor_paginate(db, stmt, User.id, limit=limit, cursor=cursor)
def list_all_coin_transactions(
db: Session,
*,
user_id: int | None = None,
biz_type: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[CoinTransaction], int | None]:
stmt = select(CoinTransaction)
if user_id is not None:
stmt = stmt.where(CoinTransaction.user_id == user_id)
if biz_type:
stmt = stmt.where(CoinTransaction.biz_type == biz_type)
return cursor_paginate(db, stmt, CoinTransaction.id, limit=limit, cursor=cursor)
def list_all_cash_transactions(
db: Session,
*,
user_id: int | None = None,
biz_type: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[CashTransaction], int | None]:
stmt = select(CashTransaction)
if user_id is not None:
stmt = stmt.where(CashTransaction.user_id == user_id)
if biz_type:
stmt = stmt.where(CashTransaction.biz_type == biz_type)
return cursor_paginate(db, stmt, CashTransaction.id, limit=limit, cursor=cursor)
def list_all_withdraw_orders(
db: Session,
*,
user_id: int | None = None,
status: str | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[WithdrawOrder], int | None]:
stmt = select(WithdrawOrder)
if user_id is not None:
stmt = stmt.where(WithdrawOrder.user_id == user_id)
if status:
stmt = stmt.where(WithdrawOrder.status == status)
return cursor_paginate(db, stmt, WithdrawOrder.id, limit=limit, cursor=cursor)
def list_feedbacks(
db: Session,
*,
status: str | None = None,
user_id: int | None = None,
limit: int = 20,
cursor: int | None = None,
) -> tuple[list[Feedback], int | None]:
stmt = select(Feedback)
if status:
stmt = stmt.where(Feedback.status == status)
if user_id is not None:
stmt = stmt.where(Feedback.user_id == user_id)
return cursor_paginate(db, stmt, Feedback.id, limit=limit, cursor=cursor)
def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
"""按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。"""
return db.execute(
select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
).scalar_one_or_none()
def get_user_overview(db: Session, user_id: int) -> dict | None:
"""用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
user = db.get(User, user_id)
if user is None:
return None
acc = db.get(CoinAccount, user_id) # 可能为 None(从未发生过金币动作)
def _count(model, *conds) -> int:
return db.execute(select(func.count(model.id)).where(*conds)).scalar_one()
return {
"user": user,
"coin_balance": acc.coin_balance if acc else 0,
"cash_balance_cents": acc.cash_balance_cents if acc else 0,
"total_coin_earned": acc.total_coin_earned if acc else 0,
"comparison_total": _count(ComparisonRecord, ComparisonRecord.user_id == user_id),
"comparison_success": _count(
ComparisonRecord,
ComparisonRecord.user_id == user_id,
ComparisonRecord.status == "success",
),
"withdraw_total": _count(WithdrawOrder, WithdrawOrder.user_id == user_id),
"withdraw_success_cents": db.execute(
select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
WithdrawOrder.user_id == user_id, WithdrawOrder.status == "success"
)
).scalar_one(),
"feedback_total": _count(Feedback, Feedback.user_id == user_id),
}
+91
View File
@@ -0,0 +1,91 @@
"""admin 大盘聚合查询(全局 count/sum/DAU/成功率)。全部只读、不改任何数据。
⚠️ 性能:这些是全表 count/sum,P0 数据量小够用;用户量上来后热点字段(user.created_at /
user.last_login_at / comparison_record.status / withdraw_order.status)要加索引,或改增量统计表。
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder
_BEIJING = timezone(timedelta(hours=8))
def _beijing_today_start_utc() -> datetime:
"""北京时间今天 0 点对应的 UTC 时刻(DAU / 今日新增按北京时区切天)。"""
now_bj = datetime.now(_BEIJING)
start_bj = now_bj.replace(hour=0, minute=0, second=0, microsecond=0)
return start_bj.astimezone(timezone.utc)
def dashboard_overview(db: Session) -> dict:
today_start = _beijing_today_start_utc()
def _count(model, *conds) -> int:
stmt = select(func.count(model.id))
if conds:
stmt = stmt.where(*conds)
return db.execute(stmt).scalar_one()
def _sum(col, *conds) -> int:
stmt = select(func.coalesce(func.sum(col), 0))
if conds:
stmt = stmt.where(*conds)
return db.execute(stmt).scalar_one()
# ===== 用户 =====
by_status = dict(
db.execute(select(User.status, func.count(User.id)).group_by(User.status)).all()
)
# ===== 提现状态分布 =====
wd_by_status = dict(
db.execute(
select(WithdrawOrder.status, func.count(WithdrawOrder.id)).group_by(
WithdrawOrder.status
)
).all()
)
# ===== 比价 =====
comparison_total = _count(ComparisonRecord)
comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0
return {
"users": {
"total": _count(User),
"active": by_status.get("active", 0),
"disabled": by_status.get("disabled", 0),
"deleted": by_status.get("deleted", 0),
"new_today": _count(User, User.created_at >= today_start),
"dau": _count(User, User.last_login_at >= today_start),
},
"coins": {
# 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
"granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
},
"cash": {
"withdraw_success_cents": _sum(
WithdrawOrder.amount_cents, WithdrawOrder.status == "success"
),
"withdraw_pending_count": wd_by_status.get("pending", 0),
"withdraw_success_count": wd_by_status.get("success", 0),
"withdraw_failed_count": wd_by_status.get("failed", 0),
},
"comparison": {
"total": comparison_total,
"success": comparison_success,
"success_rate": success_rate,
},
"feedback": {"new": _count(Feedback, Feedback.status == "new")},
# CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
"cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
}
+1
View File
@@ -0,0 +1 @@
"""admin 路由(前缀统一 /admin/api)。"""
+69
View File
@@ -0,0 +1,69 @@
"""admin 账号管理(仅 super_admin):列表 / 创建 / 改角色启停重置密码。均写审计。"""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
from app.admin.repositories import admin_user as admin_repo
from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest
from app.admin.schemas.auth import AdminOut
from app.core.security import hash_password
router = APIRouter(
prefix="/admin/api/admins",
tags=["admin-accounts"],
dependencies=[Depends(require_role())], # require_role() 无参 = 仅 super_admin 通过
)
@router.get("", response_model=list[AdminOut], summary="管理员列表")
def list_admins(db: AdminDb) -> list[AdminOut]:
return [AdminOut.model_validate(a) for a in admin_repo.list_admins(db)]
@router.post("", response_model=AdminOut, summary="创建管理员")
def create_admin(
body: AdminCreateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> AdminOut:
if admin_repo.get_by_username(db, body.username) is not None:
raise HTTPException(status_code=409, detail="用户名已存在")
new = admin_repo.create_admin(
db, username=body.username, password=body.password, role=body.role
)
write_audit(
db, admin, action="admin.create", target_type="admin", target_id=new.id,
detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True,
)
return AdminOut.model_validate(new)
@router.patch("/{admin_id}", response_model=AdminOut, summary="改角色/启停/重置密码")
def update_admin(
admin_id: int, body: AdminUpdateRequest, request: Request, admin: CurrentAdmin, db: AdminDb
) -> AdminOut:
target = admin_repo.get_by_id(db, admin_id)
if target is None:
raise HTTPException(status_code=404, detail="管理员不存在")
if admin_id == admin.id and body.status == "disabled":
raise HTTPException(status_code=400, detail="不能禁用自己")
changes: dict = {}
if body.role is not None:
target.role = body.role
changes["role"] = body.role
if body.status is not None:
target.status = body.status
changes["status"] = body.status
if body.password is not None:
target.password_hash = hash_password(body.password)
changes["password"] = "reset"
if not changes:
raise HTTPException(status_code=400, detail="无任何变更字段")
db.commit()
db.refresh(target)
write_audit(
db, admin, action="admin.update", target_type="admin", target_id=admin_id,
detail=changes, ip=get_client_ip(request), commit=True,
)
return AdminOut.model_validate(target)
+34
View File
@@ -0,0 +1,34 @@
"""admin 操作审计日志查询(所有 admin 可看:谁在何时对什么做了什么)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import audit_log as audit_repo
from app.admin.schemas.admin import AdminAuditLogOut
from app.admin.schemas.common import CursorPage
router = APIRouter(
prefix="/admin/api/audit-logs",
tags=["admin-audit"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[AdminAuditLogOut], summary="审计日志(谁改了什么)")
def list_audit_logs(
db: AdminDb,
action: Annotated[str | None, Query()] = None,
target_type: Annotated[str | None, Query()] = None,
admin_id: Annotated[int | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 50,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[AdminAuditLogOut]:
items, next_cursor = audit_repo.list_audit_logs(
db, action=action, target_type=target_type, admin_id=admin_id, limit=limit, cursor=cursor,
)
return CursorPage(
items=[AdminAuditLogOut.model_validate(x) for x in items], next_cursor=next_cursor,
)
+48
View File
@@ -0,0 +1,48 @@
"""Admin 认证:账号密码登录 → admin JWT(独立 secret)。"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, HTTPException
from app.admin.deps import AdminDb, CurrentAdmin
from app.admin.repositories import admin_user as admin_repo
from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
from app.admin.security import create_admin_token
from app.core.ratelimit import rate_limit
from app.core.security import verify_password
logger = logging.getLogger("shagua.admin.auth")
router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
@router.post(
"/login",
response_model=AdminLoginResponse,
summary="管理员登录",
dependencies=[Depends(rate_limit(10, 60, "admin-login"))], # 同 IP 每分钟≤10 次,防爆破
)
def login(req: AdminLoginRequest, db: AdminDb) -> AdminLoginResponse:
admin = admin_repo.get_by_username(db, req.username)
# 用户名不存在 / 密码错统一 401 同文案(防账号枚举)。
# disabled 账号单独 403"账号已禁用":admin 是内部少数已知账号、无枚举价值,
# 明确提示比防枚举更有运维价值(与 App 端 auth.py 对 disabled 用户的 403 一致)。
if admin is None or not verify_password(req.password, admin.password_hash):
raise HTTPException(status_code=401, detail="用户名或密码错误")
if admin.status != "active":
raise HTTPException(status_code=403, detail="账号已禁用")
token, expires_in = create_admin_token(admin_id=admin.id, role=admin.role)
admin_repo.update_last_login(db, admin)
logger.info("admin login ok id=%d username=%s role=%s", admin.id, admin.username, admin.role)
return AdminLoginResponse(
access_token=token,
expires_in=expires_in,
admin=AdminOut.model_validate(admin),
)
@router.get("/me", response_model=AdminOut, summary="当前管理员")
def me(admin: CurrentAdmin) -> AdminOut:
return AdminOut.model_validate(admin)
+19
View File
@@ -0,0 +1,19 @@
"""admin 数据大盘(只读聚合)。"""
from __future__ import annotations
from fastapi import APIRouter, Depends
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import stats
from app.admin.schemas.dashboard import DashboardOverview
router = APIRouter(
prefix="/admin/api/stats",
tags=["admin-stats"],
dependencies=[Depends(get_current_admin)],
)
@router.get("/overview", response_model=DashboardOverview, summary="大盘核心指标")
def overview(db: AdminDb) -> DashboardOverview:
return DashboardOverview.model_validate(stats.dashboard_overview(db))
+56
View File
@@ -0,0 +1,56 @@
"""admin 反馈工单:列表(读)+ 标记已处理(写,带审计)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.feedback import FeedbackOut
from app.models.admin import AdminUser
from app.models.feedback import Feedback
router = APIRouter(
prefix="/admin/api/feedbacks",
tags=["admin-feedback"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[FeedbackOut], summary="反馈工单列表")
def list_feedbacks(
db: AdminDb,
status: Annotated[str | None, Query()] = None,
user_id: Annotated[int | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[FeedbackOut]:
items, next_cursor = queries.list_feedbacks(
db, status=status, user_id=user_id, limit=limit, cursor=cursor,
)
return CursorPage(
items=[FeedbackOut.model_validate(f) for f in items], next_cursor=next_cursor,
)
@router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
def handle_feedback(
feedback_id: int,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OkResponse:
fb = db.get(Feedback, feedback_id)
if fb is None:
raise HTTPException(status_code=404, detail="反馈不存在")
before = fb.status
mutations.update_feedback_status(db, fb, status="handled", commit=False)
write_audit(
db, admin, action="feedback.handle", target_type="feedback", target_id=feedback_id,
detail={"before": before, "after": "handled"}, ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
+111
View File
@@ -0,0 +1,111 @@
"""admin 用户管理:列表 + 360 详情(读)+ 封禁/解封 + 手动调金币(写,带审计)。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import mutations, queries
from app.admin.schemas.common import CursorPage, OkResponse
from app.admin.schemas.user import (
AdminUserListItem,
AdminUserOverview,
GrantCoinsRequest,
SetUserStatusRequest,
)
from app.models.admin import AdminUser
from app.repositories import user as user_repo
from app.repositories import wallet as wallet_repo
router = APIRouter(
prefix="/admin/api/users",
tags=["admin-users"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[AdminUserListItem], summary="用户列表(筛选+分页)")
def list_users(
db: AdminDb,
phone: Annotated[str | None, Query()] = None,
register_channel: Annotated[str | None, Query()] = None,
status: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[AdminUserListItem]:
items, next_cursor = queries.list_users(
db, phone=phone, register_channel=register_channel, status=status,
limit=limit, cursor=cursor,
)
return CursorPage(
items=[AdminUserListItem.model_validate(u) for u in items],
next_cursor=next_cursor,
)
@router.get("/{user_id}", response_model=AdminUserOverview, summary="用户 360 详情")
def get_user(user_id: int, db: AdminDb) -> AdminUserOverview:
overview = queries.get_user_overview(db, user_id)
if overview is None:
raise HTTPException(status_code=404, detail="用户不存在")
return AdminUserOverview.model_validate(overview)
@router.post("/{user_id}/status", response_model=OkResponse, summary="封禁/解封用户")
def set_user_status(
user_id: int,
body: SetUserStatusRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
) -> OkResponse:
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
if user.status == "deleted":
raise HTTPException(status_code=400, detail="已注销账号不可改状态")
before = user.status
# 业务写 + 审计写同一事务(commit=False),最后一起 commit:改了就有痕、有痕就真改了
mutations.set_user_status(db, user, status=body.status, commit=False)
write_audit(
db, admin, action="user.status.set", target_type="user", target_id=user_id,
detail={"before": before, "after": body.status}, ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
@router.post("/{user_id}/coins", response_model=OkResponse, summary="手动增减金币(带审计)")
def grant_user_coins(
user_id: int,
body: GrantCoinsRequest,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> OkResponse:
if body.amount == 0:
raise HTTPException(status_code=400, detail="amount 不能为 0")
user = user_repo.get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
# 负数扣减时不允许扣成负余额(运营误操作保护)
if body.amount < 0:
acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False)
if acc_now.coin_balance + body.amount < 0:
raise HTTPException(
status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})"
)
biz_type = "admin_grant" if body.amount > 0 else "admin_deduct"
# grant_coins 只 flush 不 commit;审计同 commit=False;最后一起 commit → 原子(改钱+留痕)
acc, _ = wallet_repo.grant_coins(
db, user_id, body.amount, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
)
write_audit(
db, admin, action="user.coins.grant", target_type="user", target_id=user_id,
detail={"amount": body.amount, "balance_after": acc.coin_balance, "reason": body.reason},
ip=get_client_ip(request), commit=False,
)
db.commit()
return OkResponse()
+49
View File
@@ -0,0 +1,49 @@
"""admin 钱包:金币流水 + 现金流水(跨用户,可按 user_id 过滤)。手动调金币见 M3。"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.admin.deps import AdminDb, get_current_admin
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import CashTxnOut, CoinTxnOut
router = APIRouter(
prefix="/admin/api/wallet",
tags=["admin-wallet"],
dependencies=[Depends(get_current_admin)],
)
@router.get("/coin-transactions", response_model=CursorPage[CoinTxnOut], summary="金币流水")
def coin_transactions(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
biz_type: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CoinTxnOut]:
items, next_cursor = queries.list_all_coin_transactions(
db, user_id=user_id, biz_type=biz_type, limit=limit, cursor=cursor,
)
return CursorPage(
items=[CoinTxnOut.model_validate(t) for t in items], next_cursor=next_cursor,
)
@router.get("/cash-transactions", response_model=CursorPage[CashTxnOut], summary="现金流水")
def cash_transactions(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
biz_type: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[CashTxnOut]:
items, next_cursor = queries.list_all_cash_transactions(
db, user_id=user_id, biz_type=biz_type, limit=limit, cursor=cursor,
)
return CursorPage(
items=[CashTxnOut.model_validate(t) for t in items], next_cursor=next_cursor,
)
+85
View File
@@ -0,0 +1,85 @@
"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
审计记录"谁触发的 + 结果",事务边界比"改金币"宽松是有意的(不能塞进 wallet 的自有事务)。
"""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.repositories import queries
from app.admin.schemas.common import CursorPage
from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut
from app.integrations import wxpay
from app.models.admin import AdminUser
from app.repositories import wallet as wallet_repo
router = APIRouter(
prefix="/admin/api/withdraws",
tags=["admin-withdraw"],
dependencies=[Depends(get_current_admin)],
)
@router.get("", response_model=CursorPage[WithdrawOrderOut], summary="提现单列表")
def list_withdraws(
db: AdminDb,
user_id: Annotated[int | None, Query()] = None,
status: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int | None, Query()] = None,
) -> CursorPage[WithdrawOrderOut]:
items, next_cursor = queries.list_all_withdraw_orders(
db, user_id=user_id, status=status, limit=limit, cursor=cursor,
)
return CursorPage(
items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
)
# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
def reconcile(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
older_than_minutes: Annotated[int, Query(ge=0)] = 15,
) -> ReconcileResult:
try:
result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
detail=result, ip=get_client_ip(request), commit=True,
)
return ReconcileResult(**result)
@router.post("/{out_bill_no}/refresh", response_model=WithdrawOrderOut, summary="单笔重试查单")
def refresh_withdraw(
out_bill_no: str,
request: Request,
admin: Annotated[AdminUser, Depends(require_role("finance"))],
db: AdminDb,
) -> WithdrawOrderOut:
order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
if order is None:
raise HTTPException(status_code=404, detail="提现单不存在")
try:
refreshed = wallet_repo.refresh_withdraw_status(
db, order.user_id, out_bill_no, cancel_if_unconfirmed=True,
)
except wxpay.WxPayNotConfiguredError as e:
raise HTTPException(status_code=503, detail="微信支付未配置") from e
write_audit(
db, admin, action="withdraw.refresh", target_type="withdraw", target_id=out_bill_no,
detail={"status": refreshed.status, "wechat_state": refreshed.wechat_state},
ip=get_client_ip(request), commit=True,
)
return WithdrawOrderOut.model_validate(refreshed)
+1
View File
@@ -0,0 +1 @@
"""admin 请求/响应 schemas(snake_case,与前端约定一致)。"""
+37
View File
@@ -0,0 +1,37 @@
"""admin 账号管理 + 审计日志 schemas。"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
_Role = Literal["super_admin", "finance", "operator"]
class AdminCreateRequest(BaseModel):
username: str = Field(..., min_length=3, max_length=64)
password: str = Field(..., min_length=8, max_length=72) # bcrypt ≤72 字节
role: _Role = "operator"
class AdminUpdateRequest(BaseModel):
"""改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。"""
role: _Role | None = None
status: Literal["active", "disabled"] | None = None
password: str | None = Field(None, min_length=8, max_length=72)
class AdminAuditLogOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
admin_id: int
admin_username: str
action: str
target_type: str
target_id: str | None = None
detail: dict | None = None
ip: str | None = None
created_at: datetime
+29
View File
@@ -0,0 +1,29 @@
"""Admin 认证 schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class AdminLoginRequest(BaseModel):
username: str = Field(..., min_length=1, max_length=64)
password: str = Field(..., min_length=1, max_length=128)
class AdminOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
username: str
role: str
status: str
created_at: datetime
last_login_at: datetime | None = None
class AdminLoginResponse(BaseModel):
access_token: str
token_type: str = "Bearer"
expires_in: int = Field(..., description="access_token 剩余秒数(过期重新登录)")
admin: AdminOut
+19
View File
@@ -0,0 +1,19 @@
"""通用 schema:游标分页响应 + 通用 ok 响应。"""
from __future__ import annotations
from typing import Generic, TypeVar
from pydantic import BaseModel
T = TypeVar("T")
class CursorPage(BaseModel, Generic[T]):
"""游标分页响应:items + 下一页游标(next_cursor=None 表示末页)。"""
items: list[T]
next_cursor: int | None = None
class OkResponse(BaseModel):
ok: bool = True
+48
View File
@@ -0,0 +1,48 @@
"""admin 大盘 schemas(对应 stats.dashboard_overview 的嵌套结构)。"""
from __future__ import annotations
from pydantic import BaseModel
class DashboardUsers(BaseModel):
total: int
active: int
disabled: int
deleted: int
new_today: int
dau: int
class DashboardCoins(BaseModel):
granted_total: int
class DashboardCash(BaseModel):
withdraw_success_cents: int
withdraw_pending_count: int
withdraw_success_count: int
withdraw_failed_count: int
class DashboardComparison(BaseModel):
total: int
success: int
success_rate: float
class DashboardFeedback(BaseModel):
new: int
class DashboardCps(BaseModel):
available: bool
note: str
class DashboardOverview(BaseModel):
users: DashboardUsers
coins: DashboardCoins
cash: DashboardCash
comparison: DashboardComparison
feedback: DashboardFeedback
cps: DashboardCps
+18
View File
@@ -0,0 +1,18 @@
"""admin 反馈工单 schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
class FeedbackOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
content: str
contact: str
images: list[str] | None = None
status: str
created_at: datetime
+47
View File
@@ -0,0 +1,47 @@
"""admin 用户管理 schemas。"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class AdminUserListItem(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
phone: str
nickname: str | None = None
register_channel: str
status: str
wechat_openid: str | None = None
created_at: datetime
last_login_at: datetime
class AdminUserOverview(BaseModel):
"""用户 360 概览:基础资料 + 钱包余额 + 各项 count(历史明细走各自分页接口)。"""
model_config = ConfigDict(from_attributes=True)
user: AdminUserListItem
coin_balance: int
cash_balance_cents: int
total_coin_earned: int
comparison_total: int
comparison_success: int
withdraw_total: int
withdraw_success_cents: int
feedback_total: int
class GrantCoinsRequest(BaseModel):
amount: int = Field(..., description="金币变动:正=增加,负=扣减(不可为 0)")
reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
class SetUserStatusRequest(BaseModel):
status: Literal["active", "disabled"] = Field(
..., description="active=解封 / disabled=封禁(注销 deleted 不走此接口)"
)
+52
View File
@@ -0,0 +1,52 @@
"""admin 钱包(金币/现金流水 + 提现单)schemas。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
class CoinTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount: int
balance_after: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class CashTxnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
amount_cents: int
balance_after_cents: int
biz_type: str
ref_id: str | None = None
remark: str | None = None
created_at: datetime
class WithdrawOrderOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
out_bill_no: str
amount_cents: int
status: str
wechat_state: str | None = None
transfer_bill_no: str | None = None
fail_reason: str | None = None
created_at: datetime
updated_at: datetime
class ReconcileResult(BaseModel):
checked: int
resolved: int
+55
View File
@@ -0,0 +1,55 @@
"""Admin JWT:与 App 用户 token 完全隔离。
用独立 secret(settings.ADMIN_JWT_SECRET JWT_SECRET_KEY)+ payload typ="admin",
双重保证 App 用户的 access_token 无法当 admin token (secret 不同直接验签失败)
admin refresh:过期(默认 12h)重新登录,简单
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any
import jwt
from app.core.config import settings
_ALGO = "HS256"
class AdminTokenError(Exception):
"""admin token 解析/校验失败,api 层捕获后返回 401。"""
def _now() -> datetime:
return datetime.now(timezone.utc)
def create_admin_token(*, admin_id: int, role: str) -> tuple[str, int]:
"""签发 admin access token,返回 (token, expires_in_seconds)。"""
now = _now()
expire = now + timedelta(minutes=settings.ADMIN_JWT_EXPIRE_MINUTES)
payload: dict[str, Any] = {
"sub": str(admin_id),
"role": role,
"typ": "admin",
"iat": int(now.timestamp()),
"exp": int(expire.timestamp()),
}
token = jwt.encode(payload, settings.ADMIN_JWT_SECRET, algorithm=_ALGO)
return token, int((expire - now).total_seconds())
def decode_admin_token(token: str) -> dict[str, Any]:
"""解析校验 admin token。签名错/过期/typ 非 admin → AdminTokenError。"""
try:
payload = jwt.decode(token, settings.ADMIN_JWT_SECRET, algorithms=[_ALGO])
except jwt.ExpiredSignatureError as e:
raise AdminTokenError("token expired") from e
except jwt.InvalidTokenError as e:
raise AdminTokenError(f"invalid token: {e}") from e
if payload.get("typ") != "admin":
raise AdminTokenError("not an admin token")
if "sub" not in payload:
raise AdminTokenError("token missing sub")
return payload
+15 -4
View File
@@ -12,9 +12,10 @@ from __future__ import annotations
import logging
from fastapi import APIRouter, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status
from app.api.deps import CurrentUser, DbSession
from app.core.ratelimit import rate_limit
from app.core.security import TokenError, decode_token, issue_token_pair
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
@@ -71,19 +72,29 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
# ===================== 短信登录 =====================
@router.post("/sms/send", response_model=SmsSendResponse, summary="发送短信验证码 (mock)")
@router.post(
"/sms/send",
response_model=SmsSendResponse,
summary="发送短信验证码",
dependencies=[Depends(rate_limit(10, 60, "sms-send"))], # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
)
def sms_send(req: SmsSendRequest) -> SmsSendResponse:
try:
cooldown = send_code(req.phone)
except SmsError as e:
raise HTTPException(status_code=429, detail=str(e)) from e
raise HTTPException(status_code=e.status_code, detail=str(e)) from e
from app.core.config import settings # 局部 import 避免循环
return SmsSendResponse(sent=True, mock=settings.SMS_MOCK, cooldown_sec=cooldown)
@router.post("/sms/login", response_model=TokenWithUser, summary="手机号+验证码登录")
@router.post(
"/sms/login",
response_model=TokenWithUser,
summary="手机号+验证码登录",
dependencies=[Depends(rate_limit(20, 60, "sms-login"))], # 防撞库爆破(另有单码失败次数上限)
)
def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
if not verify_code(req.phone, req.code):
raise HTTPException(status_code=400, detail="invalid sms code")
+145
View File
@@ -0,0 +1,145 @@
"""上报更低价 endpoint。
路由前缀 `/api/v1/report`, Bearer 鉴权(上报绑登录用户)
POST / 提交上报(multipart:comparison_record_id / reported_platform_id /
reported_price() + images 1~4 )
GET /records 上报记录列表(?status= pending/approved/rejected 可选筛选)
要点:
- 截图复用 [app.core.media] 落盘到 /media/price_report/
- 原最低价由 comparison_record_id 反查比价记录 best_*(不信任客户端传的快照)
- 校验(D):上报价必须 < 原最低价,否则 400
- 提交一律 status=pending;通过发奖励是人工审核后台动作,不在此
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
from app.api.deps import CurrentUser, DbSession
from app.core import media
from app.models.comparison import ComparisonRecord
from app.repositories import report as report_repo
from app.schemas.report import (
ReportRecordCounts,
ReportRecordOut,
ReportRecordsOut,
ReportSubmitOut,
)
logger = logging.getLogger("shagua.report")
router = APIRouter(prefix="/api/v1/report", tags=["report"])
_MAX_IMAGES = 4
_VALID_STATUS = ("pending", "approved", "rejected")
# 上报平台(canonical id → 展示名),与原型 fb-platform-chip data-pf 一致
_PLATFORM_NAMES = {
"meituan-waimai": "美团外卖",
"jd-waimai": "京东外卖",
"taobao-shanguang": "淘宝闪购",
}
def _dish_summary(items: list | None) -> str | None:
"""把比价记录 items[{name,qty,...}] 拼成菜品摘要文案。"""
if not items:
return None
names = [str(it.get("name", "")).strip() for it in items if isinstance(it, dict)]
names = [n for n in names if n]
return "".join(names) if names else None
@router.post("", response_model=ReportSubmitOut, summary="提交上报更低价")
async def submit_report(
user: CurrentUser,
db: DbSession,
comparison_record_id: int = Form(...),
reported_platform_id: str = Form(...),
reported_price: str = Form(..., description="用户填的更低价(元)"),
images: list[UploadFile] = File(default=[]),
) -> ReportSubmitOut:
# 平台
reported_platform_id = reported_platform_id.strip()
platform_name = _PLATFORM_NAMES.get(reported_platform_id)
if platform_name is None:
raise HTTPException(status_code=400, detail="不支持的上报平台")
# 价格(元 → 分)
try:
price_yuan = float(reported_price)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="价格格式不正确") from None
if price_yuan <= 0:
raise HTTPException(status_code=400, detail="价格必须大于 0")
reported_price_cents = round(price_yuan * 100)
# 反查比价记录(必须属于本人)→ 取原最低价快照
rec = db.get(ComparisonRecord, comparison_record_id)
if rec is None or rec.user_id != user.id:
raise HTTPException(status_code=404, detail="比价记录不存在")
original_price_cents = rec.best_price_cents
# D:上报价必须低于原最低价
if original_price_cents is not None and reported_price_cents >= original_price_cents:
raise HTTPException(
status_code=400,
detail=f"上报价需低于原最低价 ¥{original_price_cents / 100:.2f}",
)
# 截图(至少 1 张,最多 4 张)
files = [f for f in (images or []) if f is not None and f.filename]
if not files:
raise HTTPException(status_code=400, detail="请至少上传一张截图证明")
if len(files) > _MAX_IMAGES:
raise HTTPException(status_code=400, detail=f"最多上传 {_MAX_IMAGES} 张图片")
urls: list[str] = []
for f in files:
data = await f.read()
try:
urls.append(media.save_report_image(user.id, data))
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
rep = report_repo.create_report(
db,
user_id=user.id,
comparison_record_id=comparison_record_id,
store_name=rec.store_name,
dish_summary=_dish_summary(rec.items),
original_platform_id=rec.best_platform_id,
original_platform_name=rec.best_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=platform_name,
reported_price_cents=reported_price_cents,
images=urls,
)
logger.info("price_report id=%d user_id=%d images=%d", rep.id, user.id, len(urls))
return ReportSubmitOut.model_validate(rep)
@router.get("/records", response_model=ReportRecordsOut, summary="上报记录列表")
def list_report_records(
user: CurrentUser,
db: DbSession,
status: str | None = None,
) -> ReportRecordsOut:
status = (status or "").strip() or None
if status and status not in _VALID_STATUS:
raise HTTPException(status_code=400, detail="无效的状态筛选")
rows = report_repo.list_reports(db, user.id, status)
# counts 始终基于全量(不受 status 筛选影响),供前端 chip 计数
all_rows = rows if status is None else report_repo.list_reports(db, user.id, None)
counts = ReportRecordCounts(
all=len(all_rows),
pending=sum(1 for r in all_rows if r.status == "pending"),
approved=sum(1 for r in all_rows if r.status == "approved"),
rejected=sum(1 for r in all_rows if r.status == "rejected"),
)
return ReportRecordsOut(
records=[ReportRecordOut.model_validate(r) for r in rows],
counts=counts,
)
+22
View File
@@ -40,6 +40,20 @@ class Settings(BaseSettings):
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
# ===== Admin 后台 =====
# admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
# 生产必须改成高熵随机串(同 JWT_SECRET_KEY 的要求)。
ADMIN_JWT_SECRET: str = "change-me-admin"
ADMIN_JWT_EXPIRE_MINUTES: int = 720 # admin 登录态 12 小时(无 refresh,过期重登)
# 可选 IP 白名单(逗号分隔),为空=应用层不限制(靠 nginx allow/deny 兜底)。
ADMIN_IP_ALLOWLIST: str = ""
@property
def admin_ip_allowlist(self) -> list[str]:
if not self.ADMIN_IP_ALLOWLIST.strip():
return []
return [ip.strip() for ip in self.ADMIN_IP_ALLOWLIST.split(",") if ip.strip()]
# ===== 极光 =====
JG_APP_KEY: str = ""
JG_MASTER_SECRET: str = ""
@@ -51,6 +65,14 @@ class Settings(BaseSettings):
SMS_MOCK: bool = True
SMS_CODE_TTL_SEC: int = 300
SMS_SEND_INTERVAL_SEC: int = 60
# 真实发送走极光短信 REST(自定义验证码模式:本服务生成 code,极光只负责发)。
# 复用极光一键登录的 JG_APP_KEY / JG_MASTER_SECRET(同一个极光应用)+ JG_REQUEST_TIMEOUT_SEC。
SMS_SEND_ENDPOINT: str = "https://api.sms.jpush.cn/v1/messages"
SMS_SIGN_ID: int = 31729 # 极光短信签名 ID(非机密,可被 .env 覆盖)
SMS_TEMPLATE_ID: int = 1 # 极光短信模板 ID(变量名 code,有效期 5 分钟)
SMS_CODE_LENGTH: int = 6 # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
SMS_DAILY_LIMIT_PER_PHONE: int = 10 # 单手机号每日发送上限(防刷 + 控费)
SMS_MAX_VERIFY_ATTEMPTS: int = 5 # 单个验证码最多校验失败次数,超过即作废(防爆破)
# ===== 美团联盟 CPS =====
# 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
+5
View File
@@ -62,6 +62,11 @@ def save_feedback_image(user_id: int, data: bytes) -> str:
return _save_image("feedback", user_id, data)
def save_report_image(user_id: int, data: bytes) -> str:
"""保存上报更低价截图,返回相对 URL(`/media/price_report/<file>`)。"""
return _save_image("price_report", user_id, data)
def delete_avatar(url: str | None) -> None:
"""删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
prefix = f"{settings.MEDIA_URL_PREFIX}/avatars/"
+3 -1
View File
@@ -53,8 +53,10 @@ WITHDRAW_MAX_CENTS: int = 5_000_000 # 5 万元
TASK_ENABLE_NOTIFICATION = "enable_notification"
# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 已不再 = 兑换下限(MIN_EXCHANGE_COIN=10000), test_exchange_flow 改走 grant_coins 直接供款。
TASK_REWARDS: dict[str, int] = {
TASK_ENABLE_NOTIFICATION: 10000,
TASK_ENABLE_NOTIFICATION: 1000,
}
+22
View File
@@ -14,6 +14,7 @@ from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any, Literal
import bcrypt
import jwt
from app.core.config import settings
@@ -84,3 +85,24 @@ def issue_token_pair(user_id: int) -> dict[str, Any]:
"expires_in": int((access_exp - _now()).total_seconds()),
"refresh_expires_in": int((refresh_exp - _now()).total_seconds()),
}
# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。
def hash_password(plain: str) -> str:
"""bcrypt 加盐 hash,返回可入库的字符串。
bcrypt 限制明文 72 字节,超出抛 ValueError(实测 bcrypt 5.0:25 个中文=75 字节即触发)
统一截断到 72 字节(bcrypt 标准做法)72 字节仍是强密码,verify 同样截断保证一致
"""
return bcrypt.hashpw(plain.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")
def verify_password(plain: str, hashed: str) -> bool:
"""校验明文与 bcrypt hash。明文同样截断 72 字节(与 hash_password 一致,否则超长密码永远不匹配);
hash 串损坏/格式非法时返回 False,不抛异常"""
try:
return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8"))
except (ValueError, TypeError):
return False
+175 -32
View File
@@ -1,68 +1,211 @@
"""短信验证码服务。
当前实现:**mock 模式**
- send_code(): 不真发短信, log记录 phone 发送时间( 60s 内重复发)
- verify_code(): 任意 6 位数字均通过(配合前台 demo 行为)
两种运行模式由 `SMS_MOCK` 切换:
- **mock**(开发/测试,默认):不真发短信,验证码打到日志;校验**放行任意 N 位数字**
(测试/开发便利)真实校验逻辑(比对存码 / 一次性 / 防爆破) real 分支 + 单测覆盖
- **real**(生产 `SMS_MOCK=false`):本服务生成 N 位验证码 调极光短信 REST
`/v1/messages` 发送(自定义验证码模式,极光只负责发,code 由本服务生成/保管/
校验) 鉴权复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`(同一极光应用)
后续接真供应商(阿里云 / 腾讯云):
- send_code() 改为调供应商 API,把生成的 6 位码存到 cache(redis / sqlite)
- verify_code() 比对 cache 里的码,且验过即作废
验证码存储:**进程内存**( worker uvicorn 够用)重启丢失(用户重发即可)
worker / 多机时内存不共享 冷却每日上限校验都会失效,届时迁移到 DB/Redis
docs/待办与技术债.md
进程内存方案占坑期够用;切真实供应商时一并切 redis
防刷三层(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权):
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却(本文件)
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限(本文件)
3. IP 频控(api rate_limit 依赖)+ 极光控制台 IP 白名单/防轰炸(运维侧)
:单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废(防爆破),验过即作废(一次性)
"""
from __future__ import annotations
import base64
import logging
import secrets
import time
from dataclasses import dataclass
from datetime import datetime
from threading import Lock
import httpx
from app.core.config import settings
logger = logging.getLogger("shagua.sms")
# 进程内最近发送时间表:{phone: epoch_seconds},用于 60s 内拒发
_last_sent: dict[str, float] = {}
_lock = Lock()
class SmsError(Exception):
"""业务异常:发送过频 / 验证码不对。api 层 catch 后翻成 4xx。"""
"""业务异常。`status_code` 决定 api 层翻成哪个 HTTP 码:
过频/每日超限 = 429(客户端等会再来),供应商不可用 = 503,手机号无效 = 400
"""
def __init__(self, message: str, status_code: int = 429) -> None:
super().__init__(message)
self.status_code = status_code
@dataclass
class _CodeRecord:
code: str
expires_at: float
attempts: int = 0
# 进程内存(单 worker 有效;多 worker 不共享,见模块 docstring)
_codes: dict[str, _CodeRecord] = {} # phone -> 当前有效验证码
_last_sent: dict[str, float] = {} # phone -> 上次发送 epoch(冷却)
_daily_count: dict[str, tuple[str, int]] = {} # phone -> (date_str, 当日发送数)
_lock = Lock()
_GC_THRESHOLD = 10000 # 任一内存 dict 超此阈值,send 时顺手清过期项(防无限增长,仿 ratelimit)
def _today() -> str:
return datetime.now().strftime("%Y-%m-%d")
def _gen_code() -> str:
"""生成 N 位数字验证码(用 secrets 而非 random;允许前导 0)。"""
return "".join(secrets.choice("0123456789") for _ in range(settings.SMS_CODE_LENGTH))
def _gc(now: float) -> None:
"""顺手清理过期内存项,防三个 dict 无限增长。仅在持锁时调用,且某 dict 超
_GC_THRESHOLD 才扫它(低频,开销可忽略)"""
if len(_codes) > _GC_THRESHOLD:
for p in [p for p, r in _codes.items() if now > r.expires_at]:
_codes.pop(p, None)
if len(_last_sent) > _GC_THRESHOLD:
cutoff = now - settings.SMS_SEND_INTERVAL_SEC
for p in [p for p, ts in _last_sent.items() if ts < cutoff]:
_last_sent.pop(p, None)
if len(_daily_count) > _GC_THRESHOLD:
today = _today()
for p in [p for p, (d, _c) in _daily_count.items() if d != today]:
_daily_count.pop(p, None)
def send_code(phone: str) -> int:
"""发送(或 mock 发送)验证码。
"""发送验证码。
Returns: 距下次可发的秒数(=0 表示刚刚已发,需等 SMS_SEND_INTERVAL_SEC )
Raises: SmsError 如果上次发送在 SMS_SEND_INTERVAL_SEC
Returns: 距下次可发的秒数(= SMS_SEND_INTERVAL_SEC)
Raises: SmsError(过频 429 / 当日超限 429 / 供应商失败 503 / 手机号无效 400)
"""
now = time.time()
# --- lock 内:防刷检查 + 预占(防并发重复发烧钱)---
with _lock:
last = _last_sent.get(phone, 0.0)
elapsed = now - last
_gc(now) # 顺手清过期内存(超阈值才扫)
elapsed = now - _last_sent.get(phone, 0.0)
if elapsed < settings.SMS_SEND_INTERVAL_SEC:
remain = int(settings.SMS_SEND_INTERVAL_SEC - elapsed)
raise SmsError(f"send too frequent, retry in {remain}s")
_last_sent[phone] = now
raise SmsError(f"发送过于频繁,请 {remain}s 后再试")
if settings.SMS_MOCK:
logger.info("[SMS-MOCK] send to %s****, code=any-6-digits", phone[:3])
else:
# TODO: 接真实短信供应商
logger.warning("[SMS] real provider not implemented yet, phone=%s****", phone[:3])
raise SmsError("sms provider not configured")
today = _today()
day, cnt = _daily_count.get(phone, ("", 0))
if day != today:
cnt = 0
if cnt >= settings.SMS_DAILY_LIMIT_PER_PHONE:
raise SmsError("今日验证码发送次数已达上限,请明天再试")
code = _gen_code()
# 预占:先记冷却/计数/存码,释放锁后再发网络(发失败保留冷却+计数,见下)
_last_sent[phone] = now
_daily_count[phone] = (today, cnt + 1)
_codes[phone] = _CodeRecord(code=code, expires_at=now + settings.SMS_CODE_TTL_SEC)
# --- lock 外:真正发送(网络 IO 不持锁)---
try:
if settings.SMS_MOCK:
logger.info("[SMS-MOCK] to %s**** code=%s (不真发)", phone[:3], code)
else:
_send_via_jiguang(phone, code)
logger.info("[SMS] sent to %s****", phone[:3])
except Exception as e:
# 发送失败:**保留冷却 + 每日计数**(失败也限速,挡住余额不足/签名失效时
# 前端重试狂打极光),只清掉没发出去的码(用户收不到,留着无意义且占内存)。
with _lock:
_codes.pop(phone, None)
if isinstance(e, SmsError):
raise
logger.exception("[SMS] send failed phone=%s****", phone[:3])
raise SmsError("验证码发送失败,请稍后重试", status_code=503) from e
return settings.SMS_SEND_INTERVAL_SEC
def verify_code(phone: str, code: str) -> bool:
"""校验验证码。mock 模式下任意 6 位数字均通过。"""
"""校验验证码。
- **mock 模式**:放行任意 N 位数字(测试/开发便利,不真校验)
- **real 模式**:比对本服务存的码,匹配即作废(一次性);失败累计到上限也作废(防爆破)
"""
if settings.SMS_MOCK:
if len(code) == 6 and code.isdigit():
logger.info("[SMS-MOCK] verify ok for %s****", phone[:3])
ok = len(code) == settings.SMS_CODE_LENGTH and code.isdigit()
logger.info("[SMS-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
return ok
with _lock:
rec = _codes.get(phone)
if rec is None:
return False
if time.time() > rec.expires_at:
_codes.pop(phone, None)
return False
if rec.attempts >= settings.SMS_MAX_VERIFY_ATTEMPTS:
_codes.pop(phone, None) # 试错过多,作废
return False
if secrets.compare_digest(code.encode("utf-8"), rec.code.encode("utf-8")):
_codes.pop(phone, None) # 验过即作废
return True
logger.info("[SMS-MOCK] verify fail (need 6 digits) for %s****", phone[:3])
rec.attempts += 1
return False
# TODO: 比对供应商发出的真实验证码
logger.warning("[SMS] real provider not implemented yet, phone=%s****", phone[:3])
return False
def _send_via_jiguang(phone: str, code: str) -> None:
"""调极光短信 REST /v1/messages 发送(自定义验证码模式)。失败抛 SmsError。"""
if not settings.JG_APP_KEY or not settings.JG_MASTER_SECRET:
raise SmsError("短信服务未配置(缺 JG_APP_KEY/JG_MASTER_SECRET)", status_code=503)
if not settings.SMS_SIGN_ID or not settings.SMS_TEMPLATE_ID:
raise SmsError("短信服务未配置(缺 SMS_SIGN_ID/SMS_TEMPLATE_ID)", status_code=503)
auth_b64 = base64.b64encode(
f"{settings.JG_APP_KEY}:{settings.JG_MASTER_SECRET}".encode()
).decode()
body = {
"mobile": phone,
"sign_id": settings.SMS_SIGN_ID,
"temp_id": settings.SMS_TEMPLATE_ID,
"temp_para": {"code": code},
}
try:
resp = httpx.post(
settings.SMS_SEND_ENDPOINT,
json=body,
headers={
"Content-Type": "application/json",
"Authorization": f"Basic {auth_b64}",
},
timeout=settings.JG_REQUEST_TIMEOUT_SEC,
)
except httpx.HTTPError as e:
raise SmsError(f"短信网关网络错误: {e}", status_code=503) from e
if resp.status_code == 200:
return # {"msg_id": ...}
# 极光错误码映射(节选常见;完整码表见 docs/integrations/sms.md)
try:
err = resp.json().get("error", {})
ecode, emsg = err.get("code"), err.get("message", "")
except Exception:
ecode, emsg = None, resp.text[:200]
logger.error("[SMS] jiguang error http=%s code=%s msg=%s", resp.status_code, ecode, emsg)
if ecode == 50014: # 余额不足:运维要立即告警充值
logger.critical("[SMS] 极光短信余额不足(50014),需充值!")
raise SmsError("短信服务暂不可用,请稍后重试", status_code=503)
if ecode == 50009: # 极光侧超频
raise SmsError("发送过于频繁,请稍后再试", status_code=429)
if ecode == 50006: # 手机号无效(schema 已挡格式,这里多是空号/停机)
raise SmsError("手机号无效", status_code=400)
raise SmsError(f"短信发送失败(code={ecode})", status_code=503)
+2
View File
@@ -23,6 +23,7 @@ from app.api.v1.coupon import router as coupon_router
from app.api.v1.feedback import router as feedback_router
from app.api.v1.meituan import router as meituan_router
from app.api.v1.order import router as order_router
from app.api.v1.report import router as report_router
from app.api.v1.savings import router as savings_router
from app.api.v1.signin import router as signin_router
from app.api.v1.tasks import router as tasks_router
@@ -86,6 +87,7 @@ app.include_router(tasks_router)
app.include_router(savings_router)
app.include_router(ad_router)
app.include_router(order_router)
app.include_router(report_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
+2
View File
@@ -1,9 +1,11 @@
"""所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
from app.models.ad_ecpm import AdEcpmRecord # noqa: F401
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
from app.models.comparison import ComparisonRecord # noqa: F401
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
from app.models.feedback import Feedback # noqa: F401
from app.models.price_report import PriceReport # noqa: F401
from app.models.savings import SavingsRecord # noqa: F401
from app.models.signin import SigninRecord # noqa: F401
from app.models.task import UserTask # noqa: F401
+70
View File
@@ -0,0 +1,70 @@
"""Admin 后台:管理员账号 + 操作审计日志。
App 用户(user )完全隔离:admin 走独立 JWT secret独立鉴权链( app/admin/)
- admin_user:账号密码(bcrypt)登录,带角色(super_admin / finance / operator)
- admin_audit_log:每个写操作落一条,记前后值,不可删,用于追溯"谁在何时改了谁的钱/状态"
admin_username / target_id 冗余存字符串,即使关联对象被删/改名也能追溯
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 comparison_record.raw_payload)。
_JSON = JSON().with_variant(JSONB(), "postgresql")
class AdminUser(Base):
__tablename__ = "admin_user"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
# super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
# active / disabled
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
last_login_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
def __repr__(self) -> str: # pragma: no cover
return f"<AdminUser id={self.id} username={self.username} role={self.role}>"
class AdminAuditLog(Base):
__tablename__ = "admin_audit_log"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
admin_id: Mapped[int] = mapped_column(
Integer, ForeignKey("admin_user.id"), index=True, nullable=False
)
# 冗余存操作者用户名:admin 改名/禁用后仍可追溯是谁干的
admin_username: Mapped[str] = mapped_column(String(64), nullable=False)
# 操作类型,如 user.coins.grant / user.status.set / withdraw.refresh / feedback.handle
action: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
# 被操作对象类型 + id(id 用字符串以兼容 out_bill_no 等非整型主键)
target_type: Mapped[str] = mapped_column(String(32), nullable=False)
target_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 上下文 + 前后值,如 {"amount": 1000, "reason": "...", "before": {...}, "after": {...}}
detail: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<AdminAuditLog id={self.id} admin={self.admin_username} "
f"action={self.action} target={self.target_type}:{self.target_id}>"
)
+1 -1
View File
@@ -83,7 +83,7 @@ class ComparisonRecord(Base):
# ===== 明细(JSON,越详细越好)=====
# 下单菜品 [{name, qty, specs?}]
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank}](price 单位:元,原样存)
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved}](price/coupon_saved 单位:元,原样存)
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
# 目标平台未找到、跳过的菜名
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
+67
View File
@@ -0,0 +1,67 @@
"""上报更低价记录表(price_report)。
用户在比价记录里选一条记录,上报某平台有比我们算出的最低价更低的价格,带截图证明,
人工审核通过奖励金币 feedback(自由文本反馈)不同:本表结构化关联具体比价记录
原最低价 vs 上报价审核状态奖励
- 原最低价快照(original_*)由后端按 comparison_record_id 反查比价记录的 best_* 冗余填入,
避免被关联记录后续删/改后对不上
- 价格统一存(cents), comparison_record / savings_record 一致
- status: pending(审核中)/ approved(已通过)/ rejected(未通过),对齐前端筛选与状态徽标
- images: 截图相对 URL 列表(/media/price_report/...,复用 app.core.media)
- 奖励发放(通过 钱包 +reward_coins)是人工审核后台动作,提交时一律 pending,本表只留字段
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class PriceReport(Base):
__tablename__ = "price_report"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 选中的那条比价记录(可空:记录被删后仍保留上报历史)
comparison_record_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("comparison_record.id"), index=True, nullable=True
)
# ===== 选中比价记录的快照(后端反查 comparison_record 填入,冗余存) =====
store_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
dish_summary: Mapped[str | None] = mapped_column(String(256), nullable=True)
original_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_platform_name: Mapped[str | None] = mapped_column(String(32), nullable=True)
original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
# ===== 用户上报的更低价 =====
reported_platform_id: Mapped[str] = mapped_column(String(32), nullable=False)
reported_platform_name: Mapped[str] = mapped_column(String(32), nullable=False)
reported_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
# 截图证明 URL 列表(相对路径,如 ["/media/price_report/u1_ab.jpg"])
images: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
# ===== 审核(人工后台) =====
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="pending", index=True
)
reject_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
reward_coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
reviewed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<PriceReport id={self.id} user_id={self.user_id} status={self.status}>"
)
+1 -1
View File
@@ -85,7 +85,7 @@ def grant_ad_reward(
# 发金币 + 记一笔,同事务
crud_wallet.grant_coins(
db, user_id, coin,
biz_type="ad_reward", ref_id=trans_id, remark="广告奖励",
biz_type="ad_reward", ref_id=trans_id, remark="视频奖励金币",
)
rec = AdRewardRecord(
trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
+8 -9
View File
@@ -1,8 +1,9 @@
"""比价战绩里程碑 CRUD:进度查询 + 逐档领奖。
进度 = comparison_record status='success' 的条数(crud_compare.count_success)
N 档在"成功次数 >= N"时解锁;每档领一次, comparison_milestone_claim 去重 +
grant_coins,同一事务 commit(领奖记录与加金币原子化)仿一次性任务 (task.py)
N 档在"成功次数 >= N"时解锁;每档领一次, comparison_milestone_claim 去重(标记已领)
当前产品定暂不真发金币(后续整体删除该功能):claim 只写领取记录,不调 grant_coins
不写 coin_transaction,coin_awarded 恒为 0,余额不变仿一次性任务 (task.py)
"""
from __future__ import annotations
@@ -14,7 +15,6 @@ from sqlalchemy.orm import Session
from app.core.rewards import (
RECORD_MILESTONE_COUNT,
RECORD_MILESTONES,
record_milestone_reward,
)
from app.models.comparison_milestone import ComparisonMilestoneClaim
from app.repositories import comparison as crud_compare
@@ -103,14 +103,13 @@ def claim(db: Session, user_id: int, milestone: int) -> tuple[int, int]:
if crud_compare.count_success(db, user_id) < milestone:
raise MilestoneLockedError
coin = record_milestone_reward(milestone)
# ⚠️ 比价战绩里程碑暂不真发金币(产品定,后续整体删除该功能):仍记一条领取(去重/标记已领),
# 但不调 grant_coins、不写 coin_transaction,余额不变,返回发放 0 金币 + 当前余额。
db.add(
ComparisonMilestoneClaim(
user_id=user_id, milestone=milestone, coin_awarded=coin
user_id=user_id, milestone=milestone, coin_awarded=0
)
)
acc, _ = crud_wallet.grant_coins(
db, user_id, coin, biz_type="compare_milestone", ref_id=str(milestone)
)
acc = crud_wallet.get_or_create_account(db, user_id, commit=False)
db.commit()
return coin, acc.coin_balance
return 0, acc.coin_balance
+53
View File
@@ -0,0 +1,53 @@
"""price_report 表读写。"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.price_report import PriceReport
def create_report(
db: Session,
*,
user_id: int,
comparison_record_id: int | None,
store_name: str | None,
dish_summary: str | None,
original_platform_id: str | None,
original_platform_name: str | None,
original_price_cents: int | None,
reported_platform_id: str,
reported_platform_name: str,
reported_price_cents: int,
images: list[str],
) -> PriceReport:
rep = PriceReport(
user_id=user_id,
comparison_record_id=comparison_record_id,
store_name=store_name,
dish_summary=dish_summary,
original_platform_id=original_platform_id,
original_platform_name=original_platform_name,
original_price_cents=original_price_cents,
reported_platform_id=reported_platform_id,
reported_platform_name=reported_platform_name,
reported_price_cents=reported_price_cents,
images=images,
status="pending",
)
db.add(rep)
db.commit()
db.refresh(rep)
return rep
def list_reports(
db: Session, user_id: int, status: str | None = None
) -> list[PriceReport]:
"""该用户的上报记录,按时间倒序;status 非空时按状态筛选。"""
stmt = select(PriceReport).where(PriceReport.user_id == user_id)
if status:
stmt = stmt.where(PriceReport.status == status)
stmt = stmt.order_by(PriceReport.created_at.desc(), PriceReport.id.desc())
return list(db.execute(stmt).scalars().all())
+2 -1
View File
@@ -119,7 +119,8 @@ def do_signin(db: Session, user_id: int) -> tuple[SigninRecord, int]:
)
db.add(record)
acc, _ = crud_wallet.grant_coins(
db, user_id, coin, biz_type="signin", ref_id=today.isoformat()
db, user_id, coin, biz_type="signin", ref_id=today.isoformat(),
remark=f"每日签到 第{cycle_day}",
)
db.commit()
db.refresh(record)
+5 -1
View File
@@ -33,6 +33,10 @@ class ComparisonResultIn(BaseModel):
price: float | None = None
is_source: bool = False
rank: int | None = None
# 该平台本单用红包省的**纯红包优惠额**(元, 不含配送费/代金券/满减)。None=没用/没抠到。
# 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
# 各平台抠到红包即带值(2026-06 起源平台 Phase1 意图识别也抠, 当前仅淘宝源)。见 pricebot 侧 比价红包额留痕-实现方案.md。
coupon_saved: float | None = None
class ComparisonRecordIn(BaseModel):
@@ -109,7 +113,7 @@ class ComparisonRecordCreatedOut(BaseModel):
class MilestoneStateOut(BaseModel):
milestone: int = Field(..., description="档位序号 1-based(= 解锁所需的成功比价次数)")
coin: int = Field(..., description="该档奖励金币")
coin: int = Field(..., description="该档应发金币额(产品规则值);当前领取暂不真发, 见 compare-milestone-claim 文档")
state: str = Field(..., description="claimed(已领) / active(可领) / locked(未解锁)")
+57
View File
@@ -0,0 +1,57 @@
"""上报更低价 响应 schema。
请求是 multipart 表单(comparison_record_id / reported_platform_id / reported_price / images),
在路由里直接校验,不单独建请求 schema( feedback)价格对外统一(cents)
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class ReportSubmitOut(BaseModel):
"""提交上报后的回执。"""
model_config = ConfigDict(from_attributes=True)
id: int
status: str
created_at: datetime
class ReportRecordOut(BaseModel):
"""一条上报记录(「上报记录」列表用)。价格单位:分。"""
model_config = ConfigDict(from_attributes=True)
id: int
store_name: str | None = None
dish_summary: str | None = None
# 原最低价(上报前系统给出的最低,反查比价记录 best_*)
original_platform_id: str | None = None
original_platform_name: str | None = None
original_price_cents: int | None = None
# 用户上报的更低价
reported_platform_id: str
reported_platform_name: str
reported_price_cents: int
images: list[str] = Field(default_factory=list)
status: str # pending / approved / rejected
reject_reason: str | None = None
reward_coins: int | None = None
created_at: datetime
class ReportRecordCounts(BaseModel):
"""四态计数,供前端筛选 chip 显示(全部/审核中/已通过/未通过)。"""
all: int = 0
pending: int = 0
approved: int = 0
rejected: int = 0
class ReportRecordsOut(BaseModel):
records: list[ReportRecordOut]
counts: ReportRecordCounts
+23
View File
@@ -0,0 +1,23 @@
[Unit]
Description=Shaguabijia Admin Backend (FastAPI / uvicorn, 独立进程,复用 app-server codebase)
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/shaguabijia-app-server
Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin"
EnvironmentFile=/opt/shaguabijia-app-server/.env
# 与主 App 后端(8770)复用同一 codebase / .env / .venv / DB,独立进程跑 admin app
ExecStart=/opt/shaguabijia-app-server/.venv/bin/uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 --workers 1 --log-level info
Restart=on-failure
RestartSec=3
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ReadWritePaths=/opt/shaguabijia-app-server
ProtectHome=true
[Install]
WantedBy=multi-user.target
+9 -8
View File
@@ -2,9 +2,10 @@
> 所属:比价记录组(前缀 `/api/v1/compare` | 鉴权:Bearer | [← 返回 API 索引](./README.md)
领取某一档(第 `milestone` 次)的金币奖励,真实入账金币钱包。领取记录
`comparison_milestone_claim`((user_id, milestone) 唯一)+ `grant_coins` 同一事务原子提交,
**每档只能领一次**。进度口径见 [milestones 进度接口](./compare-milestones.md)。
领取某一档(第 `milestone` 次)。⚠️ **当前不真发金币**(产品定,后续整体删除该功能):仍
`comparison_milestone_claim`((user_id, milestone) 唯一)标记该档已领、**每档只能领一次**,但不调
`grant_coins`、不写 `coin_transaction`,余额不变,`coin_awarded` 恒为 0。进度口径见
[milestones 进度接口](./compare-milestones.md)。
## 入参
- 路径参数 `milestone`(int):档位序号,1..6(= `RECORD_MILESTONES` 长度)。
@@ -15,11 +16,11 @@
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 本次领取的档位序号 |
| `coin_awarded` | int | 本次发放金币 |
| `coin_balance` | int | 领奖后金币余额 |
| `coin_awarded` | int | 本次发放金币(当前恒为 `0`,暂不真发) |
| `coin_balance` | int | 领奖后金币余额(当前不变) |
```json
{ "milestone": 1, "coin_awarded": 120, "coin_balance": 29087 }
{ "milestone": 1, "coin_awarded": 0, "coin_balance": 29087 }
```
## 错误
@@ -29,5 +30,5 @@
- `409` 该档已领过(`milestone already claimed`)
## 说明
- 领取金币,客户端刷新钱包余额([`GET /api/v1/wallet/account`](./wallet-account.md))与本档状态
- 幂等:重复领同一档返回 409 且**不会重复发金币**(唯一约束 + 领取前查重)。
- 当前不发金币,客户端领取后只需把本档状态刷成 claimed 即可;若后续恢复发奖,再按 `coin_awarded` 刷新钱包余额([`GET /api/v1/wallet/account`](./wallet-account.md))。
- 幂等:重复领同一档返回 409(唯一约束 + 领取前查重)。
+3 -1
View File
@@ -8,6 +8,8 @@
档位金额是**产品规则**,定义在后端 `app/core/rewards.py``RECORD_MILESTONES`(当前 `120/180/300/500/800/1200`),客户端**不要写死**,以本接口返回为准。
> ⚠️ **当前领取暂不真发金币**(产品定,后续整体删除该功能):`coin` 仍返回产品规则值仅供展示,但 [claim 接口](./compare-milestone-claim.md) 实际 `coin_awarded` 恒为 0、余额不变。前端展示须与之对齐,勿让用户误以为领取可到账。
## 入参
无(用户身份取自 Bearer token)。
@@ -25,7 +27,7 @@
| 字段 | 类型 | 说明 |
|---|---|---|
| `milestone` | int | 档位序号(1-based),= 解锁所需的成功比价次数 |
| `coin` | int | 该档奖励金币 |
| `coin` | int | 该档**应发**金币额(产品规则值);⚠️ 当前领取不真发,见上方说明 |
| `state` | string | `claimed`(已领) / `active`(已解锁可领) / `locked`(未解锁) |
```json
+3 -1
View File
@@ -31,7 +31,9 @@
**Item**`{ name: string, qty: int=1, specs: string[]\|null }`
**Resultcomparison_results 元素)**`{ platform_id, platform_name, package, price(元,float\|null), is_source(bool), rank(int\|null) }`
**Resultcomparison_results 元素)**`{ platform_id, platform_name, package, price(元,float\|null), is_source(bool), rank(int\|null), coupon_saved(元,float\|null) }`
- `coupon_saved`:该平台本单**平台主优惠额**(元)——美团红包 / 淘宝平台红包 / 京东优惠券·百亿补贴,**只取那一笔**,不含配送费减免/共减总额。仅外卖目标平台带值,源平台/没用为 `null`,前端记录页据此展示**「已优惠 ¥X」**(null 不展示)。
## 服务端派生(客户端不用算)
+2 -2
View File
@@ -10,7 +10,7 @@
| `id` | Integer | PK, autoincrement | |
| `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
| `milestone` | Integer | NOT NULL | 档位序号(1-based),见 `rewards.RECORD_MILESTONES` |
| `coin_awarded` | Integer | NOT NULL, default 0 | 该档发放金币 |
| `coin_awarded` | Integer | NOT NULL, default 0 | 该档应发金币额;⚠️ 当前产品定暂不真发,实际恒为 0 |
| `claimed_at` | DateTime(tz) | server_default now() | 领取时间 |
## 索引与约束
@@ -22,5 +22,5 @@
- 解锁进度依赖 `comparison_record`(`status='success'` 计数),本表不存进度本身。
## 说明
- 发金币走 `wallet.grant_coins(biz_type='compare_milestone', ref_id=str(milestone))`,与本表写入同事务
- ⚠️ 当前 claim 仅写本表标记已领(去重),**不调 `grant_coins`、不写 `coin_transaction`,`coin_awarded` 恒为 0、余额不变**(产品定暂不真发金币,后续整体删除该功能)
- 领取校验:档位越界 404、未解锁(成功数 < milestone)409、已领 409。
+1 -1
View File
@@ -29,7 +29,7 @@
| `status` | String(16) | NOT NULL, default `success` | `success`(拿到有效对比)/ `failed`(出错/没采到目标价) |
| `information` | String(256) | nullable | done 帧文案;成功=摘要,失败=具体原因(前端失败时当原因展示) |
| `items` | JSON(PG: JSONB) | NOT NULL, default [] | 下单菜品 `[{name, qty, specs?}]` |
| `comparison_results` | JSON(PG: JSONB) | NOT NULL, default [] | 逐平台对比 `[{platform_id,platform_name,package,price(元),is_source,rank}]` |
| `comparison_results` | JSON(PG: JSONB) | NOT NULL, default [] | 逐平台对比 `[{platform_id,platform_name,package,price(元),is_source,rank,coupon_saved}]`;`coupon_saved`=该平台主优惠额(元,美团红包/淘宝平台红包/京东优惠券·百亿补贴,只取那一笔,不含配送费/共减总额),各平台抠到红包即带值(2026-06 起源平台 Phase1 意图识别也抠,当前仅淘宝源),没用/没抠到为 null,前端展示「已优惠 ¥X」 |
| `skipped_dish_names` | JSON(PG: JSONB) | NOT NULL, default [] | 被跳过的菜名 |
| `raw_payload` | JSON(PG: JSONB) | nullable | 客户端原始上报(calibration + done.params 全量) |
| `created_at` | DateTime(tz) | server_default now(), index | 时间 |
+47 -12
View File
@@ -3,22 +3,57 @@
> 文件:`app/integrations/sms.py` | 关联接口:[auth-sms-send](../api/auth-sms-send.md) · [auth-sms-login](../api/auth-sms-login.md) | [← 集成索引](./README.md)
## 作用
手机号 + 验证码登录的验证码发送 / 校验。**当前是 mock 模式**(占坑),未接真实短信供应商
手机号 + 验证码登录的验证码发送 / 校验。**已接极光短信 REST**,由 `SMS_MOCK` 切 mock / real
| | mock(`SMS_MOCK=true`,默认 / 开发测试) | real(`SMS_MOCK=false`,生产) |
|---|---|---|
| 发送 | 不真发,验证码打日志 | 调极光 `/v1/messages` 真发 |
| 校验 | 放行任意 6 位数字(测试便利) | 比对本服务存的码(一次性 / 过期 / 防爆破) |
## 为什么是"自己生成码"模式
极光有两种验证码玩法:① `/v1/codes` 极光生成 + `/codes/{msg_id}/valid` 极光校验;② `/v1/messages` 本服务生成码、极光只负责发。**选 ②**:校验本地完成(不依赖极光二次往返)、有效期/错误次数/作废完全可控、跟原 mock 口子结构一致。
## 函数 / 异常
| 函数 | 行为 |
|---|---|
| `send_code(phone) -> int` | mock 下不真发、只 log;返回距下次可发的秒数。进程内存表 `_last_sent``phone → 发送时间``SMS_SEND_INTERVAL_SEC` 内再发抛 `SmsError`(发送过频)。真供应商未接时抛 `SmsError("sms provider not configured")` |
| `verify_code(phone, code) -> bool` | mock 下**任意 6 位数字**均通过(配合前台 demo);真模式未实现返回 `False` |
| 异常 `SmsError` | 发送过频 / 验证码不对,api 层 catch 后翻成 4xx(过频 `429` |
| `send_code(phone) -> int` | 防刷检查(冷却 + 每日上限)→ `secrets` 生成 N 位码 → **预占**(冷却/计数/存码)→ mock 打日志 / real 调极光 → 失败**回滚预占**。返回距下次可发秒数 |
| `verify_code(phone, code) -> bool` | mock 放行任意 6 位;real 比对存码,匹配即作废,失败累计到上限作废 |
| `SmsError(msg, status_code)` | `status_code` 决定 HTTP 码:过频/每日超限 **429**、供应商失败 **503**、号码无效 **400** |
## 配置
| 配置项 | 说明 |
|---|---|
| `SMS_MOCK` | 是否 mock(当前 `true` |
| `SMS_SEND_INTERVAL_SEC` | 同号两次发送最小间隔(防过频) |
| 配置项 | 默认 | 说明 |
|---|---|---|
| `SMS_MOCK` | `true` | mock / real 切换。**生产置 false** |
| `SMS_CODE_TTL_SEC` | 300 | 验证码有效期(秒),与极光模板文案"5分钟"一致 |
| `SMS_SEND_INTERVAL_SEC` | 60 | 单号两次发送最小间隔(冷却) |
| `SMS_SEND_ENDPOINT` | 极光 `/v1/messages` | 短信发送地址 |
| `SMS_SIGN_ID` | 31729 | 极光签名 ID |
| `SMS_TEMPLATE_ID` | 1 | 极光模板 ID(变量名 `code`) |
| `SMS_CODE_LENGTH` | 6 | 验证码位数(本服务生成;前端 code 4-8 位兼容) |
| `SMS_DAILY_LIMIT_PER_PHONE` | 10 | 单号每日发送上限(防刷 + 控费) |
| `SMS_MAX_VERIFY_ATTEMPTS` | 5 | 单码最多校验失败次数,超过作废(防爆破) |
## 后续(接真实供应商时)
- `send_code()` 改调供应商 API(阿里云 / 腾讯云),把生成的 6 位码存 cacheredis / sqlite)。
- `verify_code()` 比对 cache 里的码,且**验过即作废**。
- 进程内存方案占坑期够用;切真实供应商时一并切 redis(多进程/多机时内存表不共享,频控会失效)。
> **鉴权复用极光一键登录**:`/v1/messages``base64(JG_APP_KEY:JG_MASTER_SECRET)` 做 HTTP Basic Auth——短信与一键登录是**同一个极光应用**(同 AppKey)。**上线不需要额外凭证,只需 `SMS_MOCK=false`**(`JG_*` 一键登录已配)。
## 防刷(短信花钱 + `/sms/send` 在登录前无法 JWT 鉴权)
1. 单号 `SMS_SEND_INTERVAL_SEC` 冷却
2. 单号每日 `SMS_DAILY_LIMIT_PER_PHONE` 条上限
3. 单 IP 频控:`/sms/send``rate_limit(10,60)``/sms/login``rate_limit(20,60)`
4. 单码校验失败 `SMS_MAX_VERIFY_ATTEMPTS` 次即作废
5. 运维侧建议在极光控制台叠加:**IP 白名单**(只许服务器 IP)+ **防轰炸设置**
## 极光错误码(节选,映射在 `_send_via_jiguang`)
| code | 含义 | 处理 |
|---|---|---|
| 50014 | 余额不足 | `logger.critical` 告警 + 503 |
| 50009 | 极光侧超频 | 429 |
| 50006 | 手机号无效 | 400 |
| 其他 | — | 503 |
## 上线步骤
1. 极光控制台:企业实名 + 短信充值 + 签名审核(`sign_id=31729`)+ 模板审核(`temp_id=1`)——**均已就绪**
2. `.env``SMS_MOCK=false`(`JG_APP_KEY`/`JG_MASTER_SECRET` 一键登录已配)
3. 真机发一条验证:收到短信 + 能登录
## 已知局限
**验证码存进程内存**:单 worker uvicorn 够用;重启丢码(用户重发即可);**多 worker / 多机不共享 → 冷却 / 每日上限 / 校验失效**,扩 worker 前迁移到 DB/Redis。见 [待办与技术债](../待办与技术债.md)。
+4 -2
View File
@@ -160,7 +160,7 @@ Android SDK loginAuth → 客户端拿到 loginToken
> **密钥对配对关系是关键**:极光用控制台上传的公钥加密,后端必须用**配对的私钥**解密。私钥不配对时 `jverify-login` 报 502 `all RSA paddings failed`(密钥不对,非 padding 问题);私钥文件缺失则报 502 `private key not found`
> **2026-05-27 现状**:历史遗留的两套本地密钥均**不配对**当前 AppKey 公钥,已重新生成一对 1024-bit/PKCS#8 密钥,公钥需在极光控制台更新。⚠️ 该 AppKey 公钥为全局配置,若占坑版后端仍在用同一 AppKey,换公钥会使其旧私钥失效,需同步部署新私钥。
### 4.2 短信登录(mock 路径)
### 4.2 短信登录(已接极光,SMS_MOCK 切 mock / real)
```
POST /api/v1/auth/sms/send { phone } → 60s 冷却,不真发(SMS_MOCK=true)
@@ -169,6 +169,8 @@ POST /api/v1/auth/sms/login { phone, code } → 任意 6 位通过 → upsert
短信冷却表存进程内存(`--workers 1` 下够用,多 worker/重启即失效)。
**real 模式(`SMS_MOCK=false`,生产)**:自定义验证码——本服务 `secrets` 生成 6 位码 → 极光 `/v1/messages` 只负责发 → 本地校验(一次性 / 防爆破),鉴权**复用 `JG_APP_KEY`/`JG_MASTER_SECRET`**(短信与一键登录同一极光应用,**上线只需 `SMS_MOCK=false`**)。防刷四层(单号冷却 + 单号每日上限 + 单 IP `rate_limit` + 单码失败次数)+ 错误码 429/503/400。详见 [integrations/sms](./integrations/sms.md)。
### 4.3 Token 与刷新
- access(2h)+ refresh(30d),HS256 自包含,payload 含 `sub`/`typ`/`iat`/`exp`,**无 session 表、无 jti 黑名单**。
@@ -315,7 +317,7 @@ conda activate price # 首次:pip install -e .
| 美团接口未配凭证降级 | 未配 `MT_CPS_APP_KEY` 时 3 端点返空(不报 502),`/feed` 跟"已配但调用失败"路径无法区分——见 [integrations/meituan](./integrations/meituan.md) |
| 领券/比价依赖 pricebot | `coupon/step` / `intent/recognize` / `price/step` 仅透传,真正逻辑在 pricebot-backend;前端已接通领券链路,比价 food MVP 也已接通 |
| agent 系列接口 MVP 不鉴权 | 拿不到 user_id → 无法采集"哪个用户领了/买了什么"用户级画像(商业模式核心资产)。见 [待办与技术债.md](./待办与技术债.md) P1 |
| SMS 为 mock | 上线接真实供应商 + `SMS_MOCK=false` |
| SMS 已接极光(2026-06-03) | real 模式自定义验证码,上线只需 `SMS_MOCK=false`(复用极光凭证)。详见 [integrations/sms](./integrations/sms.md) |
| 短信冷却存内存 | 扩 worker 前需迁移到 Redis |
| `MEDIA_ROOT` 进程内 serve | 头像/反馈截图当前用 FastAPI StaticFiles,生产建议 nginx 直 serve 该目录 |
| `init_postgres.py` 已知小 bug | 5 条小坑,见 [待办与技术债.md](./待办与技术债.md) |
+1
View File
@@ -87,6 +87,7 @@
## 已解决
- ✅ **短信验证码接入极光(real 模式,2026-06-03)**:`/sms/send`+`/sms/login` 从 mock 升级为真实可用。`integrations/sms.py` 走**自定义验证码模式**(本服务 `secrets` 生成 6 位码 + 进程内存存储 + 极光 `/v1/messages` 只负责发 + 本地校验一次性/防爆破);鉴权**复用极光一键登录的 `JG_APP_KEY`/`JG_MASTER_SECRET`**(同一极光应用,**上线只需 `SMS_MOCK=false`**,无需额外凭证;`sign_id=31729`/`temp_id=1` 已审核就绪)。防刷四层(单号冷却 + 单号每日上限 + 单 IP `rate_limit` + 单码失败次数上限);`SmsError.status_code` 让发送失败按类型返回 429/503/400。**mock 保留"任意6位通过"**(不动其他 11 个测试文件的 `_login` helper);`tests/test_auth.py` 加 real 模式单测(极光协议 / 一次性 / 防爆破 / 余额回滚)。文档 [integrations/sms.md](./integrations/sms.md)。**遗留**:验证码进程内存,多 worker 需迁 DB/Redis(下方技术债)。
- ✅ **比价记录落库(server + client,2026-05-31)**:每次比价 done 后客户端上报、按 user 落库,作「我的比价记录」数据源 + 用户级画像沉淀。
- **server**(本仓):新表 `comparison_record`(独立于 `savings_record`;结构化列 + `items`/`comparison_results`/`skipped_dish_names`/`raw_payload` 四个 JSON(B) 列)+ 3 个**鉴权**端点 `POST /api/v1/compare/record`(`(user_id,trace_id)` 幂等 upsert,best/saved/is_source_best/status 服务端从 comparison_results 派生)/ `GET /api/v1/compare/records`(游标分页)/ `GET /records/{id}`(含 raw_payload);`models/comparison.py` + `repositories/comparison.py` + `schemas/compare_record.py` + `api/v1/compare_record.py` + 迁移 `comparison_record_table`(head `b2c3d4e5f6a7`)+ `tests/test_compare_record.py`(8 例全过)+ `docs/api/compare-record-*.md`
- **client**(android 仓):`PriceBotService.runTask()` 比价 done 后(`lastDoneParams!=null`,成功/引擎失败都报)用独立 IO 协程尽力上报;`Protocol.CompareRecordRequest.fromComparison()` 从 calibration+done.params 零翻译组装;`ApiClient.reportCompareRecord()` 走新建的 `authedClient`(复用 `AuthInterceptor`+`RefreshAuthenticator`,自动 Bearer+401 刷新)。领券不报(非价格对比,本期范围只外卖)。
+3
View File
@@ -31,6 +31,9 @@ dependencies = [
# multipart form (FastAPI 表单上传依赖)
"python-multipart>=0.0.9",
# admin 后台账号密码 hash(用户侧是手机号+验证码登录,不需要密码;admin 才用)
"bcrypt>=4.0.0",
]
[project.optional-dependencies]
+58
View File
@@ -0,0 +1,58 @@
"""创建 / 重置一个 admin 后台账号。
用法(在项目根 pip install -e . 的环境里跑):
python scripts/create_admin.py --username admin --role super_admin
不传 --password 自动生成强随机密码,打印一次,务必当场保存
python scripts/create_admin.py --username ops --role operator --password 'xxx'
幂等:username 已存在则重置其密码 + 角色 + 置为 active(忘记密码时用它重置)
角色:super_admin(全权+管账号)/ finance()/ operator(用户+反馈+大盘)
"""
from __future__ import annotations
import argparse
import secrets
from app.admin.repositories import admin_user as admin_repo
from app.core.security import hash_password
from app.db.session import SessionLocal
_ROLES = ("super_admin", "finance", "operator")
def main() -> None:
parser = argparse.ArgumentParser(description="创建或重置 admin 账号")
parser.add_argument("--username", required=True)
parser.add_argument("--role", default="super_admin", choices=_ROLES)
parser.add_argument("--password", default=None, help="不传则自动生成强随机密码")
args = parser.parse_args()
password = args.password or secrets.token_urlsafe(16)
generated = args.password is None
db = SessionLocal()
try:
existing = admin_repo.get_by_username(db, args.username)
if existing is None:
admin = admin_repo.create_admin(
db, username=args.username, password=password, role=args.role
)
action = "创建"
else:
existing.password_hash = hash_password(password)
existing.role = args.role
existing.status = "active"
db.commit()
db.refresh(existing)
admin = existing
action = "重置"
finally:
db.close()
print(f"✅ 已{action} admin: id={admin.id} username={admin.username} role={admin.role}")
if generated:
print(f"🔑 自动生成的密码(只显示这一次,请立即保存):\n {password}")
if __name__ == "__main__":
main()
+1
View File
@@ -18,6 +18,7 @@ _tmp_db.close()
os.environ["DATABASE_URL"] = f"sqlite:///{_tmp_db.name}"
os.environ.setdefault("JWT_SECRET_KEY", "test-secret-please-ignore-this-is-only-for-pytest-not-real")
os.environ.setdefault("ADMIN_JWT_SECRET", "test-admin-secret-please-ignore-only-for-pytest-not-real")
os.environ.setdefault("JG_APP_KEY", "test-key")
os.environ.setdefault("JG_MASTER_SECRET", "test-secret")
os.environ.setdefault("SMS_MOCK", "true")
+155
View File
@@ -0,0 +1,155 @@
"""Admin 后台 M1 测试:登录闭环 + 与 App 用户鉴权的彻底隔离。
admin app 是独立的 FastAPI(app.admin.main:admin_app),用独立 TestClient
admin 表由 conftest Base.metadata.create_all 一起建好(models/__init__ 已登记)
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.core.security import create_token, hash_password
from app.db.session import SessionLocal
@pytest.fixture()
def admin_client() -> TestClient:
return TestClient(admin_app)
def _ensure_admin(
username: str = "test_admin",
password: str = "admin-pass-123",
role: str = "super_admin",
) -> tuple[str, str]:
"""create-or-reset 一个 admin(测试 DB 跨用例共享,需幂等)。"""
db = SessionLocal()
try:
a = admin_repo.get_by_username(db, username)
if a is None:
admin_repo.create_admin(db, username=username, password=password, role=role)
else:
a.password_hash = hash_password(password)
a.role = role
a.status = "active"
db.commit()
finally:
db.close()
return username, password
def test_admin_login_and_me(admin_client: TestClient) -> None:
username, password = _ensure_admin()
r = admin_client.post(
"/admin/api/auth/login", json={"username": username, "password": password}
)
assert r.status_code == 200, r.text
data = r.json()
assert "access_token" in data
assert data["admin"]["username"] == username
assert data["admin"]["role"] == "super_admin"
token = data["access_token"]
r = admin_client.get("/admin/api/auth/me", headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 200, r.text
assert r.json()["username"] == username
def test_admin_login_wrong_password(admin_client: TestClient) -> None:
username, _ = _ensure_admin()
r = admin_client.post(
"/admin/api/auth/login", json={"username": username, "password": "definitely-wrong"}
)
assert r.status_code == 401
def test_admin_me_requires_token(admin_client: TestClient) -> None:
assert admin_client.get("/admin/api/auth/me").status_code == 401
def test_disabled_admin_cannot_login(admin_client: TestClient) -> None:
username, password = _ensure_admin(username="disabled_admin")
db = SessionLocal()
try:
a = admin_repo.get_by_username(db, username)
a.status = "disabled"
db.commit()
finally:
db.close()
r = admin_client.post(
"/admin/api/auth/login", json={"username": username, "password": password}
)
assert r.status_code == 403
# ============================ 关键:鉴权隔离 ============================
def test_app_user_token_cannot_access_admin(admin_client: TestClient) -> None:
"""App 用户的 access_token(用 JWT_SECRET_KEY 签)不能访问 admin。
admin 用独立 ADMIN_JWT_SECRET 验签 App token 直接验签失败 401
这是后台防越权的第一道线
"""
user_token, _ = create_token(user_id=1, token_type="access")
r = admin_client.get(
"/admin/api/auth/me", headers={"Authorization": f"Bearer {user_token}"}
)
assert r.status_code == 401
def test_admin_token_cannot_access_app_api(client: TestClient, admin_client: TestClient) -> None:
"""反向:admin token 也不能访问 App 用户接口(App 用 JWT_SECRET_KEY 验签,admin token 失败)。"""
username, password = _ensure_admin()
r = admin_client.post(
"/admin/api/auth/login", json={"username": username, "password": password}
)
admin_token = r.json()["access_token"]
r = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin_token}"})
assert r.status_code == 401
# ============================ 回归:review 修掉的两个 bug ============================
def test_long_password_does_not_crash(admin_client: TestClient) -> None:
""">72 UTF-8 字节的密码(如多个中文)不能让建账号/登录崩(bcrypt 72 字节限制)。"""
username = "longpw_admin"
long_pw = "超长密码测试" * 8 # 6 中文 ×8 = 48 字 ≈ 144 字节 UTF-8,远超 72
_ensure_admin(username=username, password=long_pw) # 建账号不应抛 ValueError
r = admin_client.post(
"/admin/api/auth/login", json={"username": username, "password": long_pw}
)
assert r.status_code == 200, r.text
def test_audit_log_pagination_no_gap() -> None:
"""审计游标分页跨页不丢/不重(回归 next_cursor off-by-one)。"""
from app.admin.repositories import admin_user as admin_repo
from app.admin.repositories import audit_log as audit_repo
_ensure_admin() # 确保有 test_admin 供 FK 引用
db = SessionLocal()
try:
admin = admin_repo.get_by_username(db, "test_admin")
action = "test.pagination.probe"
created_ids = []
for i in range(5):
log = audit_repo.add_audit_log(
db, admin_id=admin.id, admin_username=admin.username,
action=action, target_type="probe", target_id=str(i),
)
created_ids.append(log.id)
# limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重)
seen: list[int] = []
cursor = None
for _ in range(10): # 上限防死循环
rows, cursor = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor)
seen.extend(r.id for r in rows)
if cursor is None:
break
assert sorted(seen) == sorted(created_ids), f"分页丢/重: want={created_ids} got={seen}"
finally:
db.close()
+124
View File
@@ -0,0 +1,124 @@
"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.db.session import SessionLocal
from app.models.feedback import Feedback
from app.models.wallet import CashTransaction, WithdrawOrder
from app.repositories import user as user_repo
from app.repositories import wallet as wallet_repo
@pytest.fixture()
def admin_client() -> TestClient:
return TestClient(admin_app)
@pytest.fixture()
def admin_token() -> str:
db = SessionLocal()
try:
if admin_repo.get_by_username(db, "m2_admin") is None:
admin_repo.create_admin(db, username="m2_admin", password="m2-pass", role="super_admin")
finally:
db.close()
c = TestClient(admin_app)
r = c.post("/admin/api/auth/login", json={"username": "m2_admin", "password": "m2-pass"})
return r.json()["access_token"]
def _auth(token: str) -> dict:
return {"Authorization": f"Bearer {token}"}
def _seed_user_with_data(phone: str) -> int:
"""造一个用户 + 金币流水 + 现金流水 + 提现单 + 反馈,返回 user_id。"""
db = SessionLocal()
try:
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms")
uid = user.id
wallet_repo.grant_coins(db, uid, 5000, biz_type="signin", remark="测试发金币")
db.commit()
db.add(CashTransaction(
user_id=uid, amount_cents=-100, balance_after_cents=0, biz_type="withdraw", remark="t"
))
db.add(WithdrawOrder(
user_id=uid, out_bill_no=f"test{uid}billno0001", amount_cents=100, status="success"
))
db.add(Feedback(user_id=uid, content="测试反馈内容", contact="wx_test", status="new"))
db.commit()
return uid
finally:
db.close()
def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
_seed_user_with_data("13800000001")
r = admin_client.get("/admin/api/stats/overview", headers=_auth(admin_token))
assert r.status_code == 200, r.text
data = r.json()
assert data["users"]["total"] >= 1
assert data["coins"]["granted_total"] >= 5000
assert "success_rate" in data["comparison"]
assert data["cps"]["available"] is False
def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
uid = _seed_user_with_data("13800000002")
r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
assert r.status_code == 200, r.text
assert "items" in r.json()
r = admin_client.get(f"/admin/api/users/{uid}", headers=_auth(admin_token))
assert r.status_code == 200, r.text
d = r.json()
assert d["user"]["id"] == uid
assert d["coin_balance"] == 5000
assert d["withdraw_total"] >= 1
assert d["feedback_total"] >= 1
assert admin_client.get("/admin/api/users/999999", headers=_auth(admin_token)).status_code == 404
def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
_seed_user_with_data("13800000003")
r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
assert r.status_code == 200
assert all(u["status"] == "active" for u in r.json()["items"])
def test_wallet_and_withdraw_lists(admin_client: TestClient, admin_token: str) -> None:
uid = _seed_user_with_data("13800000004")
r = admin_client.get(
"/admin/api/wallet/coin-transactions", params={"user_id": uid}, headers=_auth(admin_token)
)
assert r.status_code == 200
assert len(r.json()["items"]) >= 1
r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token))
assert r.status_code == 200
assert len(r.json()["items"]) >= 1
assert all(o["status"] == "success" for o in r.json()["items"])
def test_feedback_list(admin_client: TestClient, admin_token: str) -> None:
_seed_user_with_data("13800000005")
r = admin_client.get("/admin/api/feedbacks", params={"status": "new"}, headers=_auth(admin_token))
assert r.status_code == 200
assert all(f["status"] == "new" for f in r.json()["items"])
def test_read_apis_require_auth(admin_client: TestClient) -> None:
"""所有 M2 读接口未带 token → 401(router 级 get_current_admin 守卫)。"""
for path in [
"/admin/api/stats/overview",
"/admin/api/users",
"/admin/api/wallet/coin-transactions",
"/admin/api/withdraws",
"/admin/api/feedbacks",
]:
assert admin_client.get(path).status_code == 401, path
+278
View File
@@ -0,0 +1,278 @@
"""Admin M3 写接口测试:调金币/封号/反馈/提现/admin账号。
验证:写操作落审计 + 金币写流水 + 扣负拒绝 + 角色守卫 + 提现复用 wallet(mock wxpay)
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
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.user import User
from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder
from app.repositories import user as user_repo
@pytest.fixture()
def admin_client() -> TestClient:
return TestClient(admin_app)
def _token(username: str, role: str) -> str:
db = SessionLocal()
try:
a = admin_repo.get_by_username(db, username)
if a is None:
admin_repo.create_admin(db, username=username, password="pass1234", role=role)
else:
a.password_hash = hash_password("pass1234")
a.role = role
a.status = "active"
db.commit()
finally:
db.close()
c = TestClient(admin_app)
return c.post(
"/admin/api/auth/login", json={"username": username, "password": "pass1234"}
).json()["access_token"]
@pytest.fixture()
def super_token() -> str:
return _token("w_super", "super_admin")
@pytest.fixture()
def finance_token() -> str:
return _token("w_finance", "finance")
@pytest.fixture()
def operator_token() -> str:
return _token("w_operator", "operator")
def _auth(token: str) -> dict:
return {"Authorization": f"Bearer {token}"}
def _seed_user(phone: str) -> int:
db = SessionLocal()
try:
return user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms").id
finally:
db.close()
def _seed_feedback(phone: str) -> int:
uid = _seed_user(phone)
db = SessionLocal()
try:
fb = Feedback(user_id=uid, content="测试", contact="wx", status="new")
db.add(fb)
db.commit()
return fb.id
finally:
db.close()
# ===== 调金币 =====
def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
uid = _seed_user("13900000001")
r = admin_client.post(
f"/admin/api/users/{uid}/coins", json={"amount": 1000, "reason": "补偿"},
headers=_auth(finance_token),
)
assert r.status_code == 200, r.text
db = SessionLocal()
try:
assert db.get(CoinAccount, uid).coin_balance == 1000
txns = db.execute(
select(CoinTransaction).where(
CoinTransaction.user_id == uid, CoinTransaction.biz_type == "admin_grant"
)
).scalars().all()
assert len(txns) == 1 and txns[0].amount == 1000
logs = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "user.coins.grant", AdminAuditLog.target_id == str(uid)
)
).scalars().all()
assert len(logs) == 1 and logs[0].detail["amount"] == 1000
finally:
db.close()
def test_deduct_below_zero_rejected(admin_client: TestClient, finance_token: str) -> None:
uid = _seed_user("13900000002")
r = admin_client.post(
f"/admin/api/users/{uid}/coins", json={"amount": -999999, "reason": ""},
headers=_auth(finance_token),
)
assert r.status_code == 400
db = SessionLocal()
try:
txns = db.execute(
select(CoinTransaction).where(CoinTransaction.user_id == uid)
).scalars().all()
assert len(txns) == 0 # 拒绝后无流水(原子:都不发生)
finally:
db.close()
def test_grant_zero_rejected(admin_client: TestClient, finance_token: str) -> None:
uid = _seed_user("13900000008")
r = admin_client.post(
f"/admin/api/users/{uid}/coins", json={"amount": 0, "reason": "x"},
headers=_auth(finance_token),
)
assert r.status_code == 400
# ===== 封号 =====
def test_set_user_status_and_audit(admin_client: TestClient, operator_token: str) -> None:
uid = _seed_user("13900000003")
r = admin_client.post(
f"/admin/api/users/{uid}/status", json={"status": "disabled"}, headers=_auth(operator_token)
)
assert r.status_code == 200, r.text
db = SessionLocal()
try:
assert db.get(User, uid).status == "disabled"
logs = db.execute(
select(AdminAuditLog).where(
AdminAuditLog.action == "user.status.set", AdminAuditLog.target_id == str(uid)
)
).scalars().all()
assert logs[0].detail == {"before": "active", "after": "disabled"}
finally:
db.close()
assert admin_client.post(
f"/admin/api/users/{uid}/status", json={"status": "active"}, headers=_auth(operator_token)
).status_code == 200
# ===== 角色守卫 =====
def test_operator_cannot_grant_coins(admin_client: TestClient, operator_token: str) -> None:
uid = _seed_user("13900000004")
r = admin_client.post(
f"/admin/api/users/{uid}/coins", json={"amount": 100, "reason": "x"},
headers=_auth(operator_token),
)
assert r.status_code == 403
def test_finance_cannot_manage_admins(admin_client: TestClient, finance_token: str) -> None:
assert admin_client.get("/admin/api/admins", headers=_auth(finance_token)).status_code == 403
def test_super_admin_can_grant_coins(admin_client: TestClient, super_token: str) -> None:
uid = _seed_user("13900000005")
r = admin_client.post(
f"/admin/api/users/{uid}/coins", json={"amount": 50, "reason": "x"},
headers=_auth(super_token),
)
assert r.status_code == 200
# ===== 反馈处理 =====
def test_handle_feedback(admin_client: TestClient, operator_token: str) -> None:
fid = _seed_feedback("13900000006")
r = admin_client.post(f"/admin/api/feedbacks/{fid}/handle", headers=_auth(operator_token))
assert r.status_code == 200
db = SessionLocal()
try:
assert db.get(Feedback, fid).status == "handled"
finally:
db.close()
# ===== admin 账号管理(super_admin) =====
def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:
r = admin_client.post(
"/admin/api/admins",
json={"username": "new_op", "password": "pass1234", "role": "operator"},
headers=_auth(super_token),
)
assert r.status_code == 200, r.text
new_id = r.json()["id"]
r = admin_client.patch(
f"/admin/api/admins/{new_id}", json={"role": "finance"}, headers=_auth(super_token)
)
assert r.status_code == 200 and r.json()["role"] == "finance"
r = admin_client.post(
"/admin/api/admins", json={"username": "new_op", "password": "pass1234"},
headers=_auth(super_token),
)
assert r.status_code == 409
def test_cannot_disable_self(admin_client: TestClient, super_token: str) -> None:
me = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
r = admin_client.patch(
f"/admin/api/admins/{me['id']}", json={"status": "disabled"}, headers=_auth(super_token)
)
assert r.status_code == 400
# ===== 提现重试 / 对账(mock wxpay,不真调微信) =====
def test_withdraw_refresh(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
uid = _seed_user("13900000007")
db = SessionLocal()
try:
db.add(WithdrawOrder(
user_id=uid, out_bill_no="adminrefresh0001", amount_cents=100, status="pending"
))
db.commit()
finally:
db.close()
from app.repositories import wallet as wr
monkeypatch.setattr(
wr.wxpay, "query_transfer",
lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
)
r = admin_client.post(
"/admin/api/withdraws/adminrefresh0001/refresh", headers=_auth(finance_token)
)
assert r.status_code == 200, r.text
assert r.json()["status"] == "success"
db = SessionLocal()
try:
logs = db.execute(
select(AdminAuditLog).where(AdminAuditLog.action == "withdraw.refresh")
).scalars().all()
assert any(x.target_id == "adminrefresh0001" for x in logs)
finally:
db.close()
def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> None:
assert admin_client.post(
"/admin/api/withdraws/nope999notexist/refresh", headers=_auth(finance_token)
).status_code == 404
def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
from app.repositories import wallet as wr
monkeypatch.setattr(
wr.wxpay, "query_transfer",
lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
)
r = admin_client.post(
"/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
headers=_auth(finance_token),
)
assert r.status_code == 200, r.text
assert "checked" in r.json() and "resolved" in r.json()
+156 -18
View File
@@ -1,48 +1,68 @@
"""auth 流程测试(不依赖真实极光,只测 sms mock 路径 + JWT 闭环)。"""
"""auth 流程测试
- mock 路径(SMS_MOCK=true,conftest 设置):验证码不真发,校验放行任意 6 , JWT 闭环
- real 路径:monkeypatch SMS_MOCK=false + httpx,测极光发送协议 + 真实校验(一次性/
防爆破/余额回滚),不真发短信
"""
from __future__ import annotations
import time
import pytest
from app.integrations import sms
def _reset(phone: str) -> None:
"""清该号的进程内存状态,隔离 real 模式用例。"""
sms._codes.pop(phone, None)
sms._last_sent.pop(phone, None)
sms._daily_count.pop(phone, None)
class _OkResp:
"""模拟极光发送成功响应。"""
status_code = 200
def json(self):
return {"msg_id": "test-msg-id"}
# ============================ mock 路径(JWT 闭环)============================
def test_sms_login_and_me_flow(client) -> None:
"""sms_send → sms_login → 用 access_token 调 /me 拿到自己"""
"""sms_send → sms_login(mock 任意6位)→ /me → refresh/logout 全闭环"""
phone = "13800138000"
# 1. 发短信(mock,任意 6 位均通过)
r = client.post("/api/v1/auth/sms/send", json={"phone": phone})
assert r.status_code == 200, r.text
body = r.json()
assert body["sent"] is True
assert body["mock"] is True
# 2. 登录
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
assert r.status_code == 200, r.text
data = r.json()
assert "access_token" in data
assert "refresh_token" in data
assert "access_token" in data and "refresh_token" in data
assert data["user"]["phone"] == phone
assert data["user"]["register_channel"] == "sms"
access = data["access_token"]
refresh = data["refresh_token"]
access, refresh = data["access_token"], data["refresh_token"]
# 3. /me
r = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
assert r.status_code == 200, r.text
assert r.json()["phone"] == phone
# 4. /me 不带 token → 401
r = client.get("/api/v1/auth/me")
assert r.status_code == 401
# 5. 用 refresh 换新 token
r = client.post("/api/v1/auth/refresh", json={"refresh_token": refresh})
assert r.status_code == 200, r.text
assert "access_token" in r.json()
# 6. 用 access 当 refresh 拒绝(防 token 类型混用)
# 用 access 当 refresh 拒绝(防 token 类型混用)
r = client.post("/api/v1/auth/refresh", json={"refresh_token": access})
assert r.status_code == 401
# 7. logout
r = client.post("/api/v1/auth/logout", headers={"Authorization": f"Bearer {access}"})
assert r.status_code == 200
assert r.json()["ok"] is True
@@ -50,14 +70,12 @@ def test_sms_login_and_me_flow(client) -> None:
def test_sms_send_too_frequent(client) -> None:
phone = "13900139000"
r1 = client.post("/api/v1/auth/sms/send", json={"phone": phone})
assert r1.status_code == 200
r2 = client.post("/api/v1/auth/sms/send", json={"phone": phone})
assert r2.status_code == 429
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 429
def test_sms_login_bad_code(client) -> None:
"""mock 模式下,5 位过 schema 长度校验,但业务层只接受 6 位 → 400。"""
"""mock 下校验位数:5 位过 schema 但业务要 6 位 → 400。"""
phone = "13700137000"
client.post("/api/v1/auth/sms/send", json={"phone": phone})
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "12345"})
@@ -68,3 +86,123 @@ def test_sms_login_bad_code(client) -> None:
def test_phone_format_validation(client) -> None:
r = client.post("/api/v1/auth/sms/send", json={"phone": "1234567"})
assert r.status_code == 422
def test_sms_daily_limit(monkeypatch) -> None:
"""单号每日上限(send 的防刷逻辑 mock/real 都跑;函数级绕开 60s 冷却)。"""
phone = "13511135000"
_reset(phone)
limit = 3
monkeypatch.setattr(sms.settings, "SMS_DAILY_LIMIT_PER_PHONE", limit)
for _ in range(limit):
sms.send_code(phone)
sms._last_sent.pop(phone, None) # 绕开冷却,只测每日上限
with pytest.raises(sms.SmsError, match="上限"):
sms.send_code(phone)
# ============================ real 路径(不真发)============================
def test_sms_real_send_calls_jiguang(monkeypatch) -> None:
"""real 模式:本服务生成 code 并按协议调极光 /v1/messages(拦 httpx,不真发)。"""
phone = "13422134000"
_reset(phone)
captured: dict = {}
def _fake_post(url, json, headers, timeout): # noqa: A002
captured.update(url=url, body=json, auth=headers.get("Authorization", ""))
return _OkResp()
monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
monkeypatch.setattr(sms.httpx, "post", _fake_post)
sms.send_code(phone)
assert captured["url"] == sms.settings.SMS_SEND_ENDPOINT
assert captured["body"]["mobile"] == phone
assert captured["body"]["sign_id"] == sms.settings.SMS_SIGN_ID
assert captured["body"]["temp_id"] == sms.settings.SMS_TEMPLATE_ID
assert captured["body"]["temp_para"]["code"] == sms._codes[phone].code
assert captured["auth"].startswith("Basic ")
def test_sms_real_verify_one_time_and_wrong(monkeypatch) -> None:
"""real 校验:错误码拒(不消费)→ 正确码成功 → 验过即作废。"""
phone = "13455134000"
_reset(phone)
monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _OkResp())
sms.send_code(phone)
code = sms._codes[phone].code
wrong = "000000" if code != "000000" else "111111"
assert sms.verify_code(phone, wrong) is False
assert sms.verify_code(phone, code) is True
assert sms.verify_code(phone, code) is False # 已作废
def test_sms_real_verify_attempts_exhausted(monkeypatch) -> None:
"""real 校验:错误次数到上限即作废,正确码也不再通过(防爆破)。"""
phone = "13466134000"
_reset(phone)
monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
monkeypatch.setattr(sms.settings, "SMS_MAX_VERIFY_ATTEMPTS", 3)
monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _OkResp())
sms.send_code(phone)
code = sms._codes[phone].code
wrong = "000000" if code != "000000" else "111111"
for _ in range(3):
assert sms.verify_code(phone, wrong) is False
assert sms.verify_code(phone, code) is False # 超限作废
def test_sms_real_balance_error_keeps_cooldown(monkeypatch) -> None:
"""极光余额不足(50014)→ SmsError(503);保留冷却(失败也限速,防重试狂打),只清未发出的码。"""
phone = "13433134000"
_reset(phone)
class _ErrResp:
status_code = 403
def json(self):
return {"error": {"code": 50014, "message": "no money"}}
monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _ErrResp())
with pytest.raises(sms.SmsError) as ei:
sms.send_code(phone)
assert ei.value.status_code == 503
assert phone not in sms._codes # 没发出去的码已清
assert phone in sms._last_sent # 冷却保留:失败也限速
# 立即重试 → 被冷却挡下(429),不会再打极光
with pytest.raises(sms.SmsError) as ei2:
sms.send_code(phone)
assert ei2.value.status_code == 429
def test_sms_gc_purges_stale_only(monkeypatch) -> None:
"""GC 清过期码 / 旧冷却 / 隔日计数,但不动今天有效的(阈值设 0 强制每次扫)。"""
monkeypatch.setattr(sms, "_GC_THRESHOLD", 0)
sms._codes.clear()
sms._last_sent.clear()
sms._daily_count.clear()
now = time.time()
sms._codes["stale"] = sms._CodeRecord(code="111111", expires_at=now - 1)
sms._codes["fresh"] = sms._CodeRecord(code="222222", expires_at=now + 999)
sms._last_sent["old"] = now - 99999
sms._last_sent["recent"] = now
sms._daily_count["yesterday"] = ("2000-01-01", 3)
sms._daily_count["today"] = (sms._today(), 1)
sms._gc(now)
assert "stale" not in sms._codes and "fresh" in sms._codes
assert "old" not in sms._last_sent and "recent" in sms._last_sent
assert "yesterday" not in sms._daily_count and "today" in sms._daily_count
+4 -4
View File
@@ -82,8 +82,8 @@ def test_unlock_progresses_with_success(client) -> None:
assert states[2] == "locked"
def test_claim_grants_coin_and_becomes_claimed(client) -> None:
"""领第 1 档:发对应金币,进金币余额,该档变 claimed。"""
def test_claim_marks_claimed_without_coin(client) -> None:
"""领第 1 档:暂不真发金币(产品定,后续删该功能),余额不变,但该档变 claimed。"""
token = _login(client, "13800003003")
_report_success(client, token, "s-1")
@@ -93,8 +93,8 @@ def test_claim_grants_coin_and_becomes_claimed(client) -> None:
assert r.status_code == 200, r.text
res = r.json()
assert res["milestone"] == 1
assert res["coin_awarded"] == RECORD_MILESTONES[0]
assert res["coin_balance"] == bal_before + RECORD_MILESTONES[0]
assert res["coin_awarded"] == 0 # 不再发金币
assert res["coin_balance"] == bal_before # 余额不变
d = client.get("/api/v1/compare/milestones", headers=_auth(token)).json()
assert d["milestones"][0]["state"] == "claimed"
+13 -1
View File
@@ -33,7 +33,7 @@ def _food_payload(trace_id: str) -> dict:
],
"comparison_results": [
{"platform_id": "meituan", "platform_name": "美团", "package": "com.sankuai.meituan",
"price": 123.50, "is_source": False, "rank": 1},
"price": 123.50, "is_source": False, "rank": 1, "coupon_saved": 7.0},
{"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "package": "com.taobao.taobao",
"price": 128.50, "is_source": True, "rank": 2},
{"platform_id": "jd_waimai", "platform_name": "京东外卖", "package": "com.jingdong.app.mall",
@@ -78,6 +78,18 @@ def test_report_and_derive(client) -> None:
assert d["raw_payload"]["trace_id"] == "trace-1"
def test_coupon_saved_passthrough(client) -> None:
"""红包优惠额 coupon_saved 上报→落库→读出;仅目标平台带值,源平台不带(null)。"""
token = _login(client, "13800002010")
rid = client.post(
"/api/v1/compare/record", json=_food_payload("trace-coupon"), headers=_auth(token)
).json()["id"]
d = client.get(f"/api/v1/compare/records/{rid}", headers=_auth(token)).json()
by_pid = {x["platform_id"]: x for x in d["comparison_results"]}
assert by_pid["meituan"]["coupon_saved"] == 7.0 # 目标平台带值
assert by_pid["taobao_flash"].get("coupon_saved") is None # 源平台不带/为 null
def test_source_is_cheapest_no_saving(client) -> None:
"""源平台本来就最便宜:saved<=0,is_source_best=True,仍记一条 success。"""
token = _login(client, "13800002002")
+16 -4
View File
@@ -12,6 +12,9 @@ from app.core.rewards import (
TASK_ENABLE_NOTIFICATION,
TASK_REWARDS,
)
from app.db.session import SessionLocal
from app.repositories import wallet as crud_wallet
from app.repositories.user import get_user_by_phone
def _login(client, phone: str) -> str:
@@ -130,10 +133,19 @@ def test_exchange_info(client) -> None:
def test_exchange_flow(client) -> None:
"""用任务领够金币 → 兑换 1 元 → 金币扣、现金加 → 现金流水有记录。"""
token = _login(client, "13800001005")
# 领"打开消息提醒"得 10000 金币(= 1 元额度)
client.post(f"/api/v1/tasks/{TASK_ENABLE_NOTIFICATION}/claim", headers=_auth(token))
"""供款够兑换下限的金币 → 兑换 1 元 → 金币扣、现金加 → 现金流水有记录。
打开消息提醒任务已降到 1000 金币(不再 = 兑换下限), 不能再靠领任务供款;
直接 grant_coins 注入 MIN_EXCHANGE_COIN(= COIN_PER_YUAN = 10000)当种子
"""
phone = "13800001005"
token = _login(client, phone)
# 供款: 直接注入 1 元额度金币(替代原先靠 notification=10000 领任务供款)
with SessionLocal() as db:
user = get_user_by_phone(db, phone)
assert user is not None
crud_wallet.grant_coins(db, user.id, MIN_EXCHANGE_COIN, biz_type="test_seed", remark="测试供款")
db.commit()
# 兑换 10000 金币 → 100 分
r = client.post(