Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 04edb50acc | |||
| 3483c1bba9 | |||
| 277f9b16a2 | |||
| 9d93b70b9b | |||
| f853938095 | |||
| 783dfd059d |
+10
-2
@@ -88,9 +88,17 @@ AUTO_EXCHANGE_ENABLED=true
|
||||
|
||||
# ===== 穿山甲激励视频(服务端发奖回调)=====
|
||||
# 看完激励视频后穿山甲服务器 S2S 回调本服务发金币(客户端不参与发奖)。
|
||||
# PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",用于验签,从后台取到后填这里;
|
||||
# 配齐并把 ENABLED=true 后,/api/v1/ad/pangle-callback 才受理回调(否则 503)。
|
||||
# 穿山甲"奖励校验密钥"(m-key),验签用,从 GroMore 后台各广告位取到后填这里。
|
||||
# 配齐(任一非空)并把 ENABLED=true 后,/api/v1/ad/pangle-callback 才受理回调(否则 503)。
|
||||
# 每个激励位 m-key 不同但共用同一回调 URL → 各位分开一行配(留空的忽略);验签逐个试、任一通过即接受。
|
||||
PANGLE_CALLBACK_ENABLED=false
|
||||
# 测试应用 激励位 104099649
|
||||
PANGLE_REWARD_SECRET_TEST=
|
||||
# 测试应用 专属激励位 104127529
|
||||
PANGLE_REWARD_SECRET_TEST_DEDICATED=
|
||||
# 正式应用 激励位 104099389
|
||||
PANGLE_REWARD_SECRET_PROD=
|
||||
# (旧用法,仍兼容:单个或逗号分隔的多个 m-key,会与上面三个合并去重)
|
||||
PANGLE_REWARD_SECRET=
|
||||
# ⚠️ 仅本地联调:true 时开放 POST /api/v1/ad/test-grant,让 debug 客户端看完广告直接发奖,
|
||||
# 验证"看广告→金币到账"全链路(未部署公网、穿山甲 S2S 打不到本地时用)。生产必须 false(绕过反作弊)。
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
"""user.username(对外展示账号 ID)+ 默认昵称,并回填存量用户
|
||||
|
||||
Revision ID: b3f1a2c4d5e6
|
||||
Revises: 45047b5a884c
|
||||
Create Date: 2026-06-16 12:00:00.000000
|
||||
|
||||
加 user.username(11 位纯数字、首位非 1 与手机号天然区分、全局唯一)。存量用户:
|
||||
生成唯一 username,昵称为空的补 9 位字母数字随机昵称。
|
||||
迁移自包含(不 import app 业务代码,逻辑冻结为当时快照)。
|
||||
"""
|
||||
import secrets
|
||||
import string
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'b3f1a2c4d5e6'
|
||||
down_revision: Union[str, Sequence[str], None] = '45047b5a884c'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
_USERNAME_FIRST = "23456789" # 首位避前导 0、避 1(手机号都以 1 开头)
|
||||
_USERNAME_DIGITS = "0123456789"
|
||||
_NICKNAME_ALPHABET = string.ascii_letters + string.digits
|
||||
|
||||
|
||||
def _gen_username() -> str:
|
||||
return secrets.choice(_USERNAME_FIRST) + "".join(
|
||||
secrets.choice(_USERNAME_DIGITS) for _ in range(10)
|
||||
)
|
||||
|
||||
|
||||
def _gen_nickname() -> str:
|
||||
return "".join(secrets.choice(_NICKNAME_ALPHABET) for _ in range(9))
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# 1. 先加可空列(存量行此刻还没值)
|
||||
with op.batch_alter_table('user', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('username', sa.String(length=11), nullable=True))
|
||||
|
||||
# 2. 回填存量:每人一个唯一 username;昵称为空的补随机昵称。
|
||||
# 表里 username 原本全空,本批用 used 防互撞即可(无既有非空值需规避)。
|
||||
# "user" 是 PostgreSQL 保留字,必须加双引号(SQLite 也接受双引号标识符)。
|
||||
bind = op.get_bind()
|
||||
rows = bind.execute(sa.text('SELECT id, nickname FROM "user"')).fetchall()
|
||||
used = set()
|
||||
for row in rows:
|
||||
uid, nick = row[0], row[1]
|
||||
while True:
|
||||
uname = _gen_username()
|
||||
if uname not in used:
|
||||
used.add(uname)
|
||||
break
|
||||
if nick is None or not str(nick).strip():
|
||||
bind.execute(
|
||||
sa.text('UPDATE "user" SET username = :u, nickname = :n WHERE id = :i'),
|
||||
{"u": uname, "n": _gen_nickname(), "i": uid},
|
||||
)
|
||||
else:
|
||||
bind.execute(
|
||||
sa.text('UPDATE "user" SET username = :u WHERE id = :i'),
|
||||
{"u": uname, "i": uid},
|
||||
)
|
||||
|
||||
# 3. 收紧:NOT NULL + 唯一索引(对齐模型 unique=True, index=True)
|
||||
with op.batch_alter_table('user', schema=None) as batch_op:
|
||||
batch_op.alter_column('username', existing_type=sa.String(length=11), nullable=False)
|
||||
batch_op.create_index('ix_user_username', ['username'], unique=True)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table('user', schema=None) as batch_op:
|
||||
batch_op.drop_index('ix_user_username')
|
||||
batch_op.drop_column('username')
|
||||
@@ -0,0 +1,61 @@
|
||||
"""cps_link / cps_click tables (群发短链 + 点击统计)
|
||||
|
||||
Revision ID: cps_link_tables
|
||||
Revises: cps_tables
|
||||
Create Date: 2026-06-16 13:20:00.000000
|
||||
|
||||
群发券链接套一层 /c/{code} 做点击统计:cps_link(短码→美团目标) + cps_click(点击事件)。
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
revision: str = "cps_link_tables"
|
||||
down_revision: Union[str, Sequence[str], None] = "cps_tables"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
_NOW = sa.text("CURRENT_TIMESTAMP")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"cps_link",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("code", sa.String(length=16), nullable=False),
|
||||
sa.Column("group_id", sa.Integer(), nullable=False),
|
||||
sa.Column("activity_id", sa.Integer(), nullable=False),
|
||||
sa.Column("sid", sa.String(length=64), nullable=False),
|
||||
sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan"),
|
||||
sa.Column("target_url", sa.String(length=1024), nullable=False),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.UniqueConstraint("code", name="uq_cps_link_code"),
|
||||
)
|
||||
op.create_index("ix_cps_link_code", "cps_link", ["code"])
|
||||
op.create_index("ix_cps_link_group_id", "cps_link", ["group_id"])
|
||||
op.create_index("ix_cps_link_activity_id", "cps_link", ["activity_id"])
|
||||
op.create_index("ix_cps_link_sid", "cps_link", ["sid"])
|
||||
|
||||
op.create_table(
|
||||
"cps_click",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("link_id", sa.Integer(), nullable=False),
|
||||
sa.Column("group_id", sa.Integer(), nullable=False),
|
||||
sa.Column("sid", sa.String(length=64), nullable=False),
|
||||
sa.Column("ip", sa.String(length=64), nullable=True),
|
||||
sa.Column("ua", sa.String(length=512), nullable=True),
|
||||
sa.Column("clicked_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
op.create_index("ix_cps_click_link_id", "cps_click", ["link_id"])
|
||||
op.create_index("ix_cps_click_group_id", "cps_click", ["group_id"])
|
||||
op.create_index("ix_cps_click_sid", "cps_click", ["sid"])
|
||||
op.create_index("ix_cps_click_clicked_at", "cps_click", ["clicked_at"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("cps_click")
|
||||
op.drop_table("cps_link")
|
||||
@@ -0,0 +1,92 @@
|
||||
"""cps_group / cps_activity / cps_order tables (CPS 分发与对账)
|
||||
|
||||
Revision ID: cps_tables
|
||||
Revises: b3f1a2c4d5e6
|
||||
Create Date: 2026-06-16 12:40:00.000000
|
||||
|
||||
群发 CPS 优惠券的分发与对账:cps_group(群=sid) + cps_activity(可推广活动池) +
|
||||
cps_order(美团 query_order 按 sid 拉回的订单明细)。金额统一存「分」。
|
||||
order_id 全局唯一(reconcile 按它 upsert,订单状态会变则更新)。
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
|
||||
revision: str = "cps_tables"
|
||||
down_revision: Union[str, Sequence[str], None] = "b3f1a2c4d5e6"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
_NOW = sa.text("CURRENT_TIMESTAMP")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"cps_group",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("sid", sa.String(length=64), nullable=False),
|
||||
sa.Column("name", sa.String(length=128), nullable=False),
|
||||
sa.Column("member_count", sa.Integer(), nullable=True),
|
||||
sa.Column("status", sa.String(length=20), nullable=False, server_default="active"),
|
||||
sa.Column("remark", sa.String(length=256), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.UniqueConstraint("sid", name="uq_cps_group_sid"),
|
||||
)
|
||||
op.create_index("ix_cps_group_sid", "cps_group", ["sid"])
|
||||
|
||||
op.create_table(
|
||||
"cps_activity",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan"),
|
||||
sa.Column("name", sa.String(length=128), nullable=False),
|
||||
sa.Column("act_id", sa.String(length=64), nullable=True),
|
||||
sa.Column("product_view_sign", sa.String(length=128), nullable=True),
|
||||
sa.Column("status", sa.String(length=20), nullable=False, server_default="active"),
|
||||
sa.Column("remark", sa.String(length=256), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
"cps_order",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("order_id", sa.String(length=64), nullable=False),
|
||||
sa.Column("sid", sa.String(length=64), nullable=True),
|
||||
sa.Column("act_id", sa.String(length=64), nullable=True),
|
||||
sa.Column("biz_line", sa.Integer(), nullable=True),
|
||||
sa.Column("trade_type", sa.Integer(), nullable=True),
|
||||
sa.Column("pay_price_cents", sa.Integer(), nullable=True),
|
||||
sa.Column("commission_cents", sa.Integer(), nullable=True),
|
||||
sa.Column("commission_rate", sa.String(length=16), nullable=True),
|
||||
sa.Column("refund_price_cents", sa.Integer(), nullable=True),
|
||||
sa.Column("refund_profit_cents", sa.Integer(), nullable=True),
|
||||
sa.Column("mt_status", sa.String(length=8), nullable=True),
|
||||
sa.Column("invalid_reason", sa.String(length=128), nullable=True),
|
||||
sa.Column("product_name", sa.String(length=512), nullable=True),
|
||||
sa.Column("pay_time", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column("mt_update_time", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column(
|
||||
"raw",
|
||||
sa.JSON().with_variant(postgresql.JSONB(), "postgresql"),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column("first_seen", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=_NOW, nullable=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.UniqueConstraint("order_id", name="uq_cps_order_order_id"),
|
||||
)
|
||||
op.create_index("ix_cps_order_order_id", "cps_order", ["order_id"])
|
||||
op.create_index("ix_cps_order_sid", "cps_order", ["sid"])
|
||||
op.create_index("ix_cps_order_act_id", "cps_order", ["act_id"])
|
||||
op.create_index("ix_cps_order_mt_status", "cps_order", ["mt_status"])
|
||||
op.create_index("ix_cps_order_pay_time", "cps_order", ["pay_time"])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("cps_order")
|
||||
op.drop_table("cps_activity")
|
||||
op.drop_table("cps_group")
|
||||
@@ -19,6 +19,7 @@ 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.config import router as config_router
|
||||
from app.admin.routers.cps import router as cps_router
|
||||
from app.admin.routers.dashboard import router as dashboard_router
|
||||
from app.admin.routers.ops_stat_config import router as ops_stat_config_router
|
||||
from app.admin.routers.feedback import router as feedback_router
|
||||
@@ -90,5 +91,6 @@ admin_app.include_router(feedback_router)
|
||||
admin_app.include_router(admins_router)
|
||||
admin_app.include_router(audit_router)
|
||||
admin_app.include_router(config_router)
|
||||
admin_app.include_router(cps_router)
|
||||
admin_app.include_router(ad_audit_router)
|
||||
admin_app.include_router(ad_revenue_router)
|
||||
|
||||
@@ -0,0 +1,287 @@
|
||||
"""admin CPS 数据访问:群/活动 CRUD + 美团订单拉单入库(对账) + 按群统计聚合。
|
||||
|
||||
美团 query_order 返回的金额是「元」字符串、时间是秒级时间戳,这里统一转「分」+ tz-aware。
|
||||
统计在 Python 侧聚合(订单量级小、admin 低频),逻辑清晰、跨 PG/SQLite 无 SQL 方言坑。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from uuid import uuid4
|
||||
|
||||
from sqlalchemy import desc, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.admin.repositories.queries import _as_utc, offset_paginate
|
||||
from app.integrations import meituan
|
||||
from app.repositories import cps_link as cps_link_repo
|
||||
from app.models.cps_activity import CpsActivity
|
||||
from app.models.cps_group import CpsGroup
|
||||
from app.models.cps_order import CpsOrder
|
||||
|
||||
# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
|
||||
_INVALID_STATUS = {"4", "5"}
|
||||
_SETTLED_STATUS = "6"
|
||||
|
||||
|
||||
# ───────────── 单位换算 ─────────────
|
||||
def _yuan_to_cents(v: object) -> int | None:
|
||||
"""「元」字符串/数 → 分。None / 空 / 字面 "null" → None。"""
|
||||
if v is None:
|
||||
return None
|
||||
s = str(v).strip()
|
||||
if not s or s.lower() == "null":
|
||||
return None
|
||||
try:
|
||||
return int((Decimal(s) * 100).to_integral_value())
|
||||
except (InvalidOperation, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _ts_to_dt(ts: object) -> datetime | None:
|
||||
"""秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
|
||||
if not ts:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromtimestamp(int(ts), tz=timezone.utc)
|
||||
except (ValueError, OSError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
# ───────────── 群 ─────────────
|
||||
def get_group(db: Session, group_id: int) -> CpsGroup | None:
|
||||
return db.get(CpsGroup, group_id)
|
||||
|
||||
|
||||
def get_group_by_sid(db: Session, sid: str) -> CpsGroup | None:
|
||||
return db.execute(select(CpsGroup).where(CpsGroup.sid == sid)).scalar_one_or_none()
|
||||
|
||||
|
||||
def list_groups(
|
||||
db: Session, *, keyword: str | None = None, status: str | None = None,
|
||||
limit: int = 20, cursor: int | None = None,
|
||||
) -> tuple[list[CpsGroup], int | None, int]:
|
||||
stmt = select(CpsGroup)
|
||||
if keyword and keyword.strip():
|
||||
kw = f"%{keyword.strip()}%"
|
||||
stmt = stmt.where(CpsGroup.name.ilike(kw) | CpsGroup.sid.ilike(kw))
|
||||
if status:
|
||||
stmt = stmt.where(CpsGroup.status == status)
|
||||
return offset_paginate(db, stmt, (desc(CpsGroup.id),), limit=limit, cursor=cursor)
|
||||
|
||||
|
||||
def create_group(
|
||||
db: Session, *, name: str, sid: str | None = None,
|
||||
member_count: int | None = None, remark: str | None = None, commit: bool = True,
|
||||
) -> CpsGroup:
|
||||
"""建群。sid 留空 → 自动 g<id>(先 flush 拿 id 再回填)。"""
|
||||
group = CpsGroup(
|
||||
name=name,
|
||||
sid=sid or f"tmp{uuid4().hex[:20]}", # 临时唯一占位,留空时下面回填 g<id>
|
||||
member_count=member_count,
|
||||
remark=remark,
|
||||
)
|
||||
db.add(group)
|
||||
db.flush()
|
||||
if not sid:
|
||||
group.sid = f"g{group.id}"
|
||||
db.flush()
|
||||
if commit:
|
||||
db.commit()
|
||||
db.refresh(group)
|
||||
return group
|
||||
|
||||
|
||||
def update_group(
|
||||
db: Session, group: CpsGroup, *, name: str | None = None,
|
||||
member_count: int | None = None, status: str | None = None,
|
||||
remark: str | None = None, commit: bool = True,
|
||||
) -> CpsGroup:
|
||||
if name is not None:
|
||||
group.name = name
|
||||
if member_count is not None:
|
||||
group.member_count = member_count
|
||||
if status is not None:
|
||||
group.status = status
|
||||
if remark is not None:
|
||||
group.remark = remark
|
||||
if commit:
|
||||
db.commit()
|
||||
db.refresh(group)
|
||||
else:
|
||||
db.flush()
|
||||
return group
|
||||
|
||||
|
||||
# ───────────── 活动 ─────────────
|
||||
def get_activity(db: Session, activity_id: int) -> CpsActivity | None:
|
||||
return db.get(CpsActivity, activity_id)
|
||||
|
||||
|
||||
def list_activities(
|
||||
db: Session, *, platform: str | None = None, status: str | None = None,
|
||||
limit: int = 20, cursor: int | None = None,
|
||||
) -> tuple[list[CpsActivity], int | None, int]:
|
||||
stmt = select(CpsActivity)
|
||||
if platform:
|
||||
stmt = stmt.where(CpsActivity.platform == platform)
|
||||
if status:
|
||||
stmt = stmt.where(CpsActivity.status == status)
|
||||
return offset_paginate(db, stmt, (desc(CpsActivity.id),), limit=limit, cursor=cursor)
|
||||
|
||||
|
||||
def create_activity(
|
||||
db: Session, *, name: str, platform: str = "meituan", act_id: str | None = None,
|
||||
product_view_sign: str | None = None, remark: str | None = None, commit: bool = True,
|
||||
) -> CpsActivity:
|
||||
activity = CpsActivity(
|
||||
name=name, platform=platform, act_id=act_id,
|
||||
product_view_sign=product_view_sign, remark=remark,
|
||||
)
|
||||
db.add(activity)
|
||||
if commit:
|
||||
db.commit()
|
||||
db.refresh(activity)
|
||||
else:
|
||||
db.flush()
|
||||
return activity
|
||||
|
||||
|
||||
# ───────────── 订单(对账) ─────────────
|
||||
def _map_order_fields(r: dict) -> dict:
|
||||
"""美团 query_order 单条 dataList → CpsOrder 字段(金额转分、时间转 datetime)。"""
|
||||
pn = r.get("productName")
|
||||
if pn and len(pn) > 500:
|
||||
pn = pn[:500]
|
||||
return {
|
||||
"sid": (r.get("sid") or None),
|
||||
"act_id": str(r["actId"]) if r.get("actId") is not None else None,
|
||||
"biz_line": r.get("businessLine"),
|
||||
"trade_type": r.get("tradeType"),
|
||||
"pay_price_cents": _yuan_to_cents(r.get("payPrice")),
|
||||
"commission_cents": _yuan_to_cents(r.get("profit")),
|
||||
"commission_rate": str(r["commissionRate"]) if r.get("commissionRate") is not None else None,
|
||||
"refund_price_cents": _yuan_to_cents(r.get("refundPrice")),
|
||||
"refund_profit_cents": _yuan_to_cents(r.get("refundProfit")),
|
||||
"mt_status": str(r["status"]) if r.get("status") is not None else None,
|
||||
"invalid_reason": (r.get("invalidReason") or None),
|
||||
"product_name": pn or None,
|
||||
"pay_time": _ts_to_dt(r.get("payTime")),
|
||||
"mt_update_time": _ts_to_dt(r.get("updateTime")),
|
||||
"raw": r,
|
||||
}
|
||||
|
||||
|
||||
def reconcile_orders(
|
||||
db: Session, *, start_time: int, end_time: int,
|
||||
query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
|
||||
) -> dict:
|
||||
"""调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
|
||||
|
||||
订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
|
||||
"""
|
||||
fetched = inserted = updated = pages = 0
|
||||
page = 1
|
||||
while page <= max_pages:
|
||||
resp = meituan.query_order(
|
||||
sid=sid, start_time=start_time, end_time=end_time,
|
||||
query_time_type=query_time_type, page=page, limit=100,
|
||||
)
|
||||
rows = ((resp.get("data") or {}).get("dataList")) or []
|
||||
if not rows:
|
||||
break
|
||||
pages += 1
|
||||
for r in rows:
|
||||
order_id = str(r.get("orderId") or "").strip()
|
||||
if not order_id:
|
||||
continue
|
||||
fetched += 1
|
||||
fields = _map_order_fields(r)
|
||||
existing = db.execute(
|
||||
select(CpsOrder).where(CpsOrder.order_id == order_id)
|
||||
).scalar_one_or_none()
|
||||
if existing is None:
|
||||
db.add(CpsOrder(order_id=order_id, **fields))
|
||||
inserted += 1
|
||||
else:
|
||||
for k, v in fields.items():
|
||||
setattr(existing, k, v)
|
||||
updated += 1
|
||||
if len(rows) < 100:
|
||||
break
|
||||
page += 1
|
||||
db.commit()
|
||||
return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
|
||||
|
||||
|
||||
def list_orders(
|
||||
db: Session, *, sid: str | None = None, mt_status: str | None = None,
|
||||
limit: int = 20, cursor: int | None = None,
|
||||
) -> tuple[list[CpsOrder], int | None, int]:
|
||||
stmt = select(CpsOrder)
|
||||
if sid:
|
||||
stmt = stmt.where(CpsOrder.sid == sid)
|
||||
if mt_status:
|
||||
stmt = stmt.where(CpsOrder.mt_status == mt_status)
|
||||
return offset_paginate(db, stmt, (desc(CpsOrder.id),), limit=limit, cursor=cursor)
|
||||
|
||||
|
||||
# ───────────── 统计(按群聚合) ─────────────
|
||||
def group_stats(
|
||||
db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
|
||||
) -> list[dict]:
|
||||
"""按 sid 聚合订单 + join 群信息。已建但本期无单的活跃群也列出(全 0)。
|
||||
未归群的 sid(历史/其它来源)单独成行 group_id=None。按预估佣金降序。
|
||||
"""
|
||||
stmt = select(CpsOrder)
|
||||
if date_from is not None:
|
||||
stmt = stmt.where(CpsOrder.pay_time >= _as_utc(date_from))
|
||||
if date_to is not None:
|
||||
stmt = stmt.where(CpsOrder.pay_time <= _as_utc(date_to))
|
||||
orders = list(db.execute(stmt).scalars().all())
|
||||
|
||||
groups = {g.sid: g for g in db.execute(select(CpsGroup)).scalars().all()}
|
||||
|
||||
buckets: dict[str | None, list[CpsOrder]] = {}
|
||||
for o in orders:
|
||||
buckets.setdefault(o.sid, []).append(o)
|
||||
|
||||
rows: list[dict] = []
|
||||
for sid, items in buckets.items():
|
||||
g = groups.get(sid) if sid else None
|
||||
valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
|
||||
settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
|
||||
canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
|
||||
rows.append({
|
||||
"group_id": g.id if g else None,
|
||||
"sid": sid,
|
||||
"name": g.name if g else (sid or "(无 sid 归属)"),
|
||||
"member_count": g.member_count if g else None,
|
||||
"order_count": len(valid),
|
||||
"settled_count": len(settled),
|
||||
"canceled_count": len(canceled),
|
||||
"gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
|
||||
"est_commission_cents": sum(o.commission_cents or 0 for o in valid),
|
||||
"settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
|
||||
})
|
||||
|
||||
# 已建的活跃群但本期无单 → 补 0 行,让运营看到全部群
|
||||
seen = set(buckets.keys())
|
||||
for sid, g in groups.items():
|
||||
if sid not in seen and g.status == "active":
|
||||
rows.append({
|
||||
"group_id": g.id, "sid": sid, "name": g.name,
|
||||
"member_count": g.member_count, "order_count": 0,
|
||||
"settled_count": 0, "canceled_count": 0, "gmv_cents": 0,
|
||||
"est_commission_cents": 0, "settled_commission_cents": 0,
|
||||
})
|
||||
|
||||
# 合并点击数据(按 group_id):点击只对生成过我们短链的群有,未归群行(group_id=None)恒 0
|
||||
clicks = cps_link_repo.click_stats_by_group(db, date_from=date_from, date_to=date_to)
|
||||
for r in rows:
|
||||
c = clicks.get(r["group_id"]) if r["group_id"] is not None else None
|
||||
r["click_pv"] = c["pv"] if c else 0
|
||||
r["click_uv"] = c["uv"] if c else 0
|
||||
|
||||
rows.sort(key=lambda x: x["est_commission_cents"], reverse=True)
|
||||
return rows
|
||||
@@ -43,6 +43,9 @@ def _validate(key: str, value: Any) -> None:
|
||||
for k, v in value.items()
|
||||
):
|
||||
raise ValueError("需为 {字符串: 整数} 映射")
|
||||
elif t == "bool":
|
||||
if not isinstance(value, bool):
|
||||
raise ValueError("需为布尔值")
|
||||
|
||||
|
||||
def _item(db, key: str) -> ConfigItemOut:
|
||||
|
||||
@@ -0,0 +1,268 @@
|
||||
"""admin CPS 分发与对账:群(sid)管理 + 活动池 + 生成带 sid 券链接 + 美团订单对账 + 统计。
|
||||
|
||||
当前仅接美团(get_referral_link / query_order)。淘宝/京东待各自联盟凭证。
|
||||
群/活动管理 = operator;订单对账(涉佣金) = finance;只读列表/统计 = 登录即可。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
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 cps as cps_repo
|
||||
from app.admin.schemas.common import CursorPage
|
||||
from app.core.config import settings
|
||||
from app.repositories import cps_link as cps_link_repo
|
||||
from app.admin.schemas.cps import (
|
||||
CpsActivityCreate,
|
||||
CpsActivityOut,
|
||||
CpsGroupCreate,
|
||||
CpsGroupOut,
|
||||
CpsGroupStat,
|
||||
CpsGroupUpdate,
|
||||
CpsOrderOut,
|
||||
CpsReconcileResult,
|
||||
CpsReferralLinkOut,
|
||||
CpsReferralLinkRequest,
|
||||
CpsStatsOut,
|
||||
)
|
||||
from app.integrations import meituan
|
||||
from app.integrations.meituan import MeituanCpsError
|
||||
from app.models.admin import AdminUser
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/admin/api/cps",
|
||||
tags=["admin-cps"],
|
||||
dependencies=[Depends(get_current_admin)],
|
||||
)
|
||||
|
||||
|
||||
# ───────────── 群(sid) ─────────────
|
||||
@router.get("/groups", response_model=CursorPage[CpsGroupOut], summary="群(sid)列表")
|
||||
def list_groups(
|
||||
db: AdminDb,
|
||||
keyword: Annotated[str | None, Query(max_length=100)] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(active|archived)$")] = None,
|
||||
limit: Annotated[int, Query(ge=1, le=100)] = 20,
|
||||
cursor: Annotated[int | None, Query()] = None,
|
||||
) -> CursorPage[CpsGroupOut]:
|
||||
items, next_cursor, total = cps_repo.list_groups(
|
||||
db, keyword=keyword, status=status, limit=limit, cursor=cursor
|
||||
)
|
||||
return CursorPage(
|
||||
items=[CpsGroupOut.model_validate(g) for g in items],
|
||||
next_cursor=next_cursor,
|
||||
total=total,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/groups", response_model=CpsGroupOut, summary="新建群(分配 sid)")
|
||||
def create_group(
|
||||
body: CpsGroupCreate,
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
) -> CpsGroupOut:
|
||||
if body.sid and cps_repo.get_group_by_sid(db, body.sid) is not None:
|
||||
raise HTTPException(status_code=409, detail="sid 已存在")
|
||||
group = cps_repo.create_group(
|
||||
db, name=body.name, sid=body.sid, member_count=body.member_count,
|
||||
remark=body.remark, commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="cps.group.create", target_type="cps_group", target_id=group.id,
|
||||
detail={"sid": group.sid, "name": group.name}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(group)
|
||||
return CpsGroupOut.model_validate(group)
|
||||
|
||||
|
||||
@router.patch("/groups/{group_id}", response_model=CpsGroupOut, summary="编辑群")
|
||||
def update_group(
|
||||
group_id: int,
|
||||
body: CpsGroupUpdate,
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
) -> CpsGroupOut:
|
||||
group = cps_repo.get_group(db, group_id)
|
||||
if group is None:
|
||||
raise HTTPException(status_code=404, detail="群不存在")
|
||||
cps_repo.update_group(
|
||||
db, group, name=body.name, member_count=body.member_count,
|
||||
status=body.status, remark=body.remark, commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="cps.group.update", target_type="cps_group", target_id=group_id,
|
||||
detail=body.model_dump(exclude_none=True), ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(group)
|
||||
return CpsGroupOut.model_validate(group)
|
||||
|
||||
|
||||
# ───────────── 活动 ─────────────
|
||||
@router.get("/activities", response_model=CursorPage[CpsActivityOut], summary="活动列表")
|
||||
def list_activities(
|
||||
db: AdminDb,
|
||||
platform: Annotated[str | None, Query(max_length=20)] = None,
|
||||
status: Annotated[str | None, Query(pattern="^(active|archived)$")] = None,
|
||||
limit: Annotated[int, Query(ge=1, le=100)] = 20,
|
||||
cursor: Annotated[int | None, Query()] = None,
|
||||
) -> CursorPage[CpsActivityOut]:
|
||||
items, next_cursor, total = cps_repo.list_activities(
|
||||
db, platform=platform, status=status, limit=limit, cursor=cursor
|
||||
)
|
||||
return CursorPage(
|
||||
items=[CpsActivityOut.model_validate(a) for a in items],
|
||||
next_cursor=next_cursor,
|
||||
total=total,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/activities", response_model=CpsActivityOut, summary="新建活动")
|
||||
def create_activity(
|
||||
body: CpsActivityCreate,
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
) -> CpsActivityOut:
|
||||
if not body.act_id and not body.product_view_sign:
|
||||
raise HTTPException(status_code=400, detail="act_id 或 product_view_sign 必填其一")
|
||||
activity = cps_repo.create_activity(
|
||||
db, name=body.name, platform=body.platform, act_id=body.act_id,
|
||||
product_view_sign=body.product_view_sign, remark=body.remark, commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="cps.activity.create", target_type="cps_activity", target_id=activity.id,
|
||||
detail={"name": activity.name, "act_id": activity.act_id, "platform": activity.platform},
|
||||
ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(activity)
|
||||
return CpsActivityOut.model_validate(activity)
|
||||
|
||||
|
||||
# ───────────── 转链 ─────────────
|
||||
@router.post("/referral-link", response_model=CpsReferralLinkOut, summary="生成群发短链(带 sid + 点击统计)")
|
||||
def generate_referral_link(
|
||||
body: CpsReferralLinkRequest,
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
) -> CpsReferralLinkOut:
|
||||
group = cps_repo.get_group(db, body.group_id)
|
||||
if group is None:
|
||||
raise HTTPException(status_code=404, detail="群不存在")
|
||||
activity = cps_repo.get_activity(db, body.activity_id)
|
||||
if activity is None:
|
||||
raise HTTPException(status_code=404, detail="活动不存在")
|
||||
if activity.platform != "meituan":
|
||||
raise HTTPException(status_code=400, detail=f"当前仅支持美团转链(活动平台={activity.platform})")
|
||||
try:
|
||||
# 全要:短链作跳转目标(微信可打开),长链/deeplink 作参考/备用
|
||||
resp = meituan.get_referral_link(
|
||||
act_id=activity.act_id,
|
||||
product_view_sign=activity.product_view_sign,
|
||||
sid=group.sid,
|
||||
link_type_list=[1, 2, 3],
|
||||
)
|
||||
except MeituanCpsError as e:
|
||||
raise HTTPException(status_code=502, detail=f"美团转链失败: {e}") from e
|
||||
|
||||
link_map = {str(k): v for k, v in (resp.get("referralLinkMap") or {}).items()}
|
||||
if not link_map and resp.get("data"):
|
||||
link_map = {"1": resp["data"]}
|
||||
# 跳转目标优先短链(2,微信可打开)→ 长链(1)→ deeplink(3)
|
||||
target_url = link_map.get("2") or link_map.get("1") or link_map.get("3")
|
||||
if not target_url:
|
||||
raise HTTPException(status_code=502, detail="美团未返回有效链接(检查 actId 是否在推广有效期)")
|
||||
|
||||
# 存我们的短链:发群用 /c/{code},用户点 → 我们记点击 → 302 跳 target_url
|
||||
link = cps_link_repo.create_link(
|
||||
db, group_id=group.id, activity_id=activity.id, sid=group.sid,
|
||||
target_url=target_url, platform=activity.platform, commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="cps.referral_link.generate", target_type="cps_link", target_id=link.id,
|
||||
detail={"sid": group.sid, "activity_id": activity.id, "code": link.code},
|
||||
ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(link)
|
||||
|
||||
base = settings.CPS_REDIRECT_BASE.rstrip("/")
|
||||
redirect_url = f"{base}/c/{link.code}" if base else f"/c/{link.code}"
|
||||
return CpsReferralLinkOut(
|
||||
redirect_url=redirect_url,
|
||||
code=link.code,
|
||||
sid=group.sid,
|
||||
group_name=group.name,
|
||||
activity_name=activity.name,
|
||||
target_url=target_url,
|
||||
link_map=link_map,
|
||||
)
|
||||
|
||||
|
||||
# ───────────── 订单对账 ─────────────
|
||||
@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
|
||||
def reconcile_orders(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("finance"))],
|
||||
db: AdminDb,
|
||||
days: Annotated[int, Query(ge=1, le=90)] = 7,
|
||||
sid: Annotated[str | None, Query(max_length=64)] = None,
|
||||
) -> CpsReconcileResult:
|
||||
now = int(time.time())
|
||||
try:
|
||||
result = cps_repo.reconcile_orders(
|
||||
db, start_time=now - days * 86400, end_time=now, sid=sid,
|
||||
)
|
||||
except MeituanCpsError as e:
|
||||
raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
|
||||
write_audit(
|
||||
db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
|
||||
detail={"days": days, "sid": sid, **result}, ip=get_client_ip(request), commit=True,
|
||||
)
|
||||
return CpsReconcileResult(**result)
|
||||
|
||||
|
||||
@router.get("/orders", response_model=CursorPage[CpsOrderOut], summary="订单明细")
|
||||
def list_orders(
|
||||
db: AdminDb,
|
||||
sid: Annotated[str | None, Query(max_length=64)] = None,
|
||||
mt_status: Annotated[str | None, Query(max_length=8)] = None,
|
||||
limit: Annotated[int, Query(ge=1, le=100)] = 20,
|
||||
cursor: Annotated[int | None, Query()] = None,
|
||||
) -> CursorPage[CpsOrderOut]:
|
||||
items, next_cursor, total = cps_repo.list_orders(
|
||||
db, sid=sid, mt_status=mt_status, limit=limit, cursor=cursor
|
||||
)
|
||||
return CursorPage(
|
||||
items=[CpsOrderOut.model_validate(o) for o in items],
|
||||
next_cursor=next_cursor,
|
||||
total=total,
|
||||
)
|
||||
|
||||
|
||||
# ───────────── 统计 ─────────────
|
||||
@router.get("/stats", response_model=CpsStatsOut, summary="按群对账统计")
|
||||
def get_stats(
|
||||
db: AdminDb,
|
||||
days: Annotated[int, Query(ge=1, le=90)] = 30,
|
||||
) -> CpsStatsOut:
|
||||
date_to = datetime.now(timezone.utc)
|
||||
date_from = date_to - timedelta(days=days)
|
||||
rows = cps_repo.group_stats(db, date_from=date_from, date_to=date_to)
|
||||
stats = [CpsGroupStat(**r) for r in rows]
|
||||
return CpsStatsOut(
|
||||
groups=stats,
|
||||
total_order_count=sum(s.order_count for s in stats),
|
||||
total_est_commission_cents=sum(s.est_commission_cents for s in stats),
|
||||
total_settled_commission_cents=sum(s.settled_commission_cents for s in stats),
|
||||
)
|
||||
@@ -34,6 +34,7 @@ from app.admin.schemas.wallet import (
|
||||
from app.core.config import settings
|
||||
from app.integrations import wxpay
|
||||
from app.models.admin import AdminUser
|
||||
from app.repositories import app_config
|
||||
from app.repositories import wallet as wallet_repo
|
||||
|
||||
router = APIRouter(
|
||||
@@ -95,7 +96,7 @@ def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
|
||||
summary="提现配置健康检查",
|
||||
dependencies=[Depends(require_role("finance"))], # 暴露密钥路径/配置,限财务+super
|
||||
)
|
||||
def withdraw_health_check() -> WxpayHealthCheckOut:
|
||||
def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
|
||||
private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) # noqa: SLF001
|
||||
public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) # noqa: SLF001
|
||||
issues: list[str] = []
|
||||
@@ -117,8 +118,15 @@ def withdraw_health_check() -> WxpayHealthCheckOut:
|
||||
issues.append("微信支付基础配置不完整")
|
||||
if not settings.WXPAY_AUTH_NOTIFY_URL:
|
||||
issues.append("免确认授权回调地址未配置")
|
||||
if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
|
||||
issues.append("自动对账未开启")
|
||||
|
||||
# 实际是否自动对账 = env 部署总闸(worker 起没起)AND 运营后台 DB 开关(本轮跑不跑)。
|
||||
worker_running = settings.WITHDRAW_AUTO_RECONCILE_ENABLED
|
||||
daily_on = bool(app_config.get_value(db, "withdraw_auto_reconcile_enabled"))
|
||||
auto_reconcile_enabled = worker_running and daily_on
|
||||
if not worker_running:
|
||||
issues.append("自动对账 worker 未启动(部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=false)")
|
||||
elif not daily_on:
|
||||
issues.append("自动对账运营开关已关闭(系统配置页可开)")
|
||||
|
||||
return WxpayHealthCheckOut(
|
||||
ok=not issues,
|
||||
@@ -131,7 +139,7 @@ def withdraw_health_check() -> WxpayHealthCheckOut:
|
||||
public_key_exists=public_path.exists(),
|
||||
public_key_loadable=public_loadable,
|
||||
auth_notify_url_configured=bool(settings.WXPAY_AUTH_NOTIFY_URL),
|
||||
auto_reconcile_enabled=settings.WITHDRAW_AUTO_RECONCILE_ENABLED,
|
||||
auto_reconcile_enabled=auto_reconcile_enabled,
|
||||
auto_reconcile_interval_sec=settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC,
|
||||
auto_reconcile_older_than_minutes=settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES,
|
||||
issues=issues,
|
||||
|
||||
@@ -10,7 +10,7 @@ class ConfigItemOut(BaseModel):
|
||||
key: str
|
||||
label: str
|
||||
group: str
|
||||
type: str # int / int_list / dict_str_int
|
||||
type: str # int / int_list / dict_str_int / bool
|
||||
help: str | None = None
|
||||
default: Any
|
||||
value: Any
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
"""admin CPS 分发与对账 schemas。金额统一「分」(cents),前端 yuan() 展示。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
# ───────────── 群(sid) ─────────────
|
||||
class CpsGroupOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
sid: str
|
||||
name: str
|
||||
member_count: int | None = None
|
||||
status: str
|
||||
remark: str | None = None
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class CpsGroupCreate(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=128)
|
||||
# 留空 → 后端自动生成 g<id>;填写则必须字母+数字(美团 sid 限制)。
|
||||
sid: str | None = Field(default=None, max_length=64, pattern=r"^[A-Za-z0-9]+$")
|
||||
member_count: int | None = Field(default=None, ge=0)
|
||||
remark: str | None = Field(default=None, max_length=256)
|
||||
|
||||
|
||||
class CpsGroupUpdate(BaseModel):
|
||||
name: str | None = Field(default=None, min_length=1, max_length=128)
|
||||
member_count: int | None = Field(default=None, ge=0)
|
||||
status: str | None = Field(default=None, pattern=r"^(active|archived)$")
|
||||
remark: str | None = Field(default=None, max_length=256)
|
||||
|
||||
|
||||
# ───────────── 活动(actId) ─────────────
|
||||
class CpsActivityOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
platform: str
|
||||
name: str
|
||||
act_id: str | None = None
|
||||
product_view_sign: str | None = None
|
||||
status: str
|
||||
remark: str | None = None
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class CpsActivityCreate(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=128)
|
||||
platform: str = Field(default="meituan", pattern=r"^(meituan|taobao|jd)$")
|
||||
act_id: str | None = Field(default=None, max_length=64)
|
||||
product_view_sign: str | None = Field(default=None, max_length=128)
|
||||
remark: str | None = Field(default=None, max_length=256)
|
||||
|
||||
|
||||
# ───────────── 转链 ─────────────
|
||||
class CpsReferralLinkRequest(BaseModel):
|
||||
group_id: int
|
||||
activity_id: int
|
||||
# 链接类型:1长链 2短链 3deeplink。默认短链(微信可打开)+deeplink。
|
||||
link_types: list[int] = Field(default_factory=lambda: [2, 3])
|
||||
|
||||
|
||||
class CpsReferralLinkOut(BaseModel):
|
||||
redirect_url: str # ★ 我们的群发短链(/c/{code}),发群就用这个,点击经我们统计
|
||||
code: str
|
||||
sid: str
|
||||
group_name: str
|
||||
activity_name: str
|
||||
target_url: str # 实际 302 跳转的美团短链
|
||||
link_map: dict[str, str] # 美团原始 1长链/2短链/3deeplink(参考/备用)
|
||||
|
||||
|
||||
# ───────────── 对账拉单 ─────────────
|
||||
class CpsReconcileResult(BaseModel):
|
||||
fetched: int # 本次美团返回的订单条数
|
||||
inserted: int # 新入库
|
||||
updated: int # 已存在→更新状态/佣金
|
||||
pages: int # 翻了几页
|
||||
|
||||
|
||||
# ───────────── 订单明细 ─────────────
|
||||
class CpsOrderOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
order_id: str
|
||||
sid: str | None = None
|
||||
act_id: str | None = None
|
||||
pay_price_cents: int | None = None
|
||||
commission_cents: int | None = None
|
||||
commission_rate: str | None = None
|
||||
mt_status: str | None = None
|
||||
invalid_reason: str | None = None
|
||||
product_name: str | None = None
|
||||
pay_time: datetime | None = None
|
||||
|
||||
|
||||
# ───────────── 统计 ─────────────
|
||||
class CpsGroupStat(BaseModel):
|
||||
group_id: int | None = None # None = 未归群的 sid(历史/其它来源)
|
||||
sid: str | None = None
|
||||
name: str
|
||||
member_count: int | None = None
|
||||
click_pv: int = 0 # 点击总次数
|
||||
click_uv: int = 0 # 独立点击(按 ip+ua 近似去重)
|
||||
order_count: int # 有效订单数(不含取消/风控)
|
||||
settled_count: int # 已结算单数
|
||||
canceled_count: int # 取消+风控单数
|
||||
gmv_cents: int # 成交额(有效单 payPrice 合计)
|
||||
est_commission_cents: int # 预估佣金(有效单 profit 合计)
|
||||
settled_commission_cents: int # 已结算佣金
|
||||
|
||||
|
||||
class CpsStatsOut(BaseModel):
|
||||
groups: list[CpsGroupStat]
|
||||
total_order_count: int
|
||||
total_est_commission_cents: int
|
||||
total_settled_commission_cents: int
|
||||
+1
-1
@@ -87,7 +87,7 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
|
||||
|
||||
params = dict(request.query_params)
|
||||
|
||||
if not pangle.verify_callback_sign(params, settings.PANGLE_REWARD_SECRET):
|
||||
if not pangle.verify_callback_sign_any(params, settings.pangle_reward_secrets):
|
||||
logger.warning("pangle callback bad sign trans_id=%s", params.get("trans_id"))
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="bad sign")
|
||||
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
"""CPS 群发短链跳转:用户点 /c/{code} → 记一条点击 → 302 跳美团。
|
||||
|
||||
公网、无鉴权(群里任何人点都要能跳)。记点击失败绝不影响跳转(用户体验优先)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, Request
|
||||
from fastapi.responses import RedirectResponse
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.db.session import get_db
|
||||
from app.repositories import cps_link as cps_link_repo
|
||||
|
||||
router = APIRouter(tags=["cps-redirect"])
|
||||
|
||||
# code 不存在/失效时的兜底落地(避免用户看到报错)
|
||||
_FALLBACK_URL = "https://www.meituan.com/"
|
||||
|
||||
|
||||
def _client_ip(request: Request) -> str | None:
|
||||
xff = request.headers.get("x-forwarded-for")
|
||||
if xff:
|
||||
return xff.split(",")[0].strip()
|
||||
return request.client.host if request.client else None
|
||||
|
||||
|
||||
@router.get("/c/{code}", summary="群发短链跳转(记点击 + 302)")
|
||||
def cps_redirect(code: str, request: Request, db: Session = Depends(get_db)):
|
||||
link = cps_link_repo.get_by_code(db, code)
|
||||
if link is None:
|
||||
return RedirectResponse(_FALLBACK_URL, status_code=302)
|
||||
try:
|
||||
cps_link_repo.record_click(
|
||||
db, link=link, ip=_client_ip(request),
|
||||
ua=request.headers.get("user-agent"),
|
||||
)
|
||||
except Exception:
|
||||
pass # 记点击失败不阻断跳转
|
||||
return RedirectResponse(link.target_url, status_code=302)
|
||||
+17
-1
@@ -2,6 +2,7 @@
|
||||
|
||||
路由前缀 `/api/v1/platform`:
|
||||
GET /stats 首页三统计(帮助用户 / 完成比价 / 累计节省),按运营后台配的模式算。
|
||||
GET /flags 客户端运营 feature flag(比价/领券期广告开关等),客户端拉取后缓存。
|
||||
|
||||
展示模式(real/manual/random,每指标独立)与计算逻辑见 app/repositories/ops_stat.py。
|
||||
"""
|
||||
@@ -12,9 +13,15 @@ import logging
|
||||
from fastapi import APIRouter, Query
|
||||
|
||||
from app.api.deps import DbSession
|
||||
from app.repositories import app_config
|
||||
from app.repositories import ops_marquee as marquee_crud
|
||||
from app.repositories import ops_stat as crud
|
||||
from app.schemas.platform import PlatformStatsOut, SavingsFeedItem, SavingsFeedOut
|
||||
from app.schemas.platform import (
|
||||
AppFlagsOut,
|
||||
PlatformStatsOut,
|
||||
SavingsFeedItem,
|
||||
SavingsFeedOut,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("shagua.platform")
|
||||
|
||||
@@ -35,3 +42,12 @@ def stats(db: DbSession) -> PlatformStatsOut:
|
||||
def savings_feed(db: DbSession, limit: int = Query(8, ge=1, le=30)) -> SavingsFeedOut:
|
||||
items = marquee_crud.get_feed(db, limit=limit)
|
||||
return SavingsFeedOut(items=[SavingsFeedItem(**it) for it in items])
|
||||
|
||||
|
||||
@router.get("/flags", response_model=AppFlagsOut, summary="客户端运营 feature flag(不鉴权)")
|
||||
def flags(db: DbSession) -> AppFlagsOut:
|
||||
"""客户端拉取运营开关并缓存(app 启动 / 每场比价开始时刷新)。不鉴权:开关非敏感,
|
||||
且比价无障碍服务取值时未必有登录态。值来自 app_config(admin 可改),空库回退默认。"""
|
||||
return AppFlagsOut(
|
||||
comparing_ad_enabled=bool(app_config.get_value(db, "comparing_ad_enabled")),
|
||||
)
|
||||
|
||||
+30
-3
@@ -136,8 +136,15 @@ class Settings(BaseSettings):
|
||||
|
||||
# ===== 穿山甲激励视频(服务端发奖回调)=====
|
||||
# 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
|
||||
# PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",验签用,从后台取到后填 .env。
|
||||
# 穿山甲后台配置的"奖励校验密钥"(m-key),验签用。每个 GroMore 广告位 m-key 不同(后台各自
|
||||
# 生成),但共用同一回调 URL → 多个位的 m-key 都要配上。验签时所有非空 m-key 逐个试、任一通过
|
||||
# 即接受(见 pangle.verify_callback_sign_any / 下面的 pangle_reward_secrets)。
|
||||
PANGLE_CALLBACK_ENABLED: bool = False
|
||||
# 推荐:每个激励位的 m-key 分开一行配,清晰不混淆(留空的忽略)。
|
||||
PANGLE_REWARD_SECRET_TEST: str = "" # 测试应用 激励位 104099649
|
||||
PANGLE_REWARD_SECRET_TEST_DEDICATED: str = "" # 测试应用 专属激励位 104127529
|
||||
PANGLE_REWARD_SECRET_PROD: str = "" # 正式应用 激励位 104099389
|
||||
# 旧用法:单个或逗号分隔的多个 m-key,仍兼容(会与上面三个命名项合并去重)。
|
||||
PANGLE_REWARD_SECRET: str = ""
|
||||
|
||||
# ⚠️ 仅本地联调:打开后开放 POST /api/v1/ad/test-grant,让(已登录的)客户端在没部署公网、
|
||||
@@ -145,10 +152,23 @@ class Settings(BaseSettings):
|
||||
# 它让客户端能自助发奖 = 绕过反作弊,**生产必须保持 False**(默认 False;只在本地 .env 设 true)。
|
||||
AD_REWARD_TEST_GRANT_ENABLED: bool = False
|
||||
|
||||
@property
|
||||
def pangle_reward_secrets(self) -> list[str]:
|
||||
"""汇总所有 m-key 成列表(去空白、去空项、去重保序)。验签时逐个试、任一通过即接受
|
||||
(见 pangle.verify_callback_sign_any)。来源可混用:三个命名项 + 旧的逗号分隔 PANGLE_REWARD_SECRET。"""
|
||||
raw = [
|
||||
*self.PANGLE_REWARD_SECRET.split(","),
|
||||
self.PANGLE_REWARD_SECRET_TEST,
|
||||
self.PANGLE_REWARD_SECRET_TEST_DEDICATED,
|
||||
self.PANGLE_REWARD_SECRET_PROD,
|
||||
]
|
||||
cleaned = [s.strip() for s in raw if s and s.strip()]
|
||||
return list(dict.fromkeys(cleaned)) # 去重保序
|
||||
|
||||
@property
|
||||
def pangle_callback_configured(self) -> bool:
|
||||
"""回调开关打开且验签密钥已配,才接受发奖回调。"""
|
||||
return bool(self.PANGLE_CALLBACK_ENABLED and self.PANGLE_REWARD_SECRET)
|
||||
"""回调开关打开且至少配了一个验签密钥,才接受发奖回调。"""
|
||||
return bool(self.PANGLE_CALLBACK_ENABLED and self.pangle_reward_secrets)
|
||||
|
||||
# ===== Pricebot 上游 (领券/比价业务透传目标) =====
|
||||
# pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step
|
||||
@@ -190,6 +210,13 @@ class Settings(BaseSettings):
|
||||
# MVP 落地页就放 app-server 的 /media 静态目录下(dl.html)。
|
||||
INVITE_LANDING_URL: str = "https://app-api.shaguabijia.com/media/dl.html"
|
||||
|
||||
# ===== CPS 群发短链跳转 =====
|
||||
# 群发券链接套一层我们的短链 /c/{code} 做点击统计后 302 跳美团。这里是写进 admin
|
||||
# 生成链接里的对外跳转域名前缀。本地填 http://localhost:8770;生产**强烈建议用独立
|
||||
# 域名**(大量群发会被微信封,别用主 API 域名连累整个 App)。留空 → 跳转端点用请求
|
||||
# 的 Host 兜底拼绝对地址。
|
||||
CPS_REDIRECT_BASE: str = ""
|
||||
|
||||
# ===== CORS =====
|
||||
CORS_ALLOW_ORIGINS: str = ""
|
||||
|
||||
|
||||
@@ -11,7 +11,7 @@ from typing import Any
|
||||
|
||||
from app.core import rewards as r
|
||||
|
||||
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int
|
||||
# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool
|
||||
CONFIG_DEFS: dict[str, dict[str, Any]] = {
|
||||
"signin_rewards": {
|
||||
"default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
|
||||
@@ -65,4 +65,23 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
|
||||
"group": "签到", "type": "int",
|
||||
"help": "Day1-Day6 签到后看完激励视频额外发放的固定金币;Day7 不展示也不允许膨胀。",
|
||||
},
|
||||
"comparing_ad_enabled": {
|
||||
"default": True, "label": "比价/领券期信息流广告",
|
||||
"group": "看广告", "type": "bool",
|
||||
"help": (
|
||||
"开启后,比价进行中 + 领券等候期会在悬浮窗展示穿山甲信息流广告(变现行为);"
|
||||
"关闭则全程不出广告。客户端按 app 启动 / 每场比价开始时拉取并缓存,故为「最终一致」的"
|
||||
"远程开关(下一场比价生效),用于出问题时无需发版即可快速止血。debug 包可用本地开关覆盖。"
|
||||
),
|
||||
},
|
||||
"withdraw_auto_reconcile_enabled": {
|
||||
"default": True, "label": "提现自动对账",
|
||||
"group": "钱包", "type": "bool",
|
||||
"help": (
|
||||
"开启后后台 worker 每隔一段时间自动扫描超时仍「打款中」的提现单并归一化"
|
||||
"(查微信/退款/撤单)。需部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=true 启动 worker "
|
||||
"进程后此开关才起效;扫描间隔/超时阈值仍由 env 控制。关掉只停自动扫描,"
|
||||
"提现页「批量对账」手动按钮不受影响。"
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -14,8 +14,11 @@ from sqlalchemy.exc import SQLAlchemyError
|
||||
from app.core.config import settings
|
||||
from app.db.session import SessionLocal
|
||||
from app.integrations.wxpay import WxPayNotConfiguredError
|
||||
from app.repositories import app_config
|
||||
from app.repositories import wallet as wallet_repo
|
||||
|
||||
_AUTO_RECONCILE_KEY = "withdraw_auto_reconcile_enabled"
|
||||
|
||||
logger = logging.getLogger("shagua.withdraw_reconcile")
|
||||
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.lock"
|
||||
|
||||
@@ -59,8 +62,15 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
|
||||
_LOCK_PATH.unlink()
|
||||
|
||||
|
||||
def _reconcile_once(older_than_minutes: int) -> dict:
|
||||
def _reconcile_once(older_than_minutes: int) -> dict | None:
|
||||
"""读运营开关(app_config):关着返回 None(本轮跳过),开着才真扫单。
|
||||
|
||||
env WITHDRAW_AUTO_RECONCILE_ENABLED 是部署级总闸(决定 worker 起不起);
|
||||
这里的 DB 开关是运营级日常开关,后台一改下一轮即生效、跨进程一致、无需重启。
|
||||
"""
|
||||
with SessionLocal() as db:
|
||||
if not app_config.get_value(db, _AUTO_RECONCILE_KEY):
|
||||
return None
|
||||
return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
|
||||
|
||||
|
||||
@@ -78,16 +88,18 @@ async def _run_loop() -> None:
|
||||
|
||||
async def _run_locked_loop(interval: int, older_than: int) -> None:
|
||||
logger.info(
|
||||
"withdraw auto reconcile started interval=%ss older_than=%sm",
|
||||
"withdraw auto reconcile worker started interval=%ss older_than=%sm "
|
||||
"(runtime on/off via app_config '%s')",
|
||||
interval,
|
||||
older_than,
|
||||
_AUTO_RECONCILE_KEY,
|
||||
)
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
_touch_lock()
|
||||
result = await asyncio.to_thread(_reconcile_once, older_than)
|
||||
if result["checked"] or result["resolved"]:
|
||||
if result is not None and (result["checked"] or result["resolved"]):
|
||||
logger.info("withdraw auto reconcile result=%s", result)
|
||||
except WxPayNotConfiguredError:
|
||||
logger.warning("withdraw auto reconcile skipped: wxpay not configured")
|
||||
@@ -103,7 +115,8 @@ async def _run_locked_loop(interval: int, older_than: int) -> None:
|
||||
|
||||
def start_withdraw_reconcile_worker() -> asyncio.Task | None:
|
||||
if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
|
||||
logger.info("withdraw auto reconcile disabled")
|
||||
# 部署级总闸关:worker 进程不启动(运营后台开关此时无效,需先在 env 打开)。
|
||||
logger.info("withdraw auto reconcile worker not started (env master switch off)")
|
||||
return None
|
||||
if not settings.wxpay_configured:
|
||||
logger.warning("withdraw auto reconcile enabled but wxpay not configured")
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""美团城市字典(地级市)读取。
|
||||
|
||||
数据源:app/integrations/data/meituan_cities.json,由 tools/gen_meituan_cities.py 从
|
||||
美团官方「城市字典」Excel 生成(359 个地级市:city_id / 城市名 / 省份)。
|
||||
实测一个地级市 cityId 已覆盖其下辖县级市供给(徐州 → 邳州 / 新沂 / 睢宁 …),故无区县层级。
|
||||
|
||||
- ETL(scripts/pull_meituan_coupons.py)遍历 all_cities() 全量抓取入库。
|
||||
- 读取侧将来「按城市过滤」时,也从这里取 city_id ↔ 城市名映射。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
_JSON = Path(__file__).resolve().parent / "data" / "meituan_cities.json"
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=1)
|
||||
def all_cities() -> list[dict]:
|
||||
"""全部城市 [{city_id, name, province}, ...](359 个地级市)。"""
|
||||
return json.loads(_JSON.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=1)
|
||||
def _by_id() -> dict[str, dict]:
|
||||
return {c["city_id"]: c for c in all_cities()}
|
||||
|
||||
|
||||
def city_name(city_id: str) -> str | None:
|
||||
"""city_id → 城市名(查不到返回 None)。"""
|
||||
c = _by_id().get(city_id)
|
||||
return c["name"] if c else None
|
||||
File diff suppressed because it is too large
Load Diff
@@ -126,13 +126,23 @@ def query_coupon(
|
||||
|
||||
def get_referral_link(
|
||||
*,
|
||||
product_view_sign: str,
|
||||
product_view_sign: str | None = None,
|
||||
act_id: str | None = None,
|
||||
platform: int = 1,
|
||||
biz_line: int | None = None,
|
||||
sid: str | None = None,
|
||||
link_type_list: list[int] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
body: dict[str, Any] = {"productViewSign": product_view_sign}
|
||||
"""换推广链接。actId(活动物料)与 productViewSign(商品券)二选一。
|
||||
用 actId 入参时,出参 data 为 null,链接全在 referralLinkMap(实测)。
|
||||
"""
|
||||
if not act_id and not product_view_sign:
|
||||
raise MeituanCpsError("act_id 或 product_view_sign 必填其一")
|
||||
body: dict[str, Any] = {}
|
||||
if act_id:
|
||||
body["actId"] = act_id
|
||||
else:
|
||||
body["productViewSign"] = product_view_sign
|
||||
if platform == 2:
|
||||
body["platform"] = 2
|
||||
if biz_line is not None:
|
||||
@@ -142,3 +152,30 @@ def get_referral_link(
|
||||
body["linkTypeList"] = link_type_list or [1, 3]
|
||||
|
||||
return _call("/cps_open/common/api/v1/get_referral_link", body)
|
||||
|
||||
|
||||
def query_order(
|
||||
*,
|
||||
sid: str | None = None,
|
||||
start_time: int,
|
||||
end_time: int,
|
||||
query_time_type: int = 1,
|
||||
page: int = 1,
|
||||
limit: int = 100,
|
||||
) -> dict[str, Any]:
|
||||
"""按时间窗(+可选 sid)拉 CPS 订单做对账。实测要点:
|
||||
- 分页是 page / limit(不是 pageNo/pageSize);startTime/endTime 是 10 位秒级时间戳
|
||||
- query_time_type: 1 按支付时间 2 按更新时间
|
||||
- 响应 data.dataList[],每条含 orderId / sid / payPrice(元) / profit(元) /
|
||||
commissionRate / status(2付款 3完成 4取消 5风控 6结算) / payTime(秒) / actId 等
|
||||
"""
|
||||
body: dict[str, Any] = {
|
||||
"queryTimeType": query_time_type,
|
||||
"startTime": start_time,
|
||||
"endTime": end_time,
|
||||
"page": page,
|
||||
"limit": limit,
|
||||
}
|
||||
if sid:
|
||||
body["sid"] = sid
|
||||
return _call("/cps_open/common/api/v1/query_order", body)
|
||||
|
||||
@@ -42,3 +42,13 @@ def verify_callback_sign(params: dict[str, str], secret: str) -> bool:
|
||||
got = params.get("sign") or ""
|
||||
expect = build_sign(trans_id, secret)
|
||||
return hmac.compare_digest(got, expect)
|
||||
|
||||
|
||||
def verify_callback_sign_any(params: dict[str, str], secrets: list[str]) -> bool:
|
||||
"""对**多个** m-key 逐个验签,任一通过即接受。
|
||||
|
||||
多广告位场景:每个 GroMore 广告位的 m-key 由后台各自生成、互不相同,但本服务用
|
||||
同一个回调 URL 接所有位的回调。配置里放多个密钥(逗号分隔),回调到达时挨个试。
|
||||
仍然安全:伪造者必须知道其中某个 m-key 才能算出合法 sign;空列表 → 一律失败。
|
||||
"""
|
||||
return any(verify_callback_sign(params, s) for s in secrets if s)
|
||||
|
||||
@@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router
|
||||
from app.api.v1.compare_milestone import router as compare_milestone_router
|
||||
from app.api.v1.compare_record import router as compare_record_router
|
||||
from app.api.v1.coupon import router as coupon_router
|
||||
from app.api.v1.cps_redirect import router as cps_redirect_router
|
||||
from app.api.internal.price import router as internal_price_router
|
||||
from app.api.internal.store import router as internal_store_router
|
||||
from app.api.v1.feedback import router as feedback_router
|
||||
@@ -107,6 +108,8 @@ app.include_router(report_router)
|
||||
app.include_router(internal_price_router)
|
||||
app.include_router(internal_store_router)
|
||||
app.include_router(platform_router)
|
||||
# CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团)
|
||||
app.include_router(cps_redirect_router)
|
||||
|
||||
# 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
|
||||
_media_root = Path(settings.MEDIA_ROOT)
|
||||
|
||||
@@ -6,6 +6,10 @@ from app.models.ad_watch_log import AdWatchLog # noqa: F401
|
||||
from app.models.admin import AdminAuditLog, AdminUser # noqa: F401
|
||||
from app.models.app_config import AppConfig # noqa: F401
|
||||
from app.models.comparison import ComparisonRecord # noqa: F401
|
||||
from app.models.cps_activity import CpsActivity # noqa: F401
|
||||
from app.models.cps_group import CpsGroup # noqa: F401
|
||||
from app.models.cps_link import CpsClick, CpsLink # noqa: F401
|
||||
from app.models.cps_order import CpsOrder # noqa: F401
|
||||
from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401
|
||||
from app.models.coupon_state import ( # noqa: F401
|
||||
CouponClaimRecord,
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
"""CPS 可推广活动池(cps_activity)。
|
||||
|
||||
运营预存常推的活动(券),生成链接时选它,省得每次记 actId。当前只接美团:
|
||||
`act_id` = 美团联盟「我要推广-活动推广」第一列的物料 ID。`platform` 预留,接
|
||||
淘宝/京东时复用本表(各自的活动标识)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, Integer, String, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class CpsActivity(Base):
|
||||
__tablename__ = "cps_activity"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
platform: Mapped[str] = mapped_column(String(20), nullable=False, default="meituan") # meituan|taobao|jd
|
||||
name: Mapped[str] = mapped_column(String(128), nullable=False)
|
||||
# 美团活动物料 ID(actId);淘宝/京东接入后存各自活动标识。
|
||||
act_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
# 美团商品券 productViewSign(与 act_id 二选一转链;多数活动用 act_id)。
|
||||
product_view_sign: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active") # active | archived
|
||||
remark: Mapped[str | None] = mapped_column(String(256), nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"<CpsActivity id={self.id} platform={self.platform} name={self.name!r}>"
|
||||
@@ -0,0 +1,32 @@
|
||||
"""CPS 推广群(cps_group)。
|
||||
|
||||
每个微信群一条记录,`sid` 是美团联盟二级渠道追踪位(get_referral_link / query_order
|
||||
共用):发券时塞进推广链接,订单回来按 sid 归群对账。美团限制 sid 仅字母+数字 ≤64。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, Integer, String, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class CpsGroup(Base):
|
||||
__tablename__ = "cps_group"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
# 每群唯一的渠道标识,建群时分配(留空则自动生成 g<id>)。字母+数字,≤64。
|
||||
sid: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
|
||||
name: Mapped[str] = mapped_column(String(128), nullable=False)
|
||||
# 群人数,运营填,作转化率分母(可空)。
|
||||
member_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
status: Mapped[str] = mapped_column(String(20), nullable=False, default="active") # active | archived
|
||||
remark: Mapped[str | None] = mapped_column(String(256), nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"<CpsGroup id={self.id} sid={self.sid!r} name={self.name!r}>"
|
||||
@@ -0,0 +1,52 @@
|
||||
"""CPS 群发短链(cps_link)+ 点击事件(cps_click)。
|
||||
|
||||
群发券链接套一层我们自己的短链 `/c/{code}`:用户点 → 记一条 cps_click → 302 跳
|
||||
target_url(美团短链,微信可打开)。据此统计点击 PV/UV(按群/活动/时段),配合 cps_order
|
||||
的下单/佣金做"点击→下单→佣金"漏斗。group_id/sid 在 cps_click 里冗余一份,统计直接按
|
||||
群聚合点击、不必 join cps_link。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, Integer, String, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class CpsLink(Base):
|
||||
__tablename__ = "cps_link"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
# 短码:发群链接 /c/{code} 的 code,全局唯一。
|
||||
code: Mapped[str] = mapped_column(String(16), unique=True, index=True, nullable=False)
|
||||
group_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
|
||||
activity_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
|
||||
sid: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
|
||||
platform: Mapped[str] = mapped_column(String(20), nullable=False, default="meituan")
|
||||
# 302 跳转目标(美团短链,微信内可打开并唤起 App)。
|
||||
target_url: Mapped[str] = mapped_column(String(1024), nullable=False)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return f"<CpsLink id={self.id} code={self.code!r} sid={self.sid!r}>"
|
||||
|
||||
|
||||
class CpsClick(Base):
|
||||
__tablename__ = "cps_click"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
link_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
|
||||
group_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False) # 冗余,按群聚合快
|
||||
sid: Mapped[str] = mapped_column(String(64), index=True, nullable=False) # 冗余
|
||||
ip: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
ua: Mapped[str | None] = mapped_column(String(512), nullable=True)
|
||||
clicked_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"<CpsClick id={self.id} link_id={self.link_id} group_id={self.group_id}>"
|
||||
@@ -0,0 +1,63 @@
|
||||
"""CPS 对账订单(cps_order)。
|
||||
|
||||
从美团联盟 query_order 按时间窗拉回、按 sid 归群的订单明细。字段对齐 query_order
|
||||
实测返回:
|
||||
- payPrice / profit 是「元」字符串 → 入库统一转「分」(与全站口径一致)
|
||||
- payTime / updateTime 是秒级时间戳 → 入库转 tz-aware datetime
|
||||
- status: 2付款 3完成 4取消 5风控 6结算(取消/风控不计佣金)
|
||||
order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import JSON, DateTime, Integer, String, func
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.db.base import Base
|
||||
|
||||
|
||||
class CpsOrder(Base):
|
||||
__tablename__ = "cps_order"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
# 美团订单号(加密串),全局唯一,upsert 幂等键。
|
||||
order_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
|
||||
# 渠道追踪位 = 群 sid(历史无 sid 订单为空)。按它归群聚合。
|
||||
sid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
act_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
|
||||
biz_line: Mapped[int | None] = mapped_column(Integer, nullable=True) # 1=外卖
|
||||
trade_type: Mapped[int | None] = mapped_column(Integer, nullable=True) # 1=cps 2=cpa
|
||||
|
||||
pay_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True) # 预估佣金(profit)
|
||||
commission_rate: Mapped[str | None] = mapped_column(String(16), nullable=True) # "300"=3% "10"=0.1%
|
||||
refund_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
refund_profit_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
|
||||
# 美团订单状态: 2付款 3完成 4取消 5风控 6结算
|
||||
mt_status: Mapped[str | None] = mapped_column(String(8), index=True, nullable=True)
|
||||
invalid_reason: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
||||
product_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
|
||||
|
||||
pay_time: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), index=True, nullable=True
|
||||
)
|
||||
mt_update_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
|
||||
raw: Mapped[dict] = mapped_column(
|
||||
JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=dict
|
||||
)
|
||||
first_seen: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
|
||||
)
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return (
|
||||
f"<CpsOrder id={self.id} order_id={self.order_id!r} "
|
||||
f"sid={self.sid!r} status={self.mt_status} profit_cents={self.commission_cents}>"
|
||||
)
|
||||
@@ -27,6 +27,10 @@ class User(Base):
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
phone: Mapped[str] = mapped_column(String(20), unique=True, index=True, nullable=False)
|
||||
|
||||
# 对外展示的账号 ID:11 位纯数字、首位非 1(与手机号天然区分——手机号都以 1 开头)、全局唯一、
|
||||
# 创建时随机生成、不可变、不参与登录(登录仍走 phone)。生成见 repositories/user._gen_username。
|
||||
username: Mapped[str] = mapped_column(String(11), unique=True, index=True, nullable=False)
|
||||
|
||||
# 注册渠道:jverify / sms。后续加 wechat / apple 时扩
|
||||
register_channel: Mapped[str] = mapped_column(String(20), nullable=False, default="jverify")
|
||||
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""CPS 群发短链 + 点击 数据访问(用户侧跳转端点 + admin 生成/统计 共用)。
|
||||
|
||||
短码生成去掉易混字符;点击按 (ip, ua) 近似去重 UV。点击量级小、admin 低频,统计在
|
||||
Python 侧聚合,跨 PG/SQLite 无方言坑(与 admin cps repo 同风格)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.cps_link import CpsClick, CpsLink
|
||||
|
||||
# 去掉易混字符 0/O/1/l/I,运营/用户肉眼复制不易错
|
||||
_CODE_ALPHABET = "abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
|
||||
|
||||
|
||||
def _gen_code(n: int = 8) -> str:
|
||||
return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(n))
|
||||
|
||||
|
||||
def get_by_code(db: Session, code: str) -> CpsLink | None:
|
||||
return db.execute(select(CpsLink).where(CpsLink.code == code)).scalar_one_or_none()
|
||||
|
||||
|
||||
def create_link(
|
||||
db: Session, *, group_id: int, activity_id: int, sid: str, target_url: str,
|
||||
platform: str = "meituan", commit: bool = True,
|
||||
) -> CpsLink:
|
||||
"""生成唯一短码并存库。冲突重试 5 次,仍冲突用 12 位兜底(概率近 0)。"""
|
||||
code = _gen_code()
|
||||
for _ in range(5):
|
||||
if get_by_code(db, code) is None:
|
||||
break
|
||||
code = _gen_code()
|
||||
else:
|
||||
code = _gen_code(12)
|
||||
link = CpsLink(
|
||||
code=code, group_id=group_id, activity_id=activity_id, sid=sid,
|
||||
target_url=target_url, platform=platform,
|
||||
)
|
||||
db.add(link)
|
||||
if commit:
|
||||
db.commit()
|
||||
db.refresh(link)
|
||||
else:
|
||||
db.flush()
|
||||
return link
|
||||
|
||||
|
||||
def record_click(db: Session, *, link: CpsLink, ip: str | None = None, ua: str | None = None) -> None:
|
||||
db.add(CpsClick(
|
||||
link_id=link.id, group_id=link.group_id, sid=link.sid,
|
||||
ip=ip, ua=(ua[:500] if ua else None),
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def click_stats_by_group(
|
||||
db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
|
||||
) -> dict[int, dict]:
|
||||
"""按 group_id 聚合点击 → {group_id: {"pv": N, "uv": M}}。uv 按 (ip, ua) 近似去重。"""
|
||||
stmt = select(CpsClick)
|
||||
if date_from is not None:
|
||||
stmt = stmt.where(CpsClick.clicked_at >= date_from)
|
||||
if date_to is not None:
|
||||
stmt = stmt.where(CpsClick.clicked_at <= date_to)
|
||||
clicks = list(db.execute(stmt).scalars().all())
|
||||
|
||||
agg: dict[int, dict] = {}
|
||||
seen: dict[int, set] = {}
|
||||
for c in clicks:
|
||||
a = agg.setdefault(c.group_id, {"pv": 0, "uv": 0})
|
||||
a["pv"] += 1
|
||||
s = seen.setdefault(c.group_id, set())
|
||||
key = (c.ip, c.ua)
|
||||
if key not in s:
|
||||
s.add(key)
|
||||
a["uv"] += 1
|
||||
return agg
|
||||
@@ -4,6 +4,8 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
import string
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
@@ -12,6 +14,50 @@ from sqlalchemy.orm import Session
|
||||
from app.models.user import User
|
||||
|
||||
|
||||
# ===== 创建时分配的标识:用户名(对外展示账号 ID)+ 默认昵称 =====
|
||||
|
||||
# 用户名 = 11 位纯数字,首位 2-9(避前导 0、避 1 与手机号区分),后 10 位 0-9。
|
||||
# 空间 8×10^10,创建时随机生成 + 查重去重,全局唯一、不可变、不参与登录。
|
||||
_USERNAME_FIRST = "23456789"
|
||||
_USERNAME_DIGITS = "0123456789"
|
||||
_USERNAME_LEN = 11
|
||||
|
||||
# 默认昵称 = 9 位大小写字母 + 数字随机串(创建时给,用户可后续改;不要求唯一)。
|
||||
_NICKNAME_ALPHABET = string.ascii_letters + string.digits
|
||||
_NICKNAME_LEN = 9
|
||||
|
||||
|
||||
def _gen_username() -> str:
|
||||
"""随机一个 11 位纯数字、首位非 0/1 的用户名(不保证唯一,唯一性由 _gen_unique_username 兜)。"""
|
||||
return secrets.choice(_USERNAME_FIRST) + "".join(
|
||||
secrets.choice(_USERNAME_DIGITS) for _ in range(_USERNAME_LEN - 1)
|
||||
)
|
||||
|
||||
|
||||
def _gen_nickname() -> str:
|
||||
"""随机一个 9 位字母+数字的默认昵称。"""
|
||||
return "".join(secrets.choice(_NICKNAME_ALPHABET) for _ in range(_NICKNAME_LEN))
|
||||
|
||||
|
||||
def get_user_by_username(db: Session, username: str) -> User | None:
|
||||
return db.execute(
|
||||
select(User).where(User.username == username)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def _gen_unique_username(db: Session) -> str:
|
||||
"""生成一个库里尚不存在的用户名(仿 invite.ensure_code 的碰撞重试)。
|
||||
|
||||
8×10^10 空间下碰撞极罕见,查重 + 重试足够;并发下的残留碰撞由 username 唯一约束
|
||||
兜底(commit 抛 IntegrityError,与现有 phone 并发同号同级别,概率 ~ 用户数/8e10)。
|
||||
"""
|
||||
for _ in range(8):
|
||||
uname = _gen_username()
|
||||
if get_user_by_username(db, uname) is None:
|
||||
return uname
|
||||
raise RuntimeError("生成用户名连续碰撞,请重试")
|
||||
|
||||
|
||||
def get_user_by_id(db: Session, user_id: int) -> User | None:
|
||||
return db.get(User, user_id)
|
||||
|
||||
@@ -36,6 +82,8 @@ def upsert_user_for_login(
|
||||
if user is None:
|
||||
user = User(
|
||||
phone=phone,
|
||||
username=_gen_unique_username(db),
|
||||
nickname=_gen_nickname(),
|
||||
register_channel=register_channel,
|
||||
last_login_at=now,
|
||||
)
|
||||
|
||||
@@ -18,6 +18,8 @@ class UserOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True) # 允许直接 UserOut.model_validate(orm_user)
|
||||
|
||||
id: int
|
||||
# 对外展示的账号 ID:11 位纯数字、首位非 1、全局唯一、创建时分配、不可变。区别于登录用的 phone。
|
||||
username: str
|
||||
phone: str
|
||||
nickname: str | None = None
|
||||
avatar_url: str | None = None
|
||||
|
||||
@@ -22,3 +22,9 @@ class SavingsFeedItem(BaseModel):
|
||||
|
||||
class SavingsFeedOut(BaseModel):
|
||||
items: list[SavingsFeedItem]
|
||||
|
||||
|
||||
class AppFlagsOut(BaseModel):
|
||||
"""客户端拉取的运营 feature flag(不鉴权,登录前也能拉)。客户端缓存后按需读。"""
|
||||
|
||||
comparing_ad_enabled: bool # 比价/领券期是否展示信息流广告(远程 kill-switch)
|
||||
|
||||
@@ -22,7 +22,7 @@ GroMore 以 GET 回调,关键参数:
|
||||
| `ecpm` | string | 本次广告 eCPM(同上,可用于收益分析) |
|
||||
| `sign` | string | 签名,见下 |
|
||||
|
||||
**验签**:`sign = SHA256("{m-key}:{trans_id}")` 十六进制(只签 `trans_id`,其余参数不参与)。算法细节、m-key 来源、为什么这样设计 → 见集成文档 [integrations/pangle](../integrations/pangle.md)。
|
||||
**验签**:`sign = SHA256("{m-key}:{trans_id}")` 十六进制(只签 `trans_id`,其余参数不参与)。多激励位共用同一回调 URL → 服务端把各位的 m-key 都配上,`verify_callback_sign_any` 逐个试、任一过即接受。算法细节、m-key 配置项(`PANGLE_REWARD_SECRET_TEST/_TEST_DEDICATED/_PROD`)、为什么这样设计 → 见集成文档 [integrations/pangle](../integrations/pangle.md)。
|
||||
|
||||
## 出参
|
||||
响应 `200`,**响应体必须是 `{"is_verify": bool, "reason": int}`**(GroMore 规范)。
|
||||
|
||||
@@ -14,8 +14,8 @@
|
||||
## 字段
|
||||
| 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
|
||||
|---|---|---|---|
|
||||
| `key` | String(64) | **PK** | 配置标识,取值见 `config_schema.CONFIG_DEFS`:`signin_rewards` / `min_exchange_coin` / `withdraw_min_cents` / `withdraw_max_cents` / `task_rewards` / `record_milestones` / `ad_reward_coin` / `ad_daily_limit` / `ad_max_coin` / `ad_round_count` / `ad_cooldown_sec` |
|
||||
| `value` | JSON(PG: JSONB) | NOT NULL | 配置值,类型随 key(`int` / `int_list` 如签到 14 档 / `dict_str_int` 如 task_rewards) |
|
||||
| `key` | String(64) | **PK** | 配置标识,取值见 `config_schema.CONFIG_DEFS`:`signin_rewards` / `min_exchange_coin` / `withdraw_min_cents` / `withdraw_max_cents` / `task_rewards` / `record_milestones` / `ad_reward_coin` / `ad_daily_limit` / `ad_max_coin` / `ad_round_count` / `ad_cooldown_sec` / `signin_boost_coin` / `withdraw_auto_reconcile_enabled` |
|
||||
| `value` | JSON(PG: JSONB) | NOT NULL | 配置值,类型随 key(`int` / `int_list` 如签到 14 档 / `dict_str_int` 如 task_rewards / `bool` 如 withdraw_auto_reconcile_enabled) |
|
||||
| `updated_by_admin_id` | Integer | nullable | 最后修改的管理员 id(= `admin_user.id`,软引用,无 FK) |
|
||||
| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | 最后修改时间 |
|
||||
|
||||
|
||||
@@ -16,12 +16,18 @@
|
||||
| 函数 | 说明 |
|
||||
|---|---|
|
||||
| `build_sign(trans_id, secret) -> str` | 计算 `SHA256("{secret}:{trans_id}")` hex。自验签测试 / 模拟回调脚本共用,保证两端一致 |
|
||||
| `verify_callback_sign(params, secret) -> bool` | 校验回调签名。密钥空 / 缺 `trans_id` 一律失败;比较用 `hmac.compare_digest` 定长比较防时序侧信道 |
|
||||
| `verify_callback_sign(params, secret) -> bool` | 用**单个** m-key 校验回调签名。密钥空 / 缺 `trans_id` 一律失败;比较用 `hmac.compare_digest` 定长比较防时序侧信道 |
|
||||
| `verify_callback_sign_any(params, secrets) -> bool` | 用**一组** m-key 逐个验签、任一通过即接受(多激励位共用同一回调 URL 时用)。空列表 → 失败 |
|
||||
|
||||
## 配置(多激励位 = 多 m-key)
|
||||
每个 GroMore 激励位的 m-key 由后台各自生成、互不相同,但本服务用**同一个回调 URL** 接所有位的回调,因此把用到的位的 m-key 都配上;验签时 `verify_callback_sign_any` 逐个试、任一过即接受(仍安全:伪造者须知道其中某个 m-key 才能造出合法 sign)。m-key 在后台「GroMore 聚合管理 → 搜广告位 ID → 编辑」处获取。`settings.pangle_reward_secrets` 把下列来源汇总去重:
|
||||
|
||||
## 配置
|
||||
| 配置项 | 说明 |
|
||||
|---|---|
|
||||
| `PANGLE_REWARD_SECRET` | m-key(安全密钥)。后台「GroMore 聚合管理 → 搜广告位 ID → 编辑」处获取 |
|
||||
| `PANGLE_REWARD_SECRET_TEST` | 测试应用 激励位 104099649 的 m-key |
|
||||
| `PANGLE_REWARD_SECRET_TEST_DEDICATED` | 测试应用 专属激励位 104127529 的 m-key |
|
||||
| `PANGLE_REWARD_SECRET_PROD` | 正式应用 激励位 104099389 的 m-key |
|
||||
| `PANGLE_REWARD_SECRET` | 旧用法:单个或逗号分隔多个 m-key,仍兼容(与上面三个命名项合并去重) |
|
||||
|
||||
## 踩坑
|
||||
- **别用联盟代码位那套**:联盟代码位层级用的是另一套 Security Key + `isValid` 响应体,与 GroMore 广告位层级不通用。我们走 GroMore,别接错。
|
||||
|
||||
+15
-2
@@ -41,18 +41,31 @@
|
||||
> 已确认走 **GroMore 广告位层级**回调(客户端 useMediation(true);规范 supportcenter/26240),
|
||||
> **不是**联盟代码位层级(5416)。验签算法 / 响应格式 / reward_amount 解析**代码已按 GroMore 规范实现**
|
||||
> (2026-05-27),剩下的是后台配置 + 填密钥。
|
||||
>
|
||||
> 🔴 **2026-06-16 实测的头号阻塞:穿山甲打线上后端会超时**。真机看广告时 SDK 日志
|
||||
> `onRewardArrived valid=false errorCode=50002`,errorMsg =
|
||||
> `Get "https://app-api.shaguabijia.com/api/v1/ad/pangle-callback...": context deadline exceeded`。
|
||||
> 而我们自己 curl 线上是秒回。说明**穿山甲机房到线上后端这条网络路径有问题**(疑似 nginx/云防火墙对
|
||||
> 机房/境外 IP 限流或拦截)。**m-key 配对了也救不了——穿山甲根本没把请求送达。** 上线前必须单独验
|
||||
> 「线上 `app-api.shaguabijia.com` 对穿山甲机房可达且响应够快」,否则用户看完发不了奖。
|
||||
>
|
||||
> ✅ **多激励位已支持**(2026-06-16):内部测试期 测试位/专属位/正式位 并存,m-key 各配一行
|
||||
> (`PANGLE_REWARD_SECRET_TEST/_TEST_DEDICATED/_PROD`),`verify_callback_sign_any` 逐个试。本地已验通。
|
||||
|
||||
- [ ] **GroMore 后台配回调**:GroMore 聚合管理 → 搜广告位ID → 编辑 → 勾选「服务端激励回调」→
|
||||
回调 URL 填 `https://app-api.shaguabijia.com/api/v1/ad/pangle-callback`(用域名,别用 IP)。
|
||||
⚠️ 广告位层级配了就**别再在代码位层级重复配**(会导致发奖出问题)。
|
||||
- [ ] **拿 m-key(安全密钥)**:就在上面"编辑广告位"页获取,填到生产 `.env` 的 `PANGLE_REWARD_SECRET`。
|
||||
- [ ] **拿 m-key(安全密钥)**:就在上面"编辑广告位"页获取(每个激励位各一把),填到生产 `.env` 的命名变量
|
||||
(正式位填 `PANGLE_REWARD_SECRET_PROD`;内部测试期 测试位/专属位 也并存就填 `_TEST`/`_TEST_DEDICATED`)。
|
||||
(注:这是 GroMore 广告位的 m-key,**不是**联盟代码位那个 Security Key。)
|
||||
- [x] ~~换验签~~:`app/integrations/pangle.py` 已实现 `sign = SHA256("{m-key}:{trans_id}")`(GroMore 真实算法)。
|
||||
- [x] ~~响应格式~~:已返回 GroMore 要求的 `{"is_verify": bool, "reason": int}`。
|
||||
- [x] ~~reward_amount~~:回调按 `reward_amount` 发金币(`rewards.resolve_ad_reward_coin`,带回退/夹紧);
|
||||
**后台广告位"奖励数量"须配成与 `AD_REWARD_COIN`(=100)一致**,保证"广告内展示/进度预告/到账"三者一致。
|
||||
- [x] ~~透传 user_id~~:客户端已 `setUserID(userId)` + `setMediaExtra("uid:...")`。
|
||||
- [ ] 生产 `.env`:`PANGLE_CALLBACK_ENABLED=true` + `PANGLE_REWARD_SECRET=<m-key>`(配齐才不返 503)
|
||||
- [ ] 生产 `.env`:`PANGLE_CALLBACK_ENABLED=true` + 至少一个 m-key 命名变量(配齐才不返 503)+ 确认
|
||||
`AD_REWARD_TEST_GRANT_ENABLED=false`(绕过反作弊,生产红线)。改 `.env` 后**重启 uvicorn**
|
||||
(watchfiles 不监听 .env)。验证:外部 curl 回调 URL 从 `503` 变 `403 bad sign` 即生效。
|
||||
|
||||
## C. 部署 + 包名
|
||||
|
||||
|
||||
+186
-52
@@ -1,45 +1,74 @@
|
||||
"""美团 CPS 券定时抓取入库(北京试点)。
|
||||
"""美团 CPS 券定时抓取入库(全国 359 个地级市)。
|
||||
|
||||
把 3 路券抓进 meituan_coupon 表,供「销量/佣金排序」从库里捞、本地排序,不再实时打美团:
|
||||
把每个城市的 3 路券抓进 meituan_coupon 表,供「智能推荐 / 销量最高」从库里捞、本地排序,
|
||||
不再实时打美团:
|
||||
1. search_waimai : 到家/外卖, 搜「外卖」 翻到尽头
|
||||
2. search_meishi : 到家/外卖, 搜「美食」 翻到尽头
|
||||
3. store_supply : 到店, 多业务线供给(到餐+到综+酒店+门票) 翻到尽头
|
||||
|
||||
城市来自 app/integrations/data/meituan_cities.json(美团官方城市字典,359 地级市)。
|
||||
实测一个地级市 cityId 已覆盖其下辖县级市(徐州 → 邳州/新沂…),故按地级市抓即可。
|
||||
|
||||
并发:城市级并发(每城内部仍顺序跑 3 路),主线程串行入库(Session 不跨线程)。
|
||||
实测单城全量 ~110s/~2300 条;15 并发抓完全国一轮 ~50–60min,402 占比 ~3% 且退避全消化。
|
||||
mentor 要求每 3h 全量一轮,窗口充裕,故默认 12 并发 + 启动错峰,把瞬时峰值与 402 压更低。
|
||||
|
||||
按 (source, product_view_sign) upsert 存最新态;last_seen 每轮刷新。带文件锁,防止
|
||||
上一轮没跑完下一轮又起(本地 5~10min、跨进程 cron 都安全)。
|
||||
上一轮没跑完下一轮又起。
|
||||
|
||||
用法:
|
||||
# 单轮(打通验证 / 给 cron 用,线上每 1h 一次)
|
||||
# 单轮全量(给 cron 用,线上每 3h 一次)
|
||||
python -m scripts.pull_meituan_coupons --once
|
||||
|
||||
# 本地循环(默认每 10min 一轮)
|
||||
python -m scripts.pull_meituan_coupons --loop --interval 600
|
||||
# 常驻循环(每 3h 一轮)
|
||||
python -m scripts.pull_meituan_coupons --loop --interval 10800
|
||||
|
||||
# 本地测试:只抓前 N 个城市 / 指定城市
|
||||
python -m scripts.pull_meituan_coupons --once --limit 3
|
||||
python -m scripts.pull_meituan_coupons --once --city-ids OCZOBCJDEXKE7KBN3BD7AYQG2Q
|
||||
|
||||
部署(服务器):推荐 cron 跑 --once(每 3h),避免长驻进程孤儿:
|
||||
0 */3 * * * cd /path/to/app-server && .venv/bin/python -m scripts.pull_meituan_coupons --once >> data/etl.log 2>&1
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
# Windows 控制台按 UTF-8 输出中文/¥
|
||||
# Windows 控制台按 UTF-8 输出中文/¥;line_buffering=True 让 print 每行即时 flush——
|
||||
# 否则 stdout 重定向到 cron/后台日志文件时是块缓冲,要攒到 ~4KB 或进程退出才落盘,
|
||||
# 常驻(--loop)时几乎看不到每轮进度。
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
|
||||
sys.stdout.reconfigure(encoding="utf-8", line_buffering=True) # type: ignore[attr-defined]
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy import delete, func, select # noqa: E402
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert # noqa: E402
|
||||
|
||||
from app.db.session import SessionLocal
|
||||
from app.integrations.meituan import MeituanCpsError, _call
|
||||
from app.models.meituan_coupon import MeituanCoupon
|
||||
from app.db.session import SessionLocal, engine # noqa: E402
|
||||
from app.integrations.cities import all_cities # noqa: E402
|
||||
from app.integrations.meituan import MeituanCpsError, _call # noqa: E402
|
||||
from app.models.meituan_coupon import MeituanCoupon # noqa: E402
|
||||
|
||||
# dev 默认 create_engine(echo=True) 会逐条打 SQL,本脚本逐城 upsert 会把日志刷爆;
|
||||
# ETL 不需要 SQL 日志,显式关掉(入库不受影响;线上 prod 本就 echo=False)。
|
||||
engine.echo = False
|
||||
|
||||
# 美团调用偶发错误(本机走代理高并发时的 SSL EOF / 上游 code=5 等)会被 meituan._call 的
|
||||
# logger.exception 打完整 traceback,并发抓取下单轮可刷数十 KB 日志。ETL 自己用 _STATS 统计
|
||||
# 「放弃页数」,无需逐条 traceback,故把美团 logger 压到 CRITICAL(线上直连少见此类错误)。
|
||||
logging.getLogger("shagua.meituan").setLevel(logging.CRITICAL)
|
||||
|
||||
CITY_BEIJING = "WKV2HMXUEK634WP64CUCUQGM64"
|
||||
QUERY_PATH = "/cps_open/common/api/v1/query_coupon"
|
||||
PAGE_SIZE = 20
|
||||
MAX_PAGES = 80 # 单路安全上限(搜索 ~52 页、供给 ~70 页)
|
||||
@@ -49,7 +78,12 @@ RETRY = 7
|
||||
# 无 sudo 部署时 data/ 常属 root,cps 写不了会导致每轮 PermissionError、cron 抓不进数据。
|
||||
# 需要指定位置时用环境变量 MEITUAN_ETL_LOCK 覆盖。
|
||||
LOCK_FILE = os.environ.get("MEITUAN_ETL_LOCK") or os.path.join(tempfile.gettempdir(), "meituan_etl.lock")
|
||||
LOCK_STALE_SEC = 30 * 60 # 锁超过 30min 视为陈旧(进程异常退出残留),自动接管
|
||||
# 多城全量一轮 ~50–60min,锁陈旧阈值放宽到 90min,避免把「正在跑的轮次」误判为残留而接管。
|
||||
LOCK_STALE_SEC = 90 * 60
|
||||
|
||||
DEFAULT_CONCURRENCY = 12 # 并发城市数(实测 15 并发 402 占 3% 可退避消化;12 更稳,3h 窗口充裕)
|
||||
STARTUP_STAGGER = 0.3 # 首批城市启动错峰间隔秒(削平瞬时峰值,实测能压低 402)
|
||||
PRUNE_FAIL_RATIO_MAX = 0.05 # 失败城占比超此值则本轮跳过 prune(避免大面积抓取失败误删库)
|
||||
|
||||
SOURCES = [
|
||||
{"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
|
||||
@@ -58,33 +92,44 @@ SOURCES = [
|
||||
"platform": 2, "biz_lines": [1, 2, 3, 4]},
|
||||
]
|
||||
|
||||
# 跨线程统计(并发抓取下汇总请求量 / 402 / 放弃页数,供观测限流)
|
||||
_STATS_LOCK = threading.Lock()
|
||||
_STATS = {"req": 0, "r402": 0, "err": 0}
|
||||
|
||||
|
||||
def _bump(key: str) -> None:
|
||||
with _STATS_LOCK:
|
||||
_STATS[key] += 1
|
||||
|
||||
|
||||
# ───────────────────────── 美团调用 ─────────────────────────
|
||||
|
||||
def _call_retry(body: dict) -> dict | None:
|
||||
"""打美团,402/频繁退避重试;其它错误打印并放弃本页。"""
|
||||
"""打美团,402/频繁退避重试;其它错误放弃本页。并发下不逐条 print(避免刷屏),计入 _STATS。"""
|
||||
for a in range(RETRY):
|
||||
_bump("req")
|
||||
try:
|
||||
return _call(QUERY_PATH, body)
|
||||
except MeituanCpsError as e:
|
||||
msg = str(e)
|
||||
if "402" in msg or "频繁" in msg:
|
||||
_bump("r402")
|
||||
time.sleep(2.5 * (a + 1))
|
||||
continue
|
||||
print(f" [warn] meituan: {msg[:80]}")
|
||||
_bump("err")
|
||||
return None
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" [warn] {type(e).__name__}: {str(e)[:60]}")
|
||||
except Exception: # noqa: BLE001
|
||||
time.sleep(2.0 * (a + 1))
|
||||
_bump("err")
|
||||
return None
|
||||
|
||||
|
||||
def _pull_search(platform: int, keyword: str) -> list[dict]:
|
||||
def _pull_search(city_id: str, platform: int, keyword: str) -> list[dict]:
|
||||
rows: list[dict] = []
|
||||
sid = None
|
||||
pg = 1
|
||||
while pg <= MAX_PAGES:
|
||||
body = {"platform": platform, "searchText": keyword, "cityId": CITY_BEIJING, "pageSize": PAGE_SIZE}
|
||||
body = {"platform": platform, "searchText": keyword, "cityId": city_id, "pageSize": PAGE_SIZE}
|
||||
if sid:
|
||||
body["searchId"] = sid
|
||||
else:
|
||||
@@ -102,14 +147,14 @@ def _pull_search(platform: int, keyword: str) -> list[dict]:
|
||||
return rows
|
||||
|
||||
|
||||
def _pull_supply(platform: int, biz_lines: list[int]) -> list[dict]:
|
||||
def _pull_supply(city_id: str, platform: int, biz_lines: list[int]) -> list[dict]:
|
||||
rows: list[dict] = []
|
||||
sid = None
|
||||
biz_param = [{"bizLine": b} for b in biz_lines]
|
||||
for _ in range(MAX_PAGES):
|
||||
body = {
|
||||
"multipleSupplyList": [{"platform": platform, "bizLineParamList": biz_param}],
|
||||
"cityId": CITY_BEIJING,
|
||||
"cityId": city_id,
|
||||
"sortField": 2, # 供给查询 sortField 必填;我们入库后本地再排,这里给个默认
|
||||
"pageSize": PAGE_SIZE,
|
||||
}
|
||||
@@ -157,7 +202,19 @@ def _to_cents(yuan) -> int | None:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_item(item: dict, source: dict) -> dict | None:
|
||||
def _strip_nul(v):
|
||||
"""递归去掉字符串里的 NUL(0x00):PostgreSQL 的 text / jsonb 字段都不接受该字节,
|
||||
美团偶有脏数据(某券文本含 NUL)会让整批 upsert 抛 DataError。入库前统一清洗。"""
|
||||
if isinstance(v, str):
|
||||
return v.replace(chr(0), "")
|
||||
if isinstance(v, dict):
|
||||
return {k: _strip_nul(x) for k, x in v.items()}
|
||||
if isinstance(v, list):
|
||||
return [_strip_nul(x) for x in v]
|
||||
return v
|
||||
|
||||
|
||||
def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
|
||||
cpd = item.get("couponPackDetail") or {}
|
||||
br = item.get("brandInfo") or {}
|
||||
ci = item.get("commissionInfo") or {}
|
||||
@@ -186,11 +243,12 @@ def _parse_item(item: dict, source: dict) -> dict | None:
|
||||
dist = None
|
||||
|
||||
dedup_raw = f"{brand}|{name}|{price_cents}"
|
||||
return {
|
||||
# 入库前递归清洗 NUL(美团脏数据偶含 0x00,PostgreSQL text/jsonb 拒绝整批 → DataError)
|
||||
return _strip_nul({
|
||||
"source": source["code"],
|
||||
"platform": source["platform"],
|
||||
"biz_line": item.get("bizLine") or cpd.get("bizLine"),
|
||||
"city_id": CITY_BEIJING,
|
||||
"city_id": city_id,
|
||||
"product_view_sign": str(sign)[:128],
|
||||
"sku_view_id": cpd.get("skuViewId"),
|
||||
"name": (name[:256] or None),
|
||||
@@ -207,7 +265,28 @@ def _parse_item(item: dict, source: dict) -> dict | None:
|
||||
"delivery_distance_m": dist,
|
||||
"dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
|
||||
"raw": item,
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
|
||||
"""抓单个城市的 3 路券并解析。worker 线程内执行(只抓取+解析,不碰 DB)。
|
||||
|
||||
启动错峰:首批并发的 `concurrency` 个城市按 index 错开首请求,削平瞬时峰值(压低 402)。
|
||||
"""
|
||||
if stagger:
|
||||
time.sleep((index % concurrency) * stagger)
|
||||
cid = city["city_id"]
|
||||
parsed: list[dict] = []
|
||||
for src in SOURCES:
|
||||
if src["kind"] == "search":
|
||||
items = _pull_search(cid, src["platform"], src["keyword"])
|
||||
else:
|
||||
items = _pull_supply(cid, src["platform"], src["biz_lines"])
|
||||
for it in items:
|
||||
p = _parse_item(it, src, cid)
|
||||
if p:
|
||||
parsed.append(p)
|
||||
return parsed
|
||||
|
||||
|
||||
# ───────────────────────── 入库(upsert) ─────────────────────────
|
||||
@@ -272,30 +351,59 @@ def _release_lock() -> None:
|
||||
|
||||
# ───────────────────────── 主流程 ─────────────────────────
|
||||
|
||||
def run_once(prune_hours: int = 24) -> None:
|
||||
def run_once(
|
||||
prune_hours: int = 24,
|
||||
concurrency: int = DEFAULT_CONCURRENCY,
|
||||
stagger: float = STARTUP_STAGGER,
|
||||
cities: list[dict] | None = None,
|
||||
) -> None:
|
||||
if not _acquire_lock():
|
||||
print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
|
||||
return
|
||||
t0 = time.time()
|
||||
now = datetime.now(timezone.utc)
|
||||
with _STATS_LOCK:
|
||||
_STATS.update(req=0, r402=0, err=0)
|
||||
cities = cities if cities is not None else all_cities()
|
||||
n_city = len(cities)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
total = 0
|
||||
for src in SOURCES:
|
||||
ts = time.time()
|
||||
if src["kind"] == "search":
|
||||
items = _pull_search(src["platform"], src["keyword"])
|
||||
else:
|
||||
items = _pull_supply(src["platform"], src["biz_lines"])
|
||||
parsed = [p for p in (_parse_item(it, src) for it in items) if p]
|
||||
up, dup = _upsert(db, parsed, now)
|
||||
total += up
|
||||
print(f" {src['label']:18} 抓{len(items):5} 解析{len(parsed):5} "
|
||||
f"入库{up:5} (本源去重{dup:3}) {time.time() - ts:4.0f}s")
|
||||
# 清理长期未再出现的陈旧券(美团 sign 轮换 / 券下架后的残留),默认 24h 宽限。
|
||||
# 护栏:仅在本轮确有入库(total>0)时才清理。美团整体故障时本轮可能 0 入库
|
||||
# (脚本不抛异常、只是抓回空),若仍照常 prune,连续故障会按 last_seen 把全表删空。
|
||||
if prune_hours and prune_hours > 0 and total > 0:
|
||||
total_up = 0
|
||||
done = 0
|
||||
fails: list[str] = []
|
||||
# 城市级并发抓取(worker 只抓取+解析),主线程逐城串行入库(Session 不跨线程)。
|
||||
with ThreadPoolExecutor(max_workers=concurrency) as pool:
|
||||
futs = {
|
||||
pool.submit(_pull_one_city, city, i, concurrency, stagger): city
|
||||
for i, city in enumerate(cities)
|
||||
}
|
||||
for fut in as_completed(futs):
|
||||
city = futs[fut]
|
||||
try:
|
||||
parsed = fut.result()
|
||||
except Exception as e: # noqa: BLE001
|
||||
fails.append(city["name"])
|
||||
print(f" [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
|
||||
continue
|
||||
try:
|
||||
up, _dup = _upsert(db, parsed, now)
|
||||
except Exception as e: # noqa: BLE001
|
||||
db.rollback() # 事务已 abort,必须 rollback 才能继续给下一城复用 Session
|
||||
fails.append(city["name"])
|
||||
print(f" [入库失败] {city['name']}: {type(e).__name__}: {str(e)[:80]}")
|
||||
continue
|
||||
total_up += up
|
||||
done += 1
|
||||
if done % 20 == 0 or done == n_city:
|
||||
print(f" [{done}/{n_city}] {city['name']:10} 抓{len(parsed):5} 入库{up:5} "
|
||||
f"(累计入库 {total_up}, 用时 {time.time() - t0:.0f}s)")
|
||||
|
||||
# 清理陈旧券(美团 sign 轮换 / 券下架残留),默认 24h 宽限。两道护栏防误删:
|
||||
# ① total_up>0:本轮 0 入库(疑似上游整体故障,脚本不抛异常只抓回空)时跳过,否则
|
||||
# 会按 last_seen 把全表删空;
|
||||
# ② 失败城占比 ≤5%:大面积城市抓取失败(限流/网络)时跳过,避免误删还在架上的券。
|
||||
fail_ratio = len(fails) / max(1, n_city)
|
||||
if prune_hours and prune_hours > 0 and total_up > 0 and fail_ratio <= PRUNE_FAIL_RATIO_MAX:
|
||||
cutoff = now - timedelta(hours=prune_hours)
|
||||
pruned = db.execute(
|
||||
delete(MeituanCoupon).where(MeituanCoupon.last_seen < cutoff)
|
||||
@@ -303,36 +411,62 @@ def run_once(prune_hours: int = 24) -> None:
|
||||
db.commit()
|
||||
if pruned:
|
||||
print(f" 清理陈旧券(>{prune_hours}h 未再出现): {pruned} 条")
|
||||
elif prune_hours and prune_hours > 0 and total == 0:
|
||||
print(" 本轮 0 入库(疑似上游故障),跳过清理以防误删全表")
|
||||
elif prune_hours and prune_hours > 0:
|
||||
reason = "本轮 0 入库" if total_up == 0 else f"失败城占比 {fail_ratio:.0%}>{PRUNE_FAIL_RATIO_MAX:.0%}"
|
||||
print(f" 跳过 prune({reason},防误删)")
|
||||
|
||||
cnt = db.execute(select(func.count()).select_from(MeituanCoupon)).scalar()
|
||||
print(f"[{datetime.now():%H:%M:%S}] 本轮完成: 入库 {total} 条, 表总计 {cnt} 行, "
|
||||
f"用时 {time.time() - t0:.0f}s")
|
||||
with _STATS_LOCK:
|
||||
req, r402, err = _STATS["req"], _STATS["r402"], _STATS["err"]
|
||||
fail_tail = (": " + ",".join(fails[:10])) if fails else ""
|
||||
print(f"[{datetime.now():%H:%M:%S}] 本轮完成: {done}/{n_city} 城, 入库 {total_up} 条, "
|
||||
f"表总计 {cnt} 行, 请求 {req}(402 {r402} / 放弃 {err}), "
|
||||
f"失败城 {len(fails)}{fail_tail}, 用时 {time.time() - t0:.0f}s")
|
||||
finally:
|
||||
db.close()
|
||||
_release_lock()
|
||||
|
||||
|
||||
def _select_cities(limit: int, city_ids: str) -> list[dict]:
|
||||
cities = all_cities()
|
||||
if city_ids.strip():
|
||||
want = {c.strip() for c in city_ids.split(",") if c.strip()}
|
||||
return [c for c in cities if c["city_id"] in want]
|
||||
if limit and limit > 0:
|
||||
return cities[:limit]
|
||||
return cities
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库")
|
||||
ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库(全国 359 地级市)")
|
||||
ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
|
||||
ap.add_argument("--loop", action="store_true", help="循环跑")
|
||||
ap.add_argument("--interval", type=int, default=600, help="循环间隔秒(默认 600=10min)")
|
||||
ap.add_argument("--interval", type=int, default=10800,
|
||||
help="循环间隔秒(默认 10800=3h,对齐每 3h 全量一轮)")
|
||||
ap.add_argument("--prune-hours", type=int, default=24,
|
||||
help="清理超过 N 小时未再出现的陈旧券(默认 24;0=不清理)")
|
||||
ap.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY,
|
||||
help=f"并发城市数(默认 {DEFAULT_CONCURRENCY};实测 15 并发 402 占 3%% 可退避消化)")
|
||||
ap.add_argument("--stagger", type=float, default=STARTUP_STAGGER,
|
||||
help=f"首批城市启动错峰间隔秒(默认 {STARTUP_STAGGER};削平瞬时峰值压低 402)")
|
||||
ap.add_argument("--limit", type=int, default=0, help="只抓前 N 个城市(测试用,0=全部)")
|
||||
ap.add_argument("--city-ids", default="", help="只抓这些 cityId(逗号分隔,测试用,优先于 --limit)")
|
||||
args = ap.parse_args()
|
||||
|
||||
cities = _select_cities(args.limit, args.city_ids)
|
||||
print(f"城市数: {len(cities)} / 并发: {args.concurrency} / 错峰: {args.stagger}s / 间隔: {args.interval}s")
|
||||
|
||||
if args.loop:
|
||||
print(f"循环模式: 每 {args.interval}s 一轮 (Ctrl-C 退出)")
|
||||
while True:
|
||||
try:
|
||||
run_once(args.prune_hours)
|
||||
run_once(args.prune_hours, args.concurrency, args.stagger, cities)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[{datetime.now():%H:%M:%S}] 本轮异常: {type(e).__name__}: {e}")
|
||||
_release_lock()
|
||||
time.sleep(args.interval)
|
||||
else:
|
||||
run_once(args.prune_hours)
|
||||
run_once(args.prune_hours, args.concurrency, args.stagger, cities)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -117,6 +117,40 @@ def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> N
|
||||
db.close()
|
||||
|
||||
|
||||
def test_update_bool_config(admin_client: TestClient, token: str) -> None:
|
||||
# 提现自动对账开关默认 True
|
||||
items = {
|
||||
i["key"]: i
|
||||
for i in admin_client.get("/admin/api/config", headers=_auth(token)).json()
|
||||
}
|
||||
assert items["withdraw_auto_reconcile_enabled"]["type"] == "bool"
|
||||
assert items["withdraw_auto_reconcile_enabled"]["value"] is True
|
||||
|
||||
# 关掉 → DB 落 False、业务读到 False
|
||||
r = admin_client.patch(
|
||||
"/admin/api/config/withdraw_auto_reconcile_enabled",
|
||||
json={"value": False},
|
||||
headers=_auth(token),
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["value"] is False and r.json()["overridden"] is True
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
from app.repositories import app_config
|
||||
|
||||
assert app_config.get_value(db, "withdraw_auto_reconcile_enabled") is False
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
# bool 项不接受非布尔值
|
||||
assert admin_client.patch(
|
||||
"/admin/api/config/withdraw_auto_reconcile_enabled",
|
||||
json={"value": 1},
|
||||
headers=_auth(token),
|
||||
).status_code == 400
|
||||
|
||||
|
||||
def test_config_validation(admin_client: TestClient, token: str) -> None:
|
||||
# 签到档位长度≠7
|
||||
assert admin_client.patch(
|
||||
|
||||
@@ -206,3 +206,40 @@ def test_sms_gc_purges_stale_only(monkeypatch) -> None:
|
||||
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
|
||||
|
||||
|
||||
# ============================ 用户名 / 默认昵称 ============================
|
||||
|
||||
def test_login_assigns_username_and_nickname(client) -> None:
|
||||
"""新用户创建即分配:11 位纯数字 username(首位非 0/1,与手机号天然区分)+ 9 位字母数字默认昵称。"""
|
||||
phone = "13600136000"
|
||||
client.post("/api/v1/auth/sms/send", json={"phone": phone})
|
||||
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
|
||||
assert r.status_code == 200, r.text
|
||||
user = r.json()["user"]
|
||||
|
||||
uname = user["username"]
|
||||
assert uname.isdigit() and len(uname) == 11 # 11 位纯数字
|
||||
assert uname[0] not in ("0", "1") # 无前导 0、与手机号(均以 1 开头)区分
|
||||
|
||||
nick = user["nickname"]
|
||||
assert nick and len(nick) == 9 and nick.isalnum() # 9 位字母+数字
|
||||
|
||||
|
||||
def test_username_stable_and_unique_on_relogin() -> None:
|
||||
"""同号重登 username 不变(不可变标识);不同号 username 不同(唯一)。
|
||||
直连 repository 绕开 HTTP 短信冷却。"""
|
||||
from app.db.session import SessionLocal
|
||||
from app.repositories import user as user_repo
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
a1 = user_repo.upsert_user_for_login(db, phone="13522220001", register_channel="sms")
|
||||
uname_a, id_a = a1.username, a1.id
|
||||
a2 = user_repo.upsert_user_for_login(db, phone="13522220001", register_channel="sms")
|
||||
assert a2.id == id_a and a2.username == uname_a # 重登:同一行、username 稳定不变
|
||||
|
||||
b = user_repo.upsert_user_for_login(db, phone="13522220002", register_channel="sms")
|
||||
assert b.username != uname_a # 不同用户、username 唯一
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
+13
-5
@@ -300,7 +300,7 @@ def test_bind_no_code_no_fingerprint(client) -> None:
|
||||
# =====================================================================
|
||||
|
||||
def test_invitees_basic(client) -> None:
|
||||
"""A 邀 B、C → 列表返 2 条、total=2、没设资料的名字=脱敏手机号、头像 null、金币对。"""
|
||||
"""A 邀 B、C → 列表返 2 条、total=2、名字=被邀请人默认昵称(创建即有)、头像 null、金币对。"""
|
||||
a = _login(client, "13800002050")
|
||||
a_code = _my_code(client, a)
|
||||
for phone in ("13800002051", "13800002052"):
|
||||
@@ -313,9 +313,14 @@ def test_invitees_basic(client) -> None:
|
||||
assert body["total"] == 2
|
||||
assert body["has_more"] is False
|
||||
assert len(body["items"]) == 2
|
||||
# 没设昵称头像 → 名字脱敏手机号、头像 null(不依赖顺序用 set)
|
||||
# 创建即分配默认昵称 → 名字=被邀请人昵称(取实际值对比)、头像 null(不依赖顺序用 set)
|
||||
with SessionLocal() as db:
|
||||
expected_names = {
|
||||
get_user_by_phone(db, "13800002051").nickname,
|
||||
get_user_by_phone(db, "13800002052").nickname,
|
||||
}
|
||||
names = {it["display_name"] for it in body["items"]}
|
||||
assert names == {"138****2051", "138****2052"}
|
||||
assert names == expected_names
|
||||
assert all(it["avatar_url"] is None for it in body["items"])
|
||||
assert all(it["coins"] == INVITE_INVITER_COINS for it in body["items"])
|
||||
|
||||
@@ -336,9 +341,12 @@ def test_invitees_order_desc(client) -> None:
|
||||
).scalar_one()
|
||||
rel_b.created_at = datetime.now(timezone.utc) - timedelta(hours=2)
|
||||
db.commit()
|
||||
nick_b = ub.nickname
|
||||
nick_c = get_user_by_phone(db, "13800002062").nickname
|
||||
items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
|
||||
assert items[0]["display_name"] == "138****2062" # C 刚邀,在前
|
||||
assert items[1]["display_name"] == "138****2061" # B 2 小时前,在后
|
||||
# 创建即有默认昵称 → display_name=被邀请人昵称;本用例核心验证倒序(C 刚邀在前、B 2h 前在后)
|
||||
assert items[0]["display_name"] == nick_c
|
||||
assert items[1]["display_name"] == nick_b
|
||||
|
||||
|
||||
def test_invitees_nickname_priority(client) -> None:
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
"""平台公开端点测试(不鉴权):/api/v1/platform/flags 等。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def test_flags_default_comparing_ad_enabled(client: TestClient) -> None:
|
||||
"""空配置库下,/flags 返回 comparing_ad_enabled 的默认值 True;不需要鉴权。"""
|
||||
r = client.get("/api/v1/platform/flags")
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert "comparing_ad_enabled" in body
|
||||
assert body["comparing_ad_enabled"] is True
|
||||
@@ -0,0 +1,57 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""从美团「城市字典」Excel 生成随仓库的 JSON(app/integrations/data/meituan_cities.json)。
|
||||
|
||||
本地一次性工具:美团每年更新城市字典时,把新 Excel 放进来重跑即可。
|
||||
Excel 不入库、不进仓库(二进制、需 openpyxl);生成的 JSON 随仓库提交,部署到服务器后
|
||||
ETL(scripts/pull_meituan_coupons.py)直接读它遍历全部城市。
|
||||
|
||||
字典口径:359 个地级市,经实测一个地级市 cityId 已覆盖其下辖县级市(如徐州→邳州/新沂),
|
||||
故无需区县层级。
|
||||
|
||||
用法:
|
||||
python tools/gen_meituan_cities.py ["城市字典xxx.xlsx" 路径]
|
||||
(不传则用默认 e:\\codes\\城市字典2025 (1).xlsx)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
import openpyxl
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
ROOT = os.path.dirname(HERE)
|
||||
DEFAULT_EXCEL = r"e:\codes\城市字典2025 (1).xlsx"
|
||||
OUT = os.path.join(ROOT, "app", "integrations", "data", "meituan_cities.json")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
excel = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_EXCEL
|
||||
wb = openpyxl.load_workbook(excel, data_only=True)
|
||||
ws = wb.worksheets[0]
|
||||
rows = list(ws.iter_rows(values_only=True))
|
||||
header = rows[0]
|
||||
|
||||
cities = []
|
||||
seen = set()
|
||||
for r in rows[1:]:
|
||||
cid = str(r[0]).strip() if r[0] else ""
|
||||
name = str(r[1]).strip() if len(r) > 1 and r[1] else ""
|
||||
prov = str(r[2]).strip() if len(r) > 2 and r[2] else ""
|
||||
if not cid or cid in seen:
|
||||
continue
|
||||
seen.add(cid)
|
||||
cities.append({"city_id": cid, "name": name, "province": prov})
|
||||
|
||||
os.makedirs(os.path.dirname(OUT), exist_ok=True)
|
||||
with open(OUT, "w", encoding="utf-8") as f:
|
||||
json.dump(cities, f, ensure_ascii=False, indent=1)
|
||||
|
||||
print(f"表头: {header}")
|
||||
print(f"写出 {len(cities)} 城 → {OUT}")
|
||||
print("样例:", cities[:3])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user