财联社实时监控50条

import requests
import json
import time
from requests.adapters import HTTPAdapter

referer_url = "https://www.cls.cn/telegraph"

cookie = ""

headers = {
"Accept": "/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
# "Cookie": cookie,
"Host": "www.cls.cn",
"Referer": referer_url,
"User-Agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.117 Safari/537.36"
}

API基础URL

api_base = "https://www.cls.cn/nodeapi/telegraphList"

生成签名

def encrypts(s_url):
"""生成API签名"""
# 第一步：SHA1加密
import hashlib
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 sign

构造带签名的URL

def build_url(last_time=None):
"""构造带签名的API请求URL"""
if last_time is None:
last_time = int(time.time())

# 构造请求参数
app = 'CailianpressWeb'
os = 'web'
rn = 100  # 每页返回100条新闻
sv = '7.7.5'# 生成签名参数字符串（按特定顺序拼接）
s_url = f'app={app}&last_time={last_time}&os={os}&rn={rn}&sv={sv}'# 生成签名
sign = encrypts(s_url)# 构造完整URL
url = f'{api_base}?{s_url}&sign={sign}'
return url

def get_json_data(headers):
"""安全获取和解析JSON数据，使用正确的API签名"""
try:
# 使用Session进行请求
s = requests.Session()
s.mount('http://', HTTPAdapter(max_retries=3))
s.mount('https://', HTTPAdapter(max_retries=3))

    # 构建带签名的URLurl = build_url()print(f"请求URL: {url}")response = s.get(url, timeout=10, headers=headers)response.raise_for_status()  # 检查请求是否成功# 确保正确的编码response.encoding = 'utf-8'# 安全解析JSON数据results = response.json()# 检查返回数据结构if 'data' not in results:print("⚠️  响应中没有'data'字段")print(f"响应内容: {results}")return []if 'roll_data' not in results['data']:print("⚠️  响应中没有'roll_data'字段")print(f"data内容: {results['data']}")return []data = results['data']['roll_data']all_list = []for i in data:try:time_tuple_1 = time.localtime(i['ctime'])bj_time = time.strftime("%Y/%m/%d %H:%M:%S", time_tuple_1)# 清理内容，防止乱码content = i.get('content', '')if isinstance(content, str):# 确保字符串编码正确content = content.encode('utf-8', errors='ignore').decode('utf-8')else:content = str(content)all_list.append([bj_time, content])except Exception as e:print(f"⚠️  处理数据项时出错: {e}")print(f"数据项: {i}")continuereturn all_listexcept requests.exceptions.RequestException as e:print(f"❌ 请求失败: {e}")return []
except json.JSONDecodeError as e:print(f"❌ JSON解析错误: {e}")print(f"响应文本: {response.text}")return []
except Exception as e:print(f"❌ 处理数据时出错: {e}")import tracebacktraceback.print_exc()return []

def get_all_list():
all_list = get_json_data(headers)
return all_list
print(get_all_list())

PyQt5相关导入

from PyQt5 import QtWidgets
from PyQt5.QtGui import QIcon, QFont
from PyQt5.QtWidgets import (
QApplication, QMessageBox, QTableWidgetItem, QHeaderView, QMainWindow,
QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QStatusBar,
QSplitter
)
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer
import sys

不需要导入Cls7X24相关模块，使用本地的get_all_list函数

不需要导入UI模块，直接在代码中创建界面

创建一个后台线程类，用于异步获取新闻数据

class NewsThread(QThread):
# 定义信号，用于传递新闻数据和状态信息
news_fetched = pyqtSignal(list)
error_occurred = pyqtSignal(str)
status_changed = pyqtSignal(str)

def run(self):try:self.status_changed.emit("正在获取新闻数据...")news_list = get_all_list()self.news_fetched.emit(news_list)self.status_changed.emit(f"成功获取到 {len(news_list)} 条新闻")except Exception as e:self.error_occurred.emit(f"获取新闻失败: {str(e)}")self.status_changed.emit("获取新闻失败")

创建主窗口类

class NewsMonitorWindow(QMainWindow):
def init(self):
super().init()
self.initThread()
self.initTimer()
self.initUI()

