simon
60f8c263f2
从 CCTV 主页抓取《新闻联播》，下载 → 转 MP3/WAV → 静音切分 → ASR 识别 → LLM 校对 → 切分为单条新闻 → 入库 MySQL。 主要内容： - 全链路：getVideo5 抓取下载、audioRead 转写、deepseek 校对与切分、newsProcess 入库 - 可移植化：配置分层，.env 只放密钥、config.yml 放模型/接入点/路由/参数 - 可换供应商：endpoints（kind/base_url/api_key_env/extra_body）+ routes 按环节选路 - 数据保真：数值事实守卫，校对改动数字/年份/届次则整片回退 ASR 原文； 识别不完整不发布该日精编，避免半天内容被当成完整一天 - 定时任务：systemd 每天 21:00，失败 21:30 / 22:00 重试； 只缺切分时只重跑切分（省掉全部 ASR），用 state/.asr_complete_* 标记判定阶段 - 隧道自愈：13306 不通时自动执行 autossh.sh（所有入口共用，systemd 托管时只等待） - 中间产物每日清理；97 项离线自检（配置/清理/事实守卫/解析/隧道）
403 lines
15 KiB
Python
403 lines
15 KiB
Python
"""非敏感配置加载器：config.yml + 环境变量覆盖 + 内置默认值
|
||
|
||
分工：
|
||
- 敏感配置（数据库口令、API Key）留在 .env，由 env.py 加载；
|
||
- 其余配置（数据库地址、目录、**各环节使用的模型**、切分参数、清理开关）都写在 config.yml；
|
||
- 可用 XWLB_CONFIG_FILE 指定其他配置文件。
|
||
|
||
优先级：**进程环境变量 > config.yml > 本文件内置默认值**
|
||
（便于临时试验，例如 `DEEPSEEK_MODEL=deepseek-reasoner python newsProcess.py`）
|
||
"""
|
||
import copy
|
||
import logging
|
||
import os
|
||
from pathlib import Path
|
||
|
||
import env # noqa: F401 — 先加载 .env，使其中的变量可覆盖 config.yml
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
|
||
# 环境变量 -> 配置项（点号路径）的覆盖映射
|
||
ENV_OVERRIDES = {
|
||
'MYSQL_HOST': 'mysql.host',
|
||
'MYSQL_PORT': 'mysql.port',
|
||
'MYSQL_USER': 'mysql.user',
|
||
'MYSQL_DATABASE': 'mysql.database',
|
||
'XWLB_VIDEO_DIR': 'paths.video_dir',
|
||
'XWLB_AUDIO_DIR': 'paths.audio_dir',
|
||
'DASHSCOPE_ASR_MODEL': 'models.asr_model',
|
||
'DASHSCOPE_LLM_MODEL': 'models.correct_model',
|
||
'DEEPSEEK_MODEL': 'models.split_model',
|
||
'LLM_CORRECT_ENABLED': 'llm_correct.enabled',
|
||
'LLM_CORRECT_RETRIES': 'llm_correct.max_retries',
|
||
'LLM_CORRECT_TIMEOUT': 'llm_correct.timeout',
|
||
'LLM_CORRECT_MAX_TOKENS': 'llm_correct.max_tokens',
|
||
# 接入点地址与路由（换供应商时也可用环境变量临时试）
|
||
'DASHSCOPE_HTTP_BASE_URL': 'endpoints.dashscope.http_base_url',
|
||
'DASHSCOPE_WEBSOCKET_BASE_URL': 'endpoints.dashscope.websocket_base_url',
|
||
'DEEPSEEK_BASE_URL': 'endpoints.deepseek.base_url',
|
||
'XWLB_ROUTE_ASR': 'routes.asr',
|
||
'XWLB_ROUTE_CORRECT': 'routes.correct',
|
||
'XWLB_ROUTE_SPLIT': 'routes.split',
|
||
}
|
||
|
||
# 各环节默认走哪个接入点
|
||
DEFAULT_ROUTES = {'asr': 'dashscope', 'correct': 'dashscope', 'split': 'deepseek'}
|
||
|
||
# 内置默认值：config.yml 缺失时项目仍可运行
|
||
DEFAULTS = {
|
||
'mysql': {'host': 'localhost', 'port': 13306, 'user': 'myquant', 'database': 'myquant'},
|
||
'paths': {'video_dir': 'xwlb_video', 'audio_dir': 'audio_processing'},
|
||
'models': {
|
||
'asr_model': 'paraformer-realtime-v2',
|
||
'correct_model': 'qwen-plus',
|
||
'split_model': 'deepseek-chat',
|
||
},
|
||
'endpoints': {
|
||
'dashscope': {
|
||
'kind': 'dashscope',
|
||
'http_base_url': 'https://dashscope.aliyuncs.com/api/v1',
|
||
'websocket_base_url': 'wss://dashscope.aliyuncs.com/api-ws/v1/inference',
|
||
'api_key_env': 'DASHSCOPE_API_KEY',
|
||
},
|
||
'deepseek': {
|
||
'kind': 'openai',
|
||
'base_url': 'https://api.deepseek.com/v1',
|
||
'chat_completions_path': '/chat/completions',
|
||
'api_key_env': 'DEEPSEEK_API_KEY',
|
||
},
|
||
},
|
||
'routes': dict(DEFAULT_ROUTES),
|
||
'asr': {'sample_rate': 16000, 'language_hints': ['zh', 'en']},
|
||
'audio_split': {
|
||
'min_silence_ms': 700,
|
||
'silence_thresh_db': -40,
|
||
'keep_silence_ms': 400,
|
||
'max_chunk_ms': 180000,
|
||
},
|
||
'llm_correct': {
|
||
'enabled': 1, 'max_retries': 2, 'timeout': 90,
|
||
'max_tokens': 8000, 'temperature': 0.1, 'top_p': 0.5,
|
||
},
|
||
'llm_split': {
|
||
'max_tokens': 20000, 'retry_max_tokens': 32000,
|
||
'temperature': 0.5, 'timeout': 60, 'max_retries': 3,
|
||
},
|
||
'cleanup': {
|
||
'after_daily_run': 1, 'only_on_success': 1,
|
||
'remove_video': 1, 'remove_audio': 1,
|
||
},
|
||
}
|
||
|
||
|
||
def config_path() -> Path:
|
||
"""配置文件路径（XWLB_CONFIG_FILE 可覆盖）"""
|
||
override = os.getenv('XWLB_CONFIG_FILE')
|
||
if override:
|
||
return Path(override).expanduser()
|
||
return BASE_DIR / 'config.yml'
|
||
|
||
|
||
def _deep_merge(base: dict, override: dict) -> dict:
|
||
"""递归合并，override 优先"""
|
||
result = copy.deepcopy(base)
|
||
for key, value in (override or {}).items():
|
||
if isinstance(value, dict) and isinstance(result.get(key), dict):
|
||
result[key] = _deep_merge(result[key], value)
|
||
else:
|
||
result[key] = value
|
||
return result
|
||
|
||
|
||
def _set_by_path(target: dict, dotted: str, value):
|
||
"""按点号路径写入嵌套 dict"""
|
||
keys = dotted.split('.')
|
||
node = target
|
||
for key in keys[:-1]:
|
||
node = node.setdefault(key, {})
|
||
node[keys[-1]] = value
|
||
|
||
|
||
def load_config(path=None) -> dict:
|
||
"""加载配置：默认值 ← config.yml ← 环境变量"""
|
||
path = Path(path) if path else config_path()
|
||
file_config = {}
|
||
if path.is_file():
|
||
import yaml # 延迟导入，未使用配置文件时不强依赖
|
||
try:
|
||
with open(path, encoding='utf-8') as f:
|
||
file_config = yaml.safe_load(f) or {}
|
||
except yaml.YAMLError as e:
|
||
raise RuntimeError(f"配置文件解析失败（{path}）：{e}") from e
|
||
if not isinstance(file_config, dict):
|
||
raise RuntimeError(f"配置文件格式错误（{path}）：顶层应为键值映射")
|
||
|
||
merged = _deep_merge(DEFAULTS, file_config)
|
||
for env_key, dotted in ENV_OVERRIDES.items():
|
||
raw = os.getenv(env_key)
|
||
if raw is None or raw == '':
|
||
continue
|
||
_set_by_path(merged, dotted, raw)
|
||
return merged
|
||
|
||
|
||
CONFIG = load_config()
|
||
LOADED_CONFIG_FILE = config_path() if config_path().is_file() else None
|
||
|
||
|
||
def reload_config():
|
||
"""重新加载（供测试使用）"""
|
||
global CONFIG, LOADED_CONFIG_FILE
|
||
CONFIG = load_config()
|
||
LOADED_CONFIG_FILE = config_path() if config_path().is_file() else None
|
||
apply_dashscope_endpoints()
|
||
return CONFIG
|
||
|
||
|
||
# --------------------------------------------------------------------- 取值
|
||
def get(dotted: str, default=None):
|
||
"""按点号路径取值，如 get('models.asr_model')"""
|
||
node = CONFIG
|
||
for key in dotted.split('.'):
|
||
if not isinstance(node, dict) or key not in node:
|
||
return default
|
||
node = node[key]
|
||
return node
|
||
|
||
|
||
def get_int(dotted: str, default: int = 0) -> int:
|
||
try:
|
||
return int(get(dotted, default))
|
||
except (TypeError, ValueError):
|
||
logger.warning("配置项 %s 不是整数，使用默认值 %s", dotted, default)
|
||
return default
|
||
|
||
|
||
def get_bool(dotted: str, default: bool = False) -> bool:
|
||
value = get(dotted, default)
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)):
|
||
return bool(value)
|
||
return str(value).strip().lower() in ('1', 'true', 'yes', 'on')
|
||
|
||
|
||
def get_dict(dotted: str, default=None):
|
||
"""取字典配置（如 extra_body）；非字典时返回默认值"""
|
||
value = get(dotted, None)
|
||
if isinstance(value, dict):
|
||
return dict(value)
|
||
if value not in (None, ''):
|
||
logger.warning("配置项 %s 不是字典，已忽略: %r", dotted, value)
|
||
return dict(default or {})
|
||
|
||
|
||
def get_list(dotted: str, default=None):
|
||
value = get(dotted, None)
|
||
if isinstance(value, (list, tuple)):
|
||
return list(value)
|
||
if isinstance(value, str) and value.strip():
|
||
return [item.strip() for item in value.split(',') if item.strip()]
|
||
return list(default or [])
|
||
|
||
|
||
# --------------------------------------------------------------------- 派生
|
||
def model(role: str) -> str:
|
||
"""取某环节实际使用的模型名：role ∈ {asr_model, correct_model, split_model}"""
|
||
return str(get(f'models.{role}', ''))
|
||
|
||
|
||
# ----------------------------------------------------------------- 接入点/路由
|
||
def endpoint(name: str) -> dict:
|
||
"""取某个接入点定义（endpoints.<name>）"""
|
||
return get(f'endpoints.{name}', {}) or {}
|
||
|
||
|
||
def route(role: str) -> str:
|
||
"""取某环节走哪个接入点：role ∈ {asr, correct, split}"""
|
||
return str(get(f'routes.{role}', DEFAULT_ROUTES.get(role, '')))
|
||
|
||
|
||
def endpoint_for(role: str):
|
||
"""
|
||
解析某环节的接入点，返回 (接入点名字, 接入点定义)
|
||
|
||
未配置或缺少该协议必需字段时抛 RuntimeError（附修复指引），避免调用时才报难懂的错。
|
||
"""
|
||
name = route(role)
|
||
if not name:
|
||
raise RuntimeError(f"config.yml 未配置 routes.{role}")
|
||
cfg = endpoint(name)
|
||
if not cfg:
|
||
raise RuntimeError(
|
||
f"config.yml 的 routes.{role} 指向接入点 '{name}'，但 endpoints 下没有该定义")
|
||
kind = str(cfg.get('kind', 'openai')).lower()
|
||
if kind == 'dashscope':
|
||
if not cfg.get('http_base_url') and not cfg.get('websocket_base_url'):
|
||
raise RuntimeError(f"接入点 '{name}' 是 dashscope 类型，但缺少 http_base_url / websocket_base_url")
|
||
elif kind == 'openai':
|
||
if not cfg.get('base_url'):
|
||
raise RuntimeError(f"接入点 '{name}' 是 openai 类型，但缺少 base_url")
|
||
else:
|
||
raise RuntimeError(f"接入点 '{name}' 的 kind='{kind}' 不支持（仅支持 dashscope / openai）")
|
||
if not cfg.get('api_key_env'):
|
||
raise RuntimeError(f"接入点 '{name}' 未配置 api_key_env（密钥在 .env 中的变量名）")
|
||
return name, cfg
|
||
|
||
|
||
def openai_url(role: str) -> str:
|
||
"""OpenAI 兼容接入点的完整请求地址（base_url + chat_completions_path 拼接）"""
|
||
_, cfg = endpoint_for(role)
|
||
base = str(cfg.get('base_url', '')).rstrip('/')
|
||
path = str(cfg.get('chat_completions_path', '/chat/completions'))
|
||
if not path.startswith('/'):
|
||
path = '/' + path
|
||
return base + path
|
||
|
||
|
||
def api_key_env(role: str) -> str:
|
||
"""该环节密钥所在的 .env 变量名"""
|
||
return str(endpoint_for(role)[1].get('api_key_env', ''))
|
||
|
||
|
||
def api_key(role: str) -> str:
|
||
"""该环节的密钥值（从环境变量取，即 .env）"""
|
||
return os.getenv(api_key_env(role), '')
|
||
|
||
|
||
def apply_dashscope_endpoints(dashscope_module=None):
|
||
"""
|
||
把 config.yml 里 dashscope 类型的接入点地址写入 SDK
|
||
|
||
dashscope SDK 在 **import 时**读取 `DASHSCOPE_HTTP_BASE_URL` /
|
||
`DASHSCOPE_WEBSOCKET_BASE_URL`，因此在 config 加载阶段就写环境变量，
|
||
并在已 import 之后再次赋值模块属性，避免依赖 import 顺序。
|
||
"""
|
||
http_url = ws_url = ''
|
||
try:
|
||
if route('correct') and endpoint(route('correct')).get('kind') == 'dashscope':
|
||
http_url = endpoint(route('correct')).get('http_base_url', '')
|
||
if route('asr') and endpoint(route('asr')).get('kind') == 'dashscope':
|
||
ws_url = endpoint(route('asr')).get('websocket_base_url', '')
|
||
except Exception: # 配置不全时不阻塞启动，调用时才会明确报错
|
||
return
|
||
if not http_url:
|
||
http_url = endpoint('dashscope').get('http_base_url', '')
|
||
if not ws_url:
|
||
ws_url = endpoint('dashscope').get('websocket_base_url', '')
|
||
|
||
if http_url and not os.environ.get('DASHSCOPE_HTTP_BASE_URL'):
|
||
os.environ['DASHSCOPE_HTTP_BASE_URL'] = http_url
|
||
if ws_url and not os.environ.get('DASHSCOPE_WEBSOCKET_BASE_URL'):
|
||
os.environ['DASHSCOPE_WEBSOCKET_BASE_URL'] = ws_url
|
||
|
||
if dashscope_module is not None:
|
||
if http_url:
|
||
dashscope_module.base_http_api_url = http_url
|
||
if ws_url:
|
||
dashscope_module.base_websocket_api_url = ws_url
|
||
|
||
|
||
def validate_endpoints():
|
||
"""返回配置问题清单（供启动日志提示，不抛异常）"""
|
||
problems = []
|
||
for role in ('asr', 'correct', 'split'):
|
||
try:
|
||
name, cfg = endpoint_for(role)
|
||
except RuntimeError as e:
|
||
problems.append(str(e))
|
||
continue
|
||
if role == 'asr' and str(cfg.get('kind', '')).lower() != 'dashscope':
|
||
problems.append(
|
||
f"routes.asr -> '{name}'（kind={cfg.get('kind')}）：实时语音识别仅支持 kind=dashscope，"
|
||
f"换 ASR 供应商需要新增适配器")
|
||
key_env = cfg.get('api_key_env', '')
|
||
if key_env and not os.getenv(key_env):
|
||
problems.append(f"接入点 '{name}' 的密钥未设置：请在 .env 中填写 {key_env}")
|
||
return problems
|
||
|
||
|
||
# ----------------------------------------------------------------- 敏感项校验
|
||
def required_secrets():
|
||
"""运行所需的全部敏感项：数据库口令 + 各环节接入点的密钥"""
|
||
keys = {'MYSQL_PASSWORD'}
|
||
for role in ('asr', 'correct', 'split'):
|
||
try:
|
||
key_env = api_key_env(role)
|
||
except RuntimeError:
|
||
continue
|
||
if key_env:
|
||
keys.add(key_env)
|
||
return sorted(keys)
|
||
|
||
|
||
def missing_secrets():
|
||
"""返回缺失的敏感项列表"""
|
||
return [key for key in required_secrets() if not os.getenv(key)]
|
||
|
||
|
||
def check_secrets():
|
||
"""校验敏感项；缺失时记录 ERROR（不抛异常，便于只做只读操作的场景）"""
|
||
missing = missing_secrets()
|
||
if missing:
|
||
logger.error("缺少敏感配置 %s；请在 %s 中填写（模板见 .env.example）", ', '.join(missing), env.env_file())
|
||
return missing
|
||
|
||
|
||
def resolve_dir(value) -> Path:
|
||
"""目录配置解析：相对路径基于项目根"""
|
||
path = Path(str(value)).expanduser()
|
||
return path if path.is_absolute() else (BASE_DIR / path)
|
||
|
||
|
||
def video_dir() -> Path:
|
||
return resolve_dir(get('paths.video_dir', 'xwlb_video'))
|
||
|
||
|
||
def audio_dir() -> Path:
|
||
return resolve_dir(get('paths.audio_dir', 'audio_processing'))
|
||
|
||
|
||
def db_config() -> dict:
|
||
"""数据库连接参数（口令来自 .env）"""
|
||
return {
|
||
'host': get('mysql.host', 'localhost'),
|
||
'port': get_int('mysql.port', 3306),
|
||
'username': get('mysql.user', 'myquant'),
|
||
'password': os.getenv('MYSQL_PASSWORD', ''),
|
||
'database': get('mysql.database', 'myquant'),
|
||
}
|
||
|
||
|
||
def log_summary():
|
||
"""启动时打印一次生效配置（不含敏感项），并校验敏感项与接入点配置"""
|
||
source = str(LOADED_CONFIG_FILE) if LOADED_CONFIG_FILE else '内置默认值（未找到 config.yml）'
|
||
logger.info(
|
||
"配置来源: %s | MySQL %s@%s:%s/%s | 模型 asr=%s correct=%s split=%s | 校对=%s | 完成后清理=%s",
|
||
source,
|
||
get('mysql.user'), get('mysql.host'), get('mysql.port'), get('mysql.database'),
|
||
model('asr_model'), model('correct_model'), model('split_model'),
|
||
'开启' if get_bool('llm_correct.enabled', True) else '关闭',
|
||
'开启' if get_bool('cleanup.after_daily_run', True) else '关闭',
|
||
)
|
||
for role in ('asr', 'correct', 'split'):
|
||
try:
|
||
name, cfg = endpoint_for(role)
|
||
except RuntimeError as e:
|
||
logger.error("接入点配置错误: %s", e)
|
||
continue
|
||
if str(cfg.get('kind', '')).lower() == 'dashscope':
|
||
url = (cfg.get('websocket_base_url', '') if role == 'asr' else '') or cfg.get('http_base_url', '')
|
||
else:
|
||
url = openai_url(role)
|
||
logger.info("接入点 %-5s -> %-10s kind=%-9s %s（密钥变量 %s）",
|
||
role, name, cfg.get('kind'), url, cfg.get('api_key_env'))
|
||
for problem in validate_endpoints():
|
||
logger.error("配置检查: %s", problem)
|
||
check_secrets()
|
||
|
||
|
||
# config 加载阶段就把 dashscope 地址写入环境变量（SDK 在 import 时读取）
|
||
apply_dashscope_endpoints() |