import requests from bs4 import BeautifulSoup import re,os,subprocess from datetime import timedelta, date import yt_dlp from audioRead import * from newsProcess import 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: 查找包含 完整版《新闻联播》 的 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']*>完整版\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}") return video_url 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): base_urls=xwlb_urls(start,end) #print(base_urls) video_urls = [] for url in base_urls: video=get_xwlb_video_link(url['url']) video_urls.append({"url":video,"date":url['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下载视频并提取音频 """ # 从URL中提取日期 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") 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} 的音频...") result = subprocess.run([ "ffmpeg", "-i", mp4_path, "-c:a", "libmp3lame", # 明确指定MP3编码器 "-q:a", "0", "-map", "a", mp3_path, "-y" # 覆盖已存在文件 ], check=True, stdout=None, stderr=None) logger.info(f"🎵 提取 {date_str} 音频完成") logger.info(f"✅ 成功处理 {date_str}: {mp4_path}, {mp3_path}") except Exception as e: logger.error(f"❌ 处理 {date_str} 时发生异常: {e}") # 在get_all_video_links函数后添加调用代码 def process_videos(start_date, end_date): """ 处理指定日期范围内的所有视频 """ video_urls = get_all_video_links(start_date, end_date) for sub_url in video_urls: if sub_url: # 确保url不为空 """ date_match = re.search(r'/(\d{4})/(\d{2})/(\d{2})/', url) if not date_match: print(f"❌ 无法从URL提取日期: {url}") return year, month, day = date_match.groups() date_str = f"{year}{month}{day}" """ date_str= sub_url['date'] url = sub_url['url'] # 创建保存目录 download_dir = "/home/simon/myquant/djapi/api/video/xwlb_video" download_and_extract_audio(url,date_str,download_dir) print("=" * 80) # MP3文件路径 mp3_path = os.path.join(download_dir, f"{date_str}.mp3") # 分析提示词 # prompt = "请总结这段由中国中央电视台新闻联播音频转为文字的文本，理解其主要内容并提取其中的关键信息。" # 输出文件夹 output_folder = "/home/simon/myquant/djapi/api/video/audio_processing" # 处理长音频 try: result = process_long_audio(mp3_path, output_folder,date_str) news_to_db(date_str) # 打印分析结果 print("分析结果:\n") print(result) except Exception as e: print(f"处理失败: {e}") print("=" * 80) # ======================== # 主程序执行 # ======================== if __name__ == "__main__": import sys import re from datetime import datetime # 检查命令行参数 if len(sys.argv) < 2: print("用法: python getVideo5.py ") 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) process_videos(start_date, end_date) exit() except ValueError as e: print(f"错误: 无效日期 - {e}") sys.exit(1)