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 项离线自检（配置/清理/事实守卫/解析/隧道）
289 lines
12 KiB
Python
289 lines
12 KiB
Python
import requests
|
||
from bs4 import BeautifulSoup
|
||
import re,os,subprocess
|
||
from datetime import timedelta, date
|
||
import yt_dlp
|
||
import env
|
||
import config
|
||
from audioRead import *
|
||
from cleanup import maybe_cleanup_after_run
|
||
from newsProcess import STATUS_FAILED, news_to_db
|
||
import logging
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def get_xwlb_video_link(url):
|
||
"""
|
||
从央视网新闻联播页面抓取历史完整版视频链接
|
||
"""
|
||
headers = {
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
|
||
'Referer': 'https://tv.cctv.com/'
|
||
}
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
response.encoding = 'utf-8'
|
||
if response.status_code != 200:
|
||
logger.error(f"请求失败，状态码: {response.status_code}")
|
||
#print(f"请求失败，状态码: {response.status_code}")
|
||
return []
|
||
except Exception as e:
|
||
logger.error(f"请求异常: {e}")
|
||
return []
|
||
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
video_links = []
|
||
|
||
# 查找所有包含“完整版《新闻联播》”的链接
|
||
# 方法1: 查找包含 <i class="sql0">完整版</i>《新闻联播》 的 a 标签
|
||
for a_tag in soup.find_all('a', href=True):
|
||
# 检查文本中是否包含“完整版”和“新闻联播”
|
||
title_text = a_tag.get_text(strip=True)
|
||
inner_html = str(a_tag)
|
||
|
||
# 判断是否是“完整版《新闻联播》”的链接
|
||
if ('完整版' in title_text and '新闻联播' in title_text) or \
|
||
(re.search(r'<i[^>]*>完整版</i>\s*《新闻联播》', inner_html)):
|
||
|
||
video_url = a_tag['href']
|
||
# 提取日期信息（从标题或链接中）
|
||
date_match = re.search(r'\d{8}', title_text)
|
||
if not date_match:
|
||
# 从链接中提取日期，如 /2025/09/25/...VID...250925.shtml
|
||
date_match = re.search(r'/(\d{4})/(\d{2})/(\d{2})/', video_url)
|
||
if date_match:
|
||
year, month, day = date_match.groups()
|
||
date_str = f"{year}{month}{day}"
|
||
else:
|
||
date_str = "未知日期"
|
||
else:
|
||
date_str = date_match.group()
|
||
|
||
video_links.append({
|
||
'date': date_str,
|
||
'title': title_text.strip(),
|
||
'url': video_url,
|
||
'page_url': url
|
||
})
|
||
logger.info(f"✅ 找到新闻联播完整版: {date_str} -> {video_url}")
|
||
|
||
# 原实现误返回循环内的单个 video_url（且无匹配时该变量未定义会 UnboundLocalError）
|
||
return video_links
|
||
|
||
def xwlb_urls(start: str, end: str):
|
||
"""
|
||
start/end 格式 '20240925'
|
||
返回列表，如 ['https://tv.cctv.com/lm/xwlb/day/20240925.shtml', ...]
|
||
"""
|
||
d0 = date(int(start[:4]), int(start[4:6]), int(start[6:8]))
|
||
|
||
d1 = date(int(end[:4]), int(end[4:6]), int(end[6:8]))
|
||
urls = []
|
||
for n in range((d1 - d0).days + 1):
|
||
day = d0 + timedelta(days=n)
|
||
urls.append({"url": f"https://tv.cctv.com/lm/xwlb/day/{day:%Y%m%d}.shtml", "date": f"{day:%Y%m%d}"})
|
||
#print(urls)
|
||
return urls
|
||
|
||
def get_all_video_links(start: str, end: str):
|
||
"""逐日解析完整版视频链接，返回 [{'url':..., 'date': 'YYYYMMDD'}, ...]
|
||
|
||
原实现把 get_xwlb_video_link() 的返回值（单个 URL 或空列表）直接包成 dict，
|
||
导致空链接也会进入下载流程；现改为按天取第一条有效链接并显式跳过失败日期。
|
||
"""
|
||
base_urls = xwlb_urls(start, end)
|
||
video_urls = []
|
||
for item in base_urls:
|
||
links = get_xwlb_video_link(item['url'])
|
||
if not links:
|
||
logger.error(f"❌ {item['date']} 未解析到「完整版《新闻联播》」链接，跳过: {item['url']}")
|
||
continue
|
||
if len(links) > 1:
|
||
logger.warning(f"⚠️ {item['date']} 解析到 {len(links)} 条完整版链接，取第一条: {links[0]['url']}")
|
||
video_urls.append({"url": links[0]['url'], "date": item['date']})
|
||
|
||
return video_urls
|
||
|
||
'''
|
||
get_xwlb_video_link() 方法获得的url，
|
||
urls like: https://tv.cctv.com/2024/10/30/VIDEUlPz1Qusy41JFQj3LMLd241030.shtml
|
||
通过yt-dlp下载视频，保存为mp4文件，并用ffmpeg提取音频为mp3文件,文件名使用url 的日期部分，如上面的url应保存为 20241030.mp4 和 20241030.mp3
|
||
文件保存路径为当前目录下的 xwlb_video 文件夹，若不存在则创建。
|
||
'''
|
||
|
||
|
||
def download_and_extract_audio(video_url,date_str,download_dir):
|
||
"""
|
||
使用yt-dlp下载视频并提取音频；已存在音频时跳过（幂等，支持重跑）
|
||
返回: mp3 路径（成功）或 None（失败）
|
||
"""
|
||
# 从URL中提取日期
|
||
|
||
if not video_url:
|
||
logger.error(f"❌ {date_str} 视频链接为空，跳过下载")
|
||
return None
|
||
|
||
download_dir = os.fspath(download_dir)
|
||
os.makedirs(download_dir, exist_ok=True)
|
||
|
||
# 构建文件路径
|
||
mp4_path = os.path.join(download_dir, f"{date_str}.mp4")
|
||
mp3_path = os.path.join(download_dir, f"{date_str}.mp3")
|
||
|
||
# 幂等：音频已抽取过就不再重复下载（原实现每次重跑都重新下载 ~100MB）
|
||
if os.path.exists(mp3_path) and os.path.getsize(mp3_path) > 0:
|
||
logger.info(f"⏭️ {date_str} 音频已存在，跳过下载: {mp3_path}")
|
||
return mp3_path
|
||
|
||
try:
|
||
# 使用yt-dlp库下载视频
|
||
logger.info(f"📥 开始下载 {date_str} 的视频...")
|
||
# 配置yt-dlp选项
|
||
ydl_opts = {
|
||
'outtmpl': mp4_path,
|
||
'format': 'best[ext=mp4]/best',
|
||
'progress_hooks': [lambda d: print(f"\r📥 下载进度: {d.get('_percent_str', 'N/A').strip()} | {d.get('_speed_str', 'N/A').strip()} | 已下载: {d.get('_downloaded_bytes_str', 'N/A')}", end='') if d['status'] == 'downloading' else None],
|
||
}
|
||
|
||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||
ydl.download([video_url])
|
||
logger.info(f"📥 下载 {date_str} 完成")
|
||
|
||
# 使用ffmpeg提取音频
|
||
logger.info(f"🎵 开始提取 {date_str} 的音频...")
|
||
# 原实现 stdout=None/stderr=None 会把 ffmpeg 完整 banner 灌进日志（日志已 34MB），
|
||
# 现仅在失败时记录 stderr 尾部
|
||
proc = subprocess.run([
|
||
"ffmpeg",
|
||
"-i", mp4_path,
|
||
"-c:a", "libmp3lame", # 明确指定MP3编码器
|
||
"-q:a", "0",
|
||
"-map", "a",
|
||
mp3_path,
|
||
"-y" # 覆盖已存在文件
|
||
], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
||
if proc.returncode != 0:
|
||
raise RuntimeError(f"ffmpeg 提取音频失败(returncode={proc.returncode}): {(proc.stderr or '')[-500:]}")
|
||
logger.info(f"🎵 提取 {date_str} 音频完成")
|
||
|
||
logger.info(f"✅ 成功处理 {date_str}: {mp4_path}, {mp3_path}")
|
||
return mp3_path
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 处理 {date_str} 时发生异常: {e}")
|
||
return None
|
||
|
||
# 在get_all_video_links函数后添加调用代码
|
||
def process_videos(start_date, end_date):
|
||
"""
|
||
处理指定日期范围内的所有视频
|
||
|
||
返回值（供定时任务判断成败）:
|
||
dict: {'total': 待处理天数, 'ok': 全链路成功天数, 'failed': 失败天数}
|
||
total=0 表示一天都没抓到链接，按失败处理（页面可能还没发布）。
|
||
"""
|
||
video_urls = get_all_video_links(start_date, end_date)
|
||
if not video_urls:
|
||
logger.error(f"❌ {start_date}~{end_date} 未获取到任何视频链接，流程结束")
|
||
return {'total': 0, 'ok': 0, 'failed': 0}
|
||
|
||
summary = {'total': 0, 'ok': 0, 'failed': 0}
|
||
|
||
# 目录来自 config.yml 的 paths.*（相对项目根，可写绝对路径）
|
||
download_dir = config.video_dir()
|
||
output_folder = config.audio_dir()
|
||
|
||
for sub_url in video_urls:
|
||
date_str = sub_url.get('date')
|
||
url = sub_url.get('url')
|
||
if not url or not date_str:
|
||
logger.error(f"❌ 跳过无效条目: {sub_url}")
|
||
summary['failed'] += 1
|
||
continue
|
||
|
||
summary['total'] += 1
|
||
print("=" * 80)
|
||
mp3_path = download_and_extract_audio(url, date_str, download_dir)
|
||
if not mp3_path or not os.path.exists(mp3_path):
|
||
logger.error(f"❌ {date_str} 音频不可用，跳过识别与入库")
|
||
print("=" * 80)
|
||
summary['failed'] += 1
|
||
continue
|
||
|
||
# 处理长音频 + 切分入库；两者都成功才算当天任务完成
|
||
day_ok = False
|
||
try:
|
||
audio_result = process_long_audio(mp3_path, output_folder, date_str)
|
||
if audio_result and audio_result.get('failed', 1) > 0:
|
||
# 识别不完整：**不执行切分**。否则会给这天写出 ext，
|
||
# 让下游把"半天内容"当成完整一天（且失败原因被掩盖）。
|
||
# 分片数据仍在 xwlb_daily 里，重试会补全。
|
||
logger.error(f"❌ {date_str} 识别不完整（{audio_result.get('ok')}/{audio_result.get('chunks')} 片），"
|
||
f"跳过切分以免发布不完整的一天")
|
||
split_status = 'skipped(incomplete-asr)'
|
||
else:
|
||
split_status = news_to_db(date_str)
|
||
day_ok = bool(audio_result and audio_result.get('failed', 1) == 0
|
||
and split_status != STATUS_FAILED)
|
||
logger.info(f"{date_str} 任务结果: 识别 {audio_result.get('ok', 0)}/{audio_result.get('chunks', 0)} 片，切分 {split_status}")
|
||
except Exception as e:
|
||
logger.error(f"❌ {date_str} 处理失败: {e}")
|
||
print("=" * 80)
|
||
|
||
summary['ok' if day_ok else 'failed'] += 1
|
||
|
||
# 当天全程任务完成后清理 mp3/mp4 与 wav（config.yml: cleanup.*）
|
||
maybe_cleanup_after_run(date_str, day_ok)
|
||
|
||
logger.info(f"本次运行汇总: 共 {summary['total']} 天，成功 {summary['ok']} 天，失败 {summary['failed']} 天")
|
||
return summary
|
||
|
||
|
||
# ========================
|
||
# 主程序执行
|
||
# ========================
|
||
if __name__ == "__main__":
|
||
|
||
import sys
|
||
import re
|
||
from datetime import datetime
|
||
|
||
# 检查命令行参数
|
||
if len(sys.argv) < 2:
|
||
print("用法: python getVideo5.py <start_date> <end_date>")
|
||
print("日期格式: YYYYMMDD")
|
||
sys.exit(1)
|
||
|
||
start_date = sys.argv[1]
|
||
end_date = sys.argv[2] if len(sys.argv) > 2 and sys.argv[2] else start_date
|
||
|
||
# 检查日期格式
|
||
date_pattern = r'^\d{8}$'
|
||
if not re.match(date_pattern, start_date) or not re.match(date_pattern, end_date):
|
||
print("错误: 日期格式必须为 YYYYMMDD")
|
||
sys.exit(1)
|
||
|
||
# 检查日期有效性
|
||
try:
|
||
#start_dt = datetime.strptime(start_date, '%Y%m%d')
|
||
#end_dt = datetime.strptime(end_date, '%Y%m%d')
|
||
|
||
if start_date > end_date:
|
||
print(f"错误: start_date {start_date} 不能大于 end_date {end_date}")
|
||
sys.exit(1)
|
||
print("正在抓取央视《新闻联播》历史完整版视频链接...")
|
||
print("=" * 80)
|
||
config.log_summary()
|
||
summary = process_videos(start_date, end_date)
|
||
# 退出码供定时任务判断: 0=当天全链路成功; 1=失败（含"没抓到链接"，交由重试）
|
||
if summary and summary['ok'] > 0 and summary['failed'] == 0:
|
||
sys.exit(0)
|
||
print(f"❌ 运行失败: {summary}")
|
||
sys.exit(1)
|
||
except ValueError as e:
|
||
print(f"错误: 无效日期 - {e}")
|
||
sys.exit(1)
|
||
|