"""全局配置。 用 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 = 60 # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀) 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 ===== # 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 移动应用,用于微信支付);这是【已认证服务号】。 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()