diff --git a/.env.example b/.env.example index d3804bf..845b3d9 100644 --- a/.env.example +++ b/.env.example @@ -58,6 +58,13 @@ MT_CPS_DEFAULT_SID=sgbjia # 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。 MT_CPS_PROXY= +# ===== 京东联盟 CPS ===== +# 京东联盟/京东宙斯开放平台创建应用后填写。AUTH_KEY 是工具商授权 key,自有应用可留空。 +JD_UNION_APP_KEY= +JD_UNION_APP_SECRET= +JD_UNION_SITE_ID= +JD_UNION_AUTH_KEY= + # ===== Pricebot 上游 (领券/比价业务透传目标) ===== # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。 diff --git a/.gitignore b/.gitignore index d473d6b..4c87d95 100644 --- a/.gitignore +++ b/.gitignore @@ -23,14 +23,17 @@ dist/ *.db-journal *.db-shm *.db-wal -# data/ 整体不入库(运行时数据/上传文件/大二进制)。例外:邀请落地页 dl.html—— -# 它既是生产落地页又是本地测试资产,纳入 git 便于同事一致测试 +# data/ 整体不入库(运行时数据/上传文件/大二进制)。例外:邀请落地页 dl.html 及其 +# 引用的静态插画(coupon-page-bg.png 底图 + sb-brand.png logo)——既是生产落地页资产 +# 又是本地测试资产,纳入 git 便于同事一致测试、且随部署进生产 /media(否则线上 404→落地页毛坯)。 # (见 docs/邀请功能-实现原理与本地测试.md)。其余(avatars/ / *.apk / app.db 等)仍忽略。 data/* !data/media/ data/media/* !data/media/dl.html !data/media/taobao_landing.jpg +!data/media/coupon-page-bg.png +!data/media/sb-brand.png secrets/* !secrets/.gitkeep diff --git a/alembic/versions/761ef181ce7c_merge_jd_cps_order_fields_and_coupon_.py b/alembic/versions/761ef181ce7c_merge_jd_cps_order_fields_and_coupon_.py new file mode 100644 index 0000000..80d5d01 --- /dev/null +++ b/alembic/versions/761ef181ce7c_merge_jd_cps_order_fields_and_coupon_.py @@ -0,0 +1,26 @@ +"""merge jd_cps_order_fields and coupon_session_origin_package heads + +Revision ID: 761ef181ce7c +Revises: coupon_session_origin_package, jd_cps_order_fields +Create Date: 2026-07-01 13:52:16.068808 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '761ef181ce7c' +down_revision: Union[str, Sequence[str], None] = ('coupon_session_origin_package', 'jd_cps_order_fields') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/alembic/versions/coupon_session_origin_package.py b/alembic/versions/coupon_session_origin_package.py new file mode 100644 index 0000000..5287e93 --- /dev/null +++ b/alembic/versions/coupon_session_origin_package.py @@ -0,0 +1,32 @@ +"""coupon_session 加 origin_package 列(发起来源 App 包名 → admin「发起平台」) + +null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。与 trace_url 同理单独成迁移, +已建表环境靠它补列、全新环境顺序应用,不重复加列。 + +Revision ID: coupon_session_origin_package +Revises: coupon_session_trace_url +Create Date: 2026-06-30 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "coupon_session_origin_package" +down_revision: str | Sequence[str] | None = "coupon_session_trace_url" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.add_column(sa.Column("origin_package", sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.drop_column("origin_package") diff --git a/alembic/versions/coupon_session_table.py b/alembic/versions/coupon_session_table.py new file mode 100644 index 0000000..db0e355 --- /dev/null +++ b/alembic/versions/coupon_session_table.py @@ -0,0 +1,81 @@ +"""coupon session table(领券任务全程流水 → admin「领券数据」看板数据源) + +一次领券一行(trace_id 唯一):客户端 POST /api/v1/coupon/session 两段上报 —— 发起建行 +(status=started)、收尾(completed/failed/abandoned)按 trace_id 更新同一行。记全程耗时 +elapsed_ms + 各平台耗时 platform_elapsed + 机型/ROM,供 admin 算发起/完成数、耗时分位、机型维度。 + +Revision ID: coupon_session_table +Revises: feedback_submit_env +Create Date: 2026-06-30 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "coupon_session_table" +down_revision: str | Sequence[str] | None = "feedback_submit_env" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 model 的 _JSON variant)。 +_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql") + + +def upgrade() -> None: + op.create_table( + "coupon_session", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("trace_id", sa.String(length=64), nullable=False), + sa.Column("device_id", sa.String(length=64), nullable=False), + sa.Column("user_id", sa.Integer(), nullable=True), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("app_env", sa.String(length=16), nullable=True), + sa.Column("platforms", _JSON, nullable=True), + sa.Column("device_model", sa.String(length=128), nullable=True), + sa.Column("rom", sa.String(length=64), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("started_date", sa.Date(), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("elapsed_ms", sa.Integer(), nullable=True), + sa.Column("platform_elapsed", _JSON, nullable=True), + sa.Column("claimed_count", sa.Integer(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("trace_id", name="uq_coupon_session_trace"), + ) + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.create_index( + batch_op.f("ix_coupon_session_user_id"), ["user_id"], unique=False + ) + batch_op.create_index( + batch_op.f("ix_coupon_session_app_env"), ["app_env"], unique=False + ) + # admin 主聚合/筛选:按上海自然日 + 环境。 + batch_op.create_index( + "ix_coupon_session_date_env", ["started_date", "app_env"], unique=False + ) + + +def downgrade() -> None: + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.drop_index("ix_coupon_session_date_env") + batch_op.drop_index(batch_op.f("ix_coupon_session_app_env")) + batch_op.drop_index(batch_op.f("ix_coupon_session_user_id")) + op.drop_table("coupon_session") diff --git a/alembic/versions/coupon_session_trace_url.py b/alembic/versions/coupon_session_trace_url.py new file mode 100644 index 0000000..ce95c73 --- /dev/null +++ b/alembic/versions/coupon_session_trace_url.py @@ -0,0 +1,32 @@ +"""coupon_session 加 trace_url 列(pricebot done 帧公网调试链接) + +建表迁移 coupon_session_table 落地后才追加本列,故单独成一个迁移:已建表的环境(本地/已 upgrade 过) +靠它补列,全新环境则「建表(无 trace_url)→ 本迁移加列」,两条路一致、不重复加列。 + +Revision ID: coupon_session_trace_url +Revises: coupon_session_table +Create Date: 2026-06-30 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "coupon_session_trace_url" +down_revision: str | Sequence[str] | None = "coupon_session_table" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.add_column(sa.Column("trace_url", sa.String(length=512), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("coupon_session", schema=None) as batch_op: + batch_op.drop_column("trace_url") diff --git a/alembic/versions/jd_cps_order_fields.py b/alembic/versions/jd_cps_order_fields.py new file mode 100644 index 0000000..c371417 --- /dev/null +++ b/alembic/versions/jd_cps_order_fields.py @@ -0,0 +1,56 @@ +"""add jd cps order fields + +Revision ID: jd_cps_order_fields +Revises: 7db22acee504 +Create Date: 2026-06-28 20:30:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "jd_cps_order_fields" +down_revision = "7db22acee504" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("cps_order") as batch_op: + batch_op.add_column( + sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan") + ) + batch_op.add_column(sa.Column("external_order_id", sa.String(length=128), nullable=True)) + batch_op.add_column(sa.Column("external_row_id", sa.String(length=128), nullable=True)) + batch_op.add_column(sa.Column("estimated_commission_cents", sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column("actual_commission_cents", sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column("jd_valid_code", sa.String(length=16), nullable=True)) + batch_op.add_column(sa.Column("settle_month", sa.String(length=16), nullable=True)) + batch_op.add_column(sa.Column("site_id", sa.String(length=128), nullable=True)) + batch_op.add_column(sa.Column("position_id", sa.String(length=128), nullable=True)) + batch_op.add_column(sa.Column("pid", sa.String(length=128), nullable=True)) + batch_op.add_column(sa.Column("sub_union_id", sa.String(length=128), nullable=True)) + batch_op.create_index("ix_cps_order_platform", ["platform"]) + batch_op.create_index("ix_cps_order_external_order_id", ["external_order_id"]) + batch_op.create_index("ix_cps_order_external_row_id", ["external_row_id"]) + batch_op.create_index("ix_cps_order_jd_valid_code", ["jd_valid_code"]) + + +def downgrade() -> None: + with op.batch_alter_table("cps_order") as batch_op: + batch_op.drop_index("ix_cps_order_jd_valid_code") + batch_op.drop_index("ix_cps_order_external_row_id") + batch_op.drop_index("ix_cps_order_external_order_id") + batch_op.drop_index("ix_cps_order_platform") + batch_op.drop_column("sub_union_id") + batch_op.drop_column("pid") + batch_op.drop_column("position_id") + batch_op.drop_column("site_id") + batch_op.drop_column("settle_month") + batch_op.drop_column("jd_valid_code") + batch_op.drop_column("actual_commission_cents") + batch_op.drop_column("estimated_commission_cents") + batch_op.drop_column("external_row_id") + batch_op.drop_column("external_order_id") + batch_op.drop_column("platform") diff --git a/app/admin/main.py b/app/admin/main.py index 57d044a..ca3a374 100644 --- a/app/admin/main.py +++ b/app/admin/main.py @@ -21,6 +21,7 @@ from app.admin.routers.audit import router as audit_router from app.admin.routers.auth import router as auth_router from app.admin.routers.comparison import router as comparison_router from app.admin.routers.config import router as config_router +from app.admin.routers.coupon_data import router as coupon_data_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.device_liveness import router as device_liveness_router @@ -101,6 +102,7 @@ admin_app.include_router(audit_router) admin_app.include_router(config_router) admin_app.include_router(comparison_router) admin_app.include_router(cps_router) +admin_app.include_router(coupon_data_router) admin_app.include_router(ad_audit_router) admin_app.include_router(ad_config_router) admin_app.include_router(ad_revenue_router) diff --git a/app/admin/repositories/ad_audit.py b/app/admin/repositories/ad_audit.py index d81f725..888467c 100644 --- a/app/admin/repositories/ad_audit.py +++ b/app/admin/repositories/ad_audit.py @@ -136,12 +136,16 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool: """该信息流记录是否落入请求的展示筛选 scene。 - scene=="feed":ad_type in ("feed", NULL)(旧数据 NULL 视为 feed,向后兼容) - scene=="draw":ad_type=="draw" + - scene=="feed_all":所有信息流(feed/draw/NULL 都要)——业务已全切 Draw 信息流,收益报表把「Draw 信息流」 + 当作整个信息流口径(含历史误标 feed/NULL),用它避免筛选漏历史。 - scene 为 None:不筛(两类都要)。 """ if scene == "feed": return rec.ad_type in (None, "feed") if scene == "draw": return rec.ad_type == "draw" + if scene == "feed_all": + return True return True @@ -184,6 +188,7 @@ def _feed_rows( "record_id": rec.id, "user_id": rec.user_id, "ad_session_id": rec.ad_session_id, + "trace_id": rec.trace_id, "app_env": rec.app_env, "our_code_id": rec.our_code_id, "created_at": rec.created_at, @@ -209,6 +214,7 @@ def _feed_rows( "record_id": rec.id, "user_id": rec.user_id, "ad_session_id": rec.ad_session_id, + "trace_id": rec.trace_id, "app_env": rec.app_env, "our_code_id": rec.our_code_id, "created_at": rec.created_at, @@ -241,7 +247,7 @@ def audit_rows( rows: list[dict] = [] if scene in (None, "reward_video"): rows.extend(_reward_video_rows(db, date=date, user_id=user_id)) - if scene in (None, "feed", "draw"): + if scene in (None, "feed", "draw", "feed_all"): rows.extend(_feed_rows(db, date=date, user_id=user_id, scene=scene)) return rows diff --git a/app/admin/repositories/ad_revenue.py b/app/admin/repositories/ad_revenue.py index ec6bd4a..8dfcf2d 100644 --- a/app/admin/repositories/ad_revenue.py +++ b/app/admin/repositories/ad_revenue.py @@ -3,9 +3,11 @@ 只读。每行 = 一次广告事件(不再按用户聚合): - **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行, 直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。 -- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id) - 与逐条展示无法对应,单独成「纯发奖」行。 -- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。 +- **信息流(比价/领券)**:一次比价 / 一次领券 = 一条整场发奖(ad_feed_reward)一行,给出 eCPM / + 发奖金币 + 应发 / 实发 / 一致;点开看金币复算因子。⚠️ draw 的逐条展示(ad_ecpm,impressionId 各自 + 独立、与整场发奖无公共键、无法归到「哪一次」)**不再单独占行**(2026-07 按「一次比价/领券放一块」调整)—— + 其展示数 / eCPM / 预估收益仍进全量统计(合计 / 趋势 / 分类大盘 / 穿山甲对照),只是主表不逐条铺开。 +- 兜底:激励视频有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)仍各自成行。 展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算 (ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计, @@ -58,14 +60,6 @@ def _date_range(date_from: str, date_to: str) -> list[str]: _AUDIT_SCENES = {"reward_video", "feed", "draw"} -def _event_ad_type(row: dict) -> str: - """纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed; - 激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。""" - if row["scene"] == "reward_video": - return "reward_video" - return row.get("ad_type") or "feed" - - # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。 _REWARD_DETAIL_KEYS = ( "record_id", "created_at", "status", "ecpm", "ecpm_factor", "units", @@ -108,9 +102,14 @@ def ad_revenue_report( # 同时保留全量列表,未被展示合并的成「纯发奖」事件。 reward_by_session: dict[tuple[int, str], list[dict]] = {} all_reward_rows: list[dict] = [] - # 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被 - # 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。 - audit_scene = ad_type if ad_type in _AUDIT_SCENES else None + # 报表 ad_type → audit scene:reward_video/feed 直传;**draw(前端「Draw 信息流」)映射成 feed_all** + # ——业务已全切 Draw,把「Draw 信息流」当作整个信息流口径(含历史误标 feed/NULL),否则筛选会漏历史。 + if ad_type == "draw": + audit_scene = "feed_all" + elif ad_type in _AUDIT_SCENES: + audit_scene = ad_type + else: + audit_scene = None if ad_type is None or audit_scene is not None: for d in _date_range(date_from, date_to): for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene): @@ -140,7 +139,10 @@ def ad_revenue_report( ) if user_id is not None: stmt = stmt.where(AdEcpmRecord.user_id == user_id) - if ad_type is not None: + if ad_type == "draw": + # draw = 所有信息流展示(业务已全 Draw,含历史误标 feed);展示行只进统计,不占主表行 + stmt = stmt.where(AdEcpmRecord.ad_type.in_(["draw", "feed"])) + elif ad_type is not None: stmt = stmt.where(AdEcpmRecord.ad_type == ad_type) for rec in db.execute(stmt).scalars(): rwd = _pop_reward(rec.user_id, rec.ad_session_id) @@ -165,6 +167,8 @@ def ad_revenue_report( ), "adn": rec.adn, "slot_id": rec.slot_id, + "sub_rewards": [], + "sub_count": 1, } if rwd is not None: ev.update({ @@ -184,33 +188,88 @@ def ad_revenue_report( }) events.append(ev) - # 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。 - # 收益恒 0(收益只算展示侧,避免与展示行重复计)。 + # 3) 未被展示合并的发奖行 → 事件: + # - 激励视频(reward_video):逐条成「纯发奖」事件(每次一个 ad_session_id;有发奖无展示等)。 + # - 信息流(feed/draw):同一次比价/领券的多条广告共享**整场 ad_session_id**(客户端整场复用), + # 按 (user_id, ad_session_id) 聚成**一次比价 / 一次领券**父事件;sub_rewards 为组内逐条明细, + # 应发/实发取组内合计;业务已全 Draw → 类型统一 "draw"。session 缺失(极少旧数据)各自单独成组。 + feed_groups: dict[tuple[int, str], list[dict]] = {} for row in all_reward_rows: if row["record_id"] in used_reward_ids: continue + if row["scene"] == "reward_video": + events.append({ + "event_key": f"rwd-{row['record_id']}", + "report_date": row["_report_date"], + "user_id": row["user_id"], + "ad_type": "reward_video", + "feed_scene": row.get("feed_scene"), + "app_env": row.get("app_env"), + "our_code_id": row.get("our_code_id"), + "created_at": row["created_at"], + "hour": _cn_hour(row["created_at"]) if by_hour else None, + "has_impression": False, + "impressions": 0, + "ecpm": row["ecpm"], + "revenue_yuan": 0.0, + "adn": None, + "slot_id": None, + "has_reward": True, + "status": row["status"], + "expected_coin": int(row["expected_coin"]), + "actual_coin": int(row["actual_coin"]), + "matched": bool(row["matched"]), + "reward_detail": _reward_detail(row), + "sub_rewards": [], + "sub_count": 1, + }) + else: + # 聚合单位 = 一次完整比价/领券流程:优先用 trace_id(比价带 comparisonTraceId、领券带 sessionTraceId, + # 整个流程不变;即使中途点广告致浮层关闭重弹、ad_session_id 变了,trace_id 仍不变 → 全流程聚成一行)。 + # 无 trace_id(历史领券未上报 / 旧数据)回退整场 ad_session_id;再无则 record_id 各自成组、不误并。 + grp_key = row.get("trace_id") or row.get("ad_session_id") or f"_rid-{row['record_id']}" + feed_groups.setdefault((row["user_id"], grp_key), []).append(row) + + # 信息流分组 → 「一次比价 / 一次领券」父事件(收益恒 0:收益只算展示侧,避免与展示行重复计)。 + for (uid, grp_key), group in feed_groups.items(): + group.sort(key=lambda r: (r["created_at"], r["record_id"])) + rep = group[-1] # 代表条(最新一条):时间/场景/应用/代码位取它 + expected_sum = sum(int(g["expected_coin"]) for g in group) + actual_sum = sum(int(g["actual_coin"]) for g in group) + # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条 + ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")] + avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm") + # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进 + # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。 + row_revenue = round(sum( + min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0 + for g in group if g.get("ecpm") + ), 6) events.append({ - "event_key": f"rwd-{row['record_id']}", - "report_date": row["_report_date"], - "user_id": row["user_id"], - "ad_type": _event_ad_type(row), - "feed_scene": row.get("feed_scene"), - "app_env": row.get("app_env"), - "our_code_id": row.get("our_code_id"), - "created_at": row["created_at"], - "hour": _cn_hour(row["created_at"]) if by_hour else None, + "event_key": f"feedgrp-{uid}-{grp_key}", + "report_date": rep["_report_date"], + "user_id": uid, + "ad_type": "draw", # 业务已全切 Draw 信息流,聚合行统一 draw + "feed_scene": rep.get("feed_scene"), + "app_env": rep.get("app_env"), + "our_code_id": rep.get("our_code_id"), + "created_at": rep["created_at"], + "hour": _cn_hour(rep["created_at"]) if by_hour else None, "has_impression": False, "impressions": 0, - "ecpm": row["ecpm"], + "ecpm": avg_ecpm, "revenue_yuan": 0.0, + "row_revenue_yuan": row_revenue, "adn": None, "slot_id": None, "has_reward": True, - "status": row["status"], - "expected_coin": int(row["expected_coin"]), - "actual_coin": int(row["actual_coin"]), - "matched": bool(row["matched"]), - "reward_detail": _reward_detail(row), + "status": rep["status"], # 代表状态(逐条见展开) + "expected_coin": expected_sum, + "actual_coin": actual_sum, + "matched": all(bool(g["matched"]) for g in group), + "reward_detail": None, + "sub_rewards": [_reward_detail(g) for g in group], + "sub_count": len(group), }) # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。 @@ -331,9 +390,20 @@ def ad_revenue_report( is_today = date_from == date_to == rewards.cn_today().isoformat() dau = admin_stats.today_dau(db) if is_today else None + # 主表「逐行」= 单次广告行为(2026-07 按「一次比价/领券放一块」聚合):激励视频 = 一次观看一行(展示+发奖 + # 按 ad_session_id 合并);一次比价 / 一次领券 = 该次整场多条广告按 ad_session_id 聚成一行(展开看逐条)。 + # 信息流(draw/feed)的逐条展示(ad_ecpm,impressionId 各自独立、与整场发奖无公共键)不再单独占行 + # ——其展示数 / eCPM / 预估收益已计入上面的全量统计(total_*、daily / hourly、type_stats、穿山甲对照), + # 只是主表不逐条铺开;逐条明细在父行展开里看(sub_rewards)。合计 / 趋势 / 分类大盘均基于全量 events, + # 不受此过滤影响;total / 分页只作用于主表行。 + main_rows = [ + e for e in events + if not (e["ad_type"] in ("draw", "feed") and e["has_impression"] and not e["has_reward"]) + ] + return { - "total": len(events), - "truncated": len(events) > offset + limit, + "total": len(main_rows), + "truncated": len(main_rows) > offset + limit, "total_impressions": total_impressions, "total_revenue_yuan": total_revenue_yuan, # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。 @@ -347,5 +417,5 @@ def ad_revenue_report( "hourly": hourly, "type_stats": type_stats, "dau": dau, - "items": events[offset:offset + limit], + "items": main_rows[offset:offset + limit], } diff --git a/app/admin/repositories/coupon_data.py b/app/admin/repositories/coupon_data.py new file mode 100644 index 0000000..05b8fbd --- /dev/null +++ b/app/admin/repositories/coupon_data.py @@ -0,0 +1,225 @@ +"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。 + +数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉 +区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。 +- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。 +- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。 +- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。 +""" +from __future__ import annotations + +from datetime import UTC, date as _date, datetime + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from app.core import rewards +from app.models.coupon_state import CouponSession +from app.models.user import User + + +def _cn_hour(dt: datetime) -> int: + """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。""" + if dt.tzinfo is None: + dt = dt.replace(tzinfo=UTC) + return dt.astimezone(rewards.CN_TZ).hour + + +def _percentile(sorted_vals: list[int], q: float) -> int | None: + """线性插值分位(q=0..100,numpy 默认法)。sorted_vals 须已升序;空返回 None。""" + if not sorted_vals: + return None + if len(sorted_vals) == 1: + return sorted_vals[0] + idx = (len(sorted_vals) - 1) * q / 100.0 + lo = int(idx) + hi = min(lo + 1, len(sorted_vals) - 1) + frac = idx - lo + return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac) + + +def _avg(vals: list[int]) -> int | None: + return round(sum(vals) / len(vals)) if vals else None + + +def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict: + """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。""" + return { + "id": r.id, + "trace_id": r.trace_id, + "user_id": r.user_id, + "user_phone": phone, + "user_nickname": nickname, + "status": r.status, + "platforms": r.platforms, + "origin_package": r.origin_package, + "elapsed_ms": r.elapsed_ms, + "platform_elapsed": r.platform_elapsed, + "device_model": r.device_model, + "rom": r.rom, + "app_env": r.app_env, + "started_at": r.started_at, + "claimed_count": r.claimed_count, + "trace_url": r.trace_url, + } + + +def _empty_result() -> dict: + return { + "summary": { + "started_count": 0, "completed_count": 0, "avg_elapsed_ms": None, + "p5_ms": None, "p50_ms": None, "p95_ms": None, "p99_ms": None, + }, + "daily": [], + "hourly": [], + "total": 0, + "items": [], + } + + +def coupon_data_report( + db: Session, + *, + date_from: str, + date_to: str, + user: str | None = None, + app_env: str | None = None, + granularity: str = "day", + limit: int = 500, + offset: int = 0, + sort: str = "time", +) -> dict: + """日期区间(北京自然日 started_date,闭区间)领券数据:汇总卡 + 趋势 + 逐条明细。 + + - user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。 + - app_env:prod/dev 精确;None=全部。 + - sort:time=发起时刻倒序(默认) / elapsed=全程耗时倒序(None 末尾)。 + """ + by_hour = granularity == "hour" + d_from = _date.fromisoformat(date_from) + d_to = _date.fromisoformat(date_to) + + # user 模糊 → 先定位匹配用户 id;匹配不到直接空结果(不全表扫)。 + user_ids: set[int] | None = None + if user: + like = f"%{user}%" + user_ids = set(db.execute( + select(User.id).where(or_(User.phone.like(like), User.nickname.like(like))) + ).scalars().all()) + if not user_ids: + return _empty_result() + + stmt = select(CouponSession).where( + CouponSession.started_date >= d_from, + CouponSession.started_date <= d_to, + ) + if app_env is not None: + stmt = stmt.where(CouponSession.app_env == app_env) + if user_ids is not None: + stmt = stmt.where(CouponSession.user_id.in_(user_ids)) + rows = list(db.execute(stmt).scalars()) + + # ── 汇总卡 ── + completed_elapsed = sorted( + r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None + ) + summary = { + "started_count": len(rows), + "completed_count": sum(1 for r in rows if r.status == "completed"), + "avg_elapsed_ms": _avg(completed_elapsed), + "p5_ms": _percentile(completed_elapsed, 5), + "p50_ms": _percentile(completed_elapsed, 50), + "p95_ms": _percentile(completed_elapsed, 95), + "p99_ms": _percentile(completed_elapsed, 99), + } + + # ── 按天趋势(柱=发起/完成数,线=平均耗时)── + daily_map: dict[str, dict] = {} + for r in rows: + d = r.started_date.isoformat() + b = daily_map.get(d) + if b is None: + b = {"date": d, "started_count": 0, "completed_count": 0, "_elapsed": []} + daily_map[d] = b + b["started_count"] += 1 + if r.status == "completed": + b["completed_count"] += 1 + if r.elapsed_ms is not None: + b["_elapsed"].append(r.elapsed_ms) + daily = [ + { + "date": b["date"], + "started_count": b["started_count"], + "completed_count": b["completed_count"], + "avg_elapsed_ms": _avg(b["_elapsed"]), + } + for b in sorted(daily_map.values(), key=lambda x: x["date"]) + ] + + # ── 按小时趋势(单日 hour 粒度)── + hourly: list[dict] = [] + if by_hour: + hour_map: dict[int, dict] = {} + for r in rows: + h = _cn_hour(r.started_at) + b = hour_map.get(h) + if b is None: + b = {"hour": h, "started_count": 0, "completed_count": 0, "_elapsed": []} + hour_map[h] = b + b["started_count"] += 1 + if r.status == "completed": + b["completed_count"] += 1 + if r.elapsed_ms is not None: + b["_elapsed"].append(r.elapsed_ms) + hourly = [ + { + "hour": b["hour"], + "started_count": b["started_count"], + "completed_count": b["completed_count"], + "avg_elapsed_ms": _avg(b["_elapsed"]), + } + for b in sorted(hour_map.values(), key=lambda x: x["hour"]) + ] + + # ── 明细:排序 + 分页 + 补用户手机号/昵称(批量,防 N+1)── + if sort == "elapsed": + rows.sort(key=lambda r: (r.elapsed_ms is None, -(r.elapsed_ms or 0))) + else: # time:发起时刻倒序 + rows.sort(key=lambda r: r.started_at, reverse=True) + page = rows[offset:offset + limit] + + uids = {r.user_id for r in page if r.user_id is not None} + user_map: dict[int, tuple[str | None, str | None]] = {} + if uids: + user_map = { + uid: (phone, nickname) + for uid, phone, nickname in db.execute( + select(User.id, User.phone, User.nickname).where(User.id.in_(uids)) + ).all() + } + items = [] + for r in page: + phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None) + items.append(_session_to_row(r, phone, nickname)) + + return { + "summary": summary, + "daily": daily, + "hourly": hourly, + "total": len(rows), + "items": items, + } + + +def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict: + """某用户全部领券记录(点手机号抽屉用):按发起时刻倒序、不限日期,total=该用户领券总次数。""" + rows = list(db.execute( + select(CouponSession) + .where(CouponSession.user_id == user_id) + .order_by(CouponSession.started_at.desc()) + .limit(limit) + ).scalars()) + total = db.execute( + select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id) + ).scalar_one() + return {"items": [_session_to_row(r) for r in rows], "total": int(total)} diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py index 68b10e0..b678c38 100644 --- a/app/admin/repositories/cps.py +++ b/app/admin/repositories/cps.py @@ -7,13 +7,14 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation +from typing import Any from uuid import uuid4 from sqlalchemy import desc, func, select from sqlalchemy.orm import Session from app.admin.repositories.queries import _as_utc, offset_paginate -from app.integrations import meituan +from app.integrations import jd_union, 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 @@ -24,6 +25,11 @@ from app.models.cps_wx_user import CpsWxUser # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账 _INVALID_STATUS = {"4", "5"} _SETTLED_STATUS = "6" +_JD_INVALID_CODES = { + "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21", + "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36", +} +_JD_UNPAID_CODES = {"15"} # CPS 点击时序按北京时区分桶(运营看的是北京时间) _BJ_TZ = timezone(timedelta(hours=8)) @@ -47,6 +53,36 @@ def _ts_to_dt(ts: object) -> datetime | None: """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。""" if not ts: return None + + +def _jd_dt_to_utc(value: object) -> datetime | None: + """京东时间字符串(北京时间) → UTC aware datetime。""" + if value is None: + return None + s = str(value).strip() + if not s: + return None + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): + try: + dt = datetime.strptime(s, fmt) + return dt.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc) + except ValueError: + continue + return None + + +def _text(value: object) -> str | None: + if value is None: + return None + s = str(value).strip() + return s or None + + +def _pick(row: dict[str, Any], *keys: str) -> Any: + for key in keys: + if key in row and row[key] is not None: + return row[key] + return None try: return datetime.fromtimestamp(int(ts), tz=timezone.utc) except (ValueError, OSError, TypeError): @@ -249,6 +285,80 @@ def _map_order_fields(r: dict) -> dict: } +def _jd_order_key(r: dict[str, Any]) -> str | None: + row_id = _text(_pick(r, "id", "rowId", "orderRowId")) + if row_id: + return f"jd:{row_id}" + order_id = _text(_pick(r, "orderId", "parentOrderId")) + sku_id = _text(_pick(r, "skuId")) + if order_id and sku_id: + return f"jd:{order_id}:{sku_id}" + if order_id: + return f"jd:{order_id}" + return None + + +def _map_jd_order_fields(r: dict[str, Any]) -> dict: + """京东 order.row.query 单条订单行 → CpsOrder 字段。""" + sku_name = _text(_pick(r, "skuName", "goodsName", "productName")) + if sku_name and len(sku_name) > 500: + sku_name = sku_name[:500] + valid_code = _text(_pick(r, "validCode", "valid_code")) + actual_fee = _yuan_to_cents(_pick(r, "actualFee", "actual_fee")) + estimate_fee = _yuan_to_cents(_pick(r, "estimateFee", "estimate_fee")) + commission = actual_fee if actual_fee not in (None, 0) else estimate_fee + order_time = _jd_dt_to_utc(_pick(r, "orderTime", "order_time")) + return { + "platform": "jd", + "external_order_id": _text(_pick(r, "orderId", "parentOrderId")), + "external_row_id": _text(_pick(r, "id", "rowId", "orderRowId")), + "sid": _text(_pick(r, "subUnionId", "sub_union_id")), + "act_id": None, + "biz_line": None, + "trade_type": None, + "pay_price_cents": _yuan_to_cents( + _pick(r, "actualCosPrice", "estimateCosPrice", "price") + ), + "commission_cents": commission, + "commission_rate": _text(_pick(r, "commissionRate", "commission_rate")), + "refund_price_cents": None, + "refund_profit_cents": None, + "estimated_commission_cents": estimate_fee, + "actual_commission_cents": actual_fee, + "mt_status": None, + "jd_valid_code": valid_code, + "invalid_reason": None if _is_jd_valid_code(valid_code) else f"validCode={valid_code}", + "product_name": sku_name, + "settle_month": _text(_pick(r, "payMonth", "settleMonth", "pay_month")), + "site_id": _text(_pick(r, "siteId", "site_id")), + "position_id": _text(_pick(r, "positionId", "position_id")), + "pid": _text(_pick(r, "pid")), + "sub_union_id": _text(_pick(r, "subUnionId", "sub_union_id")), + "pay_time": order_time, + "mt_update_time": _jd_dt_to_utc(_pick(r, "modifyTime", "updateTime", "modify_time")) + or order_time, + "raw": r, + } + + +def _is_jd_valid_code(valid_code: str | None) -> bool: + code = str(valid_code).strip() if valid_code is not None else "" + return bool(code and code not in _JD_INVALID_CODES and code not in _JD_UNPAID_CODES) + + +def is_jd_order_valid(order: CpsOrder) -> bool: + return _is_jd_valid_code(order.jd_valid_code) + + +def effective_commission_cents(order: CpsOrder) -> int: + if order.platform == "jd": + if order.actual_commission_cents not in (None, 0): + return order.actual_commission_cents or 0 + if order.estimated_commission_cents is not None: + return order.estimated_commission_cents or 0 + return order.commission_cents or 0 + + def reconcile_orders( db: Session, *, start_time: int, end_time: int, query_time_type: int = 1, sid: str | None = None, max_pages: int = 200, @@ -274,6 +384,11 @@ def reconcile_orders( continue fetched += 1 fields = _map_order_fields(r) + fields.setdefault("platform", "meituan") + fields.setdefault("external_order_id", order_id) + fields.setdefault("external_row_id", None) + fields.setdefault("estimated_commission_cents", fields.get("commission_cents")) + fields.setdefault("actual_commission_cents", None) existing = db.execute( select(CpsOrder).where(CpsOrder.order_id == order_id) ).scalar_one_or_none() @@ -291,6 +406,56 @@ def reconcile_orders( return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages} +def reconcile_jd_orders( + db: Session, *, start_time: datetime, end_time: datetime, + query_time_type: int = 3, max_pages: int = 100, +) -> dict: + """调京东 order.row.query 拉单 → 按订单行 upsert。 + + 京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。 + """ + fetched = inserted = updated = pages = 0 + cur = start_time + while cur < end_time: + win_end = min(cur + timedelta(hours=1), end_time) + page = 1 + while page <= max_pages: + resp = jd_union.query_order_rows( + start_time=cur, + end_time=win_end, + query_time_type=query_time_type, + page_index=page, + page_size=200, + ) + rows = resp.get("rows") or [] + has_more = bool(resp.get("has_more")) + if not rows: + break + pages += 1 + for r in rows: + order_id = _jd_order_key(r) + if not order_id: + continue + fetched += 1 + fields = _map_jd_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 not has_more or len(rows) < 200: + break + page += 1 + cur = win_end + 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, diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 488bcf7..51ef145 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -806,6 +806,12 @@ def get_user_overview(db: Session, user_id: int) -> dict | None: } +def _as_utc_naive(value: datetime) -> datetime: + """窗口入参 → UTC naive(= _as_utc 去时区),与库里按 naive UTC 存取的 created_at 同口径比较。 + 历史遗留:_window_conds 一直引用本函数却未定义(自定义区间会 NameError),此处补上。""" + return _as_utc(value).replace(tzinfo=None) + + def _window_conds(col, date_from: datetime | None, date_to: datetime | None) -> list: """把 [date_from, date_to] 转成对 col(created_at)的过滤条件;都为 None = 全量(注册至今)。""" conds = [] @@ -895,6 +901,13 @@ def user_reward_stats( } +def _cn_wall_to_utc(dt: datetime) -> datetime: + """coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive, + 与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示) + 口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。""" + return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None) + + def user_coin_records( db: Session, user_id: int, @@ -915,6 +928,9 @@ def user_coin_records( offset = max(cursor or 0, 0) fetch = offset + limit + 1 rows: list[dict] = [] + # coin_transaction 存北京 wall-clock(其余表存 UTC);签到窗口边界 +8h 对齐北京,过滤/计数才不偏移 8 小时 + signin_from = date_from + timedelta(hours=8) if date_from is not None else None + signin_to = date_to + timedelta(hours=8) if date_to is not None else None for rec in db.execute( select(AdRewardRecord) @@ -958,7 +974,7 @@ def user_coin_records( .where( CoinTransaction.user_id == user_id, CoinTransaction.biz_type == "signin", - *_window_conds(CoinTransaction.created_at, date_from, date_to), + *_window_conds(CoinTransaction.created_at, signin_from, signin_to), ) .order_by(CoinTransaction.created_at.desc()) .limit(fetch) @@ -966,7 +982,8 @@ def user_coin_records( rows.append({ "source": "signin", "source_label": "签到", - "created_at": rec.created_at, + # 北京 wall-clock → UTC,与广告记录统一(前端 apiTime 会 +8 回北京展示,不然签到会多 8 小时) + "created_at": _cn_wall_to_utc(rec.created_at), "ecpm": None, "coin": rec.amount, }) @@ -992,7 +1009,7 @@ def user_coin_records( + _count( CoinTransaction, CoinTransaction.user_id == user_id, CoinTransaction.biz_type == "signin", - *_window_conds(CoinTransaction.created_at, date_from, date_to), + *_window_conds(CoinTransaction.created_at, signin_from, signin_to), ) ) return rows[offset:offset + limit], (offset + limit if has_more else None), total diff --git a/app/admin/repositories/stats.py b/app/admin/repositories/stats.py index 66baead..202a5ba 100644 --- a/app/admin/repositories/stats.py +++ b/app/admin/repositories/stats.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session from app.models.ad_feed_reward import AdFeedRewardRecord from app.models.ad_reward import AdRewardRecord +from app.models.analytics_event import AnalyticsEvent from app.models.comparison import ComparisonRecord from app.models.coupon_state import CouponPromptEngagement from app.models.cps_order import CpsOrder @@ -35,6 +36,13 @@ REGULAR_TASK_EXCLUDED_BIZ_TYPES = ( ) MEITUAN_CPS_INVALID_STATUSES = ("4", "5") MEITUAN_CPS_SETTLED_STATUS = "6" +COMPARE_START_EVENT = "real_compare_start" +COUPON_START_EVENT = "real_coupon_start" +JD_CPS_INVALID_CODES = { + "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21", + "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36", +} +JD_CPS_UNPAID_CODES = {"15"} def _beijing_today_start_utc() -> datetime: @@ -45,14 +53,31 @@ def _beijing_today_start_utc() -> datetime: def today_dau(db: Session) -> int: - """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。 + """今日活跃用户数(DAU):登录 + 开始比价 + 开始领券,按用户去重。 广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。 """ + today_bj = datetime.now(_BEIJING).date() today_start = _beijing_today_start_utc() - return int( - db.execute(select(func.count(User.id)).where(User.last_login_at >= today_start)).scalar_one() + tomorrow_start = today_start + timedelta(days=1) + login_user_ids = _id_set( + db, + select(User.id).where(User.last_login_at >= today_start, User.last_login_at < tomorrow_start), ) + compare_start_user_ids = _event_user_ids( + db, (COMPARE_START_EVENT,), today_start, tomorrow_start + ) + coupon_event_user_ids = _event_user_ids( + db, (COUPON_START_EVENT,), today_start, tomorrow_start + ) + coupon_claim_user_ids = _id_set( + db, + select(CouponPromptEngagement.user_id).where( + CouponPromptEngagement.engage_date == today_bj, + CouponPromptEngagement.engage_type == "claim_started", + ), + ) + return len(login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids) def _default_period_end() -> date: @@ -91,6 +116,24 @@ def _date_range(date_from: date, date_to: date) -> list[date]: return [date_from + timedelta(days=i) for i in range(days + 1)] +def _id_set(db: Session, stmt) -> set[int]: + return {int(v) for v in db.execute(stmt).scalars().all() if v is not None} + + +def _event_user_ids( + db: Session, event_names: tuple[str, ...], start_utc: datetime, end_utc: datetime +) -> set[int]: + return _id_set( + db, + select(AnalyticsEvent.user_id).where( + AnalyticsEvent.user_id.is_not(None), + AnalyticsEvent.event.in_(event_names), + AnalyticsEvent.created_at >= start_utc, + AnalyticsEvent.created_at < end_utc, + ), + ) + + def _commission_rate_percent(raw: str | None) -> Decimal | None: """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。""" if raw is None: @@ -107,6 +150,11 @@ def _commission_rate_percent(raw: str | None) -> Decimal | None: return val / Decimal("100") +def _jd_valid_order(order: CpsOrder) -> bool: + code = str(order.jd_valid_code).strip() if order.jd_valid_code is not None else "" + return bool(code and code not in JD_CPS_INVALID_CODES and code not in JD_CPS_UNPAID_CODES) + + def dashboard_overview( db: Session, *, date_from: date | None = None, date_to: date | None = None ) -> dict: @@ -216,17 +264,22 @@ def dashboard_overview( login_user_ids = _user_id_set( select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc) ) - compare_user_ids = _user_id_set( - select(ComparisonRecord.user_id).where(*period_comparison_conds) + compare_start_user_ids = _event_user_ids( + db, (COMPARE_START_EVENT,), start_utc, end_utc ) - coupon_user_ids = _user_id_set( + coupon_event_user_ids = _event_user_ids( + db, (COUPON_START_EVENT,), start_utc, end_utc + ) + coupon_claim_user_ids = _user_id_set( select(CouponPromptEngagement.user_id).where( CouponPromptEngagement.engage_date >= period_from, CouponPromptEngagement.engage_date <= period_to, CouponPromptEngagement.engage_type == "claim_started", ) ) - period_active_user_ids = login_user_ids | compare_user_ids | coupon_user_ids + period_active_user_ids = ( + login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids + ) period_retained_new_user_ids = period_new_user_ids & period_active_user_ids period_retention_rate = ( round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4) @@ -248,10 +301,13 @@ def dashboard_overview( User.last_login_at < day_end_utc, ) ) - daily_compare_user_ids = _user_id_set( - select(ComparisonRecord.user_id).where(*daily_comparison_conds) + daily_compare_start_user_ids = _event_user_ids( + db, (COMPARE_START_EVENT,), day_start_utc, day_end_utc ) - daily_coupon_user_ids = _user_id_set( + daily_coupon_event_user_ids = _event_user_ids( + db, (COUPON_START_EVENT,), day_start_utc, day_end_utc + ) + daily_coupon_claim_user_ids = _user_id_set( select(CouponPromptEngagement.user_id).where( CouponPromptEngagement.engage_date == cur_date, CouponPromptEngagement.engage_type == "claim_started", @@ -261,7 +317,10 @@ def dashboard_overview( { "date": cur_date, "active_users": len( - daily_login_user_ids | daily_compare_user_ids | daily_coupon_user_ids + daily_login_user_ids + | daily_compare_start_user_ids + | daily_coupon_event_user_ids + | daily_coupon_claim_user_ids ), "new_users": _count( User, @@ -317,7 +376,7 @@ def dashboard_overview( *period_coin_conds, CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES), ) - period_meituan_orders = list( + period_cps_orders = list( db.execute( select(CpsOrder).where( CpsOrder.pay_time >= start_utc, @@ -325,9 +384,17 @@ def dashboard_overview( ) ).scalars() ) + period_meituan_orders = [ + o for o in period_cps_orders if (o.platform or "meituan") == "meituan" + ] period_meituan_valid_orders = [ o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES ] + period_jd_orders = [o for o in period_cps_orders if o.platform == "jd"] + period_jd_valid_orders = [o for o in period_jd_orders if _jd_valid_order(o)] + period_jd_invalid_orders = [ + o for o in period_jd_orders if o.jd_valid_code and not _jd_valid_order(o) + ] period_meituan_hit_count = 0 period_meituan_miss_count = 0 period_meituan_unknown_rate_count = 0 @@ -412,8 +479,8 @@ def dashboard_overview( "retained_new_users": len(period_retained_new_user_ids), "retention_rate": period_retention_rate, "retention_note": ( - "近似口径:登录(last_login_at)+已上报比价记录+领券claim_started;" - "尚不包含未完成上报的比价开始事件" + "口径:登录(last_login_at)+开始比价(real_compare_start)+" + "开始领券(real_coupon_start/claim_started),按用户去重" ), }, "comparison": { @@ -450,7 +517,7 @@ def dashboard_overview( }, "cps": { "available": True, - "note": "美团 CPS 读 cps_order 对账订单;淘宝/京东佣金暂空", + "note": "美团/JD CPS 读 cps_order 对账订单;淘宝佣金暂空", "meituan_order_count": len(period_meituan_valid_orders), "meituan_commission_cents": sum( o.commission_cents or 0 for o in period_meituan_valid_orders @@ -459,5 +526,17 @@ def dashboard_overview( "meituan_miss_count": period_meituan_miss_count, "meituan_unknown_rate_count": period_meituan_unknown_rate_count, "meituan_hit_rate": period_meituan_hit_rate, + "jd_order_count": len(period_jd_valid_orders), + # 数据大盘京东 CPS 只看实际佣金,不再用预估佣金兜底。 + "jd_commission_cents": sum( + o.actual_commission_cents or 0 for o in period_jd_valid_orders + ), + "jd_actual_commission_cents": sum( + o.actual_commission_cents or 0 for o in period_jd_valid_orders + ), + "jd_estimated_commission_cents": sum( + o.estimated_commission_cents or 0 for o in period_jd_valid_orders + ), + "jd_invalid_count": len(period_jd_invalid_orders), }, } diff --git a/app/admin/routers/coupon_data.py b/app/admin/routers/coupon_data.py new file mode 100644 index 0000000..f3d2548 --- /dev/null +++ b/app/admin/routers/coupon_data.py @@ -0,0 +1,106 @@ +"""admin「领券数据」看板:发起/完成数 + 领券耗时(均值 + P5/P50/P95/P99)+ 按天趋势 + 逐条明细。 + +任意已登录 admin 可看(只读)。聚合逻辑在 app/admin/repositories/coupon_data.py。 +数据源 coupon_session(客户端 /api/v1/coupon/session 两段上报)。 +""" +from __future__ import annotations + +from datetime import date as _date +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Query + +from app.admin.deps import AdminDb, get_current_admin +from app.admin.repositories import coupon_data +from app.admin.schemas.coupon_data import ( + CouponDataDaily, + CouponDataHourly, + CouponDataOut, + CouponDataRow, + CouponDataSummary, + CouponUserRecordsOut, +) +from app.core.rewards import cn_today + +router = APIRouter( + prefix="/admin/api/coupon-data", + tags=["admin-coupon-data"], + dependencies=[Depends(get_current_admin)], +) + +# 区间最大跨度(天);超出拒绝,避免一次拉过多天拖垮接口(对齐广告收益报表)。 +_MAX_RANGE_DAYS = 92 + + +def _parse_day(value: str | None, *, field: str, default: _date) -> _date: + if value is None: + return default + try: + return _date.fromisoformat(value) + except ValueError as e: + raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e + + +@router.get( + "", + response_model=CouponDataOut, + summary="领券数据看板(发起/完成数 + 耗时分位 + 按天趋势 + 逐条明细)", +) +def get_coupon_data( + db: AdminDb, + date_from: Annotated[str | None, Query(description="起始日 北京 YYYY-MM-DD,默认今天")] = None, + date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None, + user: Annotated[str | None, Query(description="用户手机号/昵称模糊搜;不传=全部")] = None, + app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod", + granularity: Annotated[ + str, Query(description="day=按天 / hour=按小时(北京);区间>1 天建议 day") + ] = "day", + limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500, + offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过条数)=(页码-1)×每页条数")] = 0, + sort: Annotated[ + str, Query(description="排序:time=发起时间倒序(默认) / elapsed=耗时倒序") + ] = "time", +) -> CouponDataOut: + today = cn_today() + d_from = _parse_day(date_from, field="date_from", default=today) + d_to = _parse_day(date_to, field="date_to", default=d_from) + if d_to < d_from: + raise HTTPException(status_code=422, detail="date_to 不能早于 date_from") + if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS: + raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天") + + # 报表默认只看 prod(对齐广告报表防串台口径);app_env=all 时不过滤、看全部环境。 + env = None if app_env == "all" else app_env + result = coupon_data.coupon_data_report( + db, date_from=d_from.isoformat(), date_to=d_to.isoformat(), + user=user, app_env=env, granularity=granularity, + limit=limit, offset=offset, sort=sort, + ) + return CouponDataOut( + date_from=d_from.isoformat(), + date_to=d_to.isoformat(), + summary=CouponDataSummary(**result["summary"]), + daily=[CouponDataDaily(**d) for d in result["daily"]], + hourly=[CouponDataHourly(**h) for h in result["hourly"]], + total=result["total"], + items=[CouponDataRow(**r) for r in result["items"]], + ) + + +@router.get( + "/user-records", + response_model=CouponUserRecordsOut, + summary="某用户全部领券记录(点手机号抽屉:领券次数 + 记录列表)", +) +def get_user_coupon_records( + db: AdminDb, + user_id: Annotated[int, Query(description="用户 id")], + limit: Annotated[int, Query(ge=1, le=500, description="最多返回条数")] = 100, + sort_by: Annotated[str, Query(description="兼容 UserRecordsDrawer 参数;固定按发起时间倒序")] = "created_at", + sort_order: Annotated[str, Query(description="兼容参数,忽略")] = "desc", +) -> CouponUserRecordsOut: + result = coupon_data.coupon_user_records(db, user_id=user_id, limit=limit) + return CouponUserRecordsOut( + items=[CouponDataRow(**r) for r in result["items"]], + total=result["total"], + ) diff --git a/app/admin/routers/cps.py b/app/admin/routers/cps.py index 433f8f6..564c4e8 100644 --- a/app/admin/routers/cps.py +++ b/app/admin/routers/cps.py @@ -1,7 +1,7 @@ -"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 美团订单对账 + 统计。 +"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 联盟订单对账 + 统计。 平台:meituan(actId+sid 转链 + query_order 对账) / taobao(整段淘口令) / jd(链接)。 -淘宝/京东无 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。 +淘宝暂未接 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。 群/活动管理 = operator;订单对账(涉佣金) = finance;只读列表/统计 = 登录即可。 """ from __future__ import annotations @@ -34,6 +34,7 @@ from app.admin.schemas.cps import ( from app.core import media from app.core.config import settings from app.integrations import meituan +from app.integrations.jd_union import JdUnionError from app.integrations.meituan import MeituanCpsError from app.models.admin import AdminUser from app.models.cps_activity import CpsActivity @@ -373,7 +374,22 @@ def _reconcile_range_to_ts( return int(start_dt.timestamp()), int(end_dt.timestamp()) -@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账") +def _reconcile_range_to_bj_dt( + date_from: _date | None, date_to: _date | None, days: int +) -> tuple[datetime, datetime]: + start_ts, end_ts = _reconcile_range_to_ts(date_from, date_to, days) + return ( + datetime.fromtimestamp(start_ts, tz=_BEIJING), + datetime.fromtimestamp(end_ts, tz=_BEIJING), + ) + + +def _merge_reconcile_result(total: dict, current: dict) -> None: + for key in ("fetched", "inserted", "updated", "pages"): + total[key] = int(total.get(key, 0)) + int(current.get(key, 0)) + + +@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取联盟订单对账") def reconcile_orders( request: Request, admin: Annotated[AdminUser, Depends(require_role("finance"))], @@ -382,26 +398,42 @@ def reconcile_orders( date_to: Annotated[str | None, Query(description="结束日 YYYY-MM-DD")] = None, days: Annotated[int, Query(ge=1, le=90, description="未传日期时默认拉近 N 天")] = 7, sid: Annotated[str | None, Query(max_length=64)] = None, - query_time_type: Annotated[int, Query(ge=1, le=2)] = 1, + query_time_type: Annotated[int, Query(ge=1, le=3)] = 1, + platform: Annotated[str, Query(pattern="^(all|meituan|jd)$")] = "all", ) -> CpsReconcileResult: - start_ts, end_ts = _reconcile_range_to_ts( - _parse_day(date_from, field="date_from"), - _parse_day(date_to, field="date_to"), - days, - ) + parsed_from = _parse_day(date_from, field="date_from") + parsed_to = _parse_day(date_to, field="date_to") + result = {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0} try: - result = cps_repo.reconcile_orders( - db, - start_time=start_ts, - end_time=end_ts, - query_time_type=query_time_type, - sid=sid, - ) + if platform in {"all", "meituan"}: + start_ts, end_ts = _reconcile_range_to_ts(parsed_from, parsed_to, days) + mt_result = cps_repo.reconcile_orders( + db, + start_time=start_ts, + end_time=end_ts, + query_time_type=query_time_type if query_time_type in (1, 2) else 2, + sid=sid, + ) + _merge_reconcile_result(result, mt_result) + if platform in {"all", "jd"}: + if sid: + raise HTTPException(status_code=422, detail="京东订单刷新不支持 sid 筛选") + start_dt, end_dt = _reconcile_range_to_bj_dt(parsed_from, parsed_to, days) + jd_result = cps_repo.reconcile_jd_orders( + db, + start_time=start_dt, + end_time=end_dt, + query_time_type=query_time_type, + ) + _merge_reconcile_result(result, jd_result) except MeituanCpsError as e: raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e + except JdUnionError 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={ + "platform": platform, "date_from": date_from, "date_to": date_to, "days": days, diff --git a/app/admin/schemas/ad_revenue.py b/app/admin/schemas/ad_revenue.py index a45ec5e..cec29b2 100644 --- a/app/admin/schemas/ad_revenue.py +++ b/app/admin/schemas/ad_revenue.py @@ -94,6 +94,11 @@ class AdRevenueRow(BaseModel): impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)") ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值") revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0") + row_revenue_yuan: float | None = Field( + None, + description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;" + "其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计", + ) adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空") slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空") # ── 发奖侧 ── @@ -106,6 +111,15 @@ class AdRevenueRow(BaseModel): None, description="发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);点行展开下钻用,纯展示为空", ) + sub_rewards: list[AdRevenueRecord] = Field( + default_factory=list, + description="一次比价/领券聚合行的组内逐条发奖明细(同一整场 ad_session_id 的多条广告);" + "点行展开渲染多行。激励视频/纯展示行为空(单条看 reward_detail)", + ) + sub_count: int = Field( + 1, + description="本行聚合的发奖条数:一次比价/领券=该次广告条数(≥1);激励视频/纯展示=1", + ) class AdRevenueReportOut(BaseModel): diff --git a/app/admin/schemas/coupon_data.py b/app/admin/schemas/coupon_data.py new file mode 100644 index 0000000..c94938d --- /dev/null +++ b/app/admin/schemas/coupon_data.py @@ -0,0 +1,83 @@ +"""admin「领券数据」看板 schemas:汇总卡 + 按天/小时趋势 + 逐条领券明细。 + +数据源 coupon_session(一次领券一行)。耗时单位 ms(前端按需折秒);均值/分位只统计 completed。 +""" +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class CouponDataSummary(BaseModel): + """汇总卡:发起/完成数 + 耗时均值与分位(P5/P50/P95/P99,基于 completed 的 elapsed_ms)。""" + + started_count: int = Field(..., description="发起数(区间内所有领券 session)") + completed_count: int = Field(..., description="完成数(status=completed)") + avg_elapsed_ms: int | None = Field(None, description="平均耗时(ms,仅 completed;无数据为空)") + p5_ms: int | None = Field(None, description="耗时 5 分位(ms)") + p50_ms: int | None = Field(None, description="耗时 50 分位(ms,中位数)") + p95_ms: int | None = Field(None, description="耗时 95 分位(ms)") + p99_ms: int | None = Field(None, description="耗时 99 分位(ms)") + + +class CouponDataDaily(BaseModel): + """按天趋势(全量,不受分页影响):柱=发起/完成数,线=平均耗时。""" + + date: str = Field(..., description="北京时间 YYYY-MM-DD") + started_count: int + completed_count: int + avg_elapsed_ms: int | None = Field(None, description="当天平均耗时(ms,仅 completed)") + + +class CouponDataHourly(BaseModel): + """按北京小时(0–23)趋势(单日 granularity=hour 时非空)。""" + + hour: int = Field(..., description="北京时间小时 0–23") + started_count: int + completed_count: int + avg_elapsed_ms: int | None = None + + +class CouponDataRow(BaseModel): + """一条领券明细(一次领券任务)。""" + + id: int = Field(..., description="coupon_session 主键(抽屉 rowKey 用)") + trace_id: str + user_id: int | None = None + user_phone: str | None = Field(None, description="手机号(admin 展示;匿名领券/查不到为空)") + user_nickname: str | None = Field(None, description="昵称") + status: str = Field(..., description="started / completed / failed / abandoned") + platforms: list[str] | None = Field(None, description="发起勾选平台") + origin_package: str | None = Field(None, description="发起来源 App 包名;null=App 内(傻瓜比价首页)发起") + elapsed_ms: int | None = Field(None, description="全程耗时(ms)") + platform_elapsed: dict[str, int] | None = Field( + None, description="各平台耗时 {meituan-waimai/taobao-shanguang/jd-waimai: ms}" + ) + device_model: str | None = None + rom: str | None = None + app_env: str | None = None + started_at: datetime = Field(..., description="发起时刻(明细「时间」列)") + claimed_count: int | None = None + trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id") + + +class CouponDataOut(BaseModel): + """领券数据看板响应:汇总卡 + 趋势 + 明细分页。""" + + date_from: str + date_to: str + summary: CouponDataSummary + daily: list[CouponDataDaily] = Field(default_factory=list, description="按天趋势(全量)") + hourly: list[CouponDataHourly] = Field( + default_factory=list, description="按小时趋势(单日 hour 粒度时非空)" + ) + total: int = Field(..., description="明细总条数(全量,不受分页)") + items: list[CouponDataRow] = Field(..., description="逐条领券明细(当前页)") + + +class CouponUserRecordsOut(BaseModel): + """某用户全部领券记录(点手机号抽屉用):total=该用户领券总次数,items=记录列表(UserRecordsDrawer 渲染)。""" + + items: list[CouponDataRow] + total: int diff --git a/app/admin/schemas/cps.py b/app/admin/schemas/cps.py index e7c3f7e..5afdad6 100644 --- a/app/admin/schemas/cps.py +++ b/app/admin/schemas/cps.py @@ -1,7 +1,7 @@ """admin CPS 分发与对账 schemas。金额统一「分」(cents),前端 yuan() 展示。 -平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接,只统计点击)。 -对账类字段对淘宝/京东为 None → 前端显示 "-"(无法对账)。 +平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接 + 订单 API 对账)。 +对账类字段对淘宝为 None → 前端显示 "-"(暂未对账)。 """ from __future__ import annotations @@ -116,15 +116,22 @@ class CpsOrderOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int + platform: str = "meituan" order_id: str + external_order_id: str | None = None + external_row_id: str | None = None sid: str | None = None act_id: str | None = None pay_price_cents: int | None = None commission_cents: int | None = None + estimated_commission_cents: int | None = None + actual_commission_cents: int | None = None commission_rate: str | None = None mt_status: str | None = None + jd_valid_code: str | None = None invalid_reason: str | None = None product_name: str | None = None + settle_month: str | None = None pay_time: datetime | None = None diff --git a/app/admin/schemas/dashboard.py b/app/admin/schemas/dashboard.py index 0ec1afb..56f2c15 100644 --- a/app/admin/schemas/dashboard.py +++ b/app/admin/schemas/dashboard.py @@ -103,6 +103,11 @@ class DashboardCps(BaseModel): meituan_miss_count: int = 0 meituan_unknown_rate_count: int = 0 meituan_hit_rate: float | None = None + jd_order_count: int = 0 + jd_commission_cents: int = 0 + jd_actual_commission_cents: int = 0 + jd_estimated_commission_cents: int = 0 + jd_invalid_count: int = 0 class DashboardOverview(BaseModel): diff --git a/app/admin/schemas/user.py b/app/admin/schemas/user.py index 8e7020e..2e554a8 100644 --- a/app/admin/schemas/user.py +++ b/app/admin/schemas/user.py @@ -17,6 +17,7 @@ class AdminUserListItem(BaseModel): status: str debug_trace_enabled: bool = False wechat_openid: str | None = None + wechat_nickname: str | None = None created_at: datetime last_login_at: datetime diff --git a/app/api/v1/coupon.py b/app/api/v1/coupon.py index 1734c79..1a5368d 100644 --- a/app/api/v1/coupon.py +++ b/app/api/v1/coupon.py @@ -29,6 +29,7 @@ from app.schemas.coupon_state import ( CouponPromptDismissIn, CouponPromptShouldShowOut, CouponPromptShownIn, + CouponSessionIn, CouponStatsOut, ) @@ -194,6 +195,34 @@ async def coupon_step( return resp_json +@router.post("/session", summary="领券任务流水上报(admin 领券数据看板数据源)") +def coupon_session(payload: CouponSessionIn, db: DbSession) -> dict[str, bool]: + """客户端两段上报一次领券流水(发起 started / 收尾 completed-failed-abandoned),按 trace_id upsert + 到 coupon_session。不鉴权(同领券循环 MVP,按 device_id/trace_id);供 admin「领券数据」看板算 + 发起/完成数、耗时分位、机型维度。写库失败不应连累客户端(本就 fire-and-forget),吞掉返回 ok。""" + try: + coupon_repo.upsert_coupon_session( + db, + trace_id=payload.trace_id, + device_id=payload.device_id, + status=payload.status, + started_at_ms=payload.started_at_ms, + user_id=payload.user_id, + platforms=payload.platforms, + origin_package=payload.origin_package, + device_model=payload.device_model, + rom=payload.rom, + app_env=payload.app_env, + elapsed_ms=payload.elapsed_ms, + platform_elapsed=payload.platform_elapsed, + claimed_count=payload.claimed_count, + trace_url=payload.trace_url, + ) + except Exception as e: # noqa: BLE001 + logger.warning("coupon session write failed: %s", e) + return {"ok": True} + + @router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)") def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]: """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。 diff --git a/app/api/v1/wallet.py b/app/api/v1/wallet.py index 8641cf2..29a0407 100644 --- a/app/api/v1/wallet.py +++ b/app/api/v1/wallet.py @@ -202,6 +202,9 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw order = crud_wallet.create_withdraw( db, user.id, req.amount_cents, source=req.source, user_name=req.user_name, out_bill_no=req.out_bill_no, + # 0.01 元调试提现:放行低于最低额的小额。双闸——客户端仅 debug 包在「0.01 元提现」开关开时 + # 连同 skip_review 一起下发;服务端仅非 prod 才认。生产恒 False,最低额校验照常。 + allow_sub_min=(req.skip_review and not settings.is_prod), ) except crud_wallet.InvalidWithdrawAmountError as e: raise HTTPException( diff --git a/app/core/config.py b/app/core/config.py index f5827a2..79a12e8 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -113,6 +113,21 @@ class Settings(BaseSettings): """美团 CPS 凭证齐全(缺则接口返空,而非 502)。""" return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET) + # ===== 京东联盟 CPS ===== + # app_key/app_secret 来自京东联盟应用;site_id 是推广管理里的 APP/网站 ID; + # auth_key 是工具商授权 key,自有应用查询可留空。 + JD_UNION_APP_KEY: str = "" + JD_UNION_APP_SECRET: str = "" + JD_UNION_SITE_ID: str = "" + JD_UNION_AUTH_KEY: str = "" + JD_UNION_GATEWAY: str = "https://api.jd.com/routerjson" + JD_UNION_TIMEOUT_SEC: int = 15 + + @property + def jd_union_configured(self) -> bool: + """京东联盟订单查询凭证齐全。""" + return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET) + # ===== 微信服务号(网页授权) ===== # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。 # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 diff --git a/app/integrations/jd_union.py b/app/integrations/jd_union.py new file mode 100644 index 0000000..ac19256 --- /dev/null +++ b/app/integrations/jd_union.py @@ -0,0 +1,163 @@ +"""京东联盟 OpenAPI 客户端。 + +当前只接数据大盘需要的订单明细接口: +`jd.union.open.order.row.query`。京东要求订单查询时间窗最长 1 小时, +调用方负责切窗分页。 +""" +from __future__ import annotations + +import hashlib +import json +import logging +from datetime import datetime, timedelta, timezone +from typing import Any + +import httpx + +from app.core.config import settings + +logger = logging.getLogger(__name__) + +_BEIJING = timezone(timedelta(hours=8)) + + +class JdUnionError(RuntimeError): + """京东联盟 API 调用失败。""" + + +def _parse_json_maybe(value: Any) -> Any: + if not isinstance(value, str): + return value + text = value.strip() + if not text: + return value + try: + return json.loads(text) + except json.JSONDecodeError: + return value + + +def _sign(params: dict[str, Any], secret: str) -> str: + pieces = [secret] + for key in sorted(k for k in params if k != "sign"): + value = params[key] + if value is None: + continue + pieces.append(f"{key}{value}") + pieces.append(secret) + raw = "".join(pieces) + return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() + + +def _unwrap_response(data: dict[str, Any]) -> dict[str, Any]: + if "error_response" in data: + err = data["error_response"] or {} + msg = err.get("zh_desc") or err.get("en_desc") or err.get("msg") or err + raise JdUnionError(f"京东 API 错误: {msg}") + + body: Any = data + for key, value in data.items(): + if key.endswith("_responce") or key.endswith("_response"): + body = value + break + + body = _parse_json_maybe(body) + if not isinstance(body, dict): + raise JdUnionError("京东 API 返回格式异常") + + result = body.get("queryResult", body.get("result", body)) + result = _parse_json_maybe(result) + if not isinstance(result, dict): + raise JdUnionError("京东 API 业务结果格式异常") + + code = str(result.get("code", result.get("resultCode", "200"))) + if code not in {"0", "200"}: + msg = result.get("message") or result.get("msg") or result.get("resultMsg") or result + raise JdUnionError(f"京东 API 业务错误: {msg}") + return result + + +def call(method: str, payload: dict[str, Any], *, version: str = "1.0") -> dict[str, Any]: + if not settings.jd_union_configured: + raise JdUnionError("京东联盟凭证未配置") + + biz_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + params: dict[str, Any] = { + "method": method, + "app_key": settings.JD_UNION_APP_KEY, + "timestamp": datetime.now(_BEIJING).strftime("%Y-%m-%d %H:%M:%S"), + "format": "json", + "v": version, + "sign_method": "md5", + "360buy_param_json": biz_json, + } + params["sign"] = _sign(params, settings.JD_UNION_APP_SECRET) + + try: + with httpx.Client(timeout=settings.JD_UNION_TIMEOUT_SEC, trust_env=False) as client: + resp = client.post(settings.JD_UNION_GATEWAY, data=params) + resp.raise_for_status() + data = resp.json() + except httpx.HTTPError as e: + raise JdUnionError(f"京东 API 网络错误: {e}") from e + except json.JSONDecodeError as e: + raise JdUnionError("京东 API 返回非 JSON") from e + + return _unwrap_response(data) + + +def _extract_rows(result: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]: + payload = _parse_json_maybe(result.get("data", result.get("result", result))) + has_more = bool(result.get("hasMore") or result.get("has_more")) + + if isinstance(payload, dict): + for key in ("orderRowResp", "orderRows", "orderList", "orders", "list", "rows"): + rows = _parse_json_maybe(payload.get(key)) + if isinstance(rows, list): + return [r for r in rows if isinstance(r, dict)], bool( + payload.get("hasMore") or payload.get("has_more") or has_more + ) + return [], bool(payload.get("hasMore") or payload.get("has_more") or has_more) + if isinstance(payload, list): + return [r for r in payload if isinstance(r, dict)], has_more + return [], has_more + + +def query_order_rows( + *, + start_time: datetime, + end_time: datetime, + query_time_type: int = 3, + page_index: int = 1, + page_size: int = 200, +) -> dict[str, Any]: + """查询京东 CPS 订单行。 + + query_time_type: 1 下单时间, 2 完成时间, 3 更新时间。 + start_time/end_time 用北京时间展示给京东;调用方需保证窗口不超过 1 小时。 + """ + start_bj = start_time.astimezone(_BEIJING) + end_bj = end_time.astimezone(_BEIJING) + if end_bj <= start_bj: + return {"rows": [], "has_more": False} + if end_bj - start_bj > timedelta(hours=1): + raise JdUnionError("京东订单查询单次时间窗不能超过 1 小时") + + order_req: dict[str, Any] = { + "pageIndex": page_index, + "pageSize": min(max(page_size, 1), 200), + "type": query_time_type, + "startTime": start_bj.strftime("%Y-%m-%d %H:%M:%S"), + "endTime": end_bj.strftime("%Y-%m-%d %H:%M:%S"), + } + if settings.JD_UNION_AUTH_KEY: + order_req["key"] = settings.JD_UNION_AUTH_KEY + result = call("jd.union.open.order.row.query", {"orderReq": order_req}) + rows, has_more = _extract_rows(result) + logger.info( + "jd.union.open.order.row.query fetched rows=%s page=%s has_more=%s", + len(rows), + page_index, + has_more, + ) + return {"rows": rows, "has_more": has_more, "raw": result} diff --git a/app/main.py b/app/main.py index 6a2e044..9ac9e27 100644 --- a/app/main.py +++ b/app/main.py @@ -2,18 +2,23 @@ 通过 `uvicorn app.main:app --reload` 启动。 """ + from __future__ import annotations import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager - from pathlib import Path from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles +from app.api.internal.app_version import router as internal_app_version_router +from app.api.internal.launch_confirm import router as internal_launch_confirm_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.ad import router as ad_router from app.api.v1.analytics import router as analytics_router from app.api.v1.auth import router as auth_router @@ -21,12 +26,8 @@ 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.device import router as device_router from app.api.v1.cps_redirect import router as cps_redirect_router -from app.api.internal.app_version import router as internal_app_version_router -from app.api.internal.launch_confirm import router as internal_launch_confirm_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.device import router as device_router from app.api.v1.feedback import router as feedback_router from app.api.v1.invite import router as invite_router from app.api.v1.meituan import router as meituan_router @@ -40,14 +41,14 @@ from app.api.v1.user import router as user_router from app.api.v1.wallet import router as wallet_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings -from app.core.heartbeat_monitor_worker import ( - start_heartbeat_monitor, - stop_heartbeat_monitor, -) from app.core.daily_exchange_worker import ( start_daily_exchange_worker, stop_daily_exchange_worker, ) +from app.core.heartbeat_monitor_worker import ( + start_heartbeat_monitor, + stop_heartbeat_monitor, +) from app.core.logging import setup_logging from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client from app.core.withdraw_reconcile_worker import ( @@ -137,6 +138,30 @@ app.include_router(cps_redirect_router) # 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。 _media_root = Path(settings.MEDIA_ROOT) _media_root.mkdir(parents=True, exist_ok=True) + + +# 官网下载的 APK 直链(落地页 dl.html「官网下载」按钮指向 /media/shaguabijia.apk)。 +# 必须在 StaticFiles 挂载【之前】注册,否则被静态挂载吃掉。 +# StaticFiles 给 .apk 的 Content-Type 不对、且无 attachment 头 → 部分国产浏览器不触发下载、转甩应用市场; +# 这里显式回 application/vnd.android.package-archive + Content-Disposition:attachment 强制浏览器下载。 +# 文件由 scripts/publish_apk.sh 编 release 包后放到 data/media/shaguabijia.apk(*.apk 不入 git,需部署时放)。 +_APK_PATH = _media_root / "shaguabijia.apk" + + +@app.get(f"{settings.MEDIA_URL_PREFIX}/shaguabijia.apk", tags=["meta"], include_in_schema=False) +def download_apk() -> FileResponse: + if not _APK_PATH.is_file(): + from fastapi import HTTPException + + raise HTTPException(status_code=404, detail="安装包未就绪") + return FileResponse( + _APK_PATH, + media_type="application/vnd.android.package-archive", + filename="shaguabijia.apk", + headers={"Content-Disposition": 'attachment; filename="shaguabijia.apk"'}, + ) + + app.mount( settings.MEDIA_URL_PREFIX, StaticFiles(directory=str(_media_root)), diff --git a/app/models/__init__.py b/app/models/__init__.py index 7f61400..9e5446c 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -19,6 +19,7 @@ from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, CouponDailyCompletion, CouponPromptEngagement, + CouponSession, ) from app.models.feedback import Feedback # noqa: F401 from app.models.invite import InviteRelation # noqa: F401 diff --git a/app/models/coupon_state.py b/app/models/coupon_state.py index 3f9364d..c7ba0b9 100644 --- a/app/models/coupon_state.py +++ b/app/models/coupon_state.py @@ -181,3 +181,79 @@ class CouponPromptEngagement(Base): f"" ) + + +class CouponSession(Base): + """一次领券任务的全程流水(admin「领券数据」看板数据源,2026-06-30)。 + + 与 coupon_claim_record(按券一天一条去重)、coupon_daily_completion(按设备一天一条)都不同: + 本表**一次领券一条**(trace_id 唯一),记从发起(started)到收尾(completed/failed/abandoned)的 + 全程耗时 + 各平台耗时 + 机型/ROM。客户端 POST /api/v1/coupon/session 两段上报:发起建行、 + 收尾按 trace_id 更新同一行。发起即落库 → admin 可算「发起数」与中途流失(started 无终态=未完成)。 + + 口径:elapsed_ms 由客户端全程计时(点发起→收尾)、权威;started_at/finished_at 为时刻留痕。 + started_date = started_at 的 Asia/Shanghai 自然日,供 admin 按天聚合 / 日期筛选(索引)。 + """ + + __tablename__ = "coupon_session" + __table_args__ = ( + # 一次领券一行:trace_id 幂等 upsert(发起建、收尾更新同一行)。 + UniqueConstraint("trace_id", name="uq_coupon_session_trace"), + # admin 主聚合/筛选:按上海自然日 + 环境(报表默认只看 prod,避免测试数据串台)。 + Index("ix_coupon_session_date_env", "started_date", "app_env"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # 一次领券唯一 id(客户端 UUID,全程贯穿),upsert 键。 + trace_id: Mapped[str] = mapped_column(String(64), nullable=False) + device_id: Mapped[str] = mapped_column(String(64), nullable=False) + # 登录态才带(admin join 用户表出手机号/昵称);匿名领券为空。 + user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + + # started / completed / failed / abandoned。started 无终态 = 中途流失(未完成)。 + status: Mapped[str] = mapped_column(String(16), nullable=False) + # prod / dev(客户端 BuildConfig.DEBUG)。admin 报表默认只看 prod(对齐广告报表防串台口径)。 + app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True) + + # 发起勾选平台 ["meituan-waimai", ...](空=全领)。 + platforms: Mapped[list | None] = mapped_column(_JSON, nullable=True) + # 发起来源外卖 App 包名;null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。 + # admin「发起平台」列据此区分(空→傻瓜比价,包名→对应平台)。 + origin_package: Mapped[str | None] = mapped_column(String(64), nullable=True) + # 机型(Build.MANUFACTURER + MODEL)与 ROM(OemDetector,如 "ColorOS 14")。明细「机型/ROM」列 + 维度筛选。 + device_model: Mapped[str | None] = mapped_column(String(128), nullable=True) + rom: Mapped[str | None] = mapped_column(String(64), nullable=True) + + # 发起时刻(客户端墙钟):明细「时间」列、趋势 X 轴。 + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + # 发起的 Asia/Shanghai 自然日:按天聚合 / 日期范围筛选(索引)。 + started_date: Mapped[date] = mapped_column(Date, nullable=False) + # 收尾时刻(服务端 now);未收尾(流失)为空。 + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + # 全程耗时(ms,客户端点发起→收尾):平均 / 分位都基于它(只统计 completed)。 + elapsed_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + # 各平台领券耗时 {"meituan-waimai":3200,...}(ms)。明细美团/淘宝/京东耗时列。 + platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True) + # 领到总张数(收尾帧带)。 + claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + # pricebot done 帧回传的公网调试链接(price.shaguabijia.com/traces/{dir});含落盘时分秒、拼不出,只能存 + # (同 ComparisonRecord.trace_url)。admin「领券数据」明细据此渲染可点 trace 链接;未到 done(failed/abandoned)为空。 + trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True) + + created_at: 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"" + ) diff --git a/app/models/cps_order.py b/app/models/cps_order.py index 306b30d..e6732b8 100644 --- a/app/models/cps_order.py +++ b/app/models/cps_order.py @@ -1,11 +1,13 @@ """CPS 对账订单(cps_order)。 -从美团联盟 query_order 按时间窗拉回、按 sid 归群的订单明细。字段对齐 query_order +从联盟 API 按时间窗拉回、按平台落库的 CPS 订单明细。字段最初对齐美团 query_order, +后续兼容京东订单报表: 实测返回: - payPrice / profit 是「元」字符串 → 入库统一转「分」(与全站口径一致) - payTime / updateTime 是秒级时间戳 → 入库转 tz-aware datetime - status: 2付款 3完成 4取消 5风控 6结算(取消/风控不计佣金) -order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。 +order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。京东订单用 +`jd:` 前缀避免与美团订单号碰撞。 """ from __future__ import annotations @@ -22,8 +24,13 @@ class CpsOrder(Base): __tablename__ = "cps_order" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - # 美团订单号(加密串),全局唯一,upsert 幂等键。 + # 平台:meituan / jd。历史数据迁移默认 meituan。 + platform: Mapped[str] = mapped_column(String(20), default="meituan", index=True, nullable=False) + # 平台订单号/行号包装后的全局唯一键,upsert 幂等。 order_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + # 平台原始订单号/行号。京东一笔订单多 SKU 时可按行号区分。 + external_order_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True) + external_row_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True) # 渠道追踪位 = 群 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) @@ -35,11 +42,21 @@ class CpsOrder(Base): 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) + # 通用佣金拆分。美团只有预估 profit;京东有预估/实际佣金。 + estimated_commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True) + actual_commission_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) + # 京东订单有效码(validCode),用于判断是否有效/已完成。 + jd_valid_code: Mapped[str | None] = mapped_column(String(16), 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) + settle_month: Mapped[str | None] = mapped_column(String(16), nullable=True) + site_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + position_id: Mapped[str | None] = mapped_column(String(128), nullable=True) + pid: Mapped[str | None] = mapped_column(String(128), nullable=True) + sub_union_id: Mapped[str | None] = mapped_column(String(128), nullable=True) pay_time: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), index=True, nullable=True @@ -59,5 +76,6 @@ class CpsOrder(Base): def __repr__(self) -> str: # pragma: no cover return ( f"" + f"platform={self.platform!r} sid={self.sid!r} " + f"status={self.mt_status or self.jd_valid_code} profit_cents={self.commission_cents}>" ) diff --git a/app/repositories/coupon_state.py b/app/repositories/coupon_state.py index e3e2765..336ca0c 100644 --- a/app/repositories/coupon_state.py +++ b/app/repositories/coupon_state.py @@ -6,7 +6,7 @@ from __future__ import annotations import logging -from datetime import date, datetime +from datetime import date, datetime, timezone from zoneinfo import ZoneInfo from sqlalchemy import delete, func, select @@ -17,6 +17,7 @@ from app.models.coupon_state import ( CouponClaimRecord, CouponDailyCompletion, CouponPromptEngagement, + CouponSession, ) logger = logging.getLogger("shagua.coupon_state") @@ -232,3 +233,91 @@ def sum_claimed_count(db: Session, user_id: int) -> int: ) ).scalar_one() return int(total or 0) + + +# ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)===== + +def upsert_coupon_session( + db: Session, + *, + trace_id: str, + device_id: str, + status: str, + started_at_ms: int, + user_id: int | None = None, + platforms: list[str] | None = None, + origin_package: str | None = None, + device_model: str | None = None, + rom: str | None = None, + app_env: str | None = None, + elapsed_ms: int | None = None, + platform_elapsed: dict | None = None, + claimed_count: int | None = None, + trace_url: str | None = None, +) -> None: + """一条领券流水按 trace_id 幂等 upsert(发起 started 建行、收尾终态更新同一行)。 + + - 乱序/重复兜底:终态(completed/failed/abandoned)先到也建行;started 重复到不覆盖已有终态 + (状态只前进,不降级)。 + - started_at 由客户端墙钟毫秒转;started_date 取其 Asia/Shanghai 自然日(admin 按天聚合/筛选)。 + - 终态帧补 finished_at=服务端 now;各字段非空才写(避免 started 帧的 None 抹掉收尾值,反之亦然)。 + 并发 IntegrityError 回滚忽略(本就幂等)。 + """ + started_at = datetime.fromtimestamp(started_at_ms / 1000, tz=timezone.utc) + started_date = started_at.astimezone(_CN_TZ).date() + is_terminal = status in ("completed", "failed", "abandoned") + + row = db.execute( + select(CouponSession).where(CouponSession.trace_id == trace_id) + ).scalar_one_or_none() + + if row is None: + db.add(CouponSession( + trace_id=trace_id, + device_id=device_id, + user_id=user_id, + status=status, + app_env=app_env, + platforms=platforms, + origin_package=origin_package, + device_model=device_model, + rom=rom, + started_at=started_at, + started_date=started_date, + finished_at=datetime.now(timezone.utc) if is_terminal else None, + elapsed_ms=elapsed_ms, + platform_elapsed=platform_elapsed, + claimed_count=claimed_count, + trace_url=trace_url, + )) + else: + # 状态只前进:started 帧重复到(如 START_STICKY 重启)不把已有终态降级回 started。 + if not (status == "started" and row.status in ("completed", "failed", "abandoned")): + row.status = status + if is_terminal: + row.finished_at = datetime.now(timezone.utc) + if user_id is not None: + row.user_id = user_id + if platforms is not None: + row.platforms = platforms + if origin_package is not None: + row.origin_package = origin_package + if device_model is not None: + row.device_model = device_model + if rom is not None: + row.rom = rom + if app_env is not None: + row.app_env = app_env + if elapsed_ms is not None: + row.elapsed_ms = elapsed_ms + if platform_elapsed is not None: + row.platform_elapsed = platform_elapsed + if claimed_count is not None: + row.claimed_count = claimed_count + if trace_url is not None: + row.trace_url = trace_url + try: + db.commit() + except IntegrityError: + # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。 + db.rollback() diff --git a/app/repositories/invite.py b/app/repositories/invite.py index 7fd7dbc..8dbd290 100644 --- a/app/repositories/invite.py +++ b/app/repositories/invite.py @@ -216,13 +216,21 @@ def try_reward_on_compare(db: Session, invitee_user_id: int) -> CompareRewardRes def get_stats(db: Session, inviter_id: int) -> tuple[int, int]: """返回 (已成功邀请人数, 累计从邀请获得的金币)。 + "已邀请好友数"口径 = 完成一次比价(已触发邀请奖励金)的被邀请人数,即 compare_reward_granted=True。 + 不再数"仅绑定未比价"的关系——否则会出现"已邀请 1、可提现余额 0"(好友下载登录但没比价), + 与产品口径"已邀请好友数 × 2元 = 累计提现 + 可提现余额"对不上。过滤后该恒等式天然成立 + (每个计入的好友都恰好发过 1 笔 2 元,钱要么在余额要么已提现)。 + 金币口径(inviter_coin 之和)自 v3 起恒 0(邀请人收益改走邀请奖励金,见 get_reward_stats / try_reward_on_compare);保留返回位兼容旧响应字段 coins_earned。 """ count = db.execute( select(func.count()) .select_from(InviteRelation) - .where(InviteRelation.inviter_user_id == inviter_id) + .where( + InviteRelation.inviter_user_id == inviter_id, + InviteRelation.compare_reward_granted.is_(True), + ) ).scalar_one() coins = db.execute( select(func.coalesce(func.sum(InviteRelation.inviter_coin), 0)) diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 611bbc9..1edd034 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -610,6 +610,7 @@ def create_withdraw( source: str = "coin_cash", user_name: str | None = None, out_bill_no: str | None = None, + allow_sub_min: bool = False, ) -> WithdrawOrder: """发起提现:原子扣款 + 建单 reviewing(待人工审核),**不打款**。 @@ -619,8 +620,13 @@ def create_withdraw( 重复发起多笔提现(审核拒绝再退回)。 #2 out_bill_no 客户端幂等键:同号重试返回该单现状(reviewing 等审核),不重复扣款建单。 实名 user_name 在此存下(WithdrawOrder.user_name),供异步审核打款时传给微信(达额需实名)。 + + allow_sub_min:放行低于"提现最低额"的小额(用于 0.01 元调试提现)。仅由 endpoint 在 + `skip_review and not is_prod`(debug 包 + 非生产双闸)时置 True;仍受 schema gt=0 与 max 上限约束。 + 生产恒为 False → 最低额校验照常,绝不可能提 0.01。 """ - if amount_cents < rewards.get_withdraw_min_cents(db) or amount_cents > rewards.get_withdraw_max_cents(db): + min_c = 0 if allow_sub_min else rewards.get_withdraw_min_cents(db) + if amount_cents < min_c or amount_cents > rewards.get_withdraw_max_cents(db): raise InvalidWithdrawAmountError # 提现即要求已绑微信:否则审核通过也打不了款,提前拦更友好 diff --git a/app/schemas/coupon_state.py b/app/schemas/coupon_state.py index 236d311..35d2cca 100644 --- a/app/schemas/coupon_state.py +++ b/app/schemas/coupon_state.py @@ -49,3 +49,28 @@ class CouponStatsOut(BaseModel): """ coupon_count: int + + +class CouponSessionIn(BaseModel): + """客户端领券流水上报体(admin「领券数据」看板数据源,POST /api/v1/coupon/session)。 + + 一次领券两段上报,按 trace_id upsert 到 coupon_session: + - 发起(status=started):带勾选平台 + 机型/ROM/app_env + started_at_ms(发起墙钟毫秒)。 + - 收尾(completed/failed/abandoned):带 elapsed_ms(全程耗时)+ platform_elapsed(各平台耗时)+ claimed_count。 + 不鉴权(同领券循环 MVP,按 device_id/trace_id),user_id 登录态带上做留痕(可空)。 + """ + + trace_id: str + device_id: str + status: str # started / completed / failed / abandoned + started_at_ms: int # 发起墙钟毫秒(客户端 System.currentTimeMillis) + user_id: int | None = None + platforms: list[str] | None = None + origin_package: str | None = None + device_model: str | None = None + rom: str | None = None + app_env: str | None = None + elapsed_ms: int | None = None + platform_elapsed: dict[str, int] | None = None + claimed_count: int | None = None + trace_url: str | None = None diff --git a/data/media/coupon-page-bg.png b/data/media/coupon-page-bg.png new file mode 100644 index 0000000..19113f6 Binary files /dev/null and b/data/media/coupon-page-bg.png differ diff --git a/data/media/dl.html b/data/media/dl.html index f580014..9c84b51 100644 --- a/data/media/dl.html +++ b/data/media/dl.html @@ -198,7 +198,7 @@ 选择「在浏览器打开」 在浏览器里按提示去应用商店下载 -
我知道了 ✕
+
@@ -263,7 +263,7 @@ var wxGuide = document.getElementById("wxGuide"); function showWxGuide() { wxGuide.classList.add("show"); } function hideWxGuide() { wxGuide.classList.remove("show"); } - document.getElementById("wxGuideDismiss").addEventListener("click", hideWxGuide); + // document.getElementById("wxGuideDismiss").addEventListener("click", hideWxGuide); // 「我知道了 ✕」已注释隐藏 if (isWeChat) showWxGuide(); // 微信里一进页面就提示去浏览器（微信内下载必被拦） function showToast(text) { @@ -273,15 +273,24 @@ showToast.timer = setTimeout(function () { toast.classList.remove("show"); }, 1400); } - // ===== 下载按钮：微信内引导去浏览器；iOS 提示；安卓写邀请码 + 跳应用商店 ===== + // ===== 应用商店下载：微信内引导去浏览器；iOS 提示；安卓写邀请码 + 跳应用商店 ===== function handleDownload() { if (isWeChat) { showWxGuide(); return; } if (isIOS) { alert("iOS 版即将上线，请前往 App Store 搜索「傻瓜比价」"); return; } copyInviteCode(); // 先把邀请码写进剪贴板（供 App 首启归因），链路丢了还有 landing-track 指纹兜底 openStore(); } + // ===== 官网下载：微信/iOS 同上引导；安卓直接跳 APK 直链 → 弹系统下载弹窗 ===== + // APK_URL 跟随页面 host：本地走 LAN、生产走 app-api.shaguabijia.com（见邀请功能文档约定）。 + var APK_URL = location.origin + "/media/shaguabijia.apk"; + function handleWebsiteDownload() { + if (isWeChat) { showWxGuide(); return; } + if (isIOS) { alert("iOS 版即将上线，请前往 App Store 搜索「傻瓜比价」"); return; } + copyInviteCode(); // 同样先写邀请码进剪贴板（供 App 首启归因） + window.location.href = APK_URL; + } document.getElementById("dlbtn").addEventListener("click", handleDownload); - document.getElementById("dlbtn2").addEventListener("click", handleDownload); + document.getElementById("dlbtn2").addEventListener("click", handleWebsiteDownload); diff --git a/data/media/sb-brand.png b/data/media/sb-brand.png new file mode 100644 index 0000000..c8f1c13 Binary files /dev/null and b/data/media/sb-brand.png differ diff --git a/deploy/pangle-revenue.md b/deploy/pangle-revenue.md new file mode 100644 index 0000000..f8ff2ba --- /dev/null +++ b/deploy/pangle-revenue.md @@ -0,0 +1,67 @@ +# 穿山甲 GroMore 收益拉取 定时任务 — 运维手册 + +> 对象:维护「每天拉穿山甲后台收益入库」这套定时任务的同事。 +> 🔒 服务器登录信息见**私密交接清单**,不入库。 + +## 它是什么 +admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查穿山甲**。穿山甲只通过 GroMore 数据 API 给数、且 **T+1**(次日约 10:00 出昨天的数),所以每天得拉一次入库,报表才会往前走。 + +- 每天 10:30 跑一轮 `scripts/sync_pangle_revenue.py`,默认 `--days 3` 回补近 3 天。 +- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。 +- **幂等 upsert**:同一(日期×应用×代码位)重跑只覆盖、不重复,故回补 / 重跑 / catch-up 都安全。 +- 穿山甲无用户/设备维度 → 只能落「汇总/趋势级」,报表带 user_id 过滤时这块收益置空(显示「-」)。 + +## 文件 +| 项 | 路径 | +|---|---| +| 脚本入口 | `scripts/sync_pangle_revenue.py` | +| 拉取 / 签名 | `app/integrations/pangle_report.py`(签名=参数字典序拼接+secure_key 后 MD5) | +| 入库表 | `ad_pangle_daily_revenue`(读写在 `app/repositories/ad_pangle_revenue.py`) | +| 凭证 | `.env` 的 `PANGLE_REPORT_USER_ID` / `PANGLE_REPORT_ROLE_ID` / `PANGLE_REPORT_SECURITY_KEY` | +| 应用映射 | `.env` 的 `PANGLE_REPORT_SITE_ID_PROD`(5830519)/ `PANGLE_REPORT_SITE_ID_TEST`(5832303) | +| systemd 单元 | `deploy/pangle-revenue.{service,timer}` | + +## 上线前置(只做一次) +1. **填凭证**:后台「接入中心 → GroMore-API」领 user_id / role_id / Security Key(`secure_key`,**≠ 发奖 m-key**),填进线上 `.env` 的 `PANGLE_REPORT_*`。三项填齐脚本才工作,缺任一 → no-op 退出。 +2. **子账号要授权**:线上若用子账号(role_id ≠ user_id),需主账号给它授「查看全部数据」,否则接口报 **118**(无权限)。本人自查自己(user_id=role_id)无此问题。 +3. 凭证泄露:后台「重置 key 值」即可,改完同步线上 `.env` 重跑。 + +## 部署(Linux 服务器,需 root) +```bash +sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/ +sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer +systemctl list-timers pangle-revenue.timer # 确认下次触发时间(应是次日 10:30) +``` + +## 怎么看健康 / 手动跑一次 +```bash +sudo systemctl start pangle-revenue.service # 立即手动跑一轮(不等 10:30) +journalctl -u pangle-revenue -n 30 --no-pager # 看日志:拉取区间 / 入库行数 / 新增更新 / 预估收益合计 +``` +成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;预估收益合计 ¥19.42`。 +> 看不到收益、提示 `PANGLE_REPORT_* 未配置`→ 回「上线前置」补 `.env`;报 118 → 子账号没授「查看全部数据」。 + +## 本机 Windows 开发(无 systemd) +直接手动跑: +``` +.venv\Scripts\python -m scripts.sync_pangle_revenue # 拉昨天 +.venv\Scripts\python -m scripts.sync_pangle_revenue --days 3 # 回补近 3 天 +.venv\Scripts\python -m scripts.sync_pangle_revenue --date 2026-06-27 # 指定单天 +.venv\Scripts\python -m scripts.sync_pangle_revenue --start 2026-06-01 --end 2026-06-27 # 区间回补 +``` + +## 脚本参数 +- 无参:拉**昨天**(北京时间)。timer 用的是 `--days 3`。 +- `--days N`:从昨天起往前回补 N 天(含昨天)。 +- `--date YYYY-MM-DD`:指定单天。 +- `--start / --end`:指定闭区间(跨度 ≤ 31 天,接口上限 1 个月,超了报 114)。 + +## 注意事项 +- **触发时间**:`OnCalendar=*-*-* 10:30:00`。穿山甲 ~10:00 出数,故别早于 10:00 跑(会拉到空/不全)。 +- **catch-up**:`Persistent=true` 补跑错过的那一轮;叠加 `--days 3`,漏一两天重新触发即自愈。 +- **今天 / 今天以前要分开查**:脚本默认只拉昨天及更早,不混查今天(接口约束),无需关心。 +- **join key 是 `ad_unit_id`(我们配的 104xxx)不是 `code_id`**:`code_id` 是底层各 ADN 代码位,对不上口径;`ad_unit_id='-1'` 是未归因桶。改维度时务必注意(详见脚本头注释)。 +- **`api_revenue` 很稀疏**:测试应用 ADN 没配 Reporting → 全 0,仅 prod 个别位有;`revenue`(预估)才是稳的主力。 +- **DB 无关**:sqlite / postgres 均可(upsert 逐行 select-then-write,不像美团 ETL 需要 PG)。 +- **别和别的触发方式双跑**:本 systemd timer 与「手动 cron / 进程内任务」二选一,虽幂等不会重复入库,纯属多余。 +- **改脚本 / 改部署**:走 git + PR,由有 root 的人部署。 diff --git a/deploy/pangle-revenue.service b/deploy/pangle-revenue.service new file mode 100644 index 0000000..dc2e2b0 --- /dev/null +++ b/deploy/pangle-revenue.service @@ -0,0 +1,38 @@ +# 每天拉穿山甲 GroMore T+1 天级收益入库 —— 单轮跑,由 pangle-revenue.timer 每天 10:30 触发。 +# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。 +# +# 仅用于 Linux 服务器;本机 Windows 开发无 systemd,直接手动跑脚本即可: +# .venv\Scripts\python -m scripts.sync_pangle_revenue # 拉昨天(北京时间) +# .venv\Scripts\python -m scripts.sync_pangle_revenue --days 3 # 回补近 3 天 +# +# 部署(服务器): +# sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/ +# sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer +# # 手动跑一次验证: sudo systemctl start pangle-revenue.service && journalctl -u pangle-revenue -n 30 +# +# 前置:.env 配好 PANGLE_REPORT_USER_ID / PANGLE_REPORT_ROLE_ID / PANGLE_REPORT_SECURITY_KEY +# (后台「接入中心 → GroMore-API」领;子账号需主账号授「查看全部数据」否则接口 118)。 +# 未配齐这三项 → 脚本自动 no-op 退出,免动 timer。 +[Unit] +Description=Sync Pangle GroMore daily revenue (T+1, one-shot, driven by timer) +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=root +WorkingDirectory=/opt/shaguabijia-app-server +Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin" +EnvironmentFile=/opt/shaguabijia-app-server/.env +# 默认拉昨天;--days 3 回补近 3 天(幂等 upsert,应对偶发漏跑 + 穿山甲对历史数据订正,重跑无害)。 +ExecStart=/opt/shaguabijia-app-server/.venv/bin/python -m scripts.sync_pangle_revenue --days 3 +SyslogIdentifier=pangle-revenue +# 仅几个 HTTP 请求 + 小批量入库,通常数秒;给 10min 硬超时防穿山甲接口卡死。 +TimeoutStartSec=600 + +# 与主服务 shaguabijia-app-server.service 同款加固。 +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ReadWritePaths=/opt/shaguabijia-app-server +ProtectHome=true diff --git a/deploy/pangle-revenue.timer b/deploy/pangle-revenue.timer new file mode 100644 index 0000000..404446e --- /dev/null +++ b/deploy/pangle-revenue.timer @@ -0,0 +1,14 @@ +# 每天 10:30 触发一次穿山甲 GroMore T+1 收益拉取入库(Linux 服务器用)。 +# 见 pangle-revenue.service 顶部注释的部署步骤。 +[Unit] +Description=Run Pangle GroMore daily revenue sync at 10:30 + +[Timer] +# 穿山甲 T+1、次日约 10:00 出数;10:30 触发留 30min 余量。要错开整点扎堆可微调到 10:35。 +OnCalendar=*-*-* 10:30:00 +# 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日);叠加 --days 3 回补,漏一两天能自愈。 +Persistent=true +AccuracySec=1min + +[Install] +WantedBy=timers.target diff --git a/h5/mine/index.html b/h5/mine/index.html index 5eafa9d..47ee637 100644 --- a/h5/mine/index.html +++ b/h5/mine/index.html @@ -9,6 +9,17 @@ 我的 · 傻瓜比价 + +