财联社24小时实时跟踪

-- coding: utf-8 --

"""
A股电报新闻24小时实时监控系统 - 专业图形化界面
监控财联社电报新闻，实时获取重要资讯
"""

import requests
import hashlib
import time
import threading
import queue
import json
from datetime import datetime
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
from tkinter.font import Font

class TelegraphNewsMonitor:
"""财联社电报新闻监控器"""

def __init__(self):"""初始化会话和配置"""# 创建会话对象，复用连接self.session = requests.Session()# 配置重试策略retry_strategy = Retry(total=3,  # 最多重试3次backoff_factor=1,  # 重试间隔时间因子status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的HTTP状态码)adapter = HTTPAdapter(max_retries=retry_strategy)self.session.mount("http://", adapter)self.session.mount("https://", adapter)# 设置请求头，模拟浏览器访问self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36','Referer': 'https://www.cls.cn/telegraph','Accept': 'application/json, text/plain, */*','Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',}# API基础URLself.base_url = "https://www.cls.cn/nodeapi/telegraphList"# GUI消息队列self.gui_queue = queue.Queue()self.running = Falseself.monitor_thread = Noneprint("✅ 财联社电报新闻监控器初始化完成")def encrypts(self, s_url):"""生成API签名"""# 第一步：SHA1加密sha1 = hashlib.sha1()sha1.update(s_url.encode('utf-8'))sha1_result = sha1.hexdigest()# 第二步：对SHA1结果进行MD5加密md5 = hashlib.md5()md5.update(sha1_result.encode('utf-8'))sign = md5.hexdigest()return signdef get_telegraph_news(self, last_time=None, count=1, max_pages=10):"""获取电报新闻数据（支持自动翻页）"""# 如果是首次请求，使用当前时间戳if last_time is None:last_time = int(time.time())# 构造请求参数app = 'CailianpressWeb'os = 'web'rn = 20  # 每页返回20条新闻sv = '7.7.5'# 生成签名参数字符串（按特定顺序拼接）s_url = f'app={app}&last_time={last_time}&os={os}&rn={rn}&sv={sv}'# 生成签名sign = self.encrypts(s_url)# 构造完整URLurl = f'{self.base_url}?{s_url}&sign={sign}'try:# 发起请求response = self.session.get(url, headers=self.headers, timeout=10)response.raise_for_status()  # 检查HTTP状态码# 解析JSON数据data = response.json()# 检查返回数据结构if 'data' not in data or 'roll_data' not in data['data']:self.gui_queue.put({'type': 'error', 'message': f'第{count}页数据格式异常'})return []roll_data = data['data']['roll_data']if not roll_data:self.gui_queue.put({'type': 'info', 'message': f'第{count}页没有更多数据'})return []# 发送页面信息到GUIself.gui_queue.put({'type': 'page_info', 'page': count, 'count': len(roll_data)})# 处理并发送本页新闻到GUIfor news in roll_data:self.gui_queue.put({'type': 'news', 'data': news})# 判断是否继续翻页if count >= max_pages:self.gui_queue.put({'type': 'info', 'message': f'已达到设置的最大页数 ({max_pages})，停止获取'})return roll_data# 获取下一页的时间参数（本页最后一条新闻的时间）next_time = roll_data[-1]['ctime']time.sleep(1)  # 避免请求过快# 递归获取下一页next_page_data = self.get_telegraph_news(last_time=next_time,count=count + 1,max_pages=max_pages)# 合并数据return roll_data + next_page_dataexcept requests.exceptions.RequestException as e:self.gui_queue.put({'type': 'error', 'message': f'请求失败: {e}'})return []except Exception as e:self.gui_queue.put({'type': 'error', 'message': f'处理数据时出错: {e}'})return []def format_beijing_time(self, timestamp):"""将Unix时间戳转换为北京时间"""try:dt = datetime.fromtimestamp(timestamp)return dt.strftime('%Y-%m-%d %H:%M:%S')except:return str(timestamp)def clean_text_encoding(self, text):"""清理文本编码，防止乱码 - 增强处理能力"""if not text:return ""try:# 确保文本是字符串类型if isinstance(text, bytes):# 尝试多种编码方式for encoding in ['utf-8', 'gbk', 'gb2312', 'latin1']:try:text = text.decode(encoding)breakexcept:continueelse:# 如果所有编码都失败，使用忽略错误的方式text = text.decode('utf-8', errors='ignore')# 清理特殊字符和空白text = str(text).strip()# 移除不可打印字符text = ''.join(char for char in text if char.isprintable() or char.isspace())# 统一编码为UTF-8text = text.encode('utf-8', errors='ignore').decode('utf-8')return textexcept Exception as e:# 如果出现异常，返回清理后的原始文本return str(text).encode('utf-8', errors='ignore').decode('utf-8')def safe_format_news(self, news):"""安全格式化新闻内容"""try:# 安全获取各字段值title = self.clean_text_encoding(news.get('title', '无标题'))content = self.clean_text_encoding(news.get('content', '无内容'))ctime = news.get('ctime', 0)level = news.get('level', '')subjects = news.get('subjects', '')# 格式化时间time_str = self.format_beijing_time(ctime)# 等级映射level_map = {'A': '⭐⭐⭐ 重要','B': '⭐⭐ 一般','C': '⭐ 普通',}level_text = level_map.get(level, level)# 特殊处理subjects字段（可能是JSON格式）if subjects:# 处理已经是Python对象的情况if isinstance(subjects, (dict, list)):subjects_json = subjectselif isinstance(subjects, str):try:# 尝试解析JSON格式的subjectssubjects_json = json.loads(subjects)except json.JSONDecodeError:# 如果不是有效的JSON，直接清理subjects = self.clean_text_encoding(subjects)subjects_json = Noneelse:# 如果不是字符串也不是对象，转换为字符串subjects = self.clean_text_encoding(str(subjects))subjects_json = None# 处理解析后的JSON对象if subjects_json is not None:if isinstance(subjects_json, dict):# 如果是字典，提取名称或关键信息subjects = subjects_json.get('subject_name', '') or subjects_json.get('name', '') or ''elif isinstance(subjects_json, list):# 如果是列表，提取所有名称subject_names = []for item in subjects_json:if isinstance(item, dict):name = item.get('subject_name', '') or item.get('name', '')if name:subject_names.append(name)subjects = ', '.join(subject_names) or ''# 清理处理后的subjectssubjects = self.clean_text_encoding(subjects)return {'title': title,'time': time_str,'level': level_text,'subjects': subjects,'content': content,'importance': level}except Exception as e:return {'title': '格式化错误','time': '未知','level': '错误','subjects': '','content': f'格式化新闻时出错: {e}','importance': 'C'}def monitor_realtime(self, interval=60):"""实时监控模式（24小时持续监控）"""self.gui_queue.put({'type': 'status', 'message': f'启动24小时实时监控模式，检查间隔: {interval}秒'})iteration = 0last_news_time = Nonetry:while self.running:iteration += 1self.gui_queue.put({'type': 'iteration', 'iteration': iteration})# 获取最新一页新闻news_list = self.get_telegraph_news(max_pages=1)if news_list and len(news_list) > 0:# 检查是否有新新闻latest_time = news_list[0]['ctime']if last_news_time is None or latest_time > last_news_time:new_count = 0for news in news_list:if last_news_time is None or news['ctime'] > last_news_time:new_count += 1if new_count > 0:self.gui_queue.put({'type': 'new_news', 'count': new_count})last_news_time = latest_time# 等待下一次检查time.sleep(interval)except Exception as e:self.gui_queue.put({'type': 'error', 'message': f'监控过程中出错: {e}'})finally:self.gui_queue.put({'type': 'status', 'message': '监控已停止'})def start_realtime_monitor(self, interval=60):"""启动实时监控"""if self.running:return Falseself.running = Trueself.monitor_thread = threading.Thread(target=self.monitor_realtime, args=(interval,))self.monitor_thread.daemon = Trueself.monitor_thread.start()return Truedef stop_realtime_monitor(self):"""停止实时监控"""self.running = Falseif self.monitor_thread:self.monitor_thread.join(timeout=5)

