refactor(core): 抽出框架无关核心层 core/,为 FastAPI 迁移铺路

第一步模块拆分(纯搬运,零行为变化):
- 新增 core/ 包:config(常量/配置) + state(运行时状态) + utils(纯函数) + security(音频签名)
- main.py 改为从 core 导入并以原下划线名再导出,250+ 既有调用点零改动

关键设计:
- 状态全部原地修改,import 即共享同一对象(21 项身份断言 + 10 项冒烟全绿)
- 音频签名密钥由启动时 _init_audio_secret(app.secret_key) 注入,
  解除 core.security 对 Flask app 的依赖,新老应用可互认 token

验证:py_compile 通过;冒烟 10/10 PASS;对象身份 22/22 PASS;
waitress 真实启动 /health 200 version=2.4.0。main.py 减 229 行。
This commit is contained in:
snowy-yang committed 2026-09-26 00:16:04 +08:00
1 parent d209af847a
commit c2a8da3e43
6 files changed
+405 -227

No files matched your search

+74
View File
@@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
"""音频直链多重防护(HMAC 签名 + 时效 + 代理流)——框架无关。
自 main.py 原样迁出,唯一改动是把密钥来源从 Flask 的 `app.secret_key`
改为**启动时注入**(`init_secret_key`),从而解除对本模块对 Flask app 的依赖;
签名/校验算法与口径保持逐字一致,迁移期新旧应用共用同一密钥即可互认 token。
"""
import base64
import hashlib
import hmac
import secrets
import time
from typing import Any, Optional
from core.config import AUDIO_HOST_MARKS, AUDIO_TOKEN_TTL, AUDIO_URL_KEYS
# 由应用启动时注入(来源:auth_system.get_secret_key() → app.secret_key)
_SECRET_KEY: bytes = b''
def init_secret_key(key) -> None:
"""注入签名密钥(str 或 bytes)。须在生成/校验音频 token 前调用一次。"""
global _SECRET_KEY
_SECRET_KEY = key.encode('utf-8') if isinstance(key, str) else key
def is_audio_url(v) -> bool:
return isinstance(v, str) and v.startswith('http') and any(m in v for m in AUDIO_HOST_MARKS)
def sign_payload(payload: str) -> str:
return hmac.new(_SECRET_KEY, payload.encode('utf-8'), hashlib.sha256).hexdigest()[:32]
def make_audio_token(url: str) -> str:
exp = int(time.time()) + AUDIO_TOKEN_TTL
nonce = secrets.token_hex(4)
payload = base64.urlsafe_b64encode(f'{exp}|{nonce}|{url}'.encode('utf-8')).decode().rstrip('=')
return payload + '.' + sign_payload(payload)
def verify_audio_token(token: str) -> Optional[str]:
try:
payload, sig = token.rsplit('.', 1)
if not hmac.compare_digest(sig, sign_payload(payload)):
return None
pad = '=' * (-len(payload) % 4)
exp, _nonce, url = base64.urlsafe_b64decode(payload + pad).decode('utf-8').split('|', 2)
if int(exp) < time.time():
return None
return url
except Exception:
return None
def protect_audio_urls(obj: Any) -> bool:
"""递归就地替换 dict/list 中的音频直链为签名代理地址,返回是否发生替换"""
hit = {'v': False}
def walk(o):
if isinstance(o, dict):
for k, v in list(o.items()):
if k in AUDIO_URL_KEYS and is_audio_url(v):
o[k] = '/audio_stream?token=' + make_audio_token(v)
hit['v'] = True
else:
walk(v)
elif isinstance(o, list):
for it in o:
walk(it)
walk(obj)
return hit['v']