Compare commits

..

9 Commits

Author SHA1 Message Date
marco d46a5cd80f chore: alembic merge revision + PG 迁移指南文档
PostgreSQL 迁移 (#6) 之后两个分支引入了独立 heads (ef96beb47b1e +
d1e2f3a4b5c6), 必须合并成单一 head 才能 alembic upgrade head 顺利跑通。
- alembic/versions/f01db5d77dac_merge_pg_jsonb_and_feedback_heads.py:
  空 op 的 merge revision, 把两个 heads 收敛为 f01db5d77dac
- docs/postgres-migration.md: SQLite → PostgreSQL 迁移指南
  (背景/动机/操作步骤/无真实用户数据假设, 上线前补 pgloader 演练等)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-30 18:08:51 +08:00
marco e69e244de7 Merge pull request 'feat(ad-reward): reward-status 加本轮 3 次冷却字段 (round_count + cooldown_until)' (#5) from feat/welfare-video-round-cooldown into main
Reviewed-on: #5
2026-05-29 15:45:51 +08:00
marco 2ebde935f9 Merge pull request 'feat: SQLite → PostgreSQL 迁移代码改动' (#6) from feat/postgres-migration-and-bootstrap-script into main
Reviewed-on: #6
2026-05-29 15:44:40 +08:00
chenshuobo b3d3fda744 feat: 新增 PG 初始化脚本 scripts/init_postgres.py
新机器一键拉起数据库, 替代手动跑 psql 建用户建库的 6 行命令.

脚本特性:
- 跨平台 Python (Windows/macOS/Linux 通吃, 不依赖 psql CLI 在 PATH 里)
- 直接用 psycopg 连 PG, 不 shell out
- 幂等: 用户已存在则重置密码, 库已存在则跳过, 重复跑不出错
- 业务密码自动 secrets.token_urlsafe(32) 强随机, 不让同事自己想
- 自动从 .env.example 复制出 .env 并写入 DATABASE_URL
- .env 里已有 DATABASE_URL 会询问是否覆盖 (不无脑覆盖)
- 跑完自动 alembic upgrade head, 一条命令到可用状态
- 全交互项支持环境变量提前指定 (PG_SUPER_PASS / APP_DB_PASS 等), CI 友好

同事新机器初始化流程从 6 步简化为 3 步:
  pip install -e .
  python scripts/init_postgres.py    # 唯一手动输入项: postgres 超级用户密码
  ./run.sh

CREATE/ALTER USER 注意点: PG 不支持参数化密码绑定, 用 sql.Literal 内联,
psycopg 自带的转义会处理引号和注入.

docs/数据库迁移.md 拆出 PG / SQLite 两套初始化流程, PG 走脚本,
SQLite 保留旧步骤 (本地临时回退用).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-29 15:21:04 +08:00
chenshuobo b76e5bd515 feat: SQLite → PostgreSQL 迁移代码改动
按 docs/数据库迁移.md 第 3 节扫雷清单落地代码层改动:

- pyproject.toml: 新增 psycopg[binary]>=3.1 依赖 (psycopg3, SQLAlchemy 2.0 时代默认, 不要再装 psycopg2)

- app/db/session.py: 非 SQLite 自动加连接池参数
  - pool_size=10, max_overflow=20, pool_recycle=3600
  - SQLite 单文件不池化, _is_sqlite 判断分支保留以便本地 dev 临时回退

- app/models/savings.py: dishes 列从 JSON 改为 JSONB
  - PG 上能用 GIN 索引和 jsonb 操作符
  - SQLite 上 JSON 实际是 TEXT, 切到 PG 后用 JSON 类型存的是 json 不是 jsonb

- alembic/versions/ef96beb47b1e_*.py: 新增迁移把 savings_record.dishes 从 json 转 jsonb
  - 用 op.get_bind().dialect.name 判断 PG 才执行 ALTER COLUMN
  - SQLite 上是 no-op (SQLite 没有 jsonb 类型)
  - 旧的 savings_shop_dishes.py 迁移保持 sa.JSON() 不动 (按文档"不改老迁移"原则)

本地切 PG 验证: 10 张表全部建出, dishes 字段为 jsonb 类型,
登录/me/美团 feed 三个真接口全 200, user 数据正确写入 PG.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-29 15:20:22 +08:00
OuYingJun1024 56f4548654 feat(ad-reward): reward-status 加本轮 3 次冷却字段 (round_count + cooldown_until)
让客户端 4 态任务行 CTA (Normal/Loading/Capped/CoolingDown) 与 ExitReward
弹窗 limit note (本轮看完/今日看完 二选一) 都由后端权威派生, 跨设备 / 重装 /
杀进程都一致, 客户端不再依赖 in-memory 本地存储。

- core/rewards.py: 加 VIDEO_ROUND_REQUIRED_COUNT=3 / VIDEO_ROUND_COOLDOWN_SECONDS=600
- repositories/ad_reward.py: today_status 从 3-tuple 扩成 5-tuple,
  从已有 created_at 字段按 ORDER BY DESC + OFFSET round_count 派生最近一个
  已完成轮末尾时间; 零新字段零新表
- schemas/ad.py: AdRewardStatusOut + TestGrantOut 加 round_count + cooldown_until
- api/v1/ad.py: reward_status / test_grant 接 5-tuple
- tests: 加 4 个覆盖 (初始 / 未达一轮 / 刚达一轮进冷却 / 冷却已过)
- docs/api/ad-reward-status.md: 补字段说明 + 客户端 4 态 CTA 推导规则 +
  "冷却仅 UX, 发奖不受影响" 注解

冷却是 UX 约束: pangle-callback 发奖只看 daily_limit, 冷却期间穿山甲
回调照常发奖, 不另设拒发分支。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-29 13:27:56 +08:00
marco 4d444b7c43 Merge pull request 'feat(user): 用户资料与反馈接口(昵称/头像/注销 + 帮助反馈)' (#4) from feat/user-profile-feedback-api into main
Reviewed-on: #4
2026-05-29 01:04:10 +08:00
陈世睿 49379fd045 feat(user): 用户资料与反馈接口(昵称/头像/注销 + 帮助反馈)
- /api/v1/user: PATCH 改昵称、POST 上传头像、DELETE 注销账号(软删除+匿名化)
- /api/v1/feedback: 提交反馈(内容/联系方式/截图入库)
- 新增 media 模块(图片落盘+魔数校验)+ /media 静态服务
- feedback 表 + Alembic 迁移(部署需 alembic upgrade head)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-28 22:27:36 +08:00
marco abd8eabf9b feat(meituan): 未配置 MT_CPS_APP_KEY 时接口降级返空,不再 502
之前未配美团 CPS 凭证时,/coupons 和 /referral-link 调用会抛 MeituanCpsError
被翻成 502,前端首页一进就拉 /feed → 看到错误(虽然 /feed 因 _fetch_topic
吞异常已经返空)。新开发机没填美团 key 时体验差。

本次让美团 3 个接口都在入口处早返空响应:
- Settings 新增 mt_cps_configured property(同 wxpay_configured 套路)
- /coupons / /feed / /referral-link 未配置时分别返空 list / 空 link
- 客户端首页 feed 看到空列表,其他业务(登录/领券/广告/提现)完全不受影响

副作用:/feed 返空现在同时表示"配置缺失"和"美团真无券"两种状态,排查时
看 settings.mt_cps_configured 区分。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-28 15:54:51 +08:00
29 changed files with 1207 additions and 367 deletions
@@ -0,0 +1,36 @@
"""convert savings_record.dishes from json to jsonb
PG only. SQLite 上 JSON/JSONB 都是 TEXT, 此迁移是 no-op。
Revision ID: ef96beb47b1e
Revises: c8d9e0f1a2b3
Create Date: 2026-05-29 10:57:21.471774
"""
from typing import Sequence, Union
from alembic import op
from sqlalchemy.dialects.postgresql import JSONB
revision: str = 'ef96beb47b1e'
down_revision: Union[str, Sequence[str], None] = 'c8d9e0f1a2b3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
if op.get_bind().dialect.name != "postgresql":
return
op.execute(
"ALTER TABLE savings_record "
"ALTER COLUMN dishes TYPE JSONB USING dishes::JSONB"
)
def downgrade() -> None:
if op.get_bind().dialect.name != "postgresql":
return
op.execute(
"ALTER TABLE savings_record "
"ALTER COLUMN dishes TYPE JSON USING dishes::JSON"
)
@@ -0,0 +1,26 @@
"""merge pg_jsonb and feedback heads
Revision ID: f01db5d77dac
Revises: ef96beb47b1e, d1e2f3a4b5c6
Create Date: 2026-05-29 16:09:58.498935
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'f01db5d77dac'
down_revision: Union[str, Sequence[str], None] = ('ef96beb47b1e', 'd1e2f3a4b5c6')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
+44
View File
@@ -0,0 +1,44 @@
"""feedback table (用户帮助与反馈)
Revision ID: d1e2f3a4b5c6
Revises: c8d9e0f1a2b3
Create Date: 2026-05-28 12:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'd1e2f3a4b5c6'
down_revision: Union[str, Sequence[str], None] = 'c8d9e0f1a2b3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
'feedback',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('content', sa.Text(), nullable=False),
sa.Column('contact', sa.String(length=128), nullable=False),
sa.Column('images', sa.JSON(), nullable=True),
sa.Column('status', sa.String(length=16), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
with op.batch_alter_table('feedback', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_feedback_user_id'), ['user_id'], unique=False)
batch_op.create_index(batch_op.f('ix_feedback_created_at'), ['created_at'], unique=False)
def downgrade() -> None:
with op.batch_alter_table('feedback', schema=None) as batch_op:
batch_op.drop_index(batch_op.f('ix_feedback_created_at'))
batch_op.drop_index(batch_op.f('ix_feedback_user_id'))
op.drop_table('feedback')
+6 -2
View File
@@ -84,12 +84,14 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
@router.get("/reward-status", response_model=AdRewardStatusOut, summary="今日看广告发奖进度")
def reward_status(user: CurrentUser, db: DbSession) -> AdRewardStatusOut:
used, limit, coin_per = crud_ad.today_status(db, user.id)
used, limit, coin_per, round_count, cooldown_until = crud_ad.today_status(db, user.id)
return AdRewardStatusOut(
used_today=used,
daily_limit=limit,
remaining=max(0, limit - used),
coin_per_ad=coin_per,
round_count=round_count,
cooldown_until=cooldown_until,
)
@@ -118,7 +120,7 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
except crud_ad.UnknownUserError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
used, limit, coin_per = crud_ad.today_status(db, user.id)
used, limit, coin_per, round_count, cooldown_until = crud_ad.today_status(db, user.id)
logger.info("ad TEST grant user_id=%d status=%s coin=%d", user.id, rec.status, rec.coin)
return TestGrantOut(
granted=(rec.status == "granted"),
@@ -128,4 +130,6 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
daily_limit=limit,
remaining=max(0, limit - used),
coin_per_ad=coin_per,
round_count=round_count,
cooldown_until=cooldown_until,
)
+63
View File
@@ -0,0 +1,63 @@
"""帮助与反馈 endpoint。
路由前缀 `/api/v1/feedback`,需 Bearer 鉴权(反馈绑到登录用户,便于回访)。
POST / 提交反馈(multipart:content / contact 必填,images 可选 ≤4 张)
截图复用 [app.core.media] 落盘到 /media/feedback/。
"""
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.repositories import feedback as feedback_repo
from app.schemas.feedback import FeedbackOut
logger = logging.getLogger("shagua.feedback")
router = APIRouter(prefix="/api/v1/feedback", tags=["feedback"])
_MAX_IMAGES = 4
_CONTENT_MAX = 2000
_CONTACT_MAX = 128
@router.post("", response_model=FeedbackOut, summary="提交反馈")
async def submit_feedback(
user: CurrentUser,
db: DbSession,
content: str = Form(...),
contact: str = Form(...),
images: list[UploadFile] = File(default=[]),
) -> FeedbackOut:
content = content.strip()
contact = contact.strip()
if not content:
raise HTTPException(status_code=400, detail="反馈内容不能为空")
if len(content) > _CONTENT_MAX:
raise HTTPException(status_code=400, detail="反馈内容过长")
if not contact:
raise HTTPException(status_code=400, detail="联系方式不能为空")
if len(contact) > _CONTACT_MAX:
raise HTTPException(status_code=400, detail="联系方式过长")
files = [f for f in (images or []) if f is not None and f.filename]
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_feedback_image(user.id, data))
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
fb = feedback_repo.create_feedback(
db, user_id=user.id, content=content, contact=contact, images=urls,
)
logger.info("feedback id=%d user_id=%d images=%d", fb.id, user.id, len(urls))
return FeedbackOut.model_validate(fb)
+52 -153
View File
@@ -5,22 +5,12 @@
from __future__ import annotations
import logging
import time
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
from concurrent.futures import ThreadPoolExecutor, as_completed
from fastapi import APIRouter, HTTPException
from app.core.ranking import (
dedup,
filter_items,
get_distance_km,
inject_billboard,
merge_category_pages,
shuffle_pages,
sort_by_sales,
split_pages,
)
from app.integrations.meituan import MeituanCpsError, _call as mt_call, get_referral_link, query_coupon
from app.core.config import settings
from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon
from app.schemas.meituan import (
CouponCard,
CouponListRequest,
@@ -35,16 +25,11 @@ logger = logging.getLogger("shagua.meituan")
router = APIRouter(prefix="/api/v1/meituan", tags=["meituan-cps"])
_MAX_RECALL_PAGES = 5
_MAX_DISTANCE_KM = 8.0
_FEED_CACHE_TTL = 300
_feed_cache: dict[str, tuple[float, list[list[dict]]]] = {}
_feed_lock = Lock()
@router.post("/coupons", response_model=CouponListResponse, summary="券列表(通用查询)")
@router.post("/coupons", response_model=CouponListResponse, summary="券列表(为您推荐)")
def list_coupons(req: CouponListRequest) -> CouponListResponse:
if not settings.mt_cps_configured:
return CouponListResponse(items=[], has_next=False, search_id=None)
logger.info("[coupons] lon=%.6f lat=%.6f topic=%s", req.longitude, req.latitude, req.list_topic_id)
try:
raw = query_coupon(
@@ -71,151 +56,65 @@ def list_coupons(req: CouponListRequest) -> CouponListResponse:
)
# ────────────────────── Feed 排序策略 (MVP) ──────────────────────
_TOPIC_ROUNDS = [
(3, 3), # 爆款筛选
(2, 2), # 今日必推
(1, 5), # 精选 + 限时筛选
]
def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
items: list[CouponCard] = []
seen: set[str] = set()
i = j = 0
while i < len(waimai) or j < len(daodian):
for _ in range(2):
if i < len(waimai):
card = CouponCard.from_raw(waimai[i]); i += 1
if card.product_view_sign not in seen:
seen.add(card.product_view_sign); items.append(card)
if j < len(daodian):
card = CouponCard.from_raw(daodian[j]); j += 1
if card.product_view_sign not in seen:
seen.add(card.product_view_sign); items.append(card)
return items
def _lbs_recall(
keyword: str,
lon: float,
lat: float,
is_daodian: bool,
) -> list[dict]:
"""searchText + sortField=6 翻页召回,收集 ≤8km 商品,遇到整页都 >8km 或翻满 10 页停止。"""
lon_i = int(lon * 1_000_000)
lat_i = int(lat * 1_000_000)
all_items: list[dict] = []
search_id: str | None = None
for _ in range(_MAX_RECALL_PAGES):
body: dict = {
"longitude": lon_i,
"latitude": lat_i,
"searchText": keyword,
"sortField": 6,
"pageSize": 20,
}
if search_id:
body["searchId"] = search_id
try:
data = mt_call("/cps_open/common/api/v1/query_coupon", body)
except MeituanCpsError:
break
items = data.get("data") or []
search_id = data.get("searchId")
has_next = data.get("hasNext", False)
if not items:
break
page_has_valid = False
for item in items:
dist = get_distance_km(item, is_daodian)
if dist is not None and dist <= _MAX_DISTANCE_KM:
all_items.append(item)
page_has_valid = True
if not page_has_valid or not has_next:
break
return dedup(all_items)
def _fetch_billboard(
platform: int,
topic_id: int,
lon: float,
lat: float,
) -> list[dict]:
"""listTopiId 拉榜单,固定 20 条,不做距离过滤。"""
try:
data = query_coupon(
longitude=lon,
latitude=lat,
platform=platform,
list_topic_id=topic_id,
)
return data.get("data") or []
except MeituanCpsError:
return []
def _build_feed(lon: float, lat: float) -> list[list[dict]]:
"""完整 pipeline:并发召回 → 过滤 → 销量排序 → 分页 → shuffle → 榜单加成 → 合并。"""
with ThreadPoolExecutor(max_workers=4) as pool:
f_wm = pool.submit(_lbs_recall, "外卖", lon, lat, False)
f_dd = pool.submit(_lbs_recall, "到店餐饮", lon, lat, True)
f_wm_bill = pool.submit(_fetch_billboard, 1, 1, lon, lat)
f_dd_bill = pool.submit(_fetch_billboard, 2, 3, lon, lat)
wm_raw = f_wm.result()
dd_raw = f_dd.result()
wm_billboard = f_wm_bill.result()
dd_billboard = f_dd_bill.result()
logger.info(
"[feed:build] waimai=%d daodian=%d wm_bill=%d dd_bill=%d",
len(wm_raw), len(dd_raw), len(wm_billboard), len(dd_billboard),
)
wm_filtered = filter_items(wm_raw, is_daodian=False, max_km=_MAX_DISTANCE_KM)
dd_filtered = filter_items(dd_raw, is_daodian=True, max_km=_MAX_DISTANCE_KM)
wm_sorted = sort_by_sales(wm_filtered)
dd_sorted = sort_by_sales(dd_filtered)
wm_pages = shuffle_pages(split_pages(wm_sorted))
dd_pages = shuffle_pages(split_pages(dd_sorted))
wm_pages = inject_billboard(wm_pages, wm_billboard, per_page=4, max_inject_pages=5)
dd_pages = inject_billboard(dd_pages, dd_billboard, per_page=4, max_inject_pages=5)
return merge_category_pages(wm_pages, dd_pages)
def _cache_key(lon: float, lat: float) -> str:
return f"{lon:.4f},{lat:.4f}"
def _get_or_build_feed(lon: float, lat: float) -> list[list[dict]]:
key = _cache_key(lon, lat)
now = time.time()
with _feed_lock:
entry = _feed_cache.get(key)
if entry and now - entry[0] < _FEED_CACHE_TTL:
return entry[1]
pages = _build_feed(lon, lat)
with _feed_lock:
_feed_cache[key] = (time.time(), pages)
expired = [k for k, (t, _) in _feed_cache.items() if now - t > _FEED_CACHE_TTL]
for k in expired:
del _feed_cache[k]
return pages
@router.post("/feed", response_model=FeedResponse, summary="首页推荐 feed (MVP 排序策略)")
@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉, 无限流)")
def feed(req: FeedRequest) -> FeedResponse:
lon, lat = req.longitude, req.latitude
if not settings.mt_cps_configured:
return FeedResponse(items=[], has_next=False, page=req.page)
page_idx = req.page - 1
lon, lat = req.longitude, req.latitude
logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
pages = _get_or_build_feed(lon, lat)
def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> list[dict]:
try:
return (query_coupon(
longitude=lon, latitude=lat,
platform=platform, biz_line=biz_line,
list_topic_id=topic, page_size=20,
).get("data") or [])
except MeituanCpsError:
return []
if page_idx >= len(pages):
if page_idx >= len(_TOPIC_ROUNDS):
return FeedResponse(items=[], has_next=False, page=req.page)
items = [CouponCard.from_raw(it) for it in pages[page_idx]]
has_next = page_idx + 1 < len(pages)
wm_topic, dd_topic = _TOPIC_ROUNDS[page_idx]
with ThreadPoolExecutor(max_workers=2) as pool:
f_wm = pool.submit(_fetch_topic, 1, None, wm_topic)
f_dd = pool.submit(_fetch_topic, 2, 1, dd_topic)
waimai, daodian = f_wm.result(), f_dd.result()
items = _interleave(waimai, daodian)
has_next = page_idx + 1 < len(_TOPIC_ROUNDS)
return FeedResponse(items=items, has_next=has_next, page=req.page)
@router.post("/referral-link", response_model=ReferralLinkResponse, summary="换取推广链接(点抢时调)")
def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
if not settings.mt_cps_configured:
return ReferralLinkResponse(link="", link_map={})
try:
raw = get_referral_link(
product_view_sign=req.product_view_sign,
+59
View File
@@ -0,0 +1,59 @@
"""用户资料 endpoint。
路由前缀 `/api/v1/user`,全部需 Bearer 鉴权(用户级数据,不同于 MVP 透传接口):
PATCH /profile 修改昵称
POST /avatar 上传头像(multipart,字段名 file)
DELETE / 注销账号(软删除 + 匿名化)
昵称/头像后端持久化到 user 表,登录后经 /auth/me 与登录响应回传客户端——这才是
"改完重登仍生效"的唯一数据源(此前客户端只存本地、退登即丢)。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, File, HTTPException, UploadFile
from app.api.deps import CurrentUser, DbSession
from app.core import media
from app.repositories import user as user_repo
from app.schemas.auth import UserOut
from app.schemas.user import OkResponse, ProfileUpdateRequest
logger = logging.getLogger("shagua.user")
router = APIRouter(prefix="/api/v1/user", tags=["user"])
@router.patch("/profile", response_model=UserOut, summary="修改昵称")
def update_profile(req: ProfileUpdateRequest, user: CurrentUser, db: DbSession) -> UserOut:
updated = user_repo.update_nickname(db, user, nickname=req.nickname)
logger.info("update nickname user_id=%d", user.id)
return UserOut.model_validate(updated)
@router.post("/avatar", response_model=UserOut, summary="上传头像")
async def upload_avatar(
user: CurrentUser,
db: DbSession,
file: UploadFile = File(..., description="头像图片(jpeg/png/webp,≤5MB)"),
) -> UserOut:
data = await file.read()
try:
url = media.save_avatar(user.id, data)
except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
old_url = user.avatar_url
updated = user_repo.set_avatar_url(db, user, avatar_url=url)
media.delete_avatar(old_url) # 落库成功后再删旧文件,避免删了新的没存上
logger.info("update avatar user_id=%d url=%s", user.id, url)
return UserOut.model_validate(updated)
@router.delete("", response_model=OkResponse, summary="注销账号(软删除)")
def delete_account(user: CurrentUser, db: DbSession) -> OkResponse:
media.delete_avatar(user.avatar_url)
user_repo.soft_delete_account(db, user)
logger.info("delete account user_id=%d", user.id)
return OkResponse()
+14
View File
@@ -53,11 +53,18 @@ class Settings(BaseSettings):
SMS_SEND_INTERVAL_SEC: int = 60
# ===== 美团联盟 CPS =====
# 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
MT_CPS_APP_KEY: str = ""
MT_CPS_APP_SECRET: str = ""
MT_CPS_HOST: str = "https://media.meituan.com"
MT_CPS_TIMEOUT_SEC: int = 15
MT_CPS_DEFAULT_SID: str = "sgbjia"
@property
def mt_cps_configured(self) -> bool:
"""美团 CPS 凭证齐全(缺则接口返空,而非 502)。"""
return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET)
# ===== 微信支付(商家转账到零钱 / 提现)=====
# 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于
# 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。
@@ -106,6 +113,13 @@ class Settings(BaseSettings):
# price/step 每帧也是 LLM,可能 >30s;对齐客户端 agent ApiClient 的 60s 读超时。
PRICEBOT_COMPARE_TIMEOUT_SEC: int = 60
# ===== 媒体文件(用户头像上传)=====
# 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
# 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。
MEDIA_ROOT: str = "./data/media"
MEDIA_URL_PREFIX: str = "/media"
AVATAR_MAX_BYTES: int = 5 * 1024 * 1024 # 头像最大 5MB
# ===== CORS =====
CORS_ALLOW_ORIGINS: str = ""
+76
View File
@@ -0,0 +1,76 @@
"""用户上传文件(头像 / 反馈截图)的本地落盘 + 对外 URL 构建。
落盘到 `settings.MEDIA_ROOT/<子目录>/`,对外以 `settings.MEDIA_URL_PREFIX` 暴露
(main.py 用 StaticFiles 挂载)。返回**相对路径** URL(如 `/media/avatars/xxx.jpg`),
由客户端按自己的 BASE_URL 解析为绝对地址——服务端不知道客户端怎么访问到自己(dev 下
可能是 10.0.2.2 / LAN IP),所以不返回绝对 URL。
安全:不信任客户端给的 content-type / 文件名,改用魔数嗅探判定真实图片类型;
文件名服务端随机生成,杜绝路径穿越与覆盖。
"""
from __future__ import annotations
import secrets
from pathlib import Path
from app.core.config import settings
class MediaError(Exception):
"""上传文件不合法(类型/大小)。调用方转 400。"""
def _media_dir(subdir: str) -> Path:
d = Path(settings.MEDIA_ROOT) / subdir
d.mkdir(parents=True, exist_ok=True)
return d
def _sniff_ext(data: bytes) -> str | None:
"""按魔数判定图片类型,返回扩展名;非支持类型返回 None。"""
if data[:3] == b"\xff\xd8\xff":
return ".jpg"
if data[:8] == b"\x89PNG\r\n\x1a\n":
return ".png"
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return ".webp"
return None
def _save_image(subdir: str, user_id: int, data: bytes) -> str:
"""通用图片落盘:校验非空 / ≤上限 / 魔数为图片,随机文件名,返回相对 URL。"""
if not data:
raise MediaError("空文件")
if len(data) > settings.AVATAR_MAX_BYTES:
raise MediaError("图片过大(上限 5MB)")
ext = _sniff_ext(data)
if ext is None:
raise MediaError("仅支持 JPEG / PNG / WebP 图片")
fname = f"u{user_id}_{secrets.token_hex(8)}{ext}"
(_media_dir(subdir) / fname).write_bytes(data)
return f"{settings.MEDIA_URL_PREFIX}/{subdir}/{fname}"
def save_avatar(user_id: int, data: bytes) -> str:
"""保存头像,返回相对 URL(`/media/avatars/<file>`)。"""
return _save_image("avatars", user_id, data)
def save_feedback_image(user_id: int, data: bytes) -> str:
"""保存反馈截图,返回相对 URL(`/media/feedback/<file>`)。"""
return _save_image("feedback", user_id, data)
def delete_avatar(url: str | None) -> None:
"""删除本服务托管的旧头像文件;外部 URL(如微信头像)或空值不处理。"""
prefix = f"{settings.MEDIA_URL_PREFIX}/avatars/"
if not url or not url.startswith(prefix):
return
fname = url[len(prefix):]
if not fname or "/" in fname or "\\" in fname or ".." in fname:
return # 防路径穿越
try:
(_media_dir("avatars") / fname).unlink(missing_ok=True)
except OSError:
pass
-143
View File
@@ -1,143 +0,0 @@
"""首页 Feed 排序策略 (MVP 版)
LBS 召回 → 距离过滤 → 销量重排 → 分页 shuffle → 榜单加成 → 双品类同页交错。
"""
from __future__ import annotations
import random
import re
from typing import Any
def parse_sale_volume(text: str | None) -> int:
"""'热销8.5万+' → 85000, '热销1k+' → 1000, None → 0"""
if not text:
return 0
m = re.search(r"([\d.]+)\s*(万|k)?", text, re.IGNORECASE)
if not m:
return 0
num = float(m.group(1))
unit = (m.group(2) or "").lower()
if unit == "":
num *= 10000
elif unit == "k":
num *= 1000
return int(num)
def get_distance_km(item: dict[str, Any], is_daodian: bool) -> float | None:
"""提取距离(km)。外卖单位千米,到店单位米需÷1000。>50km 视为脏数据返回 None。"""
raw = (item.get("deliverablePoiInfo") or {}).get("deliveryDistance")
if raw is None:
return None
try:
d = float(raw)
except (ValueError, TypeError):
return None
if is_daodian:
d /= 1000
if d > 50:
return None
return d
def get_sell_price(item: dict[str, Any]) -> float | None:
raw = (item.get("couponPackDetail") or {}).get("sellPrice")
if raw is None:
return None
try:
p = float(raw)
return p if p > 0 else None
except (ValueError, TypeError):
return None
def get_product_sign(item: dict[str, Any]) -> str:
cpd = item.get("couponPackDetail") or {}
return cpd.get("productViewSign") or cpd.get("skuViewId") or ""
def filter_items(
items: list[dict], is_daodian: bool, max_km: float = 8.0,
) -> list[dict]:
"""距离 ≤ max_km、售价 > 0、去脏数据。"""
result = []
for item in items:
if get_sell_price(item) is None:
continue
dist = get_distance_km(item, is_daodian)
if dist is None or dist > max_km:
continue
result.append(item)
return result
def dedup(items: list[dict]) -> list[dict]:
seen: set[str] = set()
result = []
for item in items:
sign = get_product_sign(item)
if sign and sign not in seen:
seen.add(sign)
result.append(item)
return result
def sort_by_sales(items: list[dict]) -> list[dict]:
def _key(item: dict) -> int:
vol = (item.get("couponPackDetail") or {}).get("saleVolume")
return parse_sale_volume(vol)
return sorted(items, key=_key, reverse=True)
def split_pages(items: list[dict], page_size: int = 20) -> list[list[dict]]:
if not items:
return []
return [items[i : i + page_size] for i in range(0, len(items), page_size)]
def shuffle_pages(pages: list[list[dict]]) -> list[list[dict]]:
"""每页内 Fisher-Yates shuffle,不跨页。"""
result = []
for page in pages:
shuffled = page[:]
random.shuffle(shuffled)
result.append(shuffled)
return result
def inject_billboard(
pages: list[list[dict]],
billboard: list[dict],
per_page: int = 4,
max_inject_pages: int = 5,
) -> list[list[dict]]:
"""将榜单商品分配到前 N 页,每页额外加 per_page 个,随机位置插入。"""
existing = {get_product_sign(it) for p in pages for it in p}
unique = [it for it in billboard if get_product_sign(it) not in existing]
idx = 0
for i in range(min(max_inject_pages, len(pages))):
batch = unique[idx : idx + per_page]
idx += per_page
for it in batch:
pages[i].insert(random.randint(0, len(pages[i])), it)
return pages
def merge_category_pages(
waimai_pages: list[list[dict]],
daodian_pages: list[list[dict]],
) -> list[list[dict]]:
"""同页合并 + shuffle,页数取两者最大值。"""
n = max(len(waimai_pages), len(daodian_pages))
result = []
for k in range(n):
merged: list[dict] = []
if k < len(waimai_pages):
merged.extend(waimai_pages[k])
if k < len(daodian_pages):
merged.extend(daodian_pages[k])
random.shuffle(merged)
result.append(merged)
return result
+11 -5
View File
@@ -59,16 +59,22 @@ TASK_REWARDS: dict[str, int] = {
# ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
# 看完一个激励视频发的金币(≈¥0.01,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币"
# 看完一个激励视频发的金币(100 金币 ≈¥0.01,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
# ③ test-grant 本地联调的发奖额。
# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
# 代码位"奖励数量"配成与本值一致(100),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
# 代码位"奖励数量"配成与本值一致(=100),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 100
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限,防刷 + 控成本。
# ⚠️ 占位值:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 10
# ⚠️ 测试阶段值=20:正式上线前按产品策略 + 广告 eCPM 实测收益重新核定,改这里即可。
DAILY_AD_REWARD_LIMIT: int = 20
# 一轮看 N 次激励视频,看完一轮强制 10 分钟冷却(reward-status 派生 cooldown_until 给客户端,
# 客户端任务行 CTA 在冷却期间显示"MM:SS"且不可点)。冷却是 UX 约束,后端发奖只看 daily_limit,
# 冷却期间穿山甲回调仍照常发奖,不另设拒发分支。
VIDEO_ROUND_REQUIRED_COUNT: int = 3
VIDEO_ROUND_COOLDOWN_SECONDS: int = 600
def resolve_ad_reward_coin(reward_amount: str | int | None) -> int:
+14 -8
View File
@@ -27,19 +27,25 @@ def _ensure_sqlite_dir(url: str) -> None:
_ensure_sqlite_dir(settings.DATABASE_URL)
_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
# SQLite 跨线程访问要 check_same_thread=False;PG/MySQL 不需要这个参数
_connect_args: dict = {}
if settings.DATABASE_URL.startswith("sqlite"):
if _is_sqlite:
_connect_args["check_same_thread"] = False
engine = create_engine(
settings.DATABASE_URL,
connect_args=_connect_args,
_engine_kwargs: dict = {
"connect_args": _connect_args,
# echo 在 dev 下打 SQL,生产关掉
echo=settings.APP_DEBUG and not settings.is_prod,
future=True,
pool_pre_ping=True,
)
"echo": settings.APP_DEBUG and not settings.is_prod,
"future": True,
"pool_pre_ping": True,
}
# SQLite 用单文件不需要池;PG/MySQL 必须显式池化 + recycle 防 idle 断连
if not _is_sqlite:
_engine_kwargs.update(pool_size=10, max_overflow=20, pool_recycle=3600)
engine = create_engine(settings.DATABASE_URL, **_engine_kwargs)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False)
+16
View File
@@ -8,17 +8,22 @@ import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from app.api.v1.ad import router as ad_router
from app.api.v1.auth import router as auth_router
from app.api.v1.compare import router as compare_router
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.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
from app.api.v1.user import router as user_router
from app.api.v1.wallet import router as wallet_router
from app.core.config import settings
from app.core.logging import setup_logging
@@ -65,6 +70,8 @@ def health() -> dict[str, str]:
app.include_router(auth_router)
app.include_router(user_router)
app.include_router(feedback_router)
app.include_router(coupon_router)
app.include_router(compare_router)
app.include_router(meituan_router)
@@ -73,3 +80,12 @@ app.include_router(signin_router)
app.include_router(tasks_router)
app.include_router(savings_router)
app.include_router(ad_router)
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
_media_root = Path(settings.MEDIA_ROOT)
_media_root.mkdir(parents=True, exist_ok=True)
app.mount(
settings.MEDIA_URL_PREFIX,
StaticFiles(directory=str(_media_root)),
name="media",
)
+1
View File
@@ -1,5 +1,6 @@
"""所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
from app.models.ad_reward import AdRewardRecord # noqa: F401
from app.models.feedback import Feedback # 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
+35
View File
@@ -0,0 +1,35 @@
"""用户反馈表(帮助与反馈)。
每条 = 用户一次提交。content 必填,contact 必填(微信/QQ/手机,便于回访),images 为可选的
截图 URL 列表(/media/feedback/...,JSON 存)。status: new(待处理)/ handled(已处理)。
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class Feedback(Base):
__tablename__ = "feedback"
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
)
content: Mapped[str] = mapped_column(Text, nullable=False)
contact: Mapped[str] = mapped_column(String(128), nullable=False)
# 截图 URL 列表(相对路径,如 ["/media/feedback/u1_ab12.jpg"]);无图为 None
images: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
# new(待处理) / handled(已处理)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="new")
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"<Feedback id={self.id} user_id={self.user_id} status={self.status}>"
+4 -3
View File
@@ -8,7 +8,8 @@ from __future__ import annotations
from datetime import datetime
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
@@ -28,8 +29,8 @@ class SavingsRecord(Base):
title: Mapped[str | None] = mapped_column(String(128), nullable=True)
# 店铺名(订单明细卡标题,如「窑鸡王(王府井店)」)
shop_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
# 菜品名列表(JSON),前 2 道直接展示,其余收进「还有 N 道菜」展开。SQLite 存为 TEXT。
dishes: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
# 菜品名列表(JSONB),PG 上可建 GIN 索引,前 2 道直接展示,其余收进「还有 N 道菜」展开。
dishes: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
# 来源:demo(演示) / compare(真实比价上报)
source: Mapped[str] = mapped_column(String(16), nullable=False, default="compare")
+55 -5
View File
@@ -10,11 +10,19 @@
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core.rewards import AD_REWARD_COIN, DAILY_AD_REWARD_LIMIT, cn_today
from app.core.rewards import (
AD_REWARD_COIN,
DAILY_AD_REWARD_LIMIT,
VIDEO_ROUND_COOLDOWN_SECONDS,
VIDEO_ROUND_REQUIRED_COUNT,
cn_today,
)
from app.repositories import wallet as crud_wallet
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
@@ -102,7 +110,49 @@ def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardR
return rec
def today_status(db: Session, user_id: int) -> tuple[int, int, int]:
"""客户端查"今日看广告发奖"进度:返回 (今日已发次数, 每日上限, 单次金币)。"""
used = _granted_today(db, user_id, cn_today().isoformat())
return used, DAILY_AD_REWARD_LIMIT, AD_REWARD_COIN
def _last_completed_round_end_at(
db: Session, user_id: int, reward_date: str, round_count: int
) -> datetime | None:
"""当日 granted 记录中**最近一个已完成轮**末尾那次的 created_at。
思路:把当日 granted 按时间倒序排,跳过当前未完成轮的 round_count 条,下一条
"上一轮最后一次"。round_count==0 且 used>=N 时跳 0 条直接取最近一条。
used<N(还没完成第一轮)调用方应直接判 None,不进这里。
"""
return db.execute(
select(AdRewardRecord.created_at)
.where(
AdRewardRecord.user_id == user_id,
AdRewardRecord.reward_date == reward_date,
AdRewardRecord.status == "granted",
)
.order_by(AdRewardRecord.created_at.desc())
.offset(round_count)
.limit(1)
).scalar_one_or_none()
def today_status(
db: Session, user_id: int
) -> tuple[int, int, int, int, datetime | None]:
"""客户端查"今日看广告发奖"进度。
返回 (今日已发次数, 每日上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC))。
- round_count = used % VIDEO_ROUND_REQUIRED_COUNT,展示用(0..N-1)
- cooldown_until 计算:取最近一个已完成轮末尾的 created_at + 10 min;若仍 > now 则返回,
否则返回 None。冷却 = UX 约束(客户端 CTA 倒计时不可点),后端发奖逻辑不受影响。
"""
today = cn_today().isoformat()
used = _granted_today(db, user_id, today)
round_count = used % VIDEO_ROUND_REQUIRED_COUNT
cooldown_until: datetime | None = None
if used >= VIDEO_ROUND_REQUIRED_COUNT:
last_end = _last_completed_round_end_at(db, user_id, today, round_count)
if last_end is not None:
# SQLAlchemy 在 SQLite 上拿到的 created_at 可能是 naive,统一按 UTC 解读再比较
if last_end.tzinfo is None:
last_end = last_end.replace(tzinfo=timezone.utc)
cd_end = last_end + timedelta(seconds=VIDEO_ROUND_COOLDOWN_SECONDS)
if cd_end > datetime.now(timezone.utc):
cooldown_until = cd_end
return used, DAILY_AD_REWARD_LIMIT, AD_REWARD_COIN, round_count, cooldown_until
+27
View File
@@ -0,0 +1,27 @@
"""feedback 表写入。"""
from __future__ import annotations
from sqlalchemy.orm import Session
from app.models.feedback import Feedback
def create_feedback(
db: Session,
*,
user_id: int,
content: str,
contact: str,
images: list[str] | None,
) -> Feedback:
fb = Feedback(
user_id=user_id,
content=content,
contact=contact,
images=images or None,
status="new",
)
db.add(fb)
db.commit()
db.refresh(fb)
return fb
+28
View File
@@ -46,3 +46,31 @@ def upsert_user_for_login(
db.commit()
db.refresh(user)
return user
def update_nickname(db: Session, user: User, *, nickname: str) -> User:
user.nickname = nickname
db.commit()
db.refresh(user)
return user
def set_avatar_url(db: Session, user: User, *, avatar_url: str) -> User:
user.avatar_url = avatar_url
db.commit()
db.refresh(user)
return user
def soft_delete_account(db: Session, user: User) -> None:
"""注销账号:软删除 + 匿名化。
把 phone 改成 `deleted_<id>` 释放唯一约束,允许同号码重新注册成全新账号;
清空 PII(昵称/头像),保留行用于审计。status=deleted 后将无法再登录该行
(登录接口校验 status==active)。
"""
user.status = "deleted"
user.phone = f"deleted_{user.id}"
user.nickname = None
user.avatar_url = None
db.commit()
+14
View File
@@ -4,6 +4,8 @@
"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field
@@ -27,6 +29,14 @@ class AdRewardStatusOut(BaseModel):
daily_limit: int = Field(..., description="每日发奖次数上限")
remaining: int = Field(..., description="今日剩余可领次数")
coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
round_count: int = Field(
..., description="本轮(3 次一组)已看次数,0..N-1。客户端可由它判断'刚刚看完一轮'"
)
cooldown_until: datetime | None = Field(
None,
description="本轮看完后 10 分钟冷却结束时间(UTC ISO),null 表示不在冷却。"
"冷却是 UX 约束,后端发奖不受影响,客户端 CTA 据此显示 MM:SS 倒计时且不可点",
)
class TestGrantOut(BaseModel):
@@ -39,3 +49,7 @@ class TestGrantOut(BaseModel):
daily_limit: int = Field(..., description="每日发奖次数上限")
remaining: int = Field(..., description="今日剩余可领次数")
coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
round_count: int = Field(..., description="本轮已看次数,详见 AdRewardStatusOut")
cooldown_until: datetime | None = Field(
None, description="本轮冷却结束时间(UTC),详见 AdRewardStatusOut"
)
+14
View File
@@ -0,0 +1,14 @@
"""反馈相关响应 schema。请求是 multipart 表单(content/contact/images),在路由里直接校验。"""
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, ConfigDict
class FeedbackOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
status: str
created_at: datetime
+3 -38
View File
@@ -1,46 +1,11 @@
"""美团 CPS 券列表 / 换链相关 schemas。"""
from __future__ import annotations
import re
from typing import Any
from pydantic import BaseModel, Field
def _format_price_label(price_lbl: dict[str, Any]) -> str | None:
history = price_lbl.get("historyPriceLabel")
if history:
return history
beat = price_lbl.get("beatMTLabel")
if not beat:
return None
m = re.match(r"比日常团购省([\d.]+)元", beat)
if m:
return f"比团购省 {m.group(1)}"
return beat
def _format_rank_label(raw: str | None) -> str | None:
if not raw:
return None
m = re.search(r"(外卖|美食|饮品|轻食|奶茶|咖啡|火锅|烧烤|甜品|快餐).*?第(\d+)名", raw)
if m:
return f"{m.group(1)}榜第 {m.group(2)}"
m2 = re.search(r"第(\d+)名", raw)
if m2:
return f"销量榜第 {m2.group(1)}"
return raw
def _format_rating_label(raw: str | None) -> str | None:
if not raw:
return None
m = re.search(r"([\d.]+)\s*分", raw)
if m:
return f"点评 {m.group(1)}"
return raw
# ───────────────── 券卡片(归一化后给客户端) ─────────────────
class CouponCard(BaseModel):
@@ -137,9 +102,9 @@ class CouponCard(BaseModel):
available_poi_num=avail.get("availablePoiNum"),
coupon_num=cpd.get("couponNum"),
valid_days=valid_info.get("couponValidDay"),
price_label=_format_price_label(price_lbl),
rank_label=_format_rank_label(label.get("productRankLabel")),
rating_label=_format_rating_label(label.get("dianPingRankLabel")),
price_label=price_lbl.get("historyPriceLabel") or price_lbl.get("beatMTLabel"),
rank_label=label.get("productRankLabel"),
rating_label=label.get("dianPingRankLabel"),
)
+20
View File
@@ -0,0 +1,20 @@
"""用户资料(昵称/头像)相关请求 schemas。响应复用 [auth.UserOut]。"""
from __future__ import annotations
from pydantic import BaseModel, Field, field_validator
class ProfileUpdateRequest(BaseModel):
nickname: str = Field(..., min_length=1, max_length=16, description="昵称,1-16 字")
@field_validator("nickname")
@classmethod
def _strip_non_blank(cls, v: str) -> str:
v = v.strip()
if not v:
raise ValueError("昵称不能为空")
return v
class OkResponse(BaseModel):
ok: bool = True
+10 -1
View File
@@ -14,6 +14,15 @@
| `daily_limit` | int | 每日发奖次数上限 |
| `remaining` | int | 今日剩余可领次数 |
| `coin_per_ad` | int | 看完一个激励视频发的金币 |
| `round_count` | int | 本轮(3 次一组)已看次数,0..N-1。客户端可据「拿到 round_count==0 && cooldown_until!=null」判定刚刚看完一轮 |
| `cooldown_until` | datetime\|null | 本轮 10 分钟冷却结束时间(UTC ISO);null 表示不在冷却 |
## 说明
福利页「看视频赚金币」用:展示「今日还能看 N 次」「看一次得 M 金币」。真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
福利页「看视频赚金币」用:
- 展示「今日还能看 N 次」「看一次得 M 金币」(`remaining`/`coin_per_ad`)
- 任务行 CTA 4 态推导:`adLoading` → Loading;`remaining==0` → Capped("明天再来");`cooldown_until` 在未来 → CoolingDown(显示 MM:SS 倒计时,不可点);否则 Normal("去赚取")
- 弹窗 limit note:`used_today >= daily_limit` 显示「今日视频已到限额,明天再来」;`round_count==0 && cooldown_until!=null` 显示「本轮视频已看完,10分钟后再来」
冷却派生算法:取当日 `status='granted'` 记录里**最近一个已完成轮**(每 N=3 条算一轮)末尾那次的 `created_at`,加 10 分钟。若仍 > now → 返回;否则返回 null。**冷却是 UX 约束**,后端发奖(`/api/v1/ad/pangle-callback`)只看 daily_limit,冷却期间穿山甲回调照常发奖。
真正发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S)。
+279
View File
@@ -0,0 +1,279 @@
# SQLite → PostgreSQL 迁移指南
> 上线前把数据库从 SQLite 切到 PostgreSQL。本服务自始就为这一步预留了路径(`db/session.py` 的 dialect 特判 / `alembic/env.py` 的 batch mode 开关 / SQLAlchemy 2.0 + Alembic),代码改动很小。
>
> **背景假设**:当前生产**无真实用户数据**(MVP 阶段),所以整个迁移本质是"切引擎",不涉及数据搬迁。如果将来已有真实数据,本文档不适用,需补充 `pgloader` 演练 + 钱表金额逐行核对 + 停服窗口 + 回滚预案。
>
> 最后更新:2026-05-29
---
## 0. 为什么切
SQLite 三个硬伤决定它不能上线:
- **并发写锁库**:整个文件一把锁,多请求并发写会排队,极易超时。当前 uvicorn `--workers 1` 勉强能跑,扩 worker 立刻撞墙。
- **类型宽松**:整数列能塞字符串、`CHECK` 约束当装饰品。钱表(`wallet` / `withdraw_order` / `cash_transaction`)用 SQLite 等于裸奔。
- **无 JSONB / 无并发索引 / 无分区**:PG 的核心能力一个都享受不到。`savings.dishes` 现在用 SQLite 的 `JSON`(实际存 TEXT),无索引、无操作符,纯展示用。
PG 默认上 16 版(工具链最齐),驱动用 **psycopg3**(SQLAlchemy 2.0 时代默认,不要再装 psycopg2)。
---
## 1. 本地起 PG + 跑通空库(半天)
### 1.1 装 PG
macOS:
```bash
brew install postgresql@16
brew services start postgresql@16
```
Linux(Ubuntu/Debian):
```bash
sudo apt install postgresql-16
sudo systemctl enable --now postgresql
```
### 1.2 建库 + 用户 + 授权
```bash
psql postgres <<'EOF'
CREATE USER shaguabijia_app WITH PASSWORD 'change-me-strong-random';
CREATE DATABASE shaguabijia OWNER shaguabijia_app ENCODING 'UTF8';
GRANT ALL PRIVILEGES ON DATABASE shaguabijia TO shaguabijia_app;
EOF
```
密码用 `python -c "import secrets; print(secrets.token_urlsafe(32))"` 生成,**不要复用 JWT_SECRET_KEY**。
### 1.3 装驱动
```bash
pip install "psycopg[binary]>=3.1"
```
`psycopg[binary]` 是预编译版,免装 libpq 头文件。生产同款,无需源码编译版。
### 1.4 改 `.env`
```ini
DATABASE_URL=postgresql+psycopg://shaguabijia_app:<password>@localhost:5432/shaguabijia
```
> ⚠️ URL scheme 必须是 `postgresql+psycopg://`(显式声明 psycopg3),不能写成 `postgresql://`——SQLAlchemy 默认会去找 psycopg2,装的不一致就报 ModuleNotFoundError。
### 1.5 建表
```bash
alembic upgrade head
```
### 1.6 验证
```bash
psql -U shaguabijia_app -d shaguabijia -c "\dt"
```
应该列出全部 9 张表 + `alembic_version`:
```
user, coin_account, coin_transaction, cash_transaction,
withdraw_order, signin_record, task_claim_record,
savings_record, ad_reward_record, alembic_version
```
再手动 INSERT 一行 user,确认 `created_at` 默认值正常落:
```sql
INSERT INTO "user" (phone, register_channel, status) VALUES ('13800138000', 'sms', 'active');
SELECT id, phone, created_at, last_login_at FROM "user";
```
`created_at` 应该是当前时间戳带时区,不是 NULL。`user` 表名是 PG 关键字,**必须用双引号**包裹。
---
## 2. 测试套件全绿(1 小时)
### 2.1 测试连 PG
测试默认连 SQLite。临时让 `tests/conftest.py``DATABASE_URL` 也指到本地 PG(可以建个 `shaguabijia_test` 库专门跑测试,免得污染 dev 库)。
```bash
createdb -O shaguabijia_app shaguabijia_test
DATABASE_URL=postgresql+psycopg://shaguabijia_app:xxx@localhost:5432/shaguabijia_test pytest
```
### 2.2 处理失败用例
**所有红色测试都不能放过**——SQLite 类型/约束都很宽松,PG 严格起来暴露的就是真 bug。常见问题:
- **字符串和整数比较**:SQLite 允许 `WHERE phone = 13800138000`(自动转字符串),PG 直接报错。改成 `:phone` 参数 + 类型严格
- **timezone 处理**:SQLite 存 naive datetime,PG 的 `TIMESTAMPTZ` 必须带时区。检查所有 `datetime.utcnow()`,改成 `datetime.now(timezone.utc)`
- **隐式 commit**:SQLite 某些情况下隐式 commit,PG 严格事务边界。如果某用例报 `current transaction is aborted`,说明业务代码缺 commit/rollback
### 2.3 验证
`pytest` 全绿,**且本地起 `./run.sh` 后用 Apifox 调通登录 / 美团 feed / 领券透传三个真接口**。
---
## 3. 代码扫雷(1 天)
### 3.1 必改清单
| 文件 / 位置 | 改动 | 原因 |
|---|---|---|
| `app/models/savings.py` | `from sqlalchemy import JSON``from sqlalchemy.dialects.postgresql import JSONB`;`mapped_column(JSON, ...)``mapped_column(JSONB, ...)` | JSONB 在 PG 上有 GIN 索引和操作符支持,是 PG 核心优势之一。SQLite 上 `JSON` 实际是 TEXT,切到 PG 后用 `JSON` 类型存的是 `json` 不是 `jsonb`,查询/索引能力差一大截 |
| `app/db/session.py``create_engine` 调用 | 增加连接池参数:`pool_size=10, max_overflow=20, pool_recycle=3600` | SQLite 单文件不需要池,PG 必须显式池化。`pool_recycle=3600` 防 PG 主动断开的 idle 连接 |
| `alembic/versions/` 下旧迁移里的 `sa.text('(CURRENT_TIMESTAMP)')` | **不要改** | PG 能接受这个语法。改了反而搞乱迁移历史 + 让 alembic 觉得 schema drift。新写的迁移统一用 `sa.func.now()` |
### 3.2 保留的 dialect 特判
下面这些**保留原样**,它们的 SQLite 分支在 PG 下自动绕过、不影响新部署,但本地开发想用 SQLite 临时跑还能用:
- `app/db/session.py` 里的 `_ensure_sqlite_dir``check_same_thread` 特判
- `alembic/env.py` 里的 `render_as_batch` 特判
### 3.3 顺便清掉
`app/main.py` 启动日志里如果打 `DATABASE_URL`,确认密码不会进日志(只打 scheme + host,不打完整 URL)。
### 3.4 验证
本地 PG 起 `./run.sh`,跑一遍下面的烟测脚本:
```bash
# 登录
curl -X POST http://localhost:8770/api/v1/auth/sms/send -d '{"phone":"13800138000"}'
curl -X POST http://localhost:8770/api/v1/auth/sms/login -d '{"phone":"13800138000","code":"123456"}'
# 拿到 access_token 后调 me
curl -H "Authorization: Bearer <token>" http://localhost:8770/api/v1/auth/me
# 美团 feed
curl -X POST http://localhost:8770/api/v1/meituan/feed -d '{"longitude":116.4,"latitude":39.9,"page":0}'
```
三条都 200 即通过。`psql``user` 表多了一条 `13800138000`
---
## 4. 生产切换(半小时,无需停服窗口)
**前提**:生产无真实数据。如已有数据,本节不适用。
### 4.1 装 PG
```bash
ssh server
sudo apt install postgresql-16
sudo systemctl enable --now postgresql
```
### 4.2 建库
```bash
sudo -u postgres psql <<'EOF'
CREATE USER shaguabijia_app WITH PASSWORD '<strong-random-password>';
CREATE DATABASE shaguabijia OWNER shaguabijia_app ENCODING 'UTF8';
GRANT ALL PRIVILEGES ON DATABASE shaguabijia TO shaguabijia_app;
EOF
```
生产密码**单独生成**,不要复用本地的。
### 4.3 装驱动到生产 venv
```bash
cd /opt/shaguabijia-app-server
.venv/bin/pip install "psycopg[binary]>=3.1"
```
### 4.4 改生产 `.env`
```ini
DATABASE_URL=postgresql+psycopg://shaguabijia_app:<password>@localhost:5432/shaguabijia
```
### 4.5 建表 + 重启
```bash
.venv/bin/alembic upgrade head
sudo systemctl restart shaguabijia-app-server
sudo journalctl -u shaguabijia-app-server -f
```
启动日志里看到 `Application startup complete` + 没有 5xx 即通过。
### 4.6 验证
- 用真账号(手机号)走一遍极光登录或短信登录
-`GET /api/v1/auth/me`,确认返回当前用户
-`POST /api/v1/meituan/feed`,确认券列表正常
- `psql -U shaguabijia_app -d shaguabijia -c 'SELECT id, phone, created_at FROM "user"'` 看到刚登录的用户
### 4.7 老 SQLite 文件归档
```bash
cd /opt/shaguabijia-app-server
mv data/app.db data/app.db.legacy-$(date +%Y%m%d)
```
不要删——万一要回查测试期残留数据(开发期可能塞了一些手工记录),保留几个月。
---
## 5. 兜底:切换失败如何回滚
万一改完起不来:
```bash
# 1. .env 改回 SQLite
sed -i 's|^DATABASE_URL=postgresql.*|DATABASE_URL=sqlite:///./data/app.db|' /opt/shaguabijia-app-server/.env
# 2. 把归档的 sqlite 文件还原(如果已经改名)
mv data/app.db.legacy-* data/app.db
# 3. 重启
sudo systemctl restart shaguabijia-app-server
```
由于 SQLite 文件没动过,无数据丢失,损失只是切换过程中的几分钟 5xx。
---
## 6. 顺便建议:同时装上 Redis
PG 上线后,Redis 是下一个明确要补的基础设施(已在 [待办与技术债.md](./待办与技术债.md) 里列了用途)。**建议本次维护窗口顺手把 Redis 也装上**,不开始用、只装服务,免得下次还要单独申请部署:
```bash
sudo apt install redis-server
sudo systemctl enable --now redis-server
# 改 /etc/redis/redis.conf:bind 127.0.0.1(只本机访问)+ requirepass <strong-password>
sudo systemctl restart redis-server
```
下面三个用途等接的时候再写代码,**本次只装服务、不接代码**:
| 用途 | 当前临时方案 | 何时必须接 Redis |
|---|---|---|
| 短信验证码冷却 | 进程内存 dict(`integrations/sms.py`) | `SMS_MOCK=false` 上线时 / 扩 worker 时 |
| pricebot trace_id session | 进程内存 dict(在 pricebot-backend) | pricebot 扩 worker 时 |
| JWT 黑名单(logout 真失效) | 当前 logout 占位无失效 | P1 鉴权改造时 |
---
## 7. 后续优化(不阻塞上线,一周内补)
- **每天凌晨 `pg_dump` 备份**:用 systemd timer 或 cron,保留 7 天。备份文件 `pg_dump -Fc shaguabijia > /backup/shaguabijia-$(date +%Y%m%d).dump`
- **金钱表加 CHECK 约束**:新迁移给 `coin_account.balance``withdraw_order.amount_yuan_x100``CHECK (... >= 0)`。PG 严格执行,SQLite 形同虚设
- **timezone 统一**:生产 PG 配 `timezone='Asia/Shanghai'` 或维持 `UTC`,**团队定一个**写进本文档,业务代码统一处理。当前模型混用 `func.now()`(server-side)和 `_utcnow()`(`models/user.py``last_login_at`,Python-side),后者依赖应用层时区,迁完 PG 后建议统一改 `func.now()`
- **慢查询观察**:`shared_preload_libraries = 'pg_stat_statements'` 打开扩展,每周看一次 Top 10 慢查询,提前优化
---
## 8. 不要做的事
- ❌ 不要搞主从复制 / 读写分离——量级远没到,先用单实例
- ❌ 不要装 PgBouncer / 连接池中间件——SQLAlchemy 自带池够用
- ❌ 不要 squash 旧迁移文件——9 个增量保持原样,让 alembic 一次跑过空库
- ❌ 不要在生产 PG 上裸跑测试套件——用单独的 `shaguabijia_test`
+25 -8
View File
@@ -15,20 +15,37 @@ alembic upgrade head
---
## 1. 首次初始化(clone 后)
### 1.1 用 PostgreSQL(推荐,与生产一致)
前置:本机已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
```bash
# (1) 装依赖(在你的虚拟环境里)
# (1) 装依赖(在你的虚拟环境里, 含 psycopg)
pip install -e .
# (2) 配 .env(从模板复制,至少填 JWT_SECRET_KEY)
cp .env.example .env
# (2) 一键建用户 + 建库 + 写 .env + 跑迁移
python scripts/init_postgres.py
# 按提示输入 PG 超级用户密码即可。脚本会:
# - 自动建业务用户 shaguabijia_app + 业务库 shaguabijia
# - 自动生成业务用户强密码,写入 .env 的 DATABASE_URL
# - 跑完 alembic upgrade head
# (3) SQLite 默认库需要 data/ 目录存在
# (3) 启动
./run.sh
```
可用环境变量提前指定(免交互):
```bash
PG_SUPER_PASS=xxx APP_DB_PASS=yyy python scripts/init_postgres.py
```
脚本幂等,重复跑会重置业务用户密码、跳过已存在的库。
### 1.2 用 SQLite(本地开发临时用,不接生产链路时)
```bash
pip install -e .
cp .env.example .env # 不动 DATABASE_URL, 默认 sqlite:///./data/app.db
mkdir -p data
# (4) 跑迁移建表 —— 关键
alembic upgrade head
# (5) 启动(run.sh 会自动重跑 3+4,所以平时直接 ./run.sh 也行)
./run.sh
```
> 也可以直接 `bash scripts/migrate.sh` 只做迁移、不启服务(部署/CI 用)。
+3
View File
@@ -17,6 +17,9 @@ dependencies = [
"sqlalchemy>=2.0.35",
"alembic>=1.13.3",
# PostgreSQL 驱动 (psycopg3, SQLAlchemy 2.0 时代默认, 不要再装 psycopg2)
"psycopg[binary]>=3.1",
# JWT 签名 / 校验
"pyjwt[crypto]>=2.9.0",
+189
View File
@@ -0,0 +1,189 @@
"""Bootstrap PostgreSQL: 建用户 + 建库 + 授权 + 写 .env + 跑迁移。
新机器初始化用。前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
用法:
python scripts/init_postgres.py
环境变量(可选,不填会交互式问):
PG_HOST PG 主机, 默认 localhost
PG_PORT PG 端口, 默认 5432
PG_SUPER_USER 超级用户, 默认 postgres
PG_SUPER_PASS 超级用户密码 (不填会 getpass 交互输入)
APP_DB_NAME 业务库名, 默认 shaguabijia
APP_DB_USER 业务用户名, 默认 shaguabijia_app
APP_DB_PASS 业务用户密码 (不填会自动生成强密码)
幂等:已存在的用户/库不会重复建,已配的 .env 会被覆盖前提示。
"""
from __future__ import annotations
import getpass
import os
import secrets
import subprocess
import sys
from pathlib import Path
try:
import psycopg
from psycopg import sql
except ImportError:
print("❌ 缺 psycopg。先跑: pip install -e .")
sys.exit(1)
ROOT = Path(__file__).resolve().parent.parent
ENV_FILE = ROOT / ".env"
ENV_EXAMPLE = ROOT / ".env.example"
def env_or_input(key: str, prompt: str, default: str = "", secret: bool = False) -> str:
val = os.environ.get(key, "").strip()
if val:
return val
if secret:
return getpass.getpass(f"{prompt}: ").strip()
suffix = f" [{default}]" if default else ""
raw = input(f"{prompt}{suffix}: ").strip()
return raw or default
def ensure_role(conn: psycopg.Connection, user: str, password: str) -> None:
# 注意: CREATE/ALTER USER 不支持参数化密码, 必须用 sql.Literal 内联
with conn.cursor() as cur:
cur.execute("SELECT 1 FROM pg_roles WHERE rolname=%s", (user,))
if cur.fetchone():
print(f" · 用户 {user} 已存在, 重置密码")
cur.execute(
sql.SQL("ALTER USER {} WITH PASSWORD {}").format(
sql.Identifier(user), sql.Literal(password)
)
)
else:
print(f" · 创建用户 {user}")
cur.execute(
sql.SQL("CREATE USER {} WITH PASSWORD {}").format(
sql.Identifier(user), sql.Literal(password)
)
)
def ensure_database(conn: psycopg.Connection, db: str, owner: str) -> None:
with conn.cursor() as cur:
cur.execute("SELECT 1 FROM pg_database WHERE datname=%s", (db,))
if cur.fetchone():
print(f" · 数据库 {db} 已存在")
return
print(f" · 创建数据库 {db} (owner={owner}, encoding=UTF8)")
cur.execute(
sql.SQL("CREATE DATABASE {} OWNER {} ENCODING 'UTF8'").format(
sql.Identifier(db), sql.Identifier(owner)
)
)
def grant_all(conn: psycopg.Connection, db: str, user: str) -> None:
with conn.cursor() as cur:
cur.execute(
sql.SQL("GRANT ALL PRIVILEGES ON DATABASE {} TO {}").format(
sql.Identifier(db), sql.Identifier(user)
)
)
def write_env(database_url: str) -> None:
if not ENV_FILE.exists():
if ENV_EXAMPLE.exists():
ENV_FILE.write_text(ENV_EXAMPLE.read_text(encoding="utf-8"), encoding="utf-8")
print(f" · 从 .env.example 复制出 .env")
else:
ENV_FILE.write_text("", encoding="utf-8")
lines = ENV_FILE.read_text(encoding="utf-8").splitlines()
new_lines = []
replaced = False
for line in lines:
if line.startswith("DATABASE_URL="):
old_url = line.split("=", 1)[1]
if old_url.strip() and old_url.strip() != database_url:
ans = input(f"\n.env 里已有 DATABASE_URL=\n {old_url}\n覆盖吗? [y/N]: ").strip().lower()
if ans != "y":
print(" · 保留原 DATABASE_URL")
new_lines.append(line)
replaced = True
continue
new_lines.append(f"DATABASE_URL={database_url}")
replaced = True
else:
new_lines.append(line)
if not replaced:
new_lines.append(f"DATABASE_URL={database_url}")
ENV_FILE.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
print(f" · 写入 .env -> DATABASE_URL")
def run_alembic_upgrade() -> bool:
print("\n[3/3] 跑 alembic upgrade head")
try:
subprocess.run(
[sys.executable, "-m", "alembic", "upgrade", "head"],
cwd=ROOT,
check=True,
)
return True
except subprocess.CalledProcessError as e:
print(f"❌ alembic 失败: {e}")
return False
def main() -> int:
print("=" * 60)
print("PostgreSQL 初始化脚本 — shaguabijia-app-server")
print("=" * 60)
host = env_or_input("PG_HOST", "PG host", "localhost")
port = env_or_input("PG_PORT", "PG port", "5432")
super_user = env_or_input("PG_SUPER_USER", "PG superuser", "postgres")
super_pass = env_or_input("PG_SUPER_PASS", "PG superuser password", secret=True)
db_name = env_or_input("APP_DB_NAME", "App database name", "shaguabijia")
db_user = env_or_input("APP_DB_USER", "App database user", "shaguabijia_app")
db_pass = os.environ.get("APP_DB_PASS", "").strip()
if not db_pass:
db_pass = secrets.token_urlsafe(32)
print(f" · 自动生成业务用户密码: {db_pass}")
print(f"\n[1/3] 连接 PG (host={host}:{port}, user={super_user})")
try:
conn = psycopg.connect(
host=host, port=int(port), user=super_user, password=super_pass,
dbname="postgres", autocommit=True,
)
except psycopg.OperationalError as e:
print(f"❌ 连不上 PG: {e}")
print(" 检查 1) PG 服务是否在跑 2) 超级用户密码是否正确 3) 端口防火墙是否放行")
return 1
print("\n[2/3] 建用户 + 建库 + 授权")
with conn:
ensure_role(conn, db_user, db_pass)
ensure_database(conn, db_name, db_user)
grant_all(conn, db_name, db_user)
database_url = f"postgresql+psycopg://{db_user}:{db_pass}@{host}:{port}/{db_name}"
write_env(database_url)
if not run_alembic_upgrade():
return 1
print("\n" + "=" * 60)
print("✅ 全部完成")
print(f" DATABASE_URL = {database_url}")
print(" 下一步: 启动服务 -> ./run.sh 或 uvicorn app.main:app --reload --port 8770")
print("=" * 60)
return 0
if __name__ == "__main__":
sys.exit(main())
+83 -1
View File
@@ -9,8 +9,15 @@ from sqlalchemy import select
from app.core.config import settings
from app.integrations import pangle
from app.core.rewards import AD_REWARD_COIN, DAILY_AD_REWARD_LIMIT, MAX_AD_REWARD_COIN
from app.core.rewards import (
AD_REWARD_COIN,
DAILY_AD_REWARD_LIMIT,
MAX_AD_REWARD_COIN,
VIDEO_ROUND_COOLDOWN_SECONDS,
VIDEO_ROUND_REQUIRED_COUNT,
)
from app.db.session import SessionLocal
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
@@ -148,6 +155,81 @@ def test_reward_status_requires_auth(client) -> None:
assert r.status_code == 401
def test_reward_status_initial_round_state(client) -> None:
"""新用户没看广告 → round_count=0, cooldown_until=None。"""
phone = "13800003101"
token = _login(client, phone)
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == 0
assert st["round_count"] == 0
assert st["cooldown_until"] is None
def test_reward_status_mid_round(client) -> None:
"""看 1 次(未到一轮)→ round_count=1, cooldown_until 仍 None。"""
phone = "13800003102"
token = _login(client, phone)
uid = _user_id(phone)
_callback(client, _signed(uid, "trans_mid_1"))
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == 1
assert st["round_count"] == 1
assert st["cooldown_until"] is None
def test_reward_status_round_complete_enters_cooldown(client) -> None:
"""刚看完一轮 N 次 → round_count=0(下一轮起点) + cooldown_until 是未来 ~10 分钟。"""
from datetime import datetime, timezone
phone = "13800003103"
token = _login(client, phone)
uid = _user_id(phone)
for i in range(VIDEO_ROUND_REQUIRED_COUNT):
_callback(client, _signed(uid, f"trans_rc_{i}"))
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
assert st["round_count"] == 0
assert st["cooldown_until"] is not None
cd = datetime.fromisoformat(st["cooldown_until"].replace("Z", "+00:00"))
if cd.tzinfo is None:
cd = cd.replace(tzinfo=timezone.utc)
now = datetime.now(timezone.utc)
delta = (cd - now).total_seconds()
# 配置 600s,允许 ±30s 容差(本机 / CI 慢 IO)
assert 0 < delta <= VIDEO_ROUND_COOLDOWN_SECONDS + 30
def test_reward_status_cooldown_expired(client) -> None:
"""看完一轮后冷却已过(手动改 created_at 到 11 分钟前)→ cooldown_until 回到 None。"""
from datetime import datetime, timedelta, timezone
from sqlalchemy import select, update
phone = "13800003104"
token = _login(client, phone)
uid = _user_id(phone)
for i in range(VIDEO_ROUND_REQUIRED_COUNT):
_callback(client, _signed(uid, f"trans_exp_{i}"))
# 把当日所有 granted 记录的 created_at 推到 11 分钟前(覆盖冷却末尾那条)
db = SessionLocal()
try:
eleven_min_ago = datetime.now(timezone.utc) - timedelta(seconds=VIDEO_ROUND_COOLDOWN_SECONDS + 60)
db.execute(
update(AdRewardRecord)
.where(AdRewardRecord.user_id == uid)
.values(created_at=eleven_min_ago)
)
db.commit()
finally:
db.close()
st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
assert st["round_count"] == 0
assert st["cooldown_until"] is None
def test_callback_disabled_returns_503(client, monkeypatch) -> None:
"""未配置回调(开关关)时 → 503。"""
monkeypatch.setattr(settings, "PANGLE_CALLBACK_ENABLED", False)