shaguabijia-app-server/app/core/config.py

"""全局配置。

用 pydantic-settings 从环境变量 + .env 文件加载。所有可调参数集中在这里,
业务代码通过 `from app.core.config import settings` 引用,不再读 os.environ。
"""
from __future__ import annotations

from functools import lru_cache
from pathlib import Path
from typing import Literal

from pydantic import Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent

# 生产环境 JWT secret 的最小可接受长度(字节)。HS256 推荐高熵随机串;<16 视为弱密钥。
_MIN_PROD_SECRET_LEN = 16
# 已知的占位默认值(代码里写死的 default),prod 下绝不能沿用。
_INSECURE_SECRET_DEFAULTS = frozenset({"change-me", "change-me-admin", ""})


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=str(_PROJECT_ROOT / ".env"),
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    # ===== 环境 =====
    APP_ENV: Literal["dev", "prod"] = "dev"
    APP_NAME: str = "shaguabijia-app-server"
    APP_DEBUG: bool = True

    # 接口限流开关(测试里关掉,避免内存计数跨用例累加误伤)
    RATE_LIMIT_ENABLED: bool = True

    # ===== 数据库 =====
    DATABASE_URL: str = "sqlite:///./data/app.db"

    # ===== JWT =====
    JWT_SECRET_KEY: str = "change-me"
    JWT_ALGORITHM: str = "HS256"
    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
    JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30

    # ===== Admin 后台 =====
    # admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
    # 生产必须改成高熵随机串(同 JWT_SECRET_KEY 的要求)。
    ADMIN_JWT_SECRET: str = "change-me-admin"
    ADMIN_JWT_EXPIRE_MINUTES: int = 720  # admin 登录态 12 小时(无 refresh,过期重登)
    # 可选 IP 白名单(逗号分隔),为空=应用层不限制(靠 nginx allow/deny 兜底)。
    ADMIN_IP_ALLOWLIST: str = ""

    @property
    def admin_ip_allowlist(self) -> list[str]:
        if not self.ADMIN_IP_ALLOWLIST.strip():
            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 后置检测;本期不接推送)
    HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
    HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
    HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期

    # ===== 短信 =====
    SMS_MOCK: bool = True
    SMS_CODE_TTL_SEC: int = 300
    SMS_SEND_INTERVAL_SEC: int = 60
    # 真实发送走极光短信 REST(自定义验证码模式:本服务生成 code,极光只负责发)。
    # 复用极光一键登录的 JG_APP_KEY / JG_MASTER_SECRET(同一个极光应用)+ JG_REQUEST_TIMEOUT_SEC。
    SMS_SEND_ENDPOINT: str = "https://api.sms.jpush.cn/v1/messages"
    SMS_SIGN_ID: int = 31729       # 极光短信签名 ID(非机密,可被 .env 覆盖)
    SMS_TEMPLATE_ID: int = 1       # 极光短信模板 ID(变量名 code,有效期 5 分钟)
    SMS_CODE_LENGTH: int = 6       # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
    SMS_DAILY_LIMIT_PER_PHONE: int = 10   # 单手机号每日发送上限(防刷 + 控费)
    SMS_MAX_VERIFY_ATTEMPTS: int = 5      # 单个验证码最多校验失败次数,超过即作废(防爆破)

    # ===== 测试账号(release 包全流程联调用)=====
    # 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
    # (real 模式下也跳过校验)、每次登录【强制重走新手引导】,并设【每日使用次数上限】防被人
    # 猜到号后脚本滥用。两个值都能随时改 .env。逻辑全在 app/core/test_account.py,与其他业务解耦。
    # ⚠️ TEST_ACCOUNT_PHONE 留空 = 整个功能关闭(生产默认安全;要启用才显式填号)。
    TEST_ACCOUNT_PHONE: str = ""           # 测试手机号(11 位,如 11111111111);空=关闭整功能
    TEST_ACCOUNT_DAILY_LIMIT: int = 500    # 该测试号每日最多登录次数,当日超过即拒绝登录

    @property
    def test_account_phone(self) -> str:
        """规整后的测试手机号(去空白);空串=功能关闭。"""
        return self.TEST_ACCOUNT_PHONE.strip()

    # ===== 美团联盟 CPS =====
    # 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
    MT_CPS_APP_KEY: str = ""
    MT_CPS_APP_SECRET: str = ""
    MT_CPS_HOST: str = "https://media.meituan.com"
    MT_CPS_TIMEOUT_SEC: int = 15
    MT_CPS_DEFAULT_SID: str = "sgbjia"
    # 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
    # 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
    MT_CPS_PROXY: str = ""

    @property
    def mt_cps_configured(self) -> bool:
        """美团 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 一致。
    WECHAT_APP_ID: str = ""              # 开放平台移动应用 appid(wx 开头)
    WECHAT_APP_SECRET: str = ""          # 开放平台 AppSecret,code 换 openid 用
    WXPAY_MCH_ID: str = ""               # 微信支付商户号
    WXPAY_MCH_SERIAL_NO: str = ""        # 商户 API 证书序列号
    WXPAY_MCH_PRIVATE_KEY_PATH: str = "./secrets/apiclient_key.pem"   # 商户 API 私钥
    WXPAY_PUBLIC_KEY_ID: str = ""        # 微信支付平台公钥 ID(PUB_KEY_ID_...)
    WXPAY_PUBLIC_KEY_PATH: str = "./secrets/pub_key.pem"             # 微信支付平台公钥
    WXPAY_TRANSFER_SCENE_ID: str = "1000"  # 转账场景 ID(1000=现金营销)
    WXPAY_REQUEST_TIMEOUT_SEC: int = 10
    WITHDRAW_AUTO_RECONCILE_ENABLED: bool = False
    WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC: int = 300
    WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES: int = 15
    # 0 点自动兑金币:由 App 进程内任务 app.core.daily_exchange_worker 跨 0 点跑一轮
    # wallet.daily_auto_exchange(原 deploy/daily-exchange.timer 已不再需要,见 deploy/daily-exchange.md)。
    # 客户端已删手动兑入口、改为「0 点自动兑现金」,故默认开;运营要临时停在 .env 置 false
    # (worker 不启动;脚本也 no-op)。
    AUTO_EXCHANGE_ENABLED: bool = True
    # 进程内自动兑换 worker 的检查间隔(秒):每隔这么久醒一次,跨过北京 0 点就跑一轮。
    # 默认 600s=10min,即 0 点后最多 10 分钟内兑完(客户端文案已注明「可能存在延迟」)。
    AUTO_EXCHANGE_CHECK_INTERVAL_SEC: int = 600
    # 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
    # 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
    # (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
    WXPAY_AUTH_NOTIFY_URL: str = ""

    @property
    def wxpay_configured(self) -> bool:
        """凭证是否齐全(缺则提现接口拒绝调用,而非启动时炸)。"""
        return bool(
            self.WECHAT_APP_ID
            and self.WXPAY_MCH_ID
            and self.WXPAY_MCH_SERIAL_NO
            and self.WXPAY_PUBLIC_KEY_ID
        )

    @property
    def wxpay_auth_configured(self) -> bool:
        """免确认收款授权可用 = 微信支付凭证齐全 + 授权回调地址已配。"""
        return bool(self.wxpay_configured and self.WXPAY_AUTH_NOTIFY_URL)

    # ===== 穿山甲激励视频(服务端发奖回调)=====
    # 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
    # 穿山甲后台配置的"奖励校验密钥"(m-key),验签用。每个 GroMore 广告位 m-key 不同(后台各自
    # 生成),但共用同一回调 URL → 多个位的 m-key 都要配上。验签时所有非空 m-key 逐个试、任一通过
    # 即接受(见 pangle.verify_callback_sign_any / 下面的 pangle_reward_secrets)。
    PANGLE_CALLBACK_ENABLED: bool = False
    # 推荐:每个激励位的 m-key 分开一行配,清晰不混淆(留空的忽略)。
    PANGLE_REWARD_SECRET_TEST: str = ""            # 测试应用 激励位 104099649
    PANGLE_REWARD_SECRET_TEST_DEDICATED: str = ""  # 测试应用 专属激励位 104127529
    PANGLE_REWARD_SECRET_PROD: str = ""            # 正式应用 激励位 104099389
    # 旧用法:单个或逗号分隔的多个 m-key,仍兼容(会与上面三个命名项合并去重)。
    PANGLE_REWARD_SECRET: str = ""

    # ⚠️ 仅本地联调:打开后开放 POST /api/v1/ad/test-grant,让(已登录的)客户端在没部署公网、
    # 穿山甲 S2S 回调打不到本地时,直接触发一次发奖,验证"看广告→金币到账"全链路。
    # 它让客户端能自助发奖 = 绕过反作弊,**生产必须保持 False**(默认 False;只在本地 .env 设 true)。
    AD_REWARD_TEST_GRANT_ENABLED: bool = False

    @property
    def pangle_reward_secrets(self) -> list[str]:
        """汇总所有 m-key 成列表(去空白、去空项、去重保序)。验签时逐个试、任一通过即接受
        (见 pangle.verify_callback_sign_any)。来源可混用:三个命名项 + 旧的逗号分隔 PANGLE_REWARD_SECRET。"""
        raw = [
            *self.PANGLE_REWARD_SECRET.split(","),
            self.PANGLE_REWARD_SECRET_TEST,
            self.PANGLE_REWARD_SECRET_TEST_DEDICATED,
            self.PANGLE_REWARD_SECRET_PROD,
        ]
        cleaned = [s.strip() for s in raw if s and s.strip()]
        return list(dict.fromkeys(cleaned))  # 去重保序

    @property
    def pangle_callback_configured(self) -> bool:
        """回调开关打开且至少配了一个验签密钥,才接受发奖回调。"""
        return bool(self.PANGLE_CALLBACK_ENABLED and self.pangle_reward_secrets)

    # ===== 穿山甲 GroMore 数据 API(报表收益拉取,T+1)=====
    # ⚠️ 与上面发奖回调的 m-key 是【两套完全不同的凭证】:这三样在穿山甲后台
    # 「接入中心 → GroMore-API → 聚合数据报告 API」文档页领取(user_id / role_id / Security Key),
    # 仅用于按天拉 GroMore 收益报表(revenue 预估收益 + api_revenue 收益Api),不参与发奖。
    # 该 API 只能查【GroMore 聚合代码位】的数据(=我们 useMediation 的口径),非穿山甲 SDK 数据;
    # 且不提供用户/设备维度(官方明确),故收益只能落到 日期×代码位 汇总,不能挂到逐条事件。
    # 子账号(role_id≠user_id)需主账号在「角色管理」授予「查看全部数据」权限,否则查不到
    # ecpm/revenue(接口返回 118);role_id 填成与 user_id 一致 = 查主账号数据。
    PANGLE_REPORT_USER_ID: int = 0        # 媒体账号 user_id
    PANGLE_REPORT_ROLE_ID: int = 0        # 子账号 role_id(=user_id 时查主账号)
    PANGLE_REPORT_SECURITY_KEY: str = ""  # 该账号的 Security Key(secure_key,≠ 发奖 m-key)
    # GroMore 聚合 AppId(报表 site_id 维度)→ 我们的应用环境。报表按 site_id 区分两个穿山甲应用,
    # 用它把每行归到 prod(傻瓜比价正式)/ test(测试应用)。默认值取自现网两个应用,部署时按需覆盖。
    PANGLE_REPORT_SITE_ID_PROD: str = "5830519"   # 傻瓜比价正式应用 AppId
    PANGLE_REPORT_SITE_ID_TEST: str = "5832303"   # 测试应用 AppId

    @property
    def pangle_report_configured(self) -> bool:
        """三样齐全(user_id / role_id / security_key)才能拉 GroMore 报表;缺任一 → 同步脚本 no-op。"""
        return bool(
            self.PANGLE_REPORT_USER_ID
            and self.PANGLE_REPORT_ROLE_ID
            and self.PANGLE_REPORT_SECURITY_KEY
        )

    @property
    def pangle_report_site_id_to_env(self) -> dict[str, str]:
        """site_id(穿山甲 AppId)→ app_env(prod/test);供同步脚本把报表行归到我们的应用。留空项忽略。"""
        out: dict[str, str] = {}
        if self.PANGLE_REPORT_SITE_ID_PROD.strip():
            out[self.PANGLE_REPORT_SITE_ID_PROD.strip()] = "prod"
        if self.PANGLE_REPORT_SITE_ID_TEST.strip():
            out[self.PANGLE_REPORT_SITE_ID_TEST.strip()] = "test"
        return out

    # ===== Pricebot 上游 (领券/比价业务透传目标) =====
    # pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step
    PRICEBOT_BASE_URL: str = "http://localhost:8000"
    # 多实例(单机多进程)时填逗号分隔的实例列表,例如:
    #   PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003
    # 透传层按 trace_id 一致性 hash 选实例(见 app/core/pricebot_router.py),保证同一次
    # 比价/领券的所有帧落同一 pricebot 进程,从而用进程内内存维护 session/coordinator,
    # 无需 Redis。留空 → 退回单实例 [PRICEBOT_BASE_URL],零改动兼容。
    PRICEBOT_INSTANCES: str = ""
    # 领券一帧最多 wait 6s,加网络往返,timeout 给 30s 比较稳
    PRICEBOT_REQUEST_TIMEOUT_SEC: int = 30
    # 比价(intent/recognize + price/step)透传超时:意图识别是大上下文 LLM、
    # price/step 每帧也是 LLM,可能 >30s;对齐客户端 agent ApiClient 的 60s 读超时。
    PRICEBOT_COMPARE_TIMEOUT_SEC: int = 60

    @property
    def pricebot_instances(self) -> list[str]:
        """pricebot 上游实例列表。空 → 单实例兜底 [PRICEBOT_BASE_URL]。"""
        if not self.PRICEBOT_INSTANCES.strip():
            return [self.PRICEBOT_BASE_URL]
        return [u.strip() for u in self.PRICEBOT_INSTANCES.split(",") if u.strip()]

    # ===== 内部(server→server)端点密钥 =====
    # pricebot 比价 done 后把价格观测 POST 到 /internal/price-observation 落库,
    # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
    # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
    INTERNAL_API_SECRET: str = ""

    # ===== 媒体文件(用户头像上传)=====
    # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
    # 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。
    MEDIA_ROOT: str = "./data/media"
    MEDIA_URL_PREFIX: str = "/media"
    AVATAR_MAX_BYTES: int = 5 * 1024 * 1024  # 头像最大 5MB

    # ===== 邀请好友 =====
    # 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。
    # MVP 落地页就放 app-server 的 /media 静态目录下(dl.html)。
    INVITE_LANDING_URL: str = "https://app-api.shaguabijia.com/media/dl.html"

    # ===== CPS 群发短链跳转 =====
    # 群发券链接套一层我们的短链 /c/{code} 做点击统计后 302 跳美团。这里是写进 admin
    # 生成链接里的对外跳转域名前缀。本地填 http://localhost:8770;生产**强烈建议用独立
    # 域名**(大量群发会被微信封,别用主 API 域名连累整个 App)。留空 → 跳转端点用请求
    # 的 Host 兜底拼绝对地址。
    CPS_REDIRECT_BASE: str = ""

    # ===== CORS =====
    CORS_ALLOW_ORIGINS: str = ""

    @property
    def cors_origins_list(self) -> list[str]:
        if not self.CORS_ALLOW_ORIGINS.strip():
            return []
        return [o.strip() for o in self.CORS_ALLOW_ORIGINS.split(",") if o.strip()]

    @property
    def is_prod(self) -> bool:
        return self.APP_ENV == "prod"

    @model_validator(mode="after")
    def _enforce_prod_secrets(self) -> "Settings":
        """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。

        只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值
        时任何人都能伪造 access/admin token → 账号与后台失陷。INTERNAL_API_SECRET 默认空 = 内部端点
        关闭(返 503),是安全的默认态,故不在此强制。dev 不触发,便于本地直接起。
        """
        if not self.is_prod:
            return self
        weak: list[str] = []
        for name in ("JWT_SECRET_KEY", "ADMIN_JWT_SECRET"):
            value = getattr(self, name)
            if value in _INSECURE_SECRET_DEFAULTS or len(value) < _MIN_PROD_SECRET_LEN:
                weak.append(name)
        if weak:
            raise ValueError(
                f"APP_ENV=prod 但检测到弱/默认密钥: {', '.join(weak)} —— 必须改成 "
                f"≥{_MIN_PROD_SECRET_LEN} 位高熵随机串(否则 JWT 可被伪造 → 用户/后台账号失陷)。"
                f"生成示例: python -c \"import secrets; print(secrets.token_urlsafe(48))\""
            )
        return self


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()


settings = get_settings()