class TelegraphNewsGUI:
"""财联社电报新闻图形化界面"""

def __init__(self):"""初始化GUI界面"""# 创建监控器实例self.monitor = TelegraphNewsMonitor()# 创建主窗口self.root = tk.Tk()self.root.title("A股电报新闻24小时实时监控系统")self.root.geometry("1200x800")self.root.minsize(1000, 700)# 设置字体 - 优化字体大小，加大标题self.title_font = Font(family="微软雅黑", size=16, weight="bold")  # 加大标题字体self.subtitle_font = Font(family="微软雅黑", size=14, weight="bold")  # 加大副标题字体self.news_title_font = Font(family="微软雅黑", size=12, weight="bold")  # 新闻标题字体self.normal_font = Font(family="微软雅黑", size=10)self.small_font = Font(family="微软雅黑", size=9)# 创建界面self.create_gui()# 启动GUI更新循环self.update_gui()def create_gui(self):"""创建GUI界面组件"""# ========== 顶部标题区域 ==========title_frame = ttk.Frame(self.root, relief=tk.RAISED, borderwidth=2)title_frame.pack(fill=tk.X, padx=10, pady=10)title_label = ttk.Label(title_frame, text="📰 A股电报新闻24小时实时监控系统", font=self.title_font, foreground="#2E86AB")title_label.pack(pady=15)subtitle_label = ttk.Label(title_frame, text="专业监控财联社电报新闻，实时获取重要资讯", font=self.subtitle_font, foreground="#666666")subtitle_label.pack(pady=5)# ========== 控制面板区域 ==========control_frame = ttk.LabelFrame(self.root, text="控制面板")control_frame.pack(fill=tk.X, padx=10, pady=5)# 添加自定义标题标签control_title_label = ttk.Label(control_frame, text="⚙️ 控制面板", font=self.subtitle_font)control_title_label.pack(pady=5)# 第一行：基本控制basic_control_frame = ttk.Frame(control_frame)basic_control_frame.pack(fill=tk.X, padx=10, pady=10)# 获取历史新闻ttk.Label(basic_control_frame, text="获取历史新闻:", font=self.normal_font).pack(side=tk.LEFT, padx=5)self.pages_var = tk.StringVar(value="3")pages_entry = ttk.Entry(basic_control_frame, textvariable=self.pages_var, width=5)pages_entry.pack(side=tk.LEFT, padx=5)ttk.Label(basic_control_frame, text="页").pack(side=tk.LEFT, padx=5)self.fetch_button = ttk.Button(basic_control_frame, text="📥 获取新闻", command=self.fetch_history_news)self.fetch_button.pack(side=tk.LEFT, padx=10)# 实时监控控制ttk.Label(basic_control_frame, text="实时监控间隔:", font=self.normal_font).pack(side=tk.LEFT, padx=5)self.interval_var = tk.StringVar(value="60")interval_entry = ttk.Entry(basic_control_frame, textvariable=self.interval_var, width=5)interval_entry.pack(side=tk.LEFT, padx=5)ttk.Label(basic_control_frame, text="秒").pack(side=tk.LEFT, padx=5)self.start_button = ttk.Button(basic_control_frame, text="▶️ 开始监控", command=self.start_monitoring)self.start_button.pack(side=tk.LEFT, padx=5)self.stop_button = ttk.Button(basic_control_frame, text="⏹️ 停止监控", command=self.stop_monitoring, state=tk.DISABLED)self.stop_button.pack(side=tk.LEFT, padx=5)# 第二行：高级控制advanced_control_frame = ttk.Frame(control_frame)advanced_control_frame.pack(fill=tk.X, padx=10, pady=5)self.clear_button = ttk.Button(advanced_control_frame, text="🗑️ 清除显示", command=self.clear_display)self.clear_button.pack(side=tk.LEFT, padx=5)self.export_button = ttk.Button(advanced_control_frame, text="💾 导出数据", command=self.export_data)self.export_button.pack(side=tk.LEFT, padx=5)# 状态显示self.status_var = tk.StringVar(value="状态: 就绪")status_label = ttk.Label(advanced_control_frame, textvariable=self.status_var, font=self.normal_font, foreground="#007ACC")status_label.pack(side=tk.RIGHT, padx=10)# ========== 新闻显示区域 ==========display_frame = ttk.LabelFrame(self.root, text="新闻显示")display_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)# 添加自定义标题标签display_title_label = ttk.Label(display_frame, text="📊 新闻显示", font=self.subtitle_font)display_title_label.pack(pady=5)# 创建标签页控件self.notebook = ttk.Notebook(display_frame)self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)# 实时新闻标签页self.realtime_tab = ttk.Frame(self.notebook)self.notebook.add(self.realtime_tab, text="📰 实时新闻")# 重要新闻标签页self.important_tab = ttk.Frame(self.notebook)self.notebook.add(self.important_tab, text="⚠️ 重要新闻")# 统计信息标签页self.stats_tab = ttk.Frame(self.notebook)self.notebook.add(self.stats_tab, text="📈 统计信息")# 设置实时新闻标签页内容self.setup_realtime_tab()self.setup_important_tab()self.setup_stats_tab()# 新闻计数器self.news_count = 0self.important_count = 0# 初始化统计信息self.update_stats()def setup_realtime_tab(self):"""设置实时新闻标签页"""# 新闻显示文本框self.news_text = scrolledtext.ScrolledText(self.realtime_tab, wrap=tk.WORD, font=self.normal_font)self.news_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)# 配置文本标签 - 优化字体大小，突出标题self.news_text.tag_configure("normal", foreground="black", font=self.normal_font)self.news_text.tag_configure("important", foreground="red", font=self.news_title_font)self.news_text.tag_configure("header", foreground="#2E86AB", font=self.subtitle_font)self.news_text.tag_configure("time", foreground="#666666", font=self.small_font)self.news_text.tag_configure("news_title", foreground="#1E3A8A", font=self.news_title_font, spacing1=5, spacing3=5)  # 新闻标题专用标签def setup_important_tab(self):"""设置重要新闻标签页"""# 重要新闻文本框self.important_text = scrolledtext.ScrolledText(self.important_tab, wrap=tk.WORD, font=self.normal_font)self.important_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)# 配置文本标签 - 优化字体大小，突出重要新闻self.important_text.tag_configure("normal", foreground="black", font=self.normal_font)self.important_text.tag_configure("important", foreground="red", font=self.news_title_font)self.important_text.tag_configure("header", foreground="#2E86AB", font=self.subtitle_font)self.important_text.tag_configure("news_title", foreground="#B91C1C", font=self.news_title_font,spacing1=5, spacing3=5)  # 重要新闻标题专用标签def setup_stats_tab(self):"""设置统计信息标签页"""# 统计信息文本框self.stats_text = scrolledtext.ScrolledText(self.stats_tab, wrap=tk.WORD, font=self.normal_font)self.stats_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)def fetch_history_news(self):"""获取历史新闻"""try:pages = int(self.pages_var.get())if pages <= 0:messagebox.showerror("错误", "页数必须大于0")returnself.status_var.set("状态: 正在获取历史新闻...")self.fetch_button.config(state=tk.DISABLED)# 在新线程中获取新闻def fetch_thread():news_list = self.monitor.get_telegraph_news(max_pages=pages)self.root.after(0, lambda: self.on_fetch_complete(news_list))threading.Thread(target=fetch_thread, daemon=True).start()except ValueError:messagebox.showerror("错误", "请输入有效的页数")def on_fetch_complete(self, news_list):"""获取新闻完成回调"""self.fetch_button.config(state=tk.NORMAL)self.status_var.set(f"状态: 获取完成，共{len(news_list)}条新闻")def start_monitoring(self):"""开始实时监控"""try:interval = int(self.interval_var.get())if interval <= 0:messagebox.showerror("错误", "间隔时间必须大于0")returnif self.monitor.start_realtime_monitor(interval):self.start_button.config(state=tk.DISABLED)self.stop_button.config(state=tk.NORMAL)self.status_var.set(f"状态: 实时监控中 (间隔: {interval}秒)")else:messagebox.showwarning("警告", "监控器已经在运行中")except ValueError:messagebox.showerror("错误", "请输入有效的时间间隔")def stop_monitoring(self):"""停止实时监控"""self.monitor.stop_realtime_monitor()self.start_button.config(state=tk.NORMAL)self.stop_button.config(state=tk.DISABLED)self.status_var.set("状态: 监控已停止")def clear_display(self):"""清除显示"""self.news_text.delete(1.0, tk.END)self.important_text.delete(1.0, tk.END)self.news_count = 0self.important_count = 0self.update_stats()def export_data(self):"""导出数据"""messagebox.showinfo("导出", "数据导出功能开发中...")def update_stats(self):"""更新统计信息"""self.stats_text.delete(1.0, tk.END)stats_info = f"""📊 统计信息

