diff --git a/.env.example b/.env.example index 53f1e80..4bbb856 100644 --- a/.env.example +++ b/.env.example @@ -18,7 +18,7 @@ JWT_ACCESS_TOKEN_EXPIRE_MINUTES=120 # refresh token 有效期(天),默认 30 天 JWT_REFRESH_TOKEN_EXPIRE_DAYS=30 -# ===== 极光一键登录 ===== +# ===== 极光一键登录 / 短信 ===== # 控制台 → 应用设置 → 应用信息 拿到 JG_APP_KEY= JG_MASTER_SECRET= @@ -27,7 +27,43 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify JG_REQUEST_TIMEOUT_SEC=15 -# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)===== +# ===== 厂商直推(无障碍保护存活告警)===== +# 敏感密钥只放 .env / 服务器环境变量,不要提交到 git。 +ANDROID_PACKAGE_NAME=com.jishisongfu.shaguabijia +PUSH_REQUEST_TIMEOUT_SEC=15 +PUSH_TIME_TO_LIVE_SEC=86400 + +HONOR_PUSH_APP_ID= +HONOR_PUSH_CLIENT_ID= +HONOR_PUSH_CLIENT_SECRET= +HONOR_PUSH_TOKEN_ENDPOINT=https://iam.developer.honor.com/auth/token +HONOR_PUSH_SEND_ENDPOINT_TEMPLATE=https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage + +VIVO_PUSH_APP_ID= +VIVO_PUSH_APP_KEY= +VIVO_PUSH_APP_SECRET= +VIVO_PUSH_AUTH_ENDPOINT=https://api-push.vivo.com.cn/message/auth +VIVO_PUSH_SEND_ENDPOINT=https://api-push.vivo.com.cn/message/send +# vivo 未上架测试时可用 push_mode=1; 上架正式推送改为 0。 +VIVO_PUSH_MODE=1 +VIVO_PUSH_NOTIFY_TYPE=4 +VIVO_PUSH_CATEGORY=DEVICE_REMINDER + +XIAOMI_PUSH_APP_SECRET= +XIAOMI_PUSH_SEND_ENDPOINT=https://api.xmpush.xiaomi.com/v3/message/regid +XIAOMI_PUSH_CHANNEL_ID= +XIAOMI_PUSH_TEMPLATE_ID= +XIAOMI_PUSH_TEMPLATE_TITLE= +XIAOMI_PUSH_TEMPLATE_DESCRIPTION= +# 可选: JSON 字符串，支持 {title}/{alert} 占位符，例如 {"title":"{title}","content":"{alert}"} +XIAOMI_PUSH_TEMPLATE_PARAM_JSON= + +OPPO_PUSH_APP_KEY= +OPPO_PUSH_MASTER_SECRET= +OPPO_PUSH_AUTH_ENDPOINT=https://api.push.oppomobile.com/server/v1/auth +OPPO_PUSH_SEND_ENDPOINT=https://api.push.oppomobile.com/server/v1/message/notification/unicast + +# ===== 无障碍保护存活监控(推送 + pull 后置兜底)===== HEARTBEAT_MONITOR_ENABLED=true HEARTBEAT_TIMEOUT_MINUTES=10 HEARTBEAT_SCAN_INTERVAL_SEC=60 @@ -58,6 +94,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)。 @@ -82,8 +125,9 @@ INTERNAL_API_SECRET= # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类 CORS_ALLOW_ORIGINS= -# ===== 微信支付(商家转账到零钱 / 提现)===== -# appid/secret 来自微信开放平台移动应用;mch/序列号/公钥ID 来自微信支付商户平台。 +# ===== 微信开放平台 App + 微信支付(登录 / 商家转账到零钱 / 提现)===== +# WECHAT_APP_ID/WECHAT_APP_SECRET 来自微信开放平台移动应用,用于 App 微信一键登录(code 换 openid)。 +# mch/序列号/公钥ID 来自微信支付商户平台,仅提现/转账需要。 # 证书 .pem 放 secrets/(已 gitignore)。 WECHAT_APP_ID=wxxxxxxxxxxxxxxxxx WECHAT_APP_SECRET=your_app_secret diff --git a/alembic/versions/direct_vendor_push_fields.py b/alembic/versions/direct_vendor_push_fields.py new file mode 100644 index 0000000..ec33ed2 --- /dev/null +++ b/alembic/versions/direct_vendor_push_fields.py @@ -0,0 +1,30 @@ +"""add direct vendor push fields + +Revision ID: direct_vendor_push_fields +Revises: jd_cps_order_fields +Create Date: 2026-07-01 16:30:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "direct_vendor_push_fields" +down_revision = "jd_cps_order_fields" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("device_liveness") as batch_op: + batch_op.add_column(sa.Column("push_vendor", sa.String(length=32), nullable=True)) + batch_op.add_column(sa.Column("push_token", sa.String(length=256), nullable=True)) + batch_op.create_index("ix_device_liveness_push_vendor", ["push_vendor"]) + + +def downgrade() -> None: + with op.batch_alter_table("device_liveness") as batch_op: + batch_op.drop_index("ix_device_liveness_push_vendor") + batch_op.drop_column("push_token") + batch_op.drop_column("push_vendor") 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/ad_revenue.py b/app/admin/repositories/ad_revenue.py index 5847856..5eff4c1 100644 --- a/app/admin/repositories/ad_revenue.py +++ b/app/admin/repositories/ad_revenue.py @@ -20,7 +20,7 @@ report_date / reward_date 归日。 """ from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta, timezone from datetime import date as _date from sqlalchemy import select @@ -36,7 +36,7 @@ from app.models.user import User def _cn_hour(dt: datetime) -> int: """created_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC 处理(sqlite),tz-aware 直接换算(pg)。""" if dt.tzinfo is None: - dt = dt.replace(tzinfo=UTC) + dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(rewards.CN_TZ).hour diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py index 68b10e0..2aad935 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)) @@ -45,8 +51,58 @@ def _yuan_to_cents(v: object) -> int | None: def _ts_to_dt(ts: object) -> datetime | None: """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。""" - if not ts: + if ts is None: return None + if isinstance(ts, datetime): + return ts if ts.tzinfo else ts.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc) + s = str(ts).strip() + if not s or s.lower() == "null": + return None + try: + seconds = float(Decimal(s)) + except (InvalidOperation, ValueError): + return None + if seconds == 0: + return None + # 美团文档是秒级时间戳，这里顺手兼容毫秒/微秒，避免上游格式变化导致时间再次落空。 + if abs(seconds) > 10_000_000_000_000: + seconds /= 1_000_000 + elif abs(seconds) > 10_000_000_000: + seconds /= 1_000 + try: + return datetime.fromtimestamp(seconds, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + 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 +305,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 +404,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 +426,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..49e810a 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 @@ -23,11 +24,13 @@ from app.models.user import User from app.models.wallet import CoinTransaction, WithdrawOrder _BEIJING = timezone(timedelta(hours=8)) -COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward") +REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward") +COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward") COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward") EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant") UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",) REGULAR_TASK_EXCLUDED_BIZ_TYPES = ( + *REWARD_VIDEO_BIZ_TYPES, *COUPON_REWARD_BIZ_TYPES, *COMPARISON_REWARD_BIZ_TYPES, *EXCLUDED_REWARD_BIZ_TYPES, @@ -35,6 +38,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 +55,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 +118,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 +152,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 +266,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 +303,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 +319,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, @@ -280,13 +341,28 @@ def dashboard_overview( period_reward_video_coin_total = _sum( CoinTransaction.amount, *period_coin_conds, - CoinTransaction.biz_type.in_(("reward_video", "ad_reward")), + CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES), ) period_feed_ad_coin_total = _sum( CoinTransaction.amount, *period_coin_conds, CoinTransaction.biz_type == "feed_ad_reward", ) + period_feed_ad_conds = ( + AdFeedRewardRecord.created_at >= start_utc, + AdFeedRewardRecord.created_at < end_utc, + AdFeedRewardRecord.status == "granted", + ) + period_coupon_feed_coin_total = _sum( + AdFeedRewardRecord.coin, + *period_feed_ad_conds, + AdFeedRewardRecord.feed_scene == "coupon", + ) + period_comparison_feed_coin_total = _sum( + AdFeedRewardRecord.coin, + *period_feed_ad_conds, + AdFeedRewardRecord.feed_scene == "comparison", + ) period_signin_coin_total = _sum( CoinTransaction.amount, *period_coin_conds, @@ -306,18 +382,24 @@ def dashboard_overview( CoinTransaction.amount, *period_coin_conds, CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES), - ) + ) + period_coupon_feed_coin_total period_comparison_reward_coin_total = _sum( CoinTransaction.amount, *period_coin_conds, CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES), - ) + ) + period_comparison_feed_coin_total period_regular_task_coin_total = _sum( CoinTransaction.amount, *period_coin_conds, CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES), ) - period_meituan_orders = list( + period_classified_coin_total = ( + period_coupon_reward_coin_total + + period_comparison_reward_coin_total + + period_reward_video_coin_total + + period_regular_task_coin_total + ) + period_cps_orders = list( db.execute( select(CpsOrder).where( CpsOrder.pay_time >= start_utc, @@ -325,9 +407,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 @@ -346,6 +436,41 @@ def dashboard_overview( else None ) + total_reward_video_coin_total = _sum( + CoinTransaction.amount, + CoinTransaction.amount > 0, + CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES), + ) + total_coupon_reward_coin_total = _sum( + CoinTransaction.amount, + CoinTransaction.amount > 0, + CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES), + ) + _sum( + AdFeedRewardRecord.coin, + AdFeedRewardRecord.status == "granted", + AdFeedRewardRecord.feed_scene == "coupon", + ) + total_comparison_reward_coin_total = _sum( + CoinTransaction.amount, + CoinTransaction.amount > 0, + CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES), + ) + _sum( + AdFeedRewardRecord.coin, + AdFeedRewardRecord.status == "granted", + AdFeedRewardRecord.feed_scene == "comparison", + ) + total_regular_task_coin_total = _sum( + CoinTransaction.amount, + CoinTransaction.amount > 0, + CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES), + ) + total_classified_coin_total = ( + total_coupon_reward_coin_total + + total_comparison_reward_coin_total + + total_reward_video_coin_total + + total_regular_task_coin_total + ) + return { "users": { "total": _count(User), @@ -356,12 +481,12 @@ def dashboard_overview( "dau": today_dau(db), }, "coins": { - # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计) - "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0), + # 累计发放金币按大盘 4 个展示分类加总;未分类/运营手动/邀请奖励不进入本指标。 + "granted_total": total_classified_coin_total, "reward_video_coin_total": _sum( CoinTransaction.amount, CoinTransaction.amount > 0, - CoinTransaction.biz_type.in_(("reward_video", "ad_reward")), + CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES), ), "reward_video_watch_count": _count( AdRewardRecord, @@ -389,6 +514,10 @@ def dashboard_overview( CoinTransaction.biz_type == "signin_boost", ), "signin_boost_watch_count": _count(SigninBoostRecord), + "task_coin_total": total_regular_task_coin_total, + "coupon_reward_coin_total": total_coupon_reward_coin_total, + "comparison_reward_coin_total": total_comparison_reward_coin_total, + "regular_task_coin_total": total_regular_task_coin_total, }, "cash": { "withdraw_success_cents": _sum( @@ -412,8 +541,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": { @@ -425,7 +554,7 @@ def dashboard_overview( "average_saved_cents": period_avg_saved_cents, }, "coins": { - "granted_total": _sum(CoinTransaction.amount, *period_coin_conds), + "granted_total": period_classified_coin_total, "reward_video_coin_total": period_reward_video_coin_total, "feed_ad_coin_total": period_feed_ad_coin_total, "signin_coin_total": period_signin_coin_total, @@ -450,7 +579,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 +588,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/admin/schemas/device.py b/app/admin/schemas/device.py index da3fe60..68568b9 100644 --- a/app/admin/schemas/device.py +++ b/app/admin/schemas/device.py @@ -24,7 +24,9 @@ class DeviceLivenessItem(BaseModel): device_model: str | None = None # 由 device_id 解析(device_<机型>_);非 DB 列 platform: str app_version: str | None = None - registration_id: str | None = None # 非空 = 拿到极光 token、可推送 + registration_id: str | None = None # 旧极光字段,仅兼容历史数据 + push_vendor: str | None = None + push_token: str | None = None ever_protected: bool # 是否开过无障碍(=该设备对功能有意义) first_protected_at: datetime | None = None # 首次开无障碍时刻(老设备为 null) diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index 3c4f4bf..f5abcac 100644 --- a/app/api/v1/auth.py +++ b/app/api/v1/auth.py @@ -2,6 +2,7 @@ 路由前缀 `/api/v1/auth`,包含: POST /jverify-login 极光一键登录(loginToken → 手机号 → 注册即登录 → 签 JWT) + POST /wechat-login 微信 App 授权登录(code → openid → 注册即登录 → 签 JWT) POST /sms/send 发短信验证码(mock 阶段任意 6 位通过) POST /sms/login 手机号 + 验证码登录 POST /refresh 用 refresh_token 换新的 token 对 @@ -20,6 +21,7 @@ from app.core.ratelimit import enforce_rate_limit, rate_limit from app.core.security import TokenError, decode_token, issue_token_pair from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone from app.integrations.sms import SmsError, send_code, verify_code +from app.integrations.wechat_login import WechatLoginError, code_to_userinfo from app.repositories import onboarding as onboarding_repo from app.repositories import user as user_repo from app.schemas.auth import ( @@ -32,6 +34,7 @@ from app.schemas.auth import ( TokenPair, TokenWithUser, UserOut, + WechatLoginRequest, ) logger = logging.getLogger("shagua.auth") @@ -84,6 +87,40 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser: return _login_response(user, onboarding_completed=completed) +# ===================== 微信一键登录 ===================== + +@router.post( + "/wechat-login", + response_model=TokenWithUser, + summary="微信一键登录", + dependencies=[Depends(rate_limit(20, 60, "wechat-login"))], +) +def wechat_login(req: WechatLoginRequest, db: DbSession) -> TokenWithUser: + try: + info = code_to_userinfo(req.code) + except WechatLoginError as e: + logger.warning("wechat_login failed: %s", e) + raise HTTPException(status_code=502, detail=str(e)) from e + + user = user_repo.upsert_user_for_wechat_login( + db, + openid=info["openid"], + nickname=info.get("nickname"), + avatar_url=info.get("avatar_url"), + ) + if user.status != "active": + raise HTTPException(status_code=403, detail="account disabled") + + completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id) + logger.info( + "wechat_login ok user_id=%d openid=%s*** onboarded=%s", + user.id, + info["openid"][:6], + completed, + ) + return _login_response(user, onboarding_completed=completed) + + # ===================== 短信登录 ===================== @router.post( diff --git a/app/api/v1/device.py b/app/api/v1/device.py index 56a5a85..893c6c5 100644 --- a/app/api/v1/device.py +++ b/app/api/v1/device.py @@ -1,19 +1,22 @@ """设备注册 / 心跳 endpoint(无障碍保护存活检测)。 路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。 - POST /register 注册设备 / 更新 registration_id(App 前台、拿到 push token 时调) + POST /register 注册设备 / 更新厂商 push token(App 前台、拿到 push token 时调) POST /heartbeat 上报心跳(无障碍服务存活时周期调,刷新存活) + POST /push-test 开发验收:延迟发送厂商通道测试推送 -后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。 +后端 heartbeat_monitor_worker 据此发现心跳超时的设备并厂商直推告警。 见 spec: spec/accessibility-liveness-push.md。 """ from __future__ import annotations import logging +import time -from fastapi import APIRouter +from fastapi import APIRouter, BackgroundTasks, HTTPException, status from app.api.deps import CurrentUser, DbSession +from app.integrations import vendor_push from app.repositories import device as device_repo from app.schemas.device import ( DeviceOut, @@ -22,6 +25,8 @@ from app.schemas.device import ( LivenessAckRequest, LivenessOut, OkResponse, + PushTestOut, + PushTestRequest, ) logger = logging.getLogger("shagua.device") @@ -29,6 +34,37 @@ logger = logging.getLogger("shagua.device") router = APIRouter(prefix="/api/v1/device", tags=["device"]) +def _send_push_test_after_delay( + push_vendor: str, + push_token: str, + delay_seconds: int, + user_id: int, + device_id: str, +) -> None: + if delay_seconds > 0: + time.sleep(delay_seconds) + try: + vendor_push.send_accessibility_disabled( + push_vendor, + push_token, + title="测试推送", + alert="这是一条厂商通道测试推送。收到它说明 App 被划掉后仍可通过系统通知栏触达。", + ) + logger.info( + "push test sent user_id=%d device_id=%s delay=%ds", + user_id, + device_id, + delay_seconds, + ) + except vendor_push.VendorPushError as e: + logger.warning( + "push test failed user_id=%d device_id=%s error=%s", + user_id, + device_id, + e, + ) + + @router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token") def register_device( req: DeviceRegisterRequest, @@ -40,13 +76,17 @@ def register_device( user_id=user.id, device_id=req.device_id, registration_id=req.registration_id, + push_vendor=req.push_vendor, + push_token=req.push_token, platform=req.platform, app_version=req.app_version, ) logger.info( - "device register user_id=%d device_id=%s reg=%s", + "device register user_id=%d device_id=%s vendor=%s token=%s legacy_reg=%s", user.id, req.device_id, + req.push_vendor, + bool(req.push_token), bool(req.registration_id), ) return DeviceOut.model_validate(device) @@ -64,10 +104,59 @@ def report_heartbeat( device_id=req.device_id, accessibility_enabled=req.accessibility_enabled, registration_id=req.registration_id, + push_vendor=req.push_vendor, + push_token=req.push_token, ) return OkResponse() +@router.post("/push-test", response_model=PushTestOut, summary="延迟发送厂商通道测试推送") +def request_push_test( + req: PushTestRequest, + background_tasks: BackgroundTasks, + user: CurrentUser, + db: DbSession, +) -> PushTestOut: + """开发验收用:App 内点一次,服务端延迟发厂商直推,验证离线通道。""" + push_vendor = req.push_vendor.strip() if req.push_vendor else None + push_token = req.push_token.strip() if req.push_token else None + if push_vendor and push_token: + device_repo.register_or_update( + db, + user_id=user.id, + device_id=req.device_id, + registration_id=req.registration_id, + push_vendor=push_vendor, + push_token=push_token, + ) + else: + device = device_repo.get_device(db, user_id=user.id, device_id=req.device_id) + push_vendor = device.push_vendor if device is not None else None + push_token = device.push_token if device is not None else None + + if not push_vendor or not push_token: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="push vendor token not ready", + ) + + background_tasks.add_task( + _send_push_test_after_delay, + push_vendor, + push_token, + req.delay_seconds, + user.id, + req.device_id, + ) + logger.info( + "push test scheduled user_id=%d device_id=%s delay=%ds", + user.id, + req.device_id, + req.delay_seconds, + ) + return PushTestOut(delay_seconds=req.delay_seconds, has_push_token=True) + + @router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)") def get_liveness( device_id: str, diff --git a/app/core/config.py b/app/core/config.py index 3976b2a..b9cbaea 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -9,7 +9,7 @@ from functools import lru_cache from pathlib import Path from typing import Literal -from pydantic import Field, model_validator +from pydantic import model_validator from pydantic_settings import BaseSettings, SettingsConfigDict _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent @@ -59,14 +59,51 @@ class Settings(BaseSettings): return [] return [ip.strip() for ip in self.ADMIN_IP_ALLOWLIST.split(",") if ip.strip()] - # ===== 极光 ===== + # ===== 极光一键登录 / 短信 ===== JG_APP_KEY: str = "" JG_MASTER_SECRET: str = "" JG_PRIVATE_KEY_PATH: str = "./secrets/jverify_rsa_private.pem" JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify" JG_REQUEST_TIMEOUT_SEC: int = 15 - # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送) + # ===== 厂商直推(无障碍保护存活告警)===== + ANDROID_PACKAGE_NAME: str = "com.jishisongfu.shaguabijia" + PUSH_REQUEST_TIMEOUT_SEC: int = 15 + PUSH_TIME_TO_LIVE_SEC: int = 86400 + + HONOR_PUSH_APP_ID: str = "" + HONOR_PUSH_CLIENT_ID: str = "" + HONOR_PUSH_CLIENT_SECRET: str = "" + HONOR_PUSH_TOKEN_ENDPOINT: str = "https://iam.developer.honor.com/auth/token" + HONOR_PUSH_SEND_ENDPOINT_TEMPLATE: str = ( + "https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage" + ) + + VIVO_PUSH_APP_ID: str = "" + VIVO_PUSH_APP_KEY: str = "" + VIVO_PUSH_APP_SECRET: str = "" + VIVO_PUSH_AUTH_ENDPOINT: str = "https://api-push.vivo.com.cn/message/auth" + VIVO_PUSH_SEND_ENDPOINT: str = "https://api-push.vivo.com.cn/message/send" + VIVO_PUSH_MODE: int = 1 # 0=正式推送,1=测试推送(未上架 vivo 时用) + VIVO_PUSH_NOTIFY_TYPE: int = 4 # 1=无,2=响铃,3=振动,4=响铃+振动 + VIVO_PUSH_CATEGORY: str = "DEVICE_REMINDER" + + XIAOMI_PUSH_APP_SECRET: str = "" + XIAOMI_PUSH_SEND_ENDPOINT: str = "https://api.xmpush.xiaomi.com/v3/message/regid" + XIAOMI_PUSH_CHANNEL_ID: str = "" + XIAOMI_PUSH_TEMPLATE_ID: str = "" + XIAOMI_PUSH_TEMPLATE_TITLE: str = "" + XIAOMI_PUSH_TEMPLATE_DESCRIPTION: str = "" + XIAOMI_PUSH_TEMPLATE_PARAM_JSON: str = "" + + OPPO_PUSH_APP_KEY: str = "" + OPPO_PUSH_MASTER_SECRET: str = "" + OPPO_PUSH_AUTH_ENDPOINT: str = "https://api.push.oppomobile.com/server/v1/auth" + OPPO_PUSH_SEND_ENDPOINT: str = ( + "https://api.push.oppomobile.com/server/v1/message/notification/unicast" + ) + + # 无障碍保护存活监控后台任务(推送 + pull 后置兜底) HEARTBEAT_MONITOR_ENABLED: bool = True # 总开关 HEARTBEAT_TIMEOUT_MINUTES: int = 10 # 多久没心跳算掉线(≈3 个客户端心跳周期) HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期 @@ -113,6 +150,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 移动应用,用于微信支付);这是【已认证服务号】。 @@ -272,7 +324,7 @@ class Settings(BaseSettings): return self.APP_ENV == "prod" @model_validator(mode="after") - def _enforce_prod_secrets(self) -> "Settings": + def _enforce_prod_secrets(self) -> Settings: """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。 只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值 diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py index 6072d26..be27124 100644 --- a/app/core/heartbeat_monitor_worker.py +++ b/app/core/heartbeat_monitor_worker.py @@ -1,7 +1,7 @@ """无障碍保护存活监控后台任务。 周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了), -**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机 +**命中即在服务器终端打印告警并尝试厂商直推**;并把状态机 推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。 结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。 @@ -22,6 +22,7 @@ from sqlalchemy.exc import SQLAlchemyError from app.core.config import settings from app.db.session import SessionLocal +from app.integrations import vendor_push from app.repositories import device as device_repo logger = logging.getLogger("shagua.heartbeat_monitor") @@ -71,32 +72,66 @@ def _silent_seconds(last: datetime | None) -> int | None: """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。""" if last is None: return None - ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow() + ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow() # noqa: UP017 return int((ref - last).total_seconds()) def _scan_once(timeout_minutes: int) -> dict: - """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。 + """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备并召回。 - 本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端, - 并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。 + 有 push_vendor + push_token 时先发厂商直推,无 token 或推送失败时仍置 + kill_alert_pending,客户端下次进 App 继续走后置提醒兜底。 """ notified = 0 + pushed = 0 + push_failed = 0 with SessionLocal() as db: overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes) for device in overdue: silent = _silent_seconds(device.last_heartbeat_at) logger.warning( "[掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)" - " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】", + " → 判定 App 已被杀/无障碍已停。", device.user_id, device.device_id, silent if silent is not None else "?", timeout_minutes, ) + if device.push_vendor and device.push_token: + try: + vendor_push.send_accessibility_disabled( + device.push_vendor, + device.push_token, + ) + pushed += 1 + logger.info( + "[掉线检测] push sent user_id=%s device_id=%s vendor=%s", + device.user_id, + device.device_id, + device.push_vendor, + ) + except vendor_push.VendorPushError as e: + push_failed += 1 + logger.warning( + "[掉线检测] push failed user_id=%s device_id=%s error=%s", + device.user_id, + device.device_id, + e, + ) + else: + logger.info( + "[掉线检测] device has no push vendor/token, skip push user_id=%s device_id=%s", + device.user_id, + device.device_id, + ) device_repo.mark_notified(db, device_id_pk=device.id) notified += 1 - return {"checked": len(overdue), "notified": notified} + return { + "checked": len(overdue), + "notified": notified, + "pushed": pushed, + "push_failed": push_failed, + } async def _run_loop() -> None: 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/integrations/vendor_push.py b/app/integrations/vendor_push.py new file mode 100644 index 0000000..92917e3 --- /dev/null +++ b/app/integrations/vendor_push.py @@ -0,0 +1,380 @@ +"""厂商直推集成。 + +服务端不再经由 JPush Push API 发送无障碍召回通知,而是按客户端上报的 +push_vendor + push_token 分发到各手机厂商的服务端 API。 +""" +from __future__ import annotations + +import hashlib +import json +import logging +import time +import uuid +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +import httpx + +from app.core.config import settings + +logger = logging.getLogger("shagua.vendor_push") + +TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled" +SUPPORTED_VENDORS = frozenset({"honor", "vivo", "xiaomi", "oppo"}) + + +class VendorPushError(Exception): + """厂商推送调用失败。""" + + +@dataclass +class _CachedToken: + value: str + expires_at: float + + +_token_cache: dict[str, _CachedToken] = {} + + +def normalize_vendor(push_vendor: str | None) -> str | None: + if not push_vendor: + return None + vendor = push_vendor.strip().lower() + aliases = { + "hihonor": "honor", + "荣耀": "honor", + "mi": "xiaomi", + "小米": "xiaomi", + "oneplus": "oppo", + "realme": "oppo", + } + return aliases.get(vendor, vendor) + + +def send_accessibility_disabled( + push_vendor: str, + push_token: str, + *, + title: str = "保护已关闭", + alert: str = "傻瓜比价的无障碍保护被关了，点此重新开启，继续帮你自动比价省钱。", +) -> dict[str, Any]: + """按厂商 token 向单台设备发送无障碍掉线通知。""" + vendor = normalize_vendor(push_vendor) + token = push_token.strip() if push_token else "" + if not vendor or vendor not in SUPPORTED_VENDORS: + raise VendorPushError(f"unsupported push vendor: {push_vendor}") + if not token: + raise VendorPushError("push token is empty") + + dispatch: dict[str, Callable[[str, str, str], dict[str, Any]]] = { + "honor": _send_honor, + "vivo": _send_vivo, + "xiaomi": _send_xiaomi, + "oppo": _send_oppo, + } + return dispatch[vendor](token, title, alert) + + +def _extras() -> dict[str, str]: + return {"type": TYPE_ACCESSIBILITY_DISABLED} + + +def _require(value: str, name: str) -> str: + if not value: + raise VendorPushError(f"{name} not configured") + return value + + +def _request_json( + method: str, + url: str, + *, + expected_status: tuple[int, ...] = (200,), + **kwargs: Any, +) -> dict[str, Any]: + try: + resp = httpx.request( + method, + url, + timeout=settings.PUSH_REQUEST_TIMEOUT_SEC, + **kwargs, + ) + except httpx.HTTPError as e: + raise VendorPushError(f"push http error: {e}") from e + + if resp.status_code not in expected_status: + logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500]) + raise VendorPushError(f"push http {resp.status_code}") + try: + return resp.json() + except ValueError as e: + raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e + + +def _request_form( + method: str, + url: str, + *, + expected_status: tuple[int, ...] = (200,), + **kwargs: Any, +) -> dict[str, Any]: + try: + resp = httpx.request( + method, + url, + timeout=settings.PUSH_REQUEST_TIMEOUT_SEC, + **kwargs, + ) + except httpx.HTTPError as e: + raise VendorPushError(f"push http error: {e}") from e + + if resp.status_code not in expected_status: + logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500]) + raise VendorPushError(f"push http {resp.status_code}") + try: + return resp.json() + except ValueError as e: + raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e + + +def _cache_get(key: str) -> str | None: + cached = _token_cache.get(key) + if cached and cached.expires_at > time.time() + 60: + return cached.value + return None + + +def _cache_put(key: str, value: str, expires_in: int | float | None) -> str: + ttl = int(expires_in or 3600) + _token_cache[key] = _CachedToken(value=value, expires_at=time.time() + max(60, ttl - 60)) + return value + + +def _honor_access_token() -> str: + cache_key = "honor" + cached = _cache_get(cache_key) + if cached: + return cached + client_id = _require(settings.HONOR_PUSH_CLIENT_ID, "HONOR_PUSH_CLIENT_ID") + client_secret = _require(settings.HONOR_PUSH_CLIENT_SECRET, "HONOR_PUSH_CLIENT_SECRET") + data = _request_form( + "POST", + settings.HONOR_PUSH_TOKEN_ENDPOINT, + data={ + "grant_type": "client_credentials", + "client_id": client_id, + "client_secret": client_secret, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + token = data.get("access_token") + if not token: + raise VendorPushError(f"honor auth failed: {data}") + return _cache_put(cache_key, str(token), data.get("expires_in")) + + +def _send_honor(token: str, title: str, alert: str) -> dict[str, Any]: + app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID") + access_token = _honor_access_token() + payload = { + "data": json.dumps(_extras(), ensure_ascii=False), + "notification": {"title": title, "body": alert}, + "android": { + "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s", + "targetUserType": 1, + "notification": { + "title": title, + "body": alert, + "clickAction": {"type": 3}, + "importance": "NORMAL", + }, + }, + "token": [token], + } + data = _request_json( + "POST", + settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), + json=payload, + headers={ + "Content-Type": "application/json; charset=UTF-8", + "Authorization": f"Bearer {access_token}", + "timestamp": str(int(time.time() * 1000)), + }, + ) + code = data.get("code") + if code is not None and int(code) != 200: + raise VendorPushError(f"honor push failed: {data}") + return data + + +def _vivo_auth_token() -> str: + cache_key = "vivo" + cached = _cache_get(cache_key) + if cached: + return cached + app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID") + app_key = _require(settings.VIVO_PUSH_APP_KEY, "VIVO_PUSH_APP_KEY") + app_secret = _require(settings.VIVO_PUSH_APP_SECRET, "VIVO_PUSH_APP_SECRET") + timestamp = str(int(time.time() * 1000)) + sign = hashlib.md5(f"{app_id}{app_key}{timestamp}{app_secret}".encode()).hexdigest() # noqa: S324 + data = _request_json( + "POST", + settings.VIVO_PUSH_AUTH_ENDPOINT, + json={ + "appId": app_id, + "appKey": app_key, + "timestamp": timestamp, + "sign": sign, + }, + headers={"Content-Type": "application/json"}, + ) + if int(data.get("result", -1)) != 0: + raise VendorPushError(f"vivo auth failed: {data}") + token = data.get("authToken") + if not token: + raise VendorPushError(f"vivo auth missing authToken: {data}") + return _cache_put(cache_key, str(token), 24 * 3600) + + +def _send_vivo(token: str, title: str, alert: str) -> dict[str, Any]: + app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID") + auth_token = _vivo_auth_token() + body: dict[str, Any] = { + "appId": app_id, + "regId": token, + "notifyType": settings.VIVO_PUSH_NOTIFY_TYPE, + "title": title, + "content": alert, + "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC, + "skipType": 1, + "requestId": uuid.uuid4().hex, + "pushMode": settings.VIVO_PUSH_MODE, + "clientCustomMap": _extras(), + } + if settings.VIVO_PUSH_CATEGORY: + body["category"] = settings.VIVO_PUSH_CATEGORY + data = _request_json( + "POST", + settings.VIVO_PUSH_SEND_ENDPOINT, + json=body, + headers={ + "Content-Type": "application/json", + "authToken": auth_token, + }, + ) + if int(data.get("result", -1)) != 0: + raise VendorPushError(f"vivo push failed: {data}") + return data + + +def _send_xiaomi(token: str, title: str, alert: str) -> dict[str, Any]: + app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET") + message_title = settings.XIAOMI_PUSH_TEMPLATE_TITLE.strip() or title + message_description = settings.XIAOMI_PUSH_TEMPLATE_DESCRIPTION.strip() or alert + body = { + "registration_id": token, + "restricted_package_name": settings.ANDROID_PACKAGE_NAME, + "title": message_title, + "description": message_description, + "payload": json.dumps(_extras(), ensure_ascii=False), + "pass_through": "0", + "notify_type": "-1", + "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000), + "extra.notify_effect": "1", + } + if settings.XIAOMI_PUSH_CHANNEL_ID: + body["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip() + if settings.XIAOMI_PUSH_TEMPLATE_ID: + body["extra.template_id"] = settings.XIAOMI_PUSH_TEMPLATE_ID.strip() + if settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON: + body["extra.template_param"] = _xiaomi_template_param(title, alert) + data = _request_form( + "POST", + settings.XIAOMI_PUSH_SEND_ENDPOINT, + data=body, + headers={"Authorization": f"key={app_secret}"}, + ) + code = data.get("code") + if code not in (0, "0", None): + raise VendorPushError(f"xiaomi push failed: {data}") + if str(data.get("result", "ok")).lower() not in ("ok", "success"): + raise VendorPushError(f"xiaomi push failed: {data}") + return data + + +def _xiaomi_template_param(title: str, alert: str) -> str: + rendered = ( + settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON + .replace("{title}", title) + .replace("{alert}", alert) + ) + try: + payload = json.loads(rendered) + except ValueError as e: + raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON invalid json") from e + if not isinstance(payload, dict): + raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON must be a json object") + for key, value in payload.items(): + if not isinstance(key, str) or not isinstance(value, str): + raise VendorPushError("xiaomi template params must be string key-value pairs") + if not value.strip() or len(value) > 128: + raise VendorPushError("xiaomi template param value length must be 1-128") + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + + +def _oppo_auth_token() -> str: + cache_key = "oppo" + cached = _cache_get(cache_key) + if cached: + return cached + app_key = _require(settings.OPPO_PUSH_APP_KEY, "OPPO_PUSH_APP_KEY") + master_secret = _require(settings.OPPO_PUSH_MASTER_SECRET, "OPPO_PUSH_MASTER_SECRET") + timestamp = str(int(time.time() * 1000)) + sign = hashlib.sha256(f"{app_key}{timestamp}{master_secret}".encode()).hexdigest() + data = _request_form( + "POST", + settings.OPPO_PUSH_AUTH_ENDPOINT, + data={ + "app_key": app_key, + "timestamp": timestamp, + "sign": sign, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + if int(data.get("code", -1)) != 0: + raise VendorPushError(f"oppo auth failed: {data}") + token = (data.get("data") or {}).get("auth_token") or data.get("auth_token") + if not token: + raise VendorPushError(f"oppo auth missing auth_token: {data}") + return _cache_put(cache_key, str(token), 24 * 3600) + + +def _send_oppo(token: str, title: str, alert: str) -> dict[str, Any]: + auth_token = _oppo_auth_token() + ttl_hours = max(1, min(72, settings.PUSH_TIME_TO_LIVE_SEC // 3600)) + message = { + "target_type": 2, + "target_value": token, + "notification": { + "app_message_id": f"accessibility_disabled_{uuid.uuid4().hex}", + "title": title, + "content": alert, + "click_action_type": 0, + "off_line": True, + "off_line_ttl": ttl_hours, + "action_parameters": json.dumps(_extras(), ensure_ascii=False), + }, + } + data = _request_form( + "POST", + settings.OPPO_PUSH_SEND_ENDPOINT, + data={ + "auth_token": auth_token, + "message": json.dumps(message, ensure_ascii=False), + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + if int(data.get("code", -1)) != 0: + raise VendorPushError(f"oppo push failed: {data}") + return data diff --git a/app/integrations/wechat_login.py b/app/integrations/wechat_login.py new file mode 100644 index 0000000..119a219 --- /dev/null +++ b/app/integrations/wechat_login.py @@ -0,0 +1,89 @@ +"""微信开放平台 App 登录集成。 + +客户端只把微信 SDK 回调拿到的临时 code 传给服务端;AppSecret、access_token +与用户 openid 都留在服务端处理,避免敏感凭证落到 APK 里。 +""" +from __future__ import annotations + +import certifi +import httpx + +from app.core.config import settings + +_API_BASE = "https://api.weixin.qq.com" + + +class WechatLoginError(Exception): + """微信授权换取用户信息失败。""" + + +def _ensure_configured() -> None: + if not settings.WECHAT_APP_ID or not settings.WECHAT_APP_SECRET: + raise WechatLoginError("wechat app not configured") + + +def _http_client() -> httpx.Client: + return httpx.Client(verify=certifi.where()) + + +def code_to_userinfo(code: str) -> dict: + """用授权 code 换 openid/unionid,尽力补昵称头像。 + + 返回字段: + - openid: 同一开放平台移动应用下的用户唯一标识 + - unionid: 同一开放平台主体下的用户唯一标识,微信可能不返回 + - nickname/avatar_url: sns/userinfo 可用时返回,失败不阻断登录 + - raw: userinfo 原始响应,便于排障 + """ + _ensure_configured() + try: + with _http_client() as client: + token_resp = client.get( + f"{_API_BASE}/sns/oauth2/access_token", + params={ + "appid": settings.WECHAT_APP_ID, + "secret": settings.WECHAT_APP_SECRET, + "code": code, + "grant_type": "authorization_code", + }, + timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, + ) + token_data = token_resp.json() + except Exception as e: # noqa: BLE001 + raise WechatLoginError("微信授权请求失败,请稍后重试") from e + if "openid" not in token_data or "access_token" not in token_data: + raise WechatLoginError(f"微信授权失败: {token_data.get('errmsg', token_data)}") + + openid = token_data["openid"] + unionid = token_data.get("unionid") + nickname = None + avatar_url = None + raw: dict = {} + + try: + with _http_client() as client: + info_resp = client.get( + f"{_API_BASE}/sns/userinfo", + params={ + "access_token": token_data["access_token"], + "openid": openid, + "lang": "zh_CN", + }, + timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, + ) + raw = info_resp.json() + if "errcode" not in raw: + nickname = raw.get("nickname") or None + avatar_url = raw.get("headimgurl") or None + unionid = unionid or raw.get("unionid") + except Exception: # noqa: BLE001 + # openid 已拿到即可登录;昵称头像只是展示增强。 + pass + + return { + "openid": openid, + "unionid": unionid, + "nickname": nickname, + "avatar_url": avatar_url, + "raw": raw, + } 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/models/device.py b/app/models/device.py index e1d1261..324e176 100644 --- a/app/models/device.py +++ b/app/models/device.py @@ -1,9 +1,9 @@ -"""设备表(无障碍保护存活检测 + 极光推送)。 +"""设备表(无障碍保护存活检测 + 厂商直推)。 每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。 客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报 -registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、 -现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。 +push_vendor + push_token(厂商推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、 +现在心跳超时」的设备,通过厂商直推提醒用户重开无障碍。 liveness_state 状态机(防刷屏,一次掉线只推一条): unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送) @@ -30,7 +30,7 @@ from app.db.base import Base class DeviceLiveness(Base): # 表名不叫 device:device 易被当成「设备信息(品牌/型号/系统)」表;本表实为**无障碍存活监控状态** - # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。 + # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 厂商推送目标),故名 device_liveness。 __tablename__ = "device_liveness" __table_args__ = ( UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"), @@ -42,8 +42,12 @@ class DeviceLiveness(Base): ) # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34) device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False) - # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发) + # 旧极光推送 registration id,仅为兼容历史客户端/数据保留;新链路使用 push_vendor + push_token。 registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # 厂商推送类型:honor/vivo/xiaomi/oppo 等;客户端按实际 SDK token 来源上报。 + push_vendor: Mapped[str | None] = mapped_column(String(32), nullable=True) + # 厂商 push token / regId / registration_id;不同厂商命名不同,后端统一存这里。 + push_token: Mapped[str | None] = mapped_column(String(256), nullable=True) platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android") app_version: Mapped[str | None] = mapped_column(String(32), nullable=True) diff --git a/app/repositories/device.py b/app/repositories/device.py index c3e9047..31b84d9 100644 --- a/app/repositories/device.py +++ b/app/repositories/device.py @@ -21,17 +21,23 @@ def register_or_update( *, user_id: int, device_id: str, - registration_id: str | None, + registration_id: str | None = None, + push_vendor: str | None = None, + push_token: str | None = None, platform: str = "android", app_version: str | None = None, ) -> DeviceLiveness: - """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。""" + """注册设备或更新其厂商 push token / 元信息。upsert by (user_id, device_id)。""" + normalized_vendor = _normalize_push_vendor(push_vendor) + normalized_token = push_token.strip() if push_token else None device = _get(db, user_id=user_id, device_id=device_id) if device is None: device = DeviceLiveness( user_id=user_id, device_id=device_id, registration_id=registration_id, + push_vendor=normalized_vendor, + push_token=normalized_token, platform=platform or "android", app_version=app_version, ) @@ -39,6 +45,10 @@ def register_or_update( else: if registration_id: device.registration_id = registration_id + if normalized_vendor: + device.push_vendor = normalized_vendor + if normalized_token: + device.push_token = normalized_token if platform: device.platform = platform if app_version: @@ -54,7 +64,9 @@ def touch_heartbeat( user_id: int, device_id: str, accessibility_enabled: bool, - registration_id: str | None, + registration_id: str | None = None, + push_vendor: str | None = None, + push_token: str | None = None, ) -> DeviceLiveness: """处理一次心跳(心跳也能自注册)。 @@ -69,6 +81,12 @@ def touch_heartbeat( if registration_id: device.registration_id = registration_id + normalized_vendor = _normalize_push_vendor(push_vendor) + normalized_token = push_token.strip() if push_token else None + if normalized_vendor: + device.push_vendor = normalized_vendor + if normalized_token: + device.push_token = normalized_token device.last_report_protection_on = accessibility_enabled if accessibility_enabled: @@ -87,7 +105,7 @@ def touch_heartbeat( def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]: """掉线设备:曾经保护过、当前 alive、心跳超时。 - 本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。 + 即使没有厂商 token 也要检出,后续由 kill_alert_pending 走客户端进 App 后兜底提醒。 """ cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes) stmt = select(DeviceLiveness).where( @@ -124,3 +142,27 @@ def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None: if device is not None and device.kill_alert_pending: device.kill_alert_pending = False db.commit() + + +def has_push_target(device: DeviceLiveness | None) -> bool: + """是否已有厂商直推所需的 vendor + token。""" + return bool(device and device.push_vendor and device.push_token) + + +def _normalize_push_vendor(push_vendor: str | None) -> str | None: + if not push_vendor: + return None + vendor = push_vendor.strip().lower() + aliases = { + "honor": "honor", + "hihonor": "honor", + "荣耀": "honor", + "vivo": "vivo", + "xiaomi": "xiaomi", + "mi": "xiaomi", + "小米": "xiaomi", + "oppo": "oppo", + "oneplus": "oppo", + "realme": "oppo", + } + return aliases.get(vendor, vendor) diff --git a/app/repositories/user.py b/app/repositories/user.py index 7397d44..1a5631b 100644 --- a/app/repositories/user.py +++ b/app/repositories/user.py @@ -6,6 +6,7 @@ from __future__ import annotations import secrets import string +import hashlib from datetime import datetime, timezone from sqlalchemy import select @@ -70,6 +71,21 @@ def get_user_by_phone(db: Session, phone: str) -> User | None: return db.execute(stmt).scalar_one_or_none() +def get_user_by_wechat_openid(db: Session, openid: str) -> User | None: + stmt = select(User).where(User.wechat_openid == openid) + return db.execute(stmt).scalar_one_or_none() + + +def _wechat_placeholder_phone(openid: str) -> str: + """给纯微信注册用户生成内部登录号。 + + 当前 user.phone 非空唯一,而微信 App 授权不会返回手机号。这里生成一个不可能 + 命中手机号正则的内部值,只用于满足既有表结构和客户端登录态判断。 + """ + digest = hashlib.sha256(openid.encode("utf-8")).hexdigest()[:16] + return f"wx_{digest}" + + def upsert_user_for_login( db: Session, *, @@ -99,6 +115,52 @@ def upsert_user_for_login( return user +def upsert_user_for_wechat_login( + db: Session, + *, + openid: str, + nickname: str | None = None, + avatar_url: str | None = None, +) -> User: + """微信 App 登录:openid 已存在则回到原账号,不存在则注册新账号。 + + 如果用户之前在提现页绑定过微信,该 openid 已在 user.wechat_openid 上,因此会 + 直接登录到同一账号;新微信用户则创建 register_channel=wechat 的账号。 + """ + user = get_user_by_wechat_openid(db, openid) + now = datetime.now(timezone.utc) + if user is None: + phone = _wechat_placeholder_phone(openid) + # 极小概率 hash 前缀碰撞时,后缀追加随机段兜底。 + if get_user_by_phone(db, phone) is not None: + phone = f"wx_{secrets.token_hex(8)}" + user = User( + phone=phone, + username=_gen_unique_username(db), + nickname=nickname or _gen_nickname(), + avatar_url=avatar_url, + register_channel="wechat", + wechat_openid=openid, + wechat_nickname=nickname, + wechat_avatar_url=avatar_url, + last_login_at=now, + ) + db.add(user) + else: + user.last_login_at = now + if nickname: + user.wechat_nickname = nickname + if not user.nickname: + user.nickname = nickname + if avatar_url: + user.wechat_avatar_url = avatar_url + if not user.avatar_url: + user.avatar_url = avatar_url + db.commit() + db.refresh(user) + return user + + def update_nickname(db: Session, user: User, *, nickname: str) -> User: user.nickname = nickname db.commit() diff --git a/app/schemas/auth.py b/app/schemas/auth.py index 3870a52..d9c18ce 100644 --- a/app/schemas/auth.py +++ b/app/schemas/auth.py @@ -67,6 +67,14 @@ class JverifyLoginRequest(BaseModel): ) +class WechatLoginRequest(BaseModel): + code: str = Field(..., min_length=1, description="微信 SDK SendAuth.Resp 返回的一次性授权 code") + device_id: str = Field( + "", max_length=64, + description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理", + ) + + # ===== 短信验证码 ===== class SmsSendRequest(BaseModel): diff --git a/app/schemas/device.py b/app/schemas/device.py index dd7a7c4..3f8cda1 100644 --- a/app/schemas/device.py +++ b/app/schemas/device.py @@ -3,12 +3,15 @@ from __future__ import annotations from datetime import datetime -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, Field class DeviceRegisterRequest(BaseModel): device_id: str + # registration_id 为旧极光字段,新推送链路统一使用 push_vendor + push_token。 registration_id: str | None = None + push_vendor: str | None = None + push_token: str | None = None platform: str = "android" app_version: str | None = None @@ -18,6 +21,8 @@ class HeartbeatRequest(BaseModel): source: str = "service" # service | app accessibility_enabled: bool = True registration_id: str | None = None + push_vendor: str | None = None + push_token: str | None = None class DeviceOut(BaseModel): @@ -26,6 +31,8 @@ class DeviceOut(BaseModel): id: int device_id: str registration_id: str | None + push_vendor: str | None + push_token: str | None ever_protected: bool liveness_state: str last_heartbeat_at: datetime | None @@ -46,3 +53,17 @@ class LivenessOut(BaseModel): class LivenessAckRequest(BaseModel): device_id: str + + +class PushTestRequest(BaseModel): + device_id: str + delay_seconds: int = Field(default=10, ge=0, le=60) + push_vendor: str | None = None + push_token: str | None = None + registration_id: str | None = None + + +class PushTestOut(BaseModel): + ok: bool = True + delay_seconds: int + has_push_token: bool 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_auth.py b/tests/test_auth.py index 4968a73..804eb5e 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -68,6 +68,64 @@ def test_sms_login_and_me_flow(client) -> None: assert r.json()["ok"] is True +def test_wechat_login_creates_and_reuses_user(client, monkeypatch) -> None: + """微信 code 登录:新 openid 注册,同 openid 重登回同一账号。""" + monkeypatch.setattr( + "app.api.v1.auth.code_to_userinfo", + lambda code: { + "openid": "openid_login_1", + "unionid": "union_1", + "nickname": "微信用户A", + "avatar_url": "https://wx/avatar-a.png", + "raw": {}, + }, + ) + + r = client.post("/api/v1/auth/wechat-login", json={"code": "code-a", "device_id": "dev-wx"}) + assert r.status_code == 200, r.text + first = r.json() + assert first["user"]["register_channel"] == "wechat" + assert first["user"]["phone"].startswith("wx_") + assert first["user"]["nickname"] == "微信用户A" + + r = client.post("/api/v1/auth/wechat-login", json={"code": "code-b", "device_id": "dev-wx"}) + assert r.status_code == 200, r.text + second = r.json() + assert second["user"]["id"] == first["user"]["id"] + + +def test_wechat_login_uses_existing_bound_user(client, monkeypatch) -> None: + """如果该微信已在提现页绑定过,微信登录应回到原手机号账号。""" + from app.db.session import SessionLocal + from app.repositories import user as user_repo + + db = SessionLocal() + try: + user = user_repo.upsert_user_for_login(db, phone="13600136010", register_channel="sms") + user.wechat_openid = "openid_bound_1" + db.commit() + uid = user.id + finally: + db.close() + + monkeypatch.setattr( + "app.api.v1.auth.code_to_userinfo", + lambda code: { + "openid": "openid_bound_1", + "unionid": None, + "nickname": "绑定微信", + "avatar_url": None, + "raw": {}, + }, + ) + + r = client.post("/api/v1/auth/wechat-login", json={"code": "code-bound", "device_id": "dev-wx"}) + assert r.status_code == 200, r.text + body = r.json() + assert body["user"]["id"] == uid + assert body["user"]["phone"] == "13600136010" + + def test_sms_send_too_frequent(client) -> None: phone = "13900139000" assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200 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 diff --git a/tests/test_device_push.py b/tests/test_device_push.py new file mode 100644 index 0000000..a35ec24 --- /dev/null +++ b/tests/test_device_push.py @@ -0,0 +1,317 @@ +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone + +from fastapi.testclient import TestClient + +from app.api.v1 import device as device_api +from app.core import heartbeat_monitor_worker +from app.db.session import SessionLocal +from app.integrations import vendor_push +from app.models.device import DeviceLiveness +from app.repositories import user as user_repo + + +class _Resp: + status_code = 200 + text = "{}" + + def __init__(self, data: dict) -> None: + self._data = data + + def json(self) -> dict: + return self._data + + +def test_xiaomi_accessibility_payload(monkeypatch) -> None: + captured: dict = {} + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + captured.update(method=method, url=url, **kwargs) + return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-msg"}}) + + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", "") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + data = vendor_push.send_accessibility_disabled("xiaomi", "xm-regid") + + assert data["data"]["id"] == "xm-msg" + assert captured["method"] == "POST" + assert captured["url"] == vendor_push.settings.XIAOMI_PUSH_SEND_ENDPOINT + assert captured["headers"]["Authorization"] == "key=xiaomi-secret" + body = captured["data"] + assert body["registration_id"] == "xm-regid" + assert body["restricted_package_name"] == "com.jishisongfu.shaguabijia" + assert json.loads(body["payload"]) == {"type": "accessibility_disabled"} + assert "extra.channel_id" not in body + + +def test_xiaomi_payload_with_channel_and_template(monkeypatch) -> None: + captured: dict = {} + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + captured.update(method=method, url=url, **kwargs) + return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-msg"}}) + + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "130") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "1001") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "{$app_name$}提醒") + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "{$content$}") + monkeypatch.setattr( + vendor_push.settings, + "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", + '{"app_name":"傻瓜比价","content":"{alert}"}', + ) + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + vendor_push.send_accessibility_disabled( + "xiaomi", + "xm-regid", + title="测试标题", + alert="测试内容", + ) + + body = captured["data"] + assert body["title"] == "{$app_name$}提醒" + assert body["description"] == "{$content$}" + assert body["extra.channel_id"] == "130" + assert body["extra.template_id"] == "1001" + assert body["extra.template_param"] == '{"app_name":"傻瓜比价","content":"测试内容"}' + + +def test_vivo_auth_and_send_payload(monkeypatch) -> None: + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT: + return _Resp({"result": 0, "authToken": "vivo-auth"}) + return _Resp({"result": 0, "taskId": "vivo-task"}) + + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_ID", "106072775") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_KEY", "vivo-key") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_SECRET", "vivo-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + data = vendor_push.send_accessibility_disabled("vivo", "vivo-regid") + + assert data["taskId"] == "vivo-task" + assert calls[0]["url"] == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT + assert calls[0]["json"]["appId"] == "106072775" + assert calls[0]["json"]["sign"] + assert calls[1]["url"] == vendor_push.settings.VIVO_PUSH_SEND_ENDPOINT + assert calls[1]["headers"]["authToken"] == "vivo-auth" + body = calls[1]["json"] + assert body["regId"] == "vivo-regid" + assert body["pushMode"] == vendor_push.settings.VIVO_PUSH_MODE + assert body["clientCustomMap"] == {"type": "accessibility_disabled"} + + +def test_oppo_auth_and_send_payload(monkeypatch) -> None: + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.OPPO_PUSH_AUTH_ENDPOINT: + return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}}) + return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}}) + + monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key") + monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + data = vendor_push.send_accessibility_disabled("oppo", "oppo-regid") + + assert data["data"]["message_id"] == "oppo-msg" + assert calls[0]["data"]["app_key"] == "oppo-key" + assert calls[0]["data"]["sign"] + message = json.loads(calls[1]["data"]["message"]) + assert calls[1]["data"]["auth_token"] == "oppo-auth" + assert message["target_type"] == 2 + assert message["target_value"] == "oppo-regid" + assert json.loads(message["notification"]["action_parameters"]) == { + "type": "accessibility_disabled" + } + + +def test_honor_auth_and_send_payload(monkeypatch) -> None: + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.HONOR_PUSH_TOKEN_ENDPOINT: + return _Resp({"access_token": "honor-access", "expires_in": 3600}) + return _Resp({"code": 200, "message": "successful!", "data": {"sendResult": True}}) + + monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_APP_ID", "104559789") + monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_ID", "honor-client") + monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_SECRET", "honor-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + data = vendor_push.send_accessibility_disabled("honor", "honor-token") + + assert data["code"] == 200 + assert calls[0]["data"]["client_id"] == "honor-client" + assert calls[1]["headers"]["Authorization"] == "Bearer honor-access" + assert calls[1]["headers"]["timestamp"] + assert calls[1]["url"].endswith("/api/v1/104559789/sendMessage") + body = calls[1]["json"] + assert body["token"] == ["honor-token"] + assert body["android"]["targetUserType"] == 1 + assert body["android"]["notification"]["clickAction"] == {"type": 3} + assert json.loads(body["data"]) == {"type": "accessibility_disabled"} + + +def _seed_overdue_device( + *, + phone: str, + device_id: str, + push_vendor: str | None, + push_token: str | None, +) -> int: + with SessionLocal() as db: + user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms") + device = DeviceLiveness( + user_id=user.id, + device_id=device_id, + push_vendor=push_vendor, + push_token=push_token, + platform="android", + ever_protected=True, + last_heartbeat_at=datetime.now(timezone.utc) - timedelta(minutes=30), # noqa: UP017 + last_report_protection_on=True, + liveness_state="alive", + kill_alert_pending=False, + ) + db.add(device) + db.commit() + db.refresh(device) + return device.id + + +def _login(client: TestClient, phone: str) -> str: + client.post("/api/v1/auth/sms/send", json={"phone": phone}) + r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"}) + assert r.status_code == 200, r.text + return r.json()["access_token"] + + +def _auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def test_heartbeat_monitor_pushes_overdue_device(monkeypatch) -> None: + device_pk = _seed_overdue_device( + phone="13900009001", + device_id="dev-push-honor", + push_vendor="honor", + push_token="honor-token-1", + ) + calls: list[tuple[str, str]] = [] + + def _fake_send(push_vendor: str, push_token: str) -> dict: + calls.append((push_vendor, push_token)) + return {"msg_id": "m1"} + + monkeypatch.setattr( + heartbeat_monitor_worker.vendor_push, + "send_accessibility_disabled", + _fake_send, + ) + + result = heartbeat_monitor_worker._scan_once(timeout_minutes=10) + + assert result["pushed"] >= 1 + assert ("honor", "honor-token-1") in calls + with SessionLocal() as db: + device = db.get(DeviceLiveness, device_pk) + assert device is not None + assert device.liveness_state == "notified" + assert device.kill_alert_pending is True + + +def test_heartbeat_monitor_skips_push_without_vendor_token(monkeypatch) -> None: + device_pk = _seed_overdue_device( + phone="13900009002", + device_id="dev-push-no-token", + push_vendor=None, + push_token=None, + ) + + def _fake_send(push_vendor: str, push_token: str) -> dict: + raise AssertionError(f"should not push without token: {push_vendor}/{push_token}") + + monkeypatch.setattr( + heartbeat_monitor_worker.vendor_push, + "send_accessibility_disabled", + _fake_send, + ) + + result = heartbeat_monitor_worker._scan_once(timeout_minutes=10) + + assert result["checked"] >= 1 + with SessionLocal() as db: + device = db.get(DeviceLiveness, device_pk) + assert device is not None + assert device.liveness_state == "notified" + assert device.kill_alert_pending is True + + +def test_push_test_endpoint_schedules_vendor_push(client: TestClient, monkeypatch) -> None: + token = _login(client, "13900009003") + calls: list[tuple[str, str, str, str]] = [] + + def _fake_send(push_vendor: str, push_token: str, *, title: str, alert: str) -> dict: + calls.append((push_vendor, push_token, title, alert)) + return {"msg_id": "m-test"} + + monkeypatch.setattr(device_api.vendor_push, "send_accessibility_disabled", _fake_send) + + r = client.post( + "/api/v1/device/push-test", + json={ + "device_id": "dev-push-test", + "push_vendor": "honor", + "push_token": "honor-test-token", + "delay_seconds": 0, + }, + headers=_auth(token), + ) + + assert r.status_code == 200, r.text + assert r.json() == { + "ok": True, + "delay_seconds": 0, + "has_push_token": True, + } + assert calls == [ + ( + "honor", + "honor-test-token", + "测试推送", + "这是一条厂商通道测试推送。收到它说明 App 被划掉后仍可通过系统通知栏触达。", + ) + ] + + +def test_push_test_endpoint_requires_vendor_token(client: TestClient) -> None: + token = _login(client, "13900009004") + + r = client.post( + "/api/v1/device/push-test", + json={"device_id": "dev-push-test-no-token", "delay_seconds": 0}, + headers=_auth(token), + ) + + assert r.status_code == 409 + assert r.json()["detail"] == "push vendor token not ready" diff --git a/tests/test_health.py b/tests/test_health.py index 38cb737..375e16e 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -14,6 +14,7 @@ def test_openapi_loads(client) -> None: paths = resp.json()["paths"] # auth endpoints 都注册了 assert "/api/v1/auth/jverify-login" in paths + assert "/api/v1/auth/wechat-login" in paths assert "/api/v1/auth/sms/send" in paths assert "/api/v1/auth/sms/login" in paths assert "/api/v1/auth/refresh" in paths