zhuzihao
8a2f72d366
Co-authored-by: zzhyyyyy <2685922758@qq.com> Reviewed-on: #63 Co-authored-by: zhuzihao <zhuzihao@wonderable.ai> Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
286 lines
15 KiB
Python
286 lines
15 KiB
Python
"""全局配置。
|
|
|
|
用 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
|
|
|
|
# ===== 短信 =====
|
|
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 # 单个验证码最多校验失败次数,超过即作废(防爆破)
|
|
|
|
# ===== 美团联盟 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)
|
|
|
|
# ===== 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()
|