{'='*40}

📰 总新闻数量: {self.news_count}
⚠️ 重要新闻数量: {self.important_count}
📈 重要新闻比例: {self.important_count/max(self.news_count, 1)*100:.1f}%

⏰ 最后更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

💡 新闻等级说明:
⭐⭐⭐ 重要 (A级) - 重大事件、政策变动
⭐⭐ 一般 (B级) - 一般性新闻
⭐ 普通 (C级) - 普通资讯

🔧 系统状态: {'运行中' if self.monitor.running else '已停止'}
"""

    self.stats_text.insert(tk.END, stats_info)def display_news(self, formatted_news):"""显示新闻 - 优化显示效果，突出标题"""# 更新计数器self.news_count += 1if formatted_news['importance'] == 'A':self.important_count += 1# 格式化新闻显示timestamp = datetime.now().strftime('%H:%M:%S')# 在实时新闻标签页显示self.news_text.insert(tk.END, f"\n{'─'*80}\n", "header")# 使用专用新闻标题标签，突出显示title_tag = "news_title" if formatted_news['importance'] != 'A' else "important"self.news_text.insert(tk.END, f"📰 ", "normal")  # 图标self.news_text.insert(tk.END, f"{formatted_news['title']}\n", title_tag)self.news_text.insert(tk.END, f"⏰ {formatted_news['time']} | {formatted_news['level']}\n", "time")if formatted_news['subjects']:self.news_text.insert(tk.END, f"🔖 {formatted_news['subjects']}\n", "normal")self.news_text.insert(tk.END, f"📝 {formatted_news['content']}\n\n", "normal")self.news_text.see(tk.END)# 如果是重要新闻，在重要新闻标签页也显示if formatted_news['importance'] == 'A':self.important_text.insert(tk.END, f"\n{'─'*80}\n", "header")self.important_text.insert(tk.END, f"📰 ", "normal")  # 图标self.important_text.insert(tk.END, f"{formatted_news['title']}\n", "news_title")self.important_text.insert(tk.END, f"⏰ {formatted_news['time']} | {formatted_news['level']}\n", "time")if formatted_news['subjects']:self.important_text.insert(tk.END, f"🔖 {formatted_news['subjects']}\n", "normal")self.important_text.insert(tk.END, f"📝 {formatted_news['content']}\n\n", "normal")self.important_text.see(tk.END)# 更新统计信息self.update_stats()def update_gui(self):"""更新GUI显示"""# 处理队列中的消息while not self.monitor.gui_queue.empty():try:message = self.monitor.gui_queue.get_nowait()if message['type'] == 'news':formatted_news = self.monitor.safe_format_news(message['data'])self.display_news(formatted_news)elif message['type'] == 'page_info':page_info = f"📄 第 {message['page']} 页 - 获取到 {message['count']} 条新闻\n"self.news_text.insert(tk.END, page_info, "header")self.news_text.see(tk.END)elif message['type'] == 'status':self.status_var.set(f"状态: {message['message']}")elif message['type'] == 'iteration':self.status_var.set(f"状态: 实时监控中 - 第{message['iteration']}轮检查")elif message['type'] == 'new_news':new_info = f"🆕 发现 {message['count']} 条新新闻！\n"self.news_text.insert(tk.END, new_info, "important")self.news_text.see(tk.END)elif message['type'] == 'info':self.news_text.insert(tk.END, f"💡 {message['message']}\n", "normal")self.news_text.see(tk.END)elif message['type'] == 'error':self.news_text.insert(tk.END, f"❌ {message['message']}\n", "important")self.news_text.see(tk.END)except queue.Empty:break# 继续更新循环self.root.after(100, self.update_gui)def run(self):"""运行GUI"""# 设置窗口关闭事件self.root.protocol("WM_DELETE_WINDOW", self.on_closing)# 启动GUIself.root.mainloop()def on_closing(self):"""窗口关闭事件"""if self.monitor.running:self.monitor.stop_realtime_monitor()self.root.destroy()

def main():
"""主函数"""
# 创建并运行GUI
gui = TelegraphNewsGUI()
gui.run()

if name == "main":
main()