From a3a14b484a45caad9ca1c32f17ff2faf8d3aaf92 Mon Sep 17 00:00:00 2001 From: chenshuobo Date: Tue, 30 Jun 2026 21:56:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8E=A5=E5=85=A5=E4=BA=AC=E4=B8=9C?= =?UTF-8?q?=E8=81=94=E7=9B=9F=E8=AE=A2=E5=8D=95=E5=88=B0=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=A4=A7=E7=9B=98=20(#90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 改了什么：新增京东联盟订单查询客户端、cps_order 京东字段迁移、CPS 对账接口 platform=all/jd、数据大盘京东订单与佣金聚合。 为什么改：数据大盘需要展示京东 CPS 真实订单佣金，不再保持占位。 验证方式：python -m pytest tests\\test_admin_read.py::test_dashboard_overview tests\\test_cps_admin.py::test_jd_reconcile_updates_dashboard -q；本地真实拉取 2026-06-27 京东订单 3 行，页面显示京东有效 1 单/佣金 ¥0.13。 --------- Co-authored-by: lowmaster-chen <1119780489@qq.com> Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/90 Co-authored-by: chenshuobo Co-committed-by: chenshuobo --- .env.example | 7 + alembic/versions/jd_cps_order_fields.py | 56 ++++++++ app/admin/repositories/cps.py | 167 +++++++++++++++++++++++- app/admin/repositories/stats.py | 109 +++++++++++++--- app/admin/routers/cps.py | 64 ++++++--- app/admin/schemas/cps.py | 11 +- app/admin/schemas/dashboard.py | 5 + app/core/config.py | 15 +++ app/integrations/jd_union.py | 163 +++++++++++++++++++++++ app/models/cps_order.py | 26 +++- tests/test_admin_read.py | 1 + tests/test_cps_admin.py | 71 ++++++++++ 12 files changed, 657 insertions(+), 38 deletions(-) create mode 100644 alembic/versions/jd_cps_order_fields.py create mode 100644 app/integrations/jd_union.py 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/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/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/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/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/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/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/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/tests/test_admin_read.py b/tests/test_admin_read.py index 3d71eae..c9d3781 100644 --- a/tests/test_admin_read.py +++ b/tests/test_admin_read.py @@ -66,6 +66,7 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None: assert "success_rate" in data["comparison"] assert data["cps"]["available"] is True assert "meituan_order_count" in data["cps"] + assert "jd_order_count" in data["cps"] def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None: diff --git a/tests/test_cps_admin.py b/tests/test_cps_admin.py index f3c04ab..cb5ee01 100644 --- a/tests/test_cps_admin.py +++ b/tests/test_cps_admin.py @@ -11,6 +11,7 @@ from app.admin.repositories import admin_user as admin_repo from app.admin.repositories import cps as cps_repo from app.db.session import SessionLocal from app.models.cps_link import CpsClick +from app.models.cps_order import CpsOrder from app.models.cps_wx_user import CpsWxUser from app.repositories import cps_link as cps_link_repo @@ -180,3 +181,73 @@ def test_day_users_cross_year(admin_client: TestClient, admin_token: str) -> Non assert len(users) == 1 assert users[0]["openid"] == openid assert users[0]["visit_count"] == 5 # 次年 01-01 那条被时间窗排除 + + +def test_jd_reconcile_updates_dashboard( + admin_client: TestClient, admin_token: str, monkeypatch: pytest.MonkeyPatch +) -> None: + """京东拉单:按订单行入库;大盘只统计有效 validCode 的佣金。""" + + def fake_query_order_rows(**kwargs): + start = kwargs["start_time"].astimezone(_BJ) + if start.hour != 10 or kwargs["page_index"] != 1: + return {"rows": [], "has_more": False} + return { + "rows": [ + { + "id": "pytest-jd-row-valid", + "orderId": "pytest-jd-order-1", + "skuId": "sku-1", + "skuName": "京东测试商品", + "orderTime": "2026-06-25 10:10:00", + "modifyTime": "2026-06-25 10:20:00", + "validCode": "16", + "estimateCosPrice": "19.90", + "estimateFee": "1.23", + "actualFee": "2.34", + "commissionRate": "10.00", + }, + { + "id": "pytest-jd-row-invalid", + "orderId": "pytest-jd-order-2", + "skuId": "sku-2", + "skuName": "京东无效订单", + "orderTime": "2026-06-25 10:15:00", + "modifyTime": "2026-06-25 10:25:00", + "validCode": "4", + "estimateCosPrice": "9.90", + "estimateFee": "0.50", + "commissionRate": "5.00", + }, + ], + "has_more": False, + } + + monkeypatch.setattr("app.integrations.jd_union.query_order_rows", fake_query_order_rows) + r = admin_client.post( + "/admin/api/cps/orders/reconcile", + params={ + "platform": "jd", + "date_from": "2026-06-25", + "date_to": "2026-06-25", + "query_time_type": 1, + }, + headers=_auth(admin_token), + ) + assert r.status_code == 200, r.text + assert r.json()["fetched"] == 2 + + with SessionLocal() as db: + rows = db.query(CpsOrder).filter(CpsOrder.platform == "jd").all() + assert len([o for o in rows if o.order_id.startswith("jd:pytest-jd-row")]) == 2 + + overview = admin_client.get( + "/admin/api/stats/overview", + params={"date_from": "2026-06-25", "date_to": "2026-06-25"}, + headers=_auth(admin_token), + ) + assert overview.status_code == 200, overview.text + cps = overview.json()["cps"] + assert cps["jd_order_count"] >= 1 + assert cps["jd_commission_cents"] >= 234 + assert cps["jd_invalid_count"] >= 1