import sys import time import requests import sqlite3 from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QLineEdit, QPushButton, QLabel, QTextEdit, QMessageBox, \ QHBoxLayout from PyQt5.QtCore import QThread, pyqtSignal class RequestThread(QThread): # 定义一个信号，用于传递请求结果 result_signal = pyqtSignal(str) def __init__(self, category_code, session_info, parent=None): super().__init__(parent) self.category_code = category_code self.session_info = session_info def run(self): my_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.131 Safari/537.36", "Host": "buff.163.com", "Referer": "https://buff.163.com/market/?game=csgo", "X-Requested-With": "XMLHttpRequest", 'Cookie': f'session={self.session_info}' } page_num = 1 page_i = 1 print("开始发送请求...") # 添加开始请求的提示 while page_i <= page_num: # 构造请求URL url = f"https://buff.163.com/api/market/goods?game=csgo&page_num={page_i}&page_size=95&itemset={self.category_code}&tab=selling&_=1745160625260" # 发送GET请求 try: response = requests.get(headers=my_headers, url=url) if response.status_code == 200: print(f"第 {page_i} 页请求成功") # 添加请求成功的提示 if page_i == 1: page_num = response.json()['data']['total_page'] page_i += 1 # 遍历 response.json()['data']['items'] for item in response.json()['data']['items']: # 获取每个item的goods_id goods_id = item['id'] # 通过信号传递结果 self.result_signal.emit(f"{goods_id}") time.sleep(20) else: print(f"第 {page_i} 页请求失败: {response}") # 添加请求失败的提示 except Exception as e: print(f"第 {page_i} 页请求失败: {e}") # 添加请求失败的提示 # 在while循环结束后发出完成信号 self.result_signal.emit("请求完成") class BuffApp(QWidget): def __init__(self): super().__init__() self.initUI() self.load_data() def initUI(self): # 设置窗口标题 self.setWindowTitle('获取buff排除商品id') # 设置窗口尺寸为500x500 self.setFixedSize(500, 500) # 创建垂直布局 layout = QVBoxLayout() # 创建类别编码标签 category_code_label = QLabel('收藏品类别编码：', self) layout.addWidget(category_code_label) # 创建类别编码输入框 self.category_code_input = QLineEdit(self) self.category_code_input.setPlaceholderText('请输入类别编码') layout.addWidget(self.category_code_input) # 创建session信息标签 session_label = QLabel('session信息：', self) layout.addWidget(session_label) # 创建session信息输入框 self.session_input = QLineEdit(self) self.session_input.setPlaceholderText('请输入session信息') layout.addWidget(self.session_input) # 创建水平布局用于放置按钮 button_layout = QHBoxLayout() # 创建请求按钮 self.request_button = QPushButton('发送请求', self) self.request_button.clicked.connect(self.send_request1) button_layout.addWidget(self.request_button) # 创建停止按钮 self.stop_button = QPushButton('停止', self) self.stop_button.clicked.connect(self.stop_request) button_layout.addWidget(self.stop_button) # 将按钮布局添加到主布局中 layout.addLayout(button_layout) # 创建文本显示区域 self.text_area = QTextEdit(self) # self.text_area.setReadOnly(True) # 设置为只读 layout.addWidget(self.text_area) # 创建保存按钮 self.save_button = QPushButton('保存', self) self.save_button.clicked.connect(self.save_data) button_layout.addWidget(self.save_button) # 设置布局 self.setLayout(layout) def save_data(self): category_code = self.category_code_input.text() session_info = self.session_input.text() # 连接SQLite数据库 conn = sqlite3.connect('buff_data.db') cursor = conn.cursor() # 创建表（如果不存在） cursor.execute('''CREATE TABLE IF NOT EXISTS settings (id INTEGER PRIMARY KEY, category_code TEXT, session_info TEXT)''') # 清空表并插入新数据 cursor.execute('DELETE FROM settings') cursor.execute('INSERT INTO settings (category_code, session_info) VALUES (?, ?)', (category_code, session_info)) # 提交更改并关闭连接 conn.commit() conn.close() QMessageBox.information(self, "保存成功", "数据已成功保存！") def load_data(self): # 连接SQLite数据库 conn = sqlite3.connect('buff_data.db') cursor = conn.cursor() # 创建表（如果不存在） cursor.execute('''CREATE TABLE IF NOT EXISTS settings (id INTEGER PRIMARY KEY, category_code TEXT, session_info TEXT)''') # 查询数据 cursor.execute('SELECT category_code, session_info FROM settings LIMIT 1') result = cursor.fetchone() if result: self.category_code_input.setText(result[0]) self.session_input.setText(result[1]) # 关闭连接 conn.close() def send_request1(self): category_code = self.category_code_input.text() session_info = self.session_input.text() # 创建并启动线程 self.request_thread = RequestThread(category_code, session_info) self.request_thread.result_signal.connect(self.update_text_area) self.request_thread.start() def stop_request(self): if hasattr(self, 'request_thread') and self.request_thread.isRunning(): self.request_thread.terminate() self.request_thread.wait() QMessageBox.information(self, "停止", "请求已停止！") def update_text_area(self, text): # 更新UI if text == "请求完成": QMessageBox.information(self, "请求完成", "所有请求已完成！") else: self.text_area.append(text) QApplication.processEvents() # 强制更新 UI if __name__ == '__main__': app = QApplication(sys.argv) ex = BuffApp() ex.show() sys.exit(app.exec_())