diff --git a/.env.example b/.env.example index 1f53db8..367226a 100644 --- a/.env.example +++ b/.env.example @@ -11,7 +11,7 @@ DATABASE_URL=sqlite:///./data/app.db # ===== JWT ===== # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))" -JWT_SECRET_KEY= +JWT_SECRET_KEY=change-me-in-prod-please-use-a-long-random-string JWT_ALGORITHM=HS256 # access token 有效期(分钟),默认 2 小时 JWT_ACCESS_TOKEN_EXPIRE_MINUTES=120 @@ -27,6 +27,18 @@ 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 +# ===== 极光推送 JPush(无障碍保护掉线告警)===== +# 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。若推送与一键登录/短信是同一个极光应用, +# JPUSH_* 留空即自动回退到上面的 JG_APP_KEY/JG_MASTER_SECRET;否则单独填那个 push 应用的密钥。 +# 运维清单(厂商通道等)见 spec/accessibility-liveness-push.md §7。 +JPUSH_APP_KEY= +JPUSH_MASTER_SECRET= +JPUSH_PUSH_ENDPOINT=https://api.jpush.cn/v3/push +# 无障碍保护存活监控后台任务 +HEARTBEAT_MONITOR_ENABLED=true +HEARTBEAT_TIMEOUT_MINUTES=10 +HEARTBEAT_SCAN_INTERVAL_SEC=60 + # ===== 短信 (mock 模式) ===== # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。 diff --git a/alembic/versions/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py b/alembic/versions/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py new file mode 100644 index 0000000..7d25e70 --- /dev/null +++ b/alembic/versions/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py @@ -0,0 +1,26 @@ +"""merge cps_wx_user 与 device_kill_alert_pending 双 head + +Revision ID: 3a8fa058789c +Revises: cps_wx_user, device_kill_alert_pending +Create Date: 2026-06-20 10:27:01.193201 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '3a8fa058789c' +down_revision: Union[str, Sequence[str], None] = ('cps_wx_user', 'device_kill_alert_pending') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/alembic/versions/cps_wx_user.py b/alembic/versions/cps_wx_user.py new file mode 100644 index 0000000..0e0db37 --- /dev/null +++ b/alembic/versions/cps_wx_user.py @@ -0,0 +1,48 @@ +"""cps 微信用户表 cps_wx_user + cps_click 加 openid + +CPS 落地页微信网页授权拿 openid/昵称头像,做用户级群统计。 + +Revision ID: cps_wx_user +Revises: comparison_debug_fields +Create Date: 2026-06-19 12:00:00.000000 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "cps_wx_user" +down_revision: Union[str, Sequence[str], None] = "comparison_debug_fields" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "cps_wx_user", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("openid", sa.String(64), nullable=False), + sa.Column("unionid", sa.String(64), nullable=True), + sa.Column("nickname", sa.String(128), nullable=True), + sa.Column("headimgurl", sa.String(512), nullable=True), + sa.Column("first_code", sa.String(16), nullable=True), + sa.Column("first_group_id", sa.Integer(), nullable=True), + sa.Column("first_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("last_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + op.create_index("ix_cps_wx_user_openid", "cps_wx_user", ["openid"], unique=True) + op.create_index("ix_cps_wx_user_unionid", "cps_wx_user", ["unionid"]) + op.create_index("ix_cps_wx_user_first_group_id", "cps_wx_user", ["first_group_id"]) + + op.add_column("cps_click", sa.Column("openid", sa.String(64), nullable=True)) + op.create_index("ix_cps_click_openid", "cps_click", ["openid"]) + + +def downgrade() -> None: + op.drop_index("ix_cps_click_openid", table_name="cps_click") + op.drop_column("cps_click", "openid") + op.drop_index("ix_cps_wx_user_first_group_id", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_unionid", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_openid", table_name="cps_wx_user") + op.drop_table("cps_wx_user") diff --git a/alembic/versions/device_kill_alert_pending.py b/alembic/versions/device_kill_alert_pending.py new file mode 100644 index 0000000..f4bc6fc --- /dev/null +++ b/alembic/versions/device_kill_alert_pending.py @@ -0,0 +1,39 @@ +"""device.kill_alert_pending (无障碍掉线「后置检测」待提醒标记) + +Revision ID: device_kill_alert_pending +Revises: merge_cps_device_liveness +Create Date: 2026-06-19 + +后置检测 pull 版(spec/accessibility-liveness-pull-prompt.md):worker 检出心跳掉线即置此标记, +客户端下次进 App 拉取到 → 弹「开启自启动」引导,交互后 ack 清回。与 liveness_state 解耦,故单列。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'device_kill_alert_pending' +down_revision: Union[str, Sequence[str], None] = 'merge_cps_device_liveness' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # server_default=sa.false() 跨方言: SQLite 渲染 0 / PG 渲染 false(别用 sa.text("0"),PG 布尔严格会报错), + # 给存量行回填 False。 + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.add_column( + sa.Column( + 'kill_alert_pending', + sa.Boolean(), + nullable=False, + server_default=sa.false(), + ) + ) + + +def downgrade() -> None: + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.drop_column('kill_alert_pending') diff --git a/alembic/versions/device_table.py b/alembic/versions/device_table.py new file mode 100644 index 0000000..795864c --- /dev/null +++ b/alembic/versions/device_table.py @@ -0,0 +1,53 @@ +"""device table (无障碍保护存活检测 + 极光推送) + +Revision ID: device_liveness_table +Revises: store_mapping_tb_dl_invalid +Create Date: 2026-06-15 12:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'device_liveness_table' +down_revision: Union[str, Sequence[str], None] = 'store_mapping_tb_dl_invalid' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'device', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('user_id', sa.Integer(), nullable=False), + sa.Column('device_id', sa.String(length=128), nullable=False), + sa.Column('registration_id', sa.String(length=64), nullable=True), + sa.Column('platform', sa.String(length=16), nullable=False), + sa.Column('app_version', sa.String(length=32), nullable=True), + sa.Column('ever_protected', sa.Boolean(), nullable=False), + sa.Column('last_heartbeat_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('last_report_protection_on', sa.Boolean(), nullable=False), + sa.Column('liveness_state', sa.String(length=16), nullable=False), + sa.Column('notified_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('user_id', 'device_id', name='uq_device_user_device'), + ) + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_device_user_id'), ['user_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_device_id'), ['device_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_last_heartbeat_at'), ['last_heartbeat_at'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_device_last_heartbeat_at')) + batch_op.drop_index(batch_op.f('ix_device_device_id')) + batch_op.drop_index(batch_op.f('ix_device_user_id')) + + op.drop_table('device') diff --git a/alembic/versions/launch_confirm_sample_table.py b/alembic/versions/launch_confirm_sample_table.py new file mode 100644 index 0000000..d9ff6fe --- /dev/null +++ b/alembic/versions/launch_confirm_sample_table.py @@ -0,0 +1,53 @@ +"""launch_confirm_sample table (启动确认窗 agent 兜底样本: LLM 放行的弹窗样本沉淀) + +Revision ID: launch_confirm_sample_table +Revises: cps_wx_user +Create Date: 2026-06-20 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "launch_confirm_sample_table" +down_revision: Union[str, Sequence[str], None] = "cps_wx_user" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "launch_confirm_sample", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False), + sa.Column("trace_id", sa.String(length=64), nullable=True), + sa.Column("device_id", sa.String(length=64), nullable=True), + sa.Column("host_package", sa.String(length=128), nullable=True), + sa.Column("target_app", sa.String(length=128), nullable=True), + sa.Column("system_locale", sa.String(length=32), nullable=True), + sa.Column("exec_success", sa.Boolean(), nullable=False), + sa.Column("dialog_title", sa.String(length=256), nullable=True), + # PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐 + sa.Column("payload", sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), "postgresql"), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("launch_confirm_sample", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_created_at"), ["created_at"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_trace_id"), ["trace_id"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_device_id"), ["device_id"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_host_package"), ["host_package"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_system_locale"), ["system_locale"], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table("launch_confirm_sample", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_system_locale")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_host_package")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_device_id")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_trace_id")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_created_at")) + + op.drop_table("launch_confirm_sample") diff --git a/alembic/versions/merge_cps_link_and_device_liveness_heads.py b/alembic/versions/merge_cps_link_and_device_liveness_heads.py new file mode 100644 index 0000000..1346b50 --- /dev/null +++ b/alembic/versions/merge_cps_link_and_device_liveness_heads.py @@ -0,0 +1,31 @@ +"""merge cps_link_tables and device_liveness_table heads + +Revision ID: merge_cps_device_liveness +Revises: cps_link_tables, device_liveness_table +Create Date: 2026-06-17 + +并行分支双 head 合并(团队常态:不同 feature 分支各自基于同一父 revision 加迁移, +集成时就出现多 head,`alembic upgrade head` 报 Multiple head revisions): +- cps_link_tables —— main 的 CPS 链路表 +- device_liveness_table —— feat/a11y-liveness-heartbeat 的设备存活心跳表 +两条链彼此独立、无 schema 冲突,这里合并成单 head。纯合并节点,不改表结构。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'merge_cps_device_liveness' +down_revision: Union[str, Sequence[str], None] = ('cps_link_tables', 'device_liveness_table') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/app/admin/main.py b/app/admin/main.py index b718441..eb46a7c 100644 --- a/app/admin/main.py +++ b/app/admin/main.py @@ -14,6 +14,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.admin.routers.ad_audit import router as ad_audit_router +from app.admin.routers.ad_config import router as ad_config_router from app.admin.routers.ad_revenue import router as ad_revenue_router from app.admin.routers.admins import router as admins_router from app.admin.routers.audit import router as audit_router @@ -95,4 +96,5 @@ admin_app.include_router(config_router) admin_app.include_router(comparison_router) admin_app.include_router(cps_router) admin_app.include_router(ad_audit_router) +admin_app.include_router(ad_config_router) admin_app.include_router(ad_revenue_router) diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py index 3271a74..4b1c7c3 100644 --- a/app/admin/repositories/cps.py +++ b/app/admin/repositories/cps.py @@ -9,7 +9,7 @@ from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from uuid import uuid4 -from sqlalchemy import desc, select +from sqlalchemy import desc, func, select from sqlalchemy.orm import Session from app.admin.repositories.queries import _as_utc, offset_paginate @@ -19,6 +19,7 @@ from app.models.cps_activity import CpsActivity from app.models.cps_group import CpsGroup from app.models.cps_link import CpsClick from app.models.cps_order import CpsOrder +from app.models.cps_wx_user import CpsWxUser # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账 _INVALID_STATUS = {"4", "5"} @@ -470,3 +471,47 @@ def group_order_daily( "settled_commission_cents": sum(o.commission_cents or 0 for o in settled), } return out + + +def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]: + """群内微信用户(首次来源 first_group_id=group_id)+ 每人在该群的 visit/copy 次数(领券画像)。 + + 按首次进入倒序。copy 次数 = 该用户在该群点了几次「复制口令」(领券意向),visit = 进落地页次数。 + """ + users = ( + db.execute( + select(CpsWxUser) + .where(CpsWxUser.first_group_id == group_id) + .order_by(desc(CpsWxUser.first_seen)) + .limit(limit) + ) + .scalars() + .all() + ) + if not users: + return [] + openids = [u.openid for u in users] + stat: dict[str, dict] = {} + rows = db.execute( + select(CpsClick.openid, CpsClick.event_type, func.count()) + .where(CpsClick.group_id == group_id) + .where(CpsClick.openid.in_(openids)) + .group_by(CpsClick.openid, CpsClick.event_type) + ).all() + for openid, event_type, cnt in rows: + s = stat.setdefault(openid, {"visit": 0, "copy": 0}) + if event_type == "copy": + s["copy"] = cnt + else: + s["visit"] = cnt + return [ + { + "openid": u.openid, + "nickname": u.nickname, + "headimgurl": u.headimgurl, + "first_seen": u.first_seen, + "visit_count": stat.get(u.openid, {}).get("visit", 0), + "copy_count": stat.get(u.openid, {}).get("copy", 0), + } + for u in users + ] diff --git a/app/admin/routers/ad_config.py b/app/admin/routers/ad_config.py new file mode 100644 index 0000000..c46068c --- /dev/null +++ b/app/admin/routers/ad_config.py @@ -0,0 +1,45 @@ +"""admin 广告配置:读/写穿山甲 app_id / 各场景广告位ID / 验签密钥 / 各场景开关。 + +存在 app_config 表的 ad_config dict(见 repositories/app_config.get_ad_config/set_ad_config)。 +客户端经 /api/v1/platform/ad-config 拉取(不含 reward_mkey)。权限 operator/finance + 审计。 +""" +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Depends, Request + +from app.admin.audit import write_audit +from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role +from app.admin.schemas.ad_config import AdConfigOut, AdConfigUpdate +from app.models.admin import AdminUser +from app.repositories import app_config + +router = APIRouter( + prefix="/admin/api/ad-config", + tags=["admin-ad-config"], + dependencies=[Depends(get_current_admin)], +) + + +@router.get("", response_model=AdConfigOut, summary="广告配置(穿山甲ID/验签密钥/各场景开关)") +def get_ad_config(db: AdminDb) -> AdConfigOut: + return AdConfigOut(**app_config.get_ad_config(db)) + + +@router.patch("", response_model=AdConfigOut, summary="改广告配置(部分更新,带审计)") +def update_ad_config( + body: AdConfigUpdate, + request: Request, + admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))], + db: AdminDb, +) -> AdConfigOut: + data = body.model_dump(exclude_none=True) + app_config.set_ad_config(db, data, admin_id=admin.id, commit=False) + # 审计只记改了哪些字段,不记 mkey 明文(机密) + write_audit( + db, admin, action="ad_config.set", target_type="ad_config", target_id="ad_config", + detail={"changed": sorted(data.keys())}, ip=get_client_ip(request), commit=False, + ) + db.commit() + return AdConfigOut(**app_config.get_ad_config(db)) diff --git a/app/admin/routers/cps.py b/app/admin/routers/cps.py index 6f892dd..a1fd267 100644 --- a/app/admin/routers/cps.py +++ b/app/admin/routers/cps.py @@ -488,3 +488,11 @@ def group_daily( "days": days, "rows": rows, } + + +@router.get("/groups/{group_id}/wx-users", summary="群内微信用户(领券画像:头像/昵称/领券次数)") +def group_wx_users(group_id: int, db: AdminDb) -> dict: + group = cps_repo.get_group(db, group_id) + if group is None: + raise HTTPException(status_code=404, detail="群不存在") + return {"users": cps_repo.group_wx_users(db, group_id=group_id)} diff --git a/app/admin/schemas/ad_config.py b/app/admin/schemas/ad_config.py new file mode 100644 index 0000000..3f5c033 --- /dev/null +++ b/app/admin/schemas/ad_config.py @@ -0,0 +1,30 @@ +"""admin 广告配置 schemas(穿山甲 app_id/各位ID/验签密钥/各场景开关)。""" +from __future__ import annotations + +from pydantic import BaseModel + + +class AdConfigOut(BaseModel): + """admin 读到的完整广告配置(含 reward_mkey;admin 有权见,但绝不经 /platform 下发客户端)。""" + + app_id: str + reward_code_id: str + compare_feed_code_id: str + coupon_feed_code_id: str + reward_mkey: str + reward_enabled: bool + compare_ad_enabled: bool + coupon_ad_enabled: bool + + +class AdConfigUpdate(BaseModel): + """部分更新:只改传入(非 None)字段。空串是合法值(如清空 mkey 回退 .env)。""" + + app_id: str | None = None + reward_code_id: str | None = None + compare_feed_code_id: str | None = None + coupon_feed_code_id: str | None = None + reward_mkey: str | None = None + reward_enabled: bool | None = None + compare_ad_enabled: bool | None = None + coupon_ad_enabled: bool | None = None diff --git a/app/api/internal/launch_confirm.py b/app/api/internal/launch_confirm.py new file mode 100644 index 0000000..5f24d3f --- /dev/null +++ b/app/api/internal/launch_confirm.py @@ -0,0 +1,63 @@ +"""启动确认窗兜底样本内部上报端点(pricebot → app-server)。 + +pricebot 的 launch_confirm_agent 每次靠 LLM 兜底放行一个"静态 PROFILES 没认出"的跨 App +启动确认窗时,把完整样本 POST 到这里落库。**不是给客户端的接口**:不走用户 JWT,靠 +server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。密钥未配置 +(默认空)时直接 503,避免裸奔写端点。 + +研发定期人工把 exec_success=true 的样本沉淀进 pricebot 的 launch_confirm.PROFILES。 +""" +from __future__ import annotations + +import hmac +import logging +from typing import Annotated + +from fastapi import APIRouter, Header, HTTPException, status + +from app.api.deps import DbSession +from app.core.config import settings +from app.repositories import launch_confirm_sample as repo +from app.schemas.launch_confirm_sample import ( + LaunchConfirmSampleIn, + LaunchConfirmSampleOut, +) + +logger = logging.getLogger("shagua.internal.launch_confirm") + +router = APIRouter(prefix="/internal", tags=["internal"]) + + +def _check_secret(x_internal_secret: str | None) -> None: + """共享密钥校验。未配置 → 503(挡裸奔写端点);不匹配 → 401(常量时间比较)。""" + configured = settings.INTERNAL_API_SECRET + if not configured: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="internal api not configured", + ) + if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid internal secret", + ) + + +@router.post( + "/launch-confirm-sample", + response_model=LaunchConfirmSampleOut, + summary="启动确认窗兜底样本上报(pricebot→app-server,落 launch_confirm_sample)", +) +def report_launch_confirm_sample( + payload: LaunchConfirmSampleIn, + db: DbSession, + x_internal_secret: Annotated[str | None, Header()] = None, +) -> LaunchConfirmSampleOut: + _check_secret(x_internal_secret) + sid = repo.insert_sample(db, payload) + logger.info( + "launch_confirm_sample id=%d host=%s locale=%s success=%s trace=%s", + sid, payload.host_package, payload.system_locale, + payload.exec_success, payload.trace_id, + ) + return LaunchConfirmSampleOut(id=sid) diff --git a/app/api/v1/ad.py b/app/api/v1/ad.py index bb05891..cf866b2 100644 --- a/app/api/v1/ad.py +++ b/app/api/v1/ad.py @@ -24,6 +24,7 @@ from app.repositories import ad_ecpm as crud_ecpm from app.repositories import ad_feed_reward as crud_feed from app.repositories import ad_reward as crud_ad from app.repositories import ad_watch as crud_watch +from app.repositories import app_config from app.repositories import signin as crud_signin from app.schemas.ad import ( AdRewardStatusOut, @@ -80,14 +81,22 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut: 验签失败 403(留给真请求重试);参数缺/坏或 user 不存在 → is_verify=false + reason(不发,不重试)。 granted / capped → is_verify=true + reason=0。 """ - if not settings.pangle_callback_configured: + if not settings.PANGLE_CALLBACK_ENABLED: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="pangle callback not configured" ) params = dict(request.query_params) - if not pangle.verify_callback_sign_any(params, settings.pangle_reward_secrets): + # 验签密钥:admin 后台配的 reward_mkey 优先,.env 的 PANGLE_REWARD_SECRET* 兜底(兼容/过渡)。 + # 换激励位时后台同步换 mkey 即可,无需改 .env。两者都空才视为未配置。 + ad_mkey = app_config.get_ad_config(db).get("reward_mkey") or "" + secrets = ([ad_mkey] if ad_mkey else []) + settings.pangle_reward_secrets + if not secrets: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="pangle callback not configured" + ) + if not pangle.verify_callback_sign_any(params, secrets): logger.warning("pangle callback bad sign trans_id=%s", params.get("trans_id")) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="bad sign") diff --git a/app/api/v1/cps_redirect.py b/app/api/v1/cps_redirect.py index 49a061d..ee7aa0b 100644 --- a/app/api/v1/cps_redirect.py +++ b/app/api/v1/cps_redirect.py @@ -8,16 +8,22 @@ from __future__ import annotations import html import json +import logging from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse from sqlalchemy.orm import Session from app.core import media +from app.core.config import settings from app.db.session import get_db +from app.integrations import wx_oauth from app.models.cps_activity import CpsActivity from app.models.cps_link import CpsLink from app.repositories import cps_link as cps_link_repo +from app.repositories import cps_wx_user as cps_wx_user_repo + +logger = logging.getLogger("shagua.cps_redirect") router = APIRouter(tags=["cps-redirect"]) @@ -27,6 +33,16 @@ _FALLBACK_URL = "https://www.meituan.com/" # 淘宝活动未设图时的兜底主视觉(存量已回填,基本只在老 link/异常时触发) _DEFAULT_TAOBAO_IMAGE = "/media/taobao_landing.jpg" +# 微信网页授权拿到的用户标识 cookie(种在 coupon 域,30 天免重复授权) +_WX_OPENID_COOKIE = "wx_openid" +_WX_UINFO_COOKIE = "wx_uinfo" # "1" = 已拿过昵称头像(userinfo),点领券不再跳授权 +_COOKIE_MAX_AGE = 30 * 86400 + + +def _is_wechat(request: Request) -> bool: + """是否微信内置浏览器(网页授权只在微信内有意义,外部浏览器不跳授权)。""" + return "micromessenger" in (request.headers.get("user-agent", "").lower()) + def _client_ip(request: Request) -> str | None: xff = request.headers.get("x-forwarded-for") @@ -35,11 +51,13 @@ def _client_ip(request: Request) -> str | None: return request.client.host if request.client else None -def _record(db: Session, link: CpsLink, request: Request, event_type: str) -> None: +def _record( + db: Session, link: CpsLink, request: Request, event_type: str, openid: str | None = None +) -> None: try: cps_link_repo.record_click( db, link=link, ip=_client_ip(request), - ua=request.headers.get("user-agent"), event_type=event_type, + ua=request.headers.get("user-agent"), event_type=event_type, openid=openid, ) except Exception: pass # 记点击失败不阻断 @@ -53,38 +71,89 @@ def wx_mp_domain_verify() -> PlainTextResponse: return PlainTextResponse("F7wnRQ7xPbhVOWC8") -@router.get("/c/{code}", summary="群发短链落地(记点击 + 跳转/淘宝落地页)") +@router.get("/c/{code}", summary="群发短链落地(微信授权拿 openid + 记点击 + 跳转/淘宝落地页)") def cps_landing(code: str, request: Request, db: Session = Depends(get_db)): link = cps_link_repo.get_by_code(db, code) if link is None: return RedirectResponse(_FALLBACK_URL, status_code=302) - _record(db, link, request, "visit") + openid = request.cookies.get(_WX_OPENID_COOKIE) + # 微信内 + 还没拿到 openid + 配了服务号 → 先 base 静默授权拿 openid,再 302 回本页。 + # 非微信 / 已有 cookie / 未配服务号 → 直接走原逻辑(openid 可能为 None,绝不阻断领券)。 + if openid is None and settings.wx_oauth_active and _is_wechat(request): + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}") + return RedirectResponse(auth_url, status_code=302) + _record(db, link, request, "visit", openid=openid) if link.platform == "taobao": activity = db.get(CpsActivity, link.activity_id) image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE - return HTMLResponse(_taobao_landing_html(link.target_url, image_url)) + has_uinfo = request.cookies.get(_WX_UINFO_COOKIE) == "1" + return HTMLResponse( + _taobao_landing_html(link.target_url, image_url, code, openid, has_uinfo) + ) # 美团短链 / 京东链接:直接 302 跳 return RedirectResponse(link.target_url, status_code=302) -@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件)") +@router.get("/wx/oauth/cb", include_in_schema=False) +def wx_oauth_cb(code: str, state: str, db: Session = Depends(get_db)): + """微信网页授权回调。state='base:{原code}'(静默拿 openid) 或 'uinfo:{原code}'(补昵称头像)。 + 换 openid(+userinfo)→ upsert 用户 → 种 cookie → 302 回落地页。任何失败兜底回落地页, + 绝不阻断用户领券。""" + kind, _, orig_code = state.partition(":") + try: + token = wx_oauth.exchange_code(code) # {openid, access_token, scope, ...} + openid = token["openid"] + link = cps_link_repo.get_by_code(db, orig_code) + group_id = link.group_id if link else None + nickname = headimgurl = unionid = None + if kind == "uinfo" and "userinfo" in (token.get("scope") or ""): + info = wx_oauth.get_userinfo(token["access_token"], openid) + nickname, headimgurl, unionid = ( + info.get("nickname"), info.get("headimgurl"), info.get("unionid"), + ) + cps_wx_user_repo.upsert( + db, openid=openid, code=orig_code, group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + except Exception: + logger.exception("[wx_oauth] callback failed state=%s", state) + return RedirectResponse(f"/c/{orig_code}" if orig_code else _FALLBACK_URL, status_code=302) + # userinfo 授权回来带 ?authed=1,落地页据此自动触发复制(把"点领券→授权→复制"衔接为一步) + target = f"/c/{orig_code}" + ("?authed=1" if kind == "uinfo" else "") + resp = RedirectResponse(target, status_code=302) + resp.set_cookie(_WX_OPENID_COOKIE, openid, max_age=_COOKIE_MAX_AGE, httponly=True, samesite="lax") + if nickname: + resp.set_cookie(_WX_UINFO_COOKIE, "1", max_age=_COOKIE_MAX_AGE, samesite="lax") + return resp + + +@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件,带 openid)") def cps_copy(code: str, request: Request, db: Session = Depends(get_db)) -> dict: link = cps_link_repo.get_by_code(db, code) if link is not None: - _record(db, link, request, "copy") + _record(db, link, request, "copy", openid=request.cookies.get(_WX_OPENID_COOKIE)) return {"ok": True} -def _taobao_landing_html(token: str, image_url: str) -> str: +def _taobao_landing_html( + token: str, image_url: str, code: str, openid: str | None, has_uinfo: bool +) -> str: """淘宝落地页 H5(主视觉图按活动传入,复制淘口令按钮在 75% 屏高)。 - image_url 转绝对(落地页 coupon 域,相对也可达,绝对更稳)后经 html.escape 嵌入 - (防属性注入);token 经 json.dumps 安全嵌入 JS。 + image_url 经 html.escape 嵌入 (防注入);token 经 json.dumps 安全嵌入 JS。 + uinfo_url:已有 openid 但还没拿过昵称头像时,生成 userinfo 授权链接 —— 用户点「领券」 + 时先跳它(交互触发,避免微信快照页),授权回来自动复制。已拿过 / 无 openid 则为空,直接复制。 """ safe_img = html.escape(media.to_abs_media_url(image_url) or image_url, quote=True) + uinfo_url = "" + if openid and not has_uinfo and settings.wx_oauth_active: + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + uinfo_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_userinfo", f"uinfo:{code}") return ( _TAOBAO_HTML .replace("__IMAGE_URL__", safe_img) + .replace("__UINFO_URL__", json.dumps(uinfo_url)) .replace("__TOKEN_JS__", json.dumps(token)) ) @@ -112,6 +181,7 @@ body{font-family:-apple-system,"PingFang SC",sans-serif;background:#fff0ef;color
""" diff --git a/app/api/v1/device.py b/app/api/v1/device.py new file mode 100644 index 0000000..56a5a85 --- /dev/null +++ b/app/api/v1/device.py @@ -0,0 +1,96 @@ +"""设备注册 / 心跳 endpoint(无障碍保护存活检测)。 + +路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。 + POST /register 注册设备 / 更新 registration_id(App 前台、拿到 push token 时调) + POST /heartbeat 上报心跳(无障碍服务存活时周期调,刷新存活) + +后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。 +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter + +from app.api.deps import CurrentUser, DbSession +from app.repositories import device as device_repo +from app.schemas.device import ( + DeviceOut, + DeviceRegisterRequest, + HeartbeatRequest, + LivenessAckRequest, + LivenessOut, + OkResponse, +) + +logger = logging.getLogger("shagua.device") + +router = APIRouter(prefix="/api/v1/device", tags=["device"]) + + +@router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token") +def register_device( + req: DeviceRegisterRequest, + user: CurrentUser, + db: DbSession, +) -> DeviceOut: + device = device_repo.register_or_update( + db, + user_id=user.id, + device_id=req.device_id, + registration_id=req.registration_id, + platform=req.platform, + app_version=req.app_version, + ) + logger.info( + "device register user_id=%d device_id=%s reg=%s", + user.id, + req.device_id, + bool(req.registration_id), + ) + return DeviceOut.model_validate(device) + + +@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳") +def report_heartbeat( + req: HeartbeatRequest, + user: CurrentUser, + db: DbSession, +) -> OkResponse: + device_repo.touch_heartbeat( + db, + user_id=user.id, + device_id=req.device_id, + accessibility_enabled=req.accessibility_enabled, + registration_id=req.registration_id, + ) + return OkResponse() + + +@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)") +def get_liveness( + device_id: str, + user: CurrentUser, + db: DbSession, +) -> LivenessOut: + """客户端进 App 时拉取: 本机是否被判定掉线过(kill_alert_pending)。 + + 设备从未注册过 → 返回默认(无告警)。命中 → 客户端弹「开启自启动」引导, 之后调 /liveness/ack 清。 + 见 spec: spec/accessibility-liveness-pull-prompt.md。 + """ + device = device_repo.get_device(db, user_id=user.id, device_id=device_id) + if device is None: + return LivenessOut() + return LivenessOut.model_validate(device) + + +@router.post("/liveness/ack", response_model=OkResponse, summary="确认已提醒(清掉线告警)") +def ack_liveness( + req: LivenessAckRequest, + user: CurrentUser, + db: DbSession, +) -> OkResponse: + """客户端已弹过「开启自启动」引导 → 清掉「待提醒」标记(下次真掉线 worker 再置)。""" + device_repo.ack_kill_alert(db, user_id=user.id, device_id=req.device_id) + return OkResponse() diff --git a/app/api/v1/platform.py b/app/api/v1/platform.py index f839a58..31908bd 100644 --- a/app/api/v1/platform.py +++ b/app/api/v1/platform.py @@ -17,6 +17,7 @@ from app.repositories import app_config from app.repositories import ops_marquee as marquee_crud from app.repositories import ops_stat as crud from app.schemas.platform import ( + AdConfigPublicOut, AppFlagsOut, AppVersionOut, PlatformStatsOut, @@ -54,6 +55,22 @@ def flags(db: DbSession) -> AppFlagsOut: ) +@router.get("/ad-config", response_model=AdConfigPublicOut, summary="客户端拉广告配置(穿山甲ID+场景开关,不鉴权)") +def ad_config(db: DbSession) -> AdConfigPublicOut: + """客户端启动/每场广告前拉,缓存后用:app_id + 各位ID + 各场景开关。 + 不含验签密钥;空库回退默认(=客户端内置值,维持现状)。""" + c = app_config.get_ad_config(db) + return AdConfigPublicOut( + app_id=c["app_id"], + reward_code_id=c["reward_code_id"], + compare_feed_code_id=c["compare_feed_code_id"], + coupon_feed_code_id=c["coupon_feed_code_id"], + reward_enabled=c["reward_enabled"], + compare_ad_enabled=c["compare_ad_enabled"], + coupon_ad_enabled=c["coupon_ad_enabled"], + ) + + @router.get("/app-version", response_model=AppVersionOut, summary="最新 App 版本(OTA 检查更新,不鉴权)") def app_version(db: DbSession) -> AppVersionOut: """客户端启动 / 手动检查更新时拉取。不鉴权:版本信息非敏感,且检查更新可能在登录前。 diff --git a/app/api/v1/user.py b/app/api/v1/user.py index 6be689f..dc23388 100644 --- a/app/api/v1/user.py +++ b/app/api/v1/user.py @@ -18,6 +18,7 @@ from app.api.deps import CurrentUser, DbSession from app.core import media from app.repositories import onboarding as onboarding_repo from app.repositories import user as user_repo +from app.repositories import wallet as wallet_repo from app.schemas.auth import UserOut from app.schemas.user import ( OkResponse, @@ -87,6 +88,17 @@ def onboarding_status( @router.delete("", response_model=OkResponse, summary="注销账号(软删除)") def delete_account(user: CurrentUser, db: DbSession) -> OkResponse: + # 资金前置校验:注销是软删 + 匿名化,余额一旦留在 deleted 行就锁死(既退不出、新号也 + # 拿不回)。有可提现现金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。 + # (pending 提现转账已在途、对账 worker 不依赖 user.status 照常到账,故不拦 pending。) + if wallet_repo.get_cash_balance_cents(db, user.id) > 0: + raise HTTPException( + status_code=409, detail="账户还有未提现的现金余额,请先提现后再注销" + ) + if wallet_repo.has_reviewing_withdraw(db, user.id): + raise HTTPException( + status_code=409, detail="有提现正在审核中,请等审核完成后再注销" + ) media.delete_avatar(user.avatar_url) user_repo.soft_delete_account(db, user) logger.info("delete account user_id=%d", user.id) diff --git a/app/core/config.py b/app/core/config.py index bfc4139..900d3e8 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -66,6 +66,32 @@ class Settings(BaseSettings): JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify" JG_REQUEST_TIMEOUT_SEC: int = 15 + # ===== 极光推送 JPush(无障碍保护掉线告警)===== + # 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d(android build.gradle manifestPlaceholder)。 + # 若推送与一键登录/短信是同一个极光应用(大概率),JPUSH_* 留空即自动回退到 JG_*。 + # 否则在 .env 单独配 JPUSH_APP_KEY / JPUSH_MASTER_SECRET(对应那个 push appkey)。 + JPUSH_APP_KEY: str = "" + JPUSH_MASTER_SECRET: str = "" + JPUSH_PUSH_ENDPOINT: str = "https://api.jpush.cn/v3/push" + + # 无障碍保护存活监控后台任务 + HEARTBEAT_MONITOR_ENABLED: bool = True # 总开关 + HEARTBEAT_TIMEOUT_MINUTES: int = 10 # 多久没心跳算掉线(≈3 个客户端心跳周期) + HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期 + + @property + def jpush_app_key(self) -> str: + return self.JPUSH_APP_KEY or self.JG_APP_KEY + + @property + def jpush_master_secret(self) -> str: + return self.JPUSH_MASTER_SECRET or self.JG_MASTER_SECRET + + @property + def jpush_configured(self) -> bool: + """推送凭证齐全(缺则 monitor 只扫不发,不报错)。""" + return bool(self.jpush_app_key and self.jpush_master_secret) + # ===== 短信 ===== SMS_MOCK: bool = True SMS_CODE_TTL_SEC: int = 300 @@ -95,6 +121,25 @@ class Settings(BaseSettings): """美团 CPS 凭证齐全(缺则接口返空,而非 502)。""" return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET) + # ===== 微信服务号(网页授权) ===== + # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。 + # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 + WX_MP_APPID: str = "" + WX_MP_SECRET: str = "" + # 落地页微信网页授权【总开关】。默认关:服务号认证审核中/未就绪时,落地页走原逻辑 + # (不跳授权、不拿 openid),整套微信代码保留。审核通过后 .env 置 true + 重启即启用,无需改代码。 + WX_MP_OAUTH_ENABLED: bool = False + + @property + def wx_mp_configured(self) -> bool: + """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" + return bool(self.WX_MP_APPID and self.WX_MP_SECRET) + + @property + def wx_oauth_active(self) -> bool: + """落地页是否真正发起微信授权 = 凭证齐全 且 总开关开。""" + return self.wx_mp_configured and self.WX_MP_OAUTH_ENABLED + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py new file mode 100644 index 0000000..6072d26 --- /dev/null +++ b/app/core/heartbeat_monitor_worker.py @@ -0,0 +1,148 @@ +"""无障碍保护存活监控后台任务。 + +周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了), +**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机 +推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。 +结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。 + +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import time +from collections.abc import Iterator +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy.exc import SQLAlchemyError + +from app.core.config import settings +from app.db.session import SessionLocal +from app.repositories import device as device_repo + +logger = logging.getLogger("shagua.heartbeat_monitor") +_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.lock" + + +def _touch_lock() -> None: + with contextlib.suppress(FileNotFoundError): + os.utime(_LOCK_PATH, None) + + +@contextlib.contextmanager +def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]: + """同机多进程保护:同一时间只允许一个监控 worker 运行。""" + _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) + fd: int | None = None + try: + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + try: + age = time.time() - _LOCK_PATH.stat().st_mtime + except FileNotFoundError: + age = stale_after_sec + 1 + if age > stale_after_sec: + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + fd = None + + if fd is None: + yield False + return + + os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii")) + yield True + finally: + if fd is not None: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + + +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() + return int((ref - last).total_seconds()) + + +def _scan_once(timeout_minutes: int) -> dict: + """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。 + + 本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端, + 并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。 + """ + notified = 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 将弹「开启自启动」引导(后置检测);推送本期未接】", + device.user_id, + device.device_id, + silent if silent is not None else "?", + timeout_minutes, + ) + device_repo.mark_notified(db, device_id_pk=device.id) + notified += 1 + return {"checked": len(overdue), "notified": notified} + + +async def _run_loop() -> None: + interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC)) + timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES)) + lock_stale_after = max(interval * 3, 600) + with _single_instance_lock(lock_stale_after) as lock_acquired: + if not lock_acquired: + logger.warning("heartbeat monitor skipped: another worker owns lock") + return + await _run_locked_loop(interval, timeout_minutes) + + +async def _run_locked_loop(interval: int, timeout_minutes: int) -> None: + logger.info( + "heartbeat monitor started interval=%ss timeout=%sm", + interval, + timeout_minutes, + ) + try: + while True: + try: + _touch_lock() + result = await asyncio.to_thread(_scan_once, timeout_minutes) + if result["notified"]: + logger.info("heartbeat monitor result=%s", result) + except SQLAlchemyError: + logger.exception("heartbeat monitor db error") + except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出 + logger.exception("heartbeat monitor unexpected error") + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("heartbeat monitor stopped") + raise + + +def start_heartbeat_monitor() -> asyncio.Task | None: + if not settings.HEARTBEAT_MONITOR_ENABLED: + logger.info("heartbeat monitor disabled") + return None + return asyncio.create_task(_run_loop(), name="heartbeat-monitor") + + +async def stop_heartbeat_monitor(task: asyncio.Task | None) -> None: + if task is None: + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task diff --git a/app/integrations/jpush.py b/app/integrations/jpush.py new file mode 100644 index 0000000..5266ac9 --- /dev/null +++ b/app/integrations/jpush.py @@ -0,0 +1,94 @@ +"""极光推送 JPush(无障碍保护掉线告警)。 + +调极光 Push REST v3 /v3/push 给指定 registration_id 推一条通知。鉴权复用极光 Basic Auth +(appKey:masterSecret),与一键登录/短信同模式(见 integrations/jiguang.py、sms.py)。 + +凭证:settings.jpush_app_key / jpush_master_secret(JPUSH_* 留空时自动回退到 JG_*, +若推送与一键登录是同一极光应用)。客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。 +厂商通道(App 被杀也能到达)由极光后台 + 客户端插件负责,本服务只管调 push 接口。 +""" +from __future__ import annotations + +import base64 +import logging + +import httpx + +from app.core.config import settings + +logger = logging.getLogger("shagua.jpush") + + +class JPushError(Exception): + """极光推送调用失败。""" + + +class JPushNotConfiguredError(JPushError): + """缺 appKey / masterSecret。""" + + +def push_to_registration_ids( + registration_ids: list[str], + *, + title: str, + alert: str, + extras: dict | None = None, +) -> dict: + """给一批 registration_id 推送通知 + 透传消息。失败抛 JPushError。 + + Returns: 极光响应 JSON(含 sendno / msg_id)。 + """ + if not settings.jpush_configured: + raise JPushNotConfiguredError( + "JPush 未配置(缺 JPUSH_APP_KEY/JPUSH_MASTER_SECRET,且 JG_* 也为空)" + ) + reg_ids = [r for r in registration_ids if r] + if not reg_ids: + raise JPushError("registration_ids 为空") + + auth_b64 = base64.b64encode( + f"{settings.jpush_app_key}:{settings.jpush_master_secret}".encode() + ).decode() + payload = { + "platform": ["android"], + "audience": {"registration_id": reg_ids}, + "notification": { + "android": { + "alert": alert, + "title": title, + "priority": 1, + "extras": extras or {}, + }, + }, + "message": { + "msg_content": alert, + "title": title, + "content_type": "text", + "extras": extras or {}, + }, + "options": {"time_to_live": 86400, "apns_production": True}, + } + + try: + resp = httpx.post( + settings.JPUSH_PUSH_ENDPOINT, + json=payload, + headers={ + "Content-Type": "application/json", + "Authorization": f"Basic {auth_b64}", + }, + timeout=settings.JG_REQUEST_TIMEOUT_SEC, + ) + except httpx.HTTPError as e: + raise JPushError(f"jpush 网络错误: {e}") from e + + if resp.status_code != 200: + body = resp.text[:300] + logger.error("[JPush] http=%s body=%s", resp.status_code, body) + raise JPushError(f"jpush http {resp.status_code}") + + data = resp.json() + if not data.get("sendno") and not data.get("msg_id"): + logger.error("[JPush] unexpected response: %s", data) + raise JPushError(f"jpush 响应异常: {data}") + return data diff --git a/app/integrations/wx_oauth.py b/app/integrations/wx_oauth.py new file mode 100644 index 0000000..0af50d6 --- /dev/null +++ b/app/integrations/wx_oauth.py @@ -0,0 +1,75 @@ +"""微信服务号网页授权:构造授权链接 + code 换 openid + 拉 userinfo。 + +用于 CPS 落地页(coupon.shaguabijia.com)在微信内拿用户身份。流程(见微信文档): + 授权链接(snsapi_base|snsapi_userinfo) → 回调带 code → sns/oauth2/access_token 换 openid + → (userinfo 时)sns/userinfo 拿昵称头像。 + +注意: + - 这里的 access_token 是【网页授权 token】,与基础 access_token(cgi-bin/token)不同, + 走 sns/* 接口,不需要 IP 白名单。 + - secret 只在服务器,不下发客户端。 + - 用 WX_MP_APPID/SECRET(已认证服务号),区别于 WECHAT_APP_ID(App 移动应用)。 +""" +from __future__ import annotations + +from urllib.parse import quote + +import httpx + +from app.core.config import settings + +_API = "https://api.weixin.qq.com" +_AUTHORIZE = "https://open.weixin.qq.com/connect/oauth2/authorize" +_TIMEOUT = 10 + + +class WxOauthError(Exception): + """网页授权调用失败(凭证/网络/code 失效等)。调用方兜底为无 openid。""" + + +def build_authorize_url(redirect_uri: str, scope: str, state: str) -> str: + """构造网页授权跳转链接。scope: 'snsapi_base'(静默,仅 openid) | 'snsapi_userinfo'(昵称头像)。 + + 微信要求 redirect_uri urlEncode、参数顺序固定、结尾 #wechat_redirect。 + """ + return ( + f"{_AUTHORIZE}?appid={settings.WX_MP_APPID}" + f"&redirect_uri={quote(redirect_uri, safe='')}" + f"&response_type=code&scope={scope}&state={quote(state, safe='')}" + f"#wechat_redirect" + ) + + +def exchange_code(code: str) -> dict: + """code 换网页授权 access_token + openid。返回含 openid/access_token/scope(/unionid)。 + + code 5 分钟内一次性有效。失败(errcode)抛 WxOauthError。 + """ + resp = httpx.get( + f"{_API}/sns/oauth2/access_token", + params={ + "appid": settings.WX_MP_APPID, + "secret": settings.WX_MP_SECRET, + "code": code, + "grant_type": "authorization_code", + }, + timeout=_TIMEOUT, + ) + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"exchange_code failed: {data}") + return data + + +def get_userinfo(access_token: str, openid: str) -> dict: + """拉用户信息(nickname/headimgurl/unionid)。需 scope=snsapi_userinfo 的 access_token。""" + resp = httpx.get( + f"{_API}/sns/userinfo", + params={"access_token": access_token, "openid": openid, "lang": "zh_CN"}, + timeout=_TIMEOUT, + ) + resp.encoding = "utf-8" # 昵称含中文/emoji,强制 utf-8 避免乱码 + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"get_userinfo failed: {data}") + return data diff --git a/app/main.py b/app/main.py index 702f19b..f4fde37 100644 --- a/app/main.py +++ b/app/main.py @@ -20,8 +20,10 @@ from app.api.v1.compare import router as compare_router from app.api.v1.compare_milestone import router as compare_milestone_router from app.api.v1.compare_record import router as compare_record_router from app.api.v1.coupon import router as coupon_router +from app.api.v1.device import router as device_router from app.api.v1.cps_redirect import router as cps_redirect_router from app.api.internal.app_version import router as internal_app_version_router +from app.api.internal.launch_confirm import router as internal_launch_confirm_router from app.api.internal.price import router as internal_price_router from app.api.internal.store import router as internal_store_router from app.api.v1.feedback import router as feedback_router @@ -37,6 +39,10 @@ from app.api.v1.user import router as user_router from app.api.v1.wallet import router as wallet_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings +from app.core.heartbeat_monitor_worker import ( + start_heartbeat_monitor, + stop_heartbeat_monitor, +) from app.core.logging import setup_logging from app.core.withdraw_reconcile_worker import ( start_withdraw_reconcile_worker, @@ -58,9 +64,11 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: settings.DATABASE_URL.split("://", 1)[0], ) reconcile_task = start_withdraw_reconcile_worker() + heartbeat_task = start_heartbeat_monitor() try: yield finally: + await stop_heartbeat_monitor(heartbeat_task) await stop_withdraw_reconcile_worker(reconcile_task) logger.info("shutting down") @@ -93,6 +101,7 @@ app.include_router(user_router) app.include_router(feedback_router) app.include_router(invite_router) app.include_router(coupon_router) +app.include_router(device_router) app.include_router(compare_router) app.include_router(compare_record_router) app.include_router(compare_milestone_router) @@ -109,6 +118,7 @@ app.include_router(report_router) app.include_router(internal_price_router) app.include_router(internal_store_router) app.include_router(internal_app_version_router) +app.include_router(internal_launch_confirm_router) app.include_router(platform_router) # CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团) app.include_router(cps_redirect_router) diff --git a/app/models/__init__.py b/app/models/__init__.py index 1955e31..f3491f1 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -10,7 +10,9 @@ from app.models.cps_activity import CpsActivity # noqa: F401 from app.models.cps_group import CpsGroup # noqa: F401 from app.models.cps_link import CpsClick, CpsLink # noqa: F401 from app.models.cps_order import CpsOrder # noqa: F401 +from app.models.cps_wx_user import CpsWxUser # noqa: F401 from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401 +from app.models.device import Device # noqa: F401 from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, CouponDailyCompletion, @@ -19,6 +21,7 @@ from app.models.coupon_state import ( # noqa: F401 from app.models.feedback import Feedback # noqa: F401 from app.models.invite import InviteRelation # noqa: F401 from app.models.invite_fingerprint import InviteFingerprint # noqa: F401 +from app.models.launch_confirm_sample import LaunchConfirmSample # noqa: F401 from app.models.meituan_coupon import MeituanCoupon # noqa: F401 from app.models.onboarding import OnboardingCompletion # noqa: F401 from app.models.ops_marquee_seed import OpsMarqueeSeed # noqa: F401 diff --git a/app/models/cps_link.py b/app/models/cps_link.py index 2702c8a..bb88e6b 100644 --- a/app/models/cps_link.py +++ b/app/models/cps_link.py @@ -49,6 +49,8 @@ class CpsClick(Base): ) ip: Mapped[str | None] = mapped_column(String(64), nullable=True) ua: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计 + openid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) clicked_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True, nullable=False ) diff --git a/app/models/cps_wx_user.py b/app/models/cps_wx_user.py new file mode 100644 index 0000000..669b26c --- /dev/null +++ b/app/models/cps_wx_user.py @@ -0,0 +1,42 @@ +"""CPS 落地页微信用户(cps_wx_user)。 + +用户在微信内打开群发短链落地页 /c/{code},经服务号网页授权: + - base 静默:拿 openid(唯一标识,统计主力) + - userinfo(点领券触发):补 nickname/headimgurl/unionid + +按 openid 唯一,记录首次来源群(first_group_id),用于群内用户级统计(谁领了券)。 +下单归因到人需 user-level sid(另一期),本表只承载身份 + 领券/点击侧。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +class CpsWxUser(Base): + __tablename__ = "cps_wx_user" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + # 服务号下用户唯一标识(网页授权拿到) + openid: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + # 开放平台 unionid(服务号绑开放平台 + scope=userinfo 才有);跨 App/服务号统一用户 + unionid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + # 昵称/头像:userinfo 授权后才有(base 阶段为空) + nickname: Mapped[str | None] = mapped_column(String(128), nullable=True) + headimgurl: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 首次进入来源(从哪个群的链接授权进来) + first_code: Mapped[str | None] = mapped_column(String(16), nullable=True) + first_group_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + first_seen: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + last_seen: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False + ) + + def __repr__(self) -> str: # pragma: no cover + return f"" diff --git a/app/models/device.py b/app/models/device.py new file mode 100644 index 0000000..290bc9f --- /dev/null +++ b/app/models/device.py @@ -0,0 +1,89 @@ +"""设备表(无障碍保护存活检测 + 极光推送)。 + +每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。 +客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报 +registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、 +现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。 + +liveness_state 状态机(防刷屏,一次掉线只推一条): + unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送) + 心跳恢复时 handler 重置回 alive。 +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Integer, + String, + UniqueConstraint, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +class Device(Base): + __tablename__ = "device" + __table_args__ = ( + UniqueConstraint("user_id", "device_id", name="uq_device_user_device"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column( + Integer, ForeignKey("user.id"), index=True, nullable=False + ) + # 客户端 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: Mapped[str | None] = mapped_column(String(64), nullable=True) + platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android") + app_version: Mapped[str | None] = mapped_column(String(32), nullable=True) + + # 收到过 service 心跳即 true(=该设备开过无障碍,功能对它有意义) + ever_protected: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线 + last_heartbeat_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), index=True, nullable=True + ) + # 最近一次上报的无障碍开关状态(观测用) + last_report_protection_on: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + # unknown / alive / silent / notified + liveness_state: Mapped[str] = mapped_column( + String(16), nullable=False, default="unknown" + ) + # 最近一次推送告警时间 + notified_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # 掉线告警「待客户端提醒」标记(后置检测 pull 版, 见 spec accessibility-liveness-pull-prompt.md)。 + # 与 liveness_state 解耦: worker 检出掉线即置 True; touch_heartbeat(心跳恢复)不动它 + # → 规避「服务随 App 重启先发心跳、state 被重置回 alive → 客户端进 App 漏看」竞态; 只由客户端 ack 清。 + kill_alert_pending: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + def __repr__(self) -> str: # pragma: no cover + return ( + f"" + ) diff --git a/app/models/launch_confirm_sample.py b/app/models/launch_confirm_sample.py new file mode 100644 index 0000000..de225b0 --- /dev/null +++ b/app/models/launch_confirm_sample.py @@ -0,0 +1,61 @@ +"""启动确认窗 agent 兜底样本表(launch_confirm_sample)。 + +pricebot 的 launch_confirm_agent 每次"静态 PROFILES 没认出、靠 LLM 兜底放行"一个跨 App +启动确认窗(繁体 / 英文 / 小语种 / ROM 改版)时,把完整样本 server→server 落这里。研发定期 +**人工**把 exec_success=true 的样本沉淀进 pricebot 的 launch_confirm.PROFILES(静态快路径), +让以后同款窗不必再调 LLM。 + +设计: +- 与登录无关、不鉴权(server→server 共享密钥头 X-Internal-Secret)。 +- **都上报、不去重**(by design):同一种窗多台机器上报多条,研发按 host_package / locale + 聚合后人工筛。 +- 大字段(弹窗树快照 + LLM plan + 设备信息)塞 payload(JSONB)兜底,免得加字段就迁移 schema。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, Boolean, DateTime, Integer, String, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + +# PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON(同 price_observation)。 +_JSON = JSON().with_variant(JSONB(), "postgresql") + + +class LaunchConfirmSample(Base): + __tablename__ = "launch_confirm_sample" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), index=True, nullable=False + ) + + # pricebot 侧 trace_id:回指原始 trace,能去 price.shaguabijia.com 查完整上下文。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + device_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + + # 弹窗宿主包(= 触发判据, 也是研发沉淀进 PROFILES.host_packages 的键)。按它聚合人工筛。 + host_package: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True) + # 被打开的目标 App 包名(标题里"想要打开 X"的 X 是变量, 据此把它从 sign 里挖空)。 + target_app: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 系统语言地区(zh-TW / en-US…)—— 多语言问题的核心维度。 + # ⚠️ 前端 AgentStepRequest 暂未上报设备信息 → 现恒 None,待前端加 device_info 字段后回填。 + system_locale: Mapped[str | None] = mapped_column(String(32), index=True, nullable=True) + + # 这次兜底是否成功放行(弹窗消失=True;放弃 / LLM 没认出=False)。研发只沉淀 True 的。 + exec_success: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + # 弹窗标题全文(研发据此提 launch_sign),如"「傻瓜比价」想要開啟「京東」"。 + dialog_title: Mapped[str | None] = mapped_column(String(256), nullable=True) + + # 大 JSON:完整样本。{is_launch_confirm, llm_model, llm_reason, plan, dialog_tree, + # device_info: {screen, locale, brand, model, android_version, rom_version}} + payload: Mapped[dict | None] = mapped_column(_JSON, nullable=True) + + def __repr__(self) -> str: # pragma: no cover + return ( + f"" + ) diff --git a/app/repositories/app_config.py b/app/repositories/app_config.py index b6f67a4..40bfc80 100644 --- a/app/repositories/app_config.py +++ b/app/repositories/app_config.py @@ -91,3 +91,51 @@ def set_app_version(db: Session, data: dict, *, commit: bool = True) -> AppConfi else: db.flush() return row + + +# ── 穿山甲广告配置(admin 可配,客户端动态拉)────────────────────────────────── +# 同 app_version:结构化 dict,复用 AppConfig 表不进 CONFIG_DEFS(它是一组关联配置,不是散项)。 +# 默认值 = 客户端当前内置的硬编码 ID(AdConfig.kt),空库时下发默认 = 维持现状。 +# reward_mkey 是 GroMore S2S 验签密钥(机密),只后端验签用、绝不下发客户端(见 platform/ad-config)。 +AD_CONFIG_KEY = "ad_config" +_AD_CONFIG_DEFAULTS: dict[str, Any] = { + "app_id": "5830519", # 穿山甲应用ID(正式) + "reward_code_id": "104099389", # 福利页激励视频位 + "compare_feed_code_id": "104090333", # 比价信息流位 + "coupon_feed_code_id": "104090333", # 领券信息流位(初始同比价,运营可拆) + "reward_mkey": "", # 激励位 GroMore 验签密钥(空则回退 .env PANGLE_REWARD_SECRET*) + "reward_enabled": True, # 福利激励视频开关 + "compare_ad_enabled": True, # 比价广告开关 + "coupon_ad_enabled": True, # 领券广告开关 +} + + +def get_ad_config(db: Session) -> dict: + """读广告配置。DB 无则返回默认(=客户端内置值,维持现状);DB 有则与默认 merge + (新增字段向后兼容,老记录缺的字段用默认补)。""" + row = db.get(AppConfig, AD_CONFIG_KEY) + merged = dict(_AD_CONFIG_DEFAULTS) + if row is not None and isinstance(row.value, dict): + merged.update(row.value) + return merged + + +def set_ad_config(db: Session, data: dict, *, admin_id: int, commit: bool = True) -> AppConfig: + """admin 写广告配置。与现有值 merge(部分更新),只接受已知字段(防脏键)。""" + row = db.get(AppConfig, AD_CONFIG_KEY) + merged = dict(_AD_CONFIG_DEFAULTS) + if row is not None and isinstance(row.value, dict): + merged.update(row.value) + merged.update({k: v for k, v in data.items() if k in _AD_CONFIG_DEFAULTS}) + if row is None: + row = AppConfig(key=AD_CONFIG_KEY, value=merged, updated_by_admin_id=admin_id) + db.add(row) + else: + row.value = merged + row.updated_by_admin_id = admin_id + if commit: + db.commit() + db.refresh(row) + else: + db.flush() + return row diff --git a/app/repositories/cps_link.py b/app/repositories/cps_link.py index 2e34970..00b9c1f 100644 --- a/app/repositories/cps_link.py +++ b/app/repositories/cps_link.py @@ -52,12 +52,13 @@ def create_link( def record_click( db: Session, *, link: CpsLink, ip: str | None = None, ua: str | None = None, - event_type: str = "visit", + event_type: str = "visit", openid: str | None = None, ) -> None: - """记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。""" + """记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。 + openid: 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计。""" db.add(CpsClick( link_id=link.id, group_id=link.group_id, sid=link.sid, event_type=event_type, - ip=ip, ua=(ua[:500] if ua else None), + ip=ip, ua=(ua[:500] if ua else None), openid=openid, )) db.commit() diff --git a/app/repositories/cps_wx_user.py b/app/repositories/cps_wx_user.py new file mode 100644 index 0000000..9fde31e --- /dev/null +++ b/app/repositories/cps_wx_user.py @@ -0,0 +1,43 @@ +"""cps_wx_user 数据访问:按 openid upsert。""" +from __future__ import annotations + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models.cps_wx_user import CpsWxUser + + +def get_by_openid(db: Session, openid: str) -> CpsWxUser | None: + return db.execute( + select(CpsWxUser).where(CpsWxUser.openid == openid) + ).scalar_one_or_none() + + +def upsert( + db: Session, *, openid: str, code: str | None = None, group_id: int | None = None, + nickname: str | None = None, headimgurl: str | None = None, unionid: str | None = None, +) -> CpsWxUser: + """按 openid upsert。 + + - 首次:建行,记来源 code/group。 + - 已存在:仅在传入非 None 时更新昵称/头像/unionid(base 阶段为 None,不覆盖已有画像); + 并刷新 last_seen(显式 set,因 onupdate 仅在字段有变更时触发)。 + """ + u = get_by_openid(db, openid) + if u is None: + u = CpsWxUser( + openid=openid, first_code=code, first_group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + db.add(u) + else: + if nickname is not None: + u.nickname = nickname + if headimgurl is not None: + u.headimgurl = headimgurl + if unionid is not None: + u.unionid = unionid + u.last_seen = func.now() + db.commit() + db.refresh(u) + return u diff --git a/app/repositories/device.py b/app/repositories/device.py new file mode 100644 index 0000000..b1c40d8 --- /dev/null +++ b/app/repositories/device.py @@ -0,0 +1,124 @@ +"""device 表读写(设备注册 / 心跳 / 超时扫描)。""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models.device import Device + + +def _get(db: Session, *, user_id: int, device_id: str) -> Device | None: + stmt = select(Device).where( + Device.user_id == user_id, Device.device_id == device_id + ) + return db.execute(stmt).scalar_one_or_none() + + +def register_or_update( + db: Session, + *, + user_id: int, + device_id: str, + registration_id: str | None, + platform: str = "android", + app_version: str | None = None, +) -> Device: + """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。""" + device = _get(db, user_id=user_id, device_id=device_id) + if device is None: + device = Device( + user_id=user_id, + device_id=device_id, + registration_id=registration_id, + platform=platform or "android", + app_version=app_version, + ) + db.add(device) + else: + if registration_id: + device.registration_id = registration_id + if platform: + device.platform = platform + if app_version: + device.app_version = app_version + db.commit() + db.refresh(device) + return device + + +def touch_heartbeat( + db: Session, + *, + user_id: int, + device_id: str, + accessibility_enabled: bool, + registration_id: str | None, +) -> Device: + """处理一次心跳(心跳也能自注册)。 + + service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、 + 清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。 + """ + now = datetime.now(timezone.utc) + device = _get(db, user_id=user_id, device_id=device_id) + if device is None: + device = Device(user_id=user_id, device_id=device_id) + db.add(device) + + if registration_id: + device.registration_id = registration_id + device.last_report_protection_on = accessibility_enabled + + if accessibility_enabled: + device.last_heartbeat_at = now + device.ever_protected = True + device.liveness_state = "alive" + device.notified_at = None + + db.commit() + db.refresh(device) + return device + + +def list_overdue(db: Session, *, timeout_minutes: int) -> list[Device]: + """掉线设备:曾经保护过、当前 alive、心跳超时。 + + 本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。 + """ + cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes) + stmt = select(Device).where( + Device.ever_protected.is_(True), + Device.liveness_state == "alive", + Device.last_heartbeat_at.is_not(None), + Device.last_heartbeat_at < cutoff, + ) + return list(db.execute(stmt).scalars().all()) + + +def mark_notified(db: Session, *, device_id_pk: int) -> None: + """标记掉线已检出(状态机进入 notified,避免每轮重复处理)+ 置「待客户端提醒」标记。 + + kill_alert_pending 与 state 解耦(见 spec accessibility-liveness-pull-prompt.md): + 心跳恢复(touch_heartbeat)只重置 state、不动此标记,故客户端下次进 App 必能拉到这次掉线。 + """ + device = db.get(Device, device_id_pk) + if device is not None: + device.liveness_state = "notified" + device.notified_at = datetime.now(timezone.utc) + device.kill_alert_pending = True + db.commit() + + +def get_device(db: Session, *, user_id: int, device_id: str) -> Device | None: + """按 (user_id, device_id) 取设备(liveness 查询用)。""" + return _get(db, user_id=user_id, device_id=device_id) + + +def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None: + """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。""" + device = _get(db, user_id=user_id, device_id=device_id) + if device is not None and device.kill_alert_pending: + device.kill_alert_pending = False + db.commit() diff --git a/app/repositories/launch_confirm_sample.py b/app/repositories/launch_confirm_sample.py new file mode 100644 index 0000000..4d0b226 --- /dev/null +++ b/app/repositories/launch_confirm_sample.py @@ -0,0 +1,35 @@ +"""启动确认窗兜底样本落库:单条 insert(都上报、不去重,by design)。""" +from __future__ import annotations + +import logging +from typing import Optional + +from sqlalchemy.orm import Session + +from app.models.launch_confirm_sample import LaunchConfirmSample +from app.schemas.launch_confirm_sample import LaunchConfirmSampleIn + +logger = logging.getLogger("shagua.launch_confirm") + + +def _trunc(s: Optional[str], n: int) -> Optional[str]: + """截断到列长上限(PG varchar 超长会报错;繁体长标题等需要)。空 → None。""" + return s[:n] if s else None + + +def insert_sample(db: Session, payload: LaunchConfirmSampleIn) -> int: + """写入一条样本,返回新行 id。""" + row = LaunchConfirmSample( + trace_id=_trunc(payload.trace_id, 64), + device_id=_trunc(payload.device_id, 64), + host_package=_trunc(payload.host_package, 128), + target_app=_trunc(payload.target_app, 128), + system_locale=_trunc(payload.system_locale, 32), + exec_success=bool(payload.exec_success), + dialog_title=_trunc(payload.dialog_title, 256), + payload=payload.payload, + ) + db.add(row) + db.commit() + db.refresh(row) + return row.id diff --git a/app/repositories/user.py b/app/repositories/user.py index 33c53ae..7753e33 100644 --- a/app/repositories/user.py +++ b/app/repositories/user.py @@ -111,14 +111,27 @@ def set_avatar_url(db: Session, user: User, *, avatar_url: str) -> User: def soft_delete_account(db: Session, user: User) -> None: - """注销账号:软删除 + 匿名化。 + """注销账号:软删除 + 匿名化 + 释放唯一约束/外部绑定。 - 把 phone 改成 `deleted_` 释放唯一约束,允许同号码重新注册成全新账号; + 把 phone 改成 `deleted_` 释放手机号唯一约束,允许同号码重新注册成全新账号; 清空 PII(昵称/头像),保留行用于审计。status=deleted 后将无法再登录该行 (登录接口校验 status==active)。 + + ⚠️ 注销必须连同释放 user 上其余唯一约束/外部身份,否则 deleted 行永久占着槽位, + 对应资源再也无法被复用: + - wechat_openid: 不清 → 这个微信再也无法被任何账号绑定(提现通道彻底锁死); + - invite_code: 不清 → 邀请码槽位被占 + 裸查仍命中死号(发奖已被 bind 的 + status==active 校验兜住,清掉更干净)。 + 资金安全(未提现现金 / 在审提现单)由调用方 delete_account 端点前置校验,这里只匿名化。 """ user.status = "deleted" user.phone = f"deleted_{user.id}" user.nickname = None user.avatar_url = None + # 等价"解绑微信":释放 openid 唯一槽 + 清微信昵称头像 + user.wechat_openid = None + user.wechat_nickname = None + user.wechat_avatar_url = None + # 释放邀请码唯一槽 + user.invite_code = None db.commit() diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 615f0eb..a38ac88 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -340,6 +340,15 @@ def bind_wechat_openid(db: Session, user_id: int, code: str) -> dict: return info +def get_cash_balance_cents(db: Session, user_id: int) -> int: + """只读查现金余额(分)。账户不存在返回 0,**不创建账户**——注销前置校验用, + 不该给一个即将被删除的用户凭空建一行空账户(get_or_create_account 会建)。""" + val = db.execute( + select(CoinAccount.cash_balance_cents).where(CoinAccount.user_id == user_id) + ).scalar_one_or_none() + return val or 0 + + def has_reviewing_withdraw(db: Session, user_id: int) -> bool: """用户是否有「待审核(reviewing)」的提现单。 diff --git a/app/schemas/device.py b/app/schemas/device.py new file mode 100644 index 0000000..3254e06 --- /dev/null +++ b/app/schemas/device.py @@ -0,0 +1,50 @@ +"""设备注册 / 心跳相关 schema(无障碍保护存活检测)。""" +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class DeviceRegisterRequest(BaseModel): + device_id: str + registration_id: str | None = None + platform: str = "android" + app_version: str | None = None + + +class HeartbeatRequest(BaseModel): + device_id: str + source: str = "service" # service | app + accessibility_enabled: bool = True + registration_id: str | None = None + + +class DeviceOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: int + device_id: str + registration_id: str | None + ever_protected: bool + liveness_state: str + last_heartbeat_at: datetime | None + updated_at: datetime | None + + +class OkResponse(BaseModel): + ok: bool = True + + +class LivenessOut(BaseModel): + """本机掉线告警状态(后置检测 pull 版)。设备从未注册过 → 全默认(= 无告警)。""" + model_config = ConfigDict(from_attributes=True) + + kill_alert_pending: bool = False + liveness_state: str = "unknown" + ever_protected: bool = False + last_heartbeat_at: datetime | None = None + + +class LivenessAckRequest(BaseModel): + device_id: str diff --git a/app/schemas/launch_confirm_sample.py b/app/schemas/launch_confirm_sample.py new file mode 100644 index 0000000..e2ed223 --- /dev/null +++ b/app/schemas/launch_confirm_sample.py @@ -0,0 +1,26 @@ +"""启动确认窗兜底样本的内部上报模型(pricebot → app-server)。""" +from __future__ import annotations + +from pydantic import BaseModel + + +class LaunchConfirmSampleIn(BaseModel): + """一次 agent 兜底的完整样本。 + + 字段一律宽松可空:落盘是辅助数据,绝不能因个别字段缺失就拒收(否则连累 pricebot 兜底)。 + """ + + trace_id: str | None = None + device_id: str | None = None + host_package: str | None = None + target_app: str | None = None + system_locale: str | None = None + exec_success: bool = False + dialog_title: str | None = None + payload: dict | None = None + + +class LaunchConfirmSampleOut(BaseModel): + """落库结果。""" + + id: int diff --git a/app/schemas/platform.py b/app/schemas/platform.py index 6968271..4c97bd6 100644 --- a/app/schemas/platform.py +++ b/app/schemas/platform.py @@ -30,6 +30,21 @@ class AppFlagsOut(BaseModel): comparing_ad_enabled: bool # 比价/领券期是否展示信息流广告(远程 kill-switch) +class AdConfigPublicOut(BaseModel): + """下发给客户端的穿山甲广告配置(不鉴权,客户端缓存后用)。 + + ⚠️ 不含 reward_mkey(GroMore 验签密钥,机密,只后端验签用,绝不下发)。 + """ + + app_id: str # 穿山甲应用ID(改了客户端需冷启才生效,SDK init 一次性读) + reward_code_id: str # 福利页激励视频位 + compare_feed_code_id: str # 比价信息流位 + coupon_feed_code_id: str # 领券信息流位 + reward_enabled: bool # 福利激励视频开关 + compare_ad_enabled: bool # 比价广告开关 + coupon_ad_enabled: bool # 领券广告开关 + + class AppVersionOut(BaseModel): """最新 App 版本信息(OTA 检查更新,不鉴权)。 diff --git a/docs/后端技术实现.md b/docs/后端技术实现.md index 8828ed7..e106985 100644 --- a/docs/后端技术实现.md +++ b/docs/后端技术实现.md @@ -222,6 +222,16 @@ POST /api/v1/auth/sms/login { phone, code } → 任意 6 位通过 → upsert --- +## 6.5 pricebot 回写:内部端点(server→server) + +§6 是 app-server 透传给 pricebot;反过来 pricebot 也会把少量数据回写 app-server 落库,走 `app/api/internal/*` —— **不走用户 JWT**,靠共享密钥头 `X-Internal-Secret`(== `INTERNAL_API_SECRET`,未配则 503,防裸奔写端点)校验。 + +**启动确认窗兜底样本**(2026-06):国产 ROM 打开别的 App 时弹"想要打开 XX"确认窗,pricebot 对没见过文案(繁体/英文/ROM 改版)的窗用 LLM 兜底放行后,把样本 POST 到 [`/internal/launch-confirm-sample`](../app/api/internal/launch_confirm.py) 落 `launch_confirm_sample` 表(host 包 + 弹窗树 + LLM plan + 设备 locale/机型),供研发定期人工沉淀回 pricebot 的规则 yaml(回到快路径)。**需 pricebot 与 app-server 两边 `.env` 配同一 `INTERNAL_API_SECRET` 才生效**(未配则 pricebot 侧跳过上报、不影响比价/领券)。 + +> 同类内部回写端点还有 `/internal/price-observation`(价格观测)、`/internal/store-mapping`(跨平台店铺映射)等 pricebot 比价资产沉淀。 + +--- + ## 7. 数据模型 业务表(下表)+ `alembic_version` 框架表。生产 PG / 开发可回退 SQLite。 diff --git a/tests/test_delete_account.py b/tests/test_delete_account.py new file mode 100644 index 0000000..464b02c --- /dev/null +++ b/tests/test_delete_account.py @@ -0,0 +1,141 @@ +"""注销账号(DELETE /api/v1/user):软删 + 释放唯一约束 + 资金前置校验。 + +覆盖: + - 干净注销:status=deleted + phone 匿名化 + wechat_openid/invite_code 等唯一槽全部释放 + - 回归原始 bug:注销释放 openid 后,别的账号能绑同一个微信 + - 资金前置闸:有可提现现金 / 在审提现单 → 409 拒绝注销,账号不被删 + - 注销后旧 token 即时失效(status==active 闸) +""" +from __future__ import annotations + +from sqlalchemy import select + +from app.db.session import SessionLocal +from app.models.user import User +from app.models.wallet import CoinAccount + + +def _login(client, 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 _patch_userinfo(monkeypatch, openid: str) -> None: + monkeypatch.setattr( + "app.integrations.wxpay.code_to_userinfo", + lambda code: {"openid": openid, "nickname": "微信昵称", "avatar_url": "https://x/a.png", "raw": {}}, + ) + + +def _seed_cash(client, token: str, phone: str, cents: int) -> None: + """先访问 /account 触发建账户,再用 DB 直接灌现金余额。""" + client.get("/api/v1/wallet/account", headers=_auth(token)) + db = SessionLocal() + try: + user = db.execute(select(User).where(User.phone == phone)).scalar_one() + acc = db.get(CoinAccount, user.id) + acc.cash_balance_cents = cents + db.commit() + finally: + db.close() + + +def _get_user(phone: str) -> User | None: + db = SessionLocal() + try: + return db.execute(select(User).where(User.phone == phone)).scalar_one_or_none() + finally: + db.close() + + +def test_delete_clean_account_releases_all_unique_slots(client) -> None: + phone = "13900000001" + token = _login(client, phone) + # 预置:给该用户造昵称 + 邀请码(占 invite_code 唯一槽),验证注销时一并释放 + db = SessionLocal() + try: + u = db.execute(select(User).where(User.phone == phone)).scalar_one() + u.nickname = "张三" + u.invite_code = "INVCODE0001" + db.commit() + uid = u.id + finally: + db.close() + + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 200, r.text + + u = _get_user(f"deleted_{uid}") + assert u is not None + assert u.status == "deleted" + assert u.phone == f"deleted_{uid}" + assert u.nickname is None + assert u.avatar_url is None + assert u.wechat_openid is None + assert u.wechat_nickname is None + assert u.wechat_avatar_url is None + assert u.invite_code is None + + +def test_delete_releases_wechat_so_another_account_can_bind(client, monkeypatch) -> None: + """原始 bug 回归:A 绑微信 → 注销,B 必须能绑同一个 openid(修复前会 409)。""" + openid = "openid_delreuse_a1" # 独特值,避免跨测试共享库污染(如 test_withdraw 也用 openid_shared_xyz) + # A 绑微信 + _patch_userinfo(monkeypatch, openid) + token_a = _login(client, "13900000002") + r = client.post("/api/v1/wallet/bind-wechat", json={"code": "ca"}, headers=_auth(token_a)) + assert r.status_code == 200, r.text + # A 注销(无现金、无提现单 → 允许) + r = client.delete("/api/v1/user", headers=_auth(token_a)) + assert r.status_code == 200, r.text + # B 绑同一个 openid → 必须成功(修复前会 409 "该微信已绑定其他账号") + token_b = _login(client, "13900000003") + r = client.post("/api/v1/wallet/bind-wechat", json={"code": "cb"}, headers=_auth(token_b)) + assert r.status_code == 200, r.text + assert r.json()["bound"] is True + + +def test_delete_blocked_when_cash_balance_positive(client) -> None: + phone = "13900000004" + token = _login(client, phone) + _seed_cash(client, token, phone, 100) # 1 元未提现 + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 409, r.text + assert "现金" in r.json()["detail"] + u = _get_user(phone) # 账号未被删 + assert u is not None and u.status == "active" + + +def test_delete_blocked_when_reviewing_withdraw_exists(client, monkeypatch) -> None: + """cash 提光(=0)但有在审提现单 → 仍拒绝(命中 reviewing 闸,非 cash 闸)。""" + phone = "13900000005" + _patch_userinfo(monkeypatch, "openid_delrev_b2") + token = _login(client, phone) + _seed_cash(client, token, phone, 50) + client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) + # 提光 50 → cash 归 0,单进 reviewing + r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token)) + assert r.status_code == 200 and r.json()["status"] == "reviewing", r.text + assert r.json()["cash_balance_cents"] == 0 + + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 409, r.text + assert "审核" in r.json()["detail"] + u = _get_user(phone) + assert u is not None and u.status == "active" + + +def test_deleted_user_old_token_is_rejected(client) -> None: + phone = "13900000006" + token = _login(client, phone) + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 200, r.text + # 注销后旧 token 立即失效(get_current_user 校验 status==active) + r = client.get("/api/v1/user/onboarding/status", params={"device_id": "d"}, headers=_auth(token)) + assert r.status_code == 401, r.text