def initUI(self):# 设置窗口属性self.setWindowTitle("实时新闻监控系统")self.setGeometry(100, 100, 1200, 800)self.setWindowIcon(QIcon())# 创建中心部件central_widget = QWidget()self.setCentralWidget(central_widget)# 创建主布局main_layout = QVBoxLayout(central_widget)# 创建标题title_label = QLabel("财联社实时新闻监控")title_font = QFont("微软雅黑", 18, QFont.Bold)title_label.setFont(title_font)title_label.setAlignment(Qt.AlignCenter)title_label.setStyleSheet("color: #2c3e50; margin: 10px 0;")main_layout.addWidget(title_label)# 创建控制栏control_layout = QHBoxLayout()# 刷新按钮self.refresh_btn = QPushButton("刷新新闻")self.refresh_btn.setFont(QFont("微软雅黑", 10))self.refresh_btn.setStyleSheet("padding: 8px 16px; background-color: #3498db; color: white; border-radius: 4px;")self.refresh_btn.clicked.connect(self.refreshNews)control_layout.addWidget(self.refresh_btn)# 自动刷新复选框self.auto_refresh_check = QtWidgets.QCheckBox("自动刷新")self.auto_refresh_check.setFont(QFont("微软雅黑", 10))self.auto_refresh_check.setChecked(False)self.auto_refresh_check.stateChanged.connect(self.toggleAutoRefresh)control_layout.addWidget(self.auto_refresh_check)# 自动刷新间隔self.interval_label = QLabel("间隔：")self.interval_label.setFont(QFont("微软雅黑", 10))control_layout.addWidget(self.interval_label)self.interval_combo = QtWidgets.QComboBox()self.interval_combo.addItems(["1秒", "3秒", "5秒", "10秒", "30秒", "1分钟", "5分钟", "10分钟"])self.interval_combo.setFont(QFont("微软雅黑", 10))self.interval_combo.currentIndexChanged.connect(self.changeInterval)control_layout.addWidget(self.interval_combo)# 状态栏信息self.status_label = QLabel("就绪")self.status_label.setFont(QFont("微软雅黑", 10))self.status_label.setStyleSheet("color: #7f8c8d;")control_layout.addStretch()control_layout.addWidget(self.status_label)main_layout.addLayout(control_layout)# 创建新闻表格self.news_table = QtWidgets.QTableWidget()self.news_table.setColumnCount(2)self.news_table.setHorizontalHeaderLabels(["时间", "新闻内容"])# 设置表格属性self.news_table.setAlternatingRowColors(True)self.news_table.setEditTriggers(QtWidgets.QAbstractItemView.NoEditTriggers)self.news_table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectRows)self.news_table.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection)# 设置表头header = self.news_table.horizontalHeader()header.setSectionResizeMode(0, QHeaderView.ResizeToContents)header.setSectionResizeMode(1, QHeaderView.Stretch)header.setFont(QFont("微软雅黑", 11, QFont.Bold))# 设置表格字体self.news_table.setFont(QFont("微软雅黑", 10))main_layout.addWidget(self.news_table)# 创建状态栏self.statusBar = QStatusBar()self.setStatusBar(self.statusBar)self.statusBar.showMessage("就绪")# 初始加载新闻self.refreshNews()def initThread(self):# 创建新闻获取线程self.news_thread = NewsThread()self.news_thread.news_fetched.connect(self.displayNews)self.news_thread.error_occurred.connect(self.showError)self.news_thread.status_changed.connect(self.updateStatus)def initTimer(self):# 创建自动刷新计时器self.auto_refresh_timer = QTimer()self.auto_refresh_timer.timeout.connect(self.refreshNews)self.auto_refresh_timer.setInterval(30000)  # 默认30秒def refreshNews(self):# 禁用刷新按钮self.refresh_btn.setEnabled(False)self.refresh_btn.setText("刷新中...")# 启动新闻获取线程if not self.news_thread.isRunning():self.news_thread.start()def displayNews(self, news_list):# 清空表格self.news_table.setRowCount(0)# 设置行数self.news_table.setRowCount(len(news_list))# 填充数据for row, (news_time, content) in enumerate(news_list):# 时间列time_item = QTableWidgetItem(news_time)time_item.setFont(QFont("微软雅黑", 10))time_item.setTextAlignment(Qt.AlignCenter | Qt.AlignVCenter)self.news_table.setItem(row, 0, time_item)# 内容列content_item = QTableWidgetItem(content)content_item.setFont(QFont("微软雅黑", 10))content_item.setTextAlignment(Qt.AlignLeft | Qt.AlignVCenter)self.news_table.setItem(row, 1, content_item)# 设置行高self.news_table.resizeRowToContents(row)# 恢复刷新按钮self.refresh_btn.setEnabled(True)self.refresh_btn.setText("刷新新闻")def showError(self, error_msg):# 显示错误消息QMessageBox.critical(self, "错误", error_msg)# 恢复刷新按钮self.refresh_btn.setEnabled(True)self.refresh_btn.setText("刷新新闻")def updateStatus(self, status_msg):# 更新状态栏self.statusBar.showMessage(status_msg)self.status_label.setText(status_msg)def toggleAutoRefresh(self, state):# 切换自动刷新if state == Qt.Checked:self.auto_refresh_timer.start()self.updateStatus(f"自动刷新已开启，间隔：{self.interval_combo.currentText()}")else:self.auto_refresh_timer.stop()self.updateStatus("自动刷新已关闭")def changeInterval(self):# 改变自动刷新间隔interval_text = self.interval_combo.currentText()if interval_text == "1秒":interval = 1000elif interval_text == "3秒":interval = 3000elif interval_text == "5秒":interval = 5000elif interval_text == "10秒":interval = 10000elif interval_text == "30秒":interval = 30000elif interval_text == "1分钟":interval = 60000elif interval_text == "5分钟":interval = 300000elif interval_text == "10分钟":interval = 600000else:interval = 30000  # 默认30秒self.auto_refresh_timer.setInterval(interval)# 如果自动刷新已开启，更新状态if self.auto_refresh_check.isChecked():self.updateStatus(f"自动刷新已开启，间隔：{interval_text}")

GUI入口

if name == 'main':
app = QApplication(sys.argv)
window = NewsMonitorWindow()
window.show()
sys.exit(app.exec_())