今日头条微头条API批量发文桌面工具
基于 Python + PyQt6 开发，对接今日头条创作者开放平台官方API，合规安全无封号风险。
环境要求
- 依赖库：PyQt6、requests、Pillow
安装依赖
pip install PyQt6 requests Pillow
运行
python toutiao_publisher.py
今日头条创作者开放平台 API 申请步骤
- 创建应用，获取
client_id 和 client_secret - 按照平台文档完成 OAuth2.0 授权流程，获取
access_token
注意：access_token 有有效期，过期后需重新获取。
打包为 EXE（Windows）
pip install pyinstallerpyinstaller --onefile --windowed --name="头条微头条发文工具" --icon=app.ico toutiao_publisher.py
生成的 exe 文件位于 dist/ 目录下。
如果没有 ico 图标文件，可省略 --icon 参数：
pyinstaller --onefile --windowed --name="头条微头条发文工具" toutiao_publisher.py
功能模块说明
| |
|---|
| 输入/保存/读取/清空 Client ID、Secret、Token |
| |
| |
| |
| |
代码结构
toutiao_publisher.py # 完整单文件，包含：├── 配置管理模块 # load_config / save_config├── 图片处理模块 # validate_image / compress_image├── API请求封装模块 # make_api_request / test / upload / publish├── 异步线程模块 # PublishWorker / TestConnectionWorker├── GUI主窗口 # MainWindow 类└── 全局异常捕获 # global_exception_handler
扩展说明
代码已预留 batch_publish_from_data(data_list) 接口，后续可直接对接 Excel/CSV 批量导入，无需重构核心逻辑。
阿里百炼AI自动生成发布工具
基于阿里百炼大模型API（流式输出）+ 今日头条微头条发布，实现AI批量生成内容并自动发文。
环境要求
安装依赖
pip install PyQt6 requests
运行
python ali_bailian_publisher.py
阿里百炼 API Key 获取步骤
- 进入「API-KEY管理」页面，点击「创建API-KEY」
- 将密钥填入工具「设置」页面的 API Key 输入框并保存
支持的模型：qwen-turbo、qwen-plus、qwen-max、qwen-long、qwen2.5-72b-instruct 等，也可手动输入自定义模型名。
今日头条 Access Token 获取
同上方「今日头条创作者开放平台 API 申请步骤」，获取 access_token 后填入设置页即可。
如果不填写头条 Token，工具将只生成内容并保存到本地，不执行发布。
打包为 EXE（Windows）
pyinstaller --onefile --windowed --name="百炼AI自动发文工具" ali_bailian_publisher.py
功能模块说明（多Tab页）
| |
|---|
| 配置阿里百炼 API Key、模型选择、头条 Token，支持本地持久化 |
| 多行输入问题列表（每行一个），提供测试生成/测试发布/一键生成发布/暂停/继续/停止按钮 |
| |
| |
操作流程
1. 配置API
- 填入阿里百炼 API Key（sk-xxx 格式）
2. 输入问题
- 每行输入一个问题/提示词，例如：
写一篇关于春天的微头条分享一个Python编程技巧推荐3本值得阅读的科技书籍
3. 测试验证
- 点击「测试生成（第一条）」：仅用第一条问题调用百炼API，流式输出到预览区，验证Key是否有效
- 点击「测试发布」：发送测试内容到头条，验证Token是否有效
4. 一键生成发布
5. 定时任务
文件存储规则
生成的内容按以下路径结构保存：
当前工作目录/└── 20240615/ # 年月日文件夹 ├── 143025_写一篇关于春天的微头条.txt ├── 143128_分享一个Python编程技巧.txt └── 143231_推荐3本科技书籍.txt
每个文件包含：问题原文、生成时间、分割线、AI生成的完整内容。
代码结构
ali_bailian_publisher.py # 完整单文件，包含：├── 配置管理模块 # load_config / save_config（ali_config.json）├── 文件存储模块 # save_generated_content（按日期文件夹存储）├── 阿里百炼API流式调用 # call_bailian_stream（SSE流式生成器）├── 头条发布模块 # publish_to_toutiao├── 异步线程模块 # GenerateWorker（单条）/ BatchWorker（批量调度）├── GUI主窗口（多Tab） # MainWindow 类└── 全局异常捕获 # global_exception_handler
注意事项
- 阿里百炼 API 有调用频率限制，建议定时间隔不低于10秒
- 头条 Access Token 有有效期，过期后需重新获取
- 生成内容质量取决于问题/提示词的质量，建议明确具体
- 首次使用建议先用「测试生成」和「测试发布」验证配置正确性
#!/usr/bin/env python3-- coding: utf-8 --"""阿里百炼AI + 今日头条微头条自动发文工具功能：调用阿里百炼大模型API（流式输出）批量生成内容定时任务自动逐条生成并发布生成内容按时间+问题存储到日期文件夹对接头条微头条发布接口"""import sysimport osimport jsonimport timeimport refrom datetime import datetimefrom pathlib import Pathfrom typing import Optionalimport requestsfrom PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,QLabel, QLineEdit, QTextEdit, QPushButton, QTabWidget,QGroupBox, QMessageBox, QTextBrowser, QSpinBox, QComboBox,QPlainTextEdit)from PyQt6.QtCore import Qt, QThread, pyqtSignal, QTimerfrom PyQt6.QtGui import QFont, QTextCursor============================================================常量============================================================CONFIG_FILE = "ali_config.json"WINDOW_TITLE = "阿里百炼AI自动生成发布工具"WINDOW_WIDTH = 860WINDOW_HEIGHT = 680阿里百炼API地址（兼容OpenAI格式）BAILIAN_API_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"头条发布相关（复用之前的接口）BASE_API_URL = "https://open.snssdk.com"PUBLISH_URL = f"{BASE_API_URL}/data/content/publish/"REQUEST_TIMEOUT = 60============================================================配置管理============================================================def load_config() -> dict:"""加载本地配置"""default = {"bailian_api_key": "","bailian_model": "qwen-turbo","toutiao_access_token": "","interval_seconds": 60}if not os.path.exists(CONFIG_FILE):return defaulttry:with open(CONFIG_FILE, "r", encoding="utf-8") as f:data = json.load(f)for k, v in default.items():if k not in data:data[k] = vreturn dataexcept (json.JSONDecodeError, IOError):return defaultdef save_config(config: dict) -> bool:"""保存配置到本地"""try:with open(CONFIG_FILE, "w", encoding="utf-8") as f:json.dump(config, f, ensure_ascii=False, indent=2)return Trueexcept IOError:return False============================================================文件存储============================================================def save_generated_content(question: str, content: str) -> str:"""将生成内容保存到 当前目录/年月日/时分秒_问题.txt返回保存的文件路径"""now = datetime.now()date_folder = now.strftime("%Y%m%d")time_prefix = now.strftime("%H%M%S")# 清理文件名中不合法的字符safe_question = re.sub(r'[\/:*?"<>|]', '', question)[:50]filename = f"{time_prefix}{safe_question}.txt"folder_path = Path(date_folder)folder_path.mkdir(exist_ok=True)file_path = folder_path / filenamewith open(file_path, "w", encoding="utf-8") as f: f.write(f"问题: {question}\n") f.write(f"生成时间: {now.strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"{'='*50}\n") f.write(content)return str(file_path)============================================================阿里百炼API流式调用============================================================def call_bailian_stream(api_key: str, model: str, prompt: str):"""调用阿里百炼API（流式输出）生成器，逐块yield文本片段"""headers = {"Authorization": f"Bearer {api_key}","Content-Type": "application/json"}payload = {"model": model,"messages": [{"role": "system", "content": "你是一个专业的内容创作者，擅长撰写有吸引力的短文和微头条内容。"},{"role": "user", "content": prompt}],"stream": True}try: resp = requests.post(BAILIAN_API_URL, headers=headers, json=payload, stream=True, timeout=REQUEST_TIMEOUT) resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): if not line: continue if line.startswith("data: "): data_str = line[6:] if data_str.strip() == "[DONE]": break try: chunk = json.loads(data_str) delta = chunk.get("choices", [{}])[0].get("delta", {}) text = delta.get("content", "") if text: yield text except json.JSONDecodeError: continueexcept requests.exceptions.Timeout: yield "\n[错误] 请求超时，请检查网络"except requests.exceptions.ConnectionError: yield "\n[错误] 网络连接失败"except requests.exceptions.HTTPError as e: yield f"\n[错误] HTTP {e.response.status_code}: {e.response.text[:200]}"except Exception as e: yield f"\n[错误] {str(e)}"============================================================头条发布============================================================def publish_to_toutiao(content: str, access_token: str) -> dict:"""发布微头条到今日头条"""params = {"access_token": access_token}data = {"content": content, "content_type": "text"}try:resp = requests.post(PUBLISH_URL, params=params, data=data, timeout=15)return resp.json()except Exception as e:return {"error_code": -1, "message": str(e)}============================================================工作线程：生成单条内容============================================================class GenerateWorker(QThread):"""单条问题生成线程（流式输出）"""chunk_signal = pyqtSignal(str) # 流式文本片段finished_signal = pyqtSignal(str, str) # (问题, 完整内容)error_signal = pyqtSignal(str)def __init__(self, api_key: str, model: str, question: str): super().__init__() self.api_key = api_key self.model = model self.question = question self._is_paused = False self._is_stopped = Falsedef run(self): full_content = "" for chunk in call_bailian_stream(self.api_key, self.model, self.question): if self._is_stopped: return # 暂停循环等待 while self._is_paused and not self._is_stopped: time.sleep(0.2) full_content += chunk self.chunk_signal.emit(chunk) if not self._is_stopped and full_content.strip(): self.finished_signal.emit(self.question, full_content) elif not full_content.strip(): self.error_signal.emit(f"问题「{self.question}」生成内容为空")def pause(self): self._is_paused = Truedef resume(self): self._is_paused = Falsedef stop(self): self._is_stopped = True self._is_paused = False============================================================批量任务调度线程============================================================class BatchWorker(QThread):"""批量生成+发布调度线程"""log_signal = pyqtSignal(str, str) # (消息, 级别)chunk_signal = pyqtSignal(str) # 流式片段progress_signal = pyqtSignal(int, int) # (当前索引, 总数)finished_signal = pyqtSignal()def __init__(self, api_key: str, model: str, questions: list, access_token: str, auto_publish: bool = True): super().__init__() self.api_key = api_key self.model = model self.questions = questions self.access_token = access_token self.auto_publish = auto_publish self._is_paused = False self._is_stopped = Falsedef run(self): total = len(self.questions) for idx, question in enumerate(self.questions): if self._is_stopped: self.log_signal.emit("任务已停止", "info") break while self._is_paused and not self._is_stopped: time.sleep(0.3) if self._is_stopped: break self.progress_signal.emit(idx + 1, total) self.log_signal.emit(f"[{idx+1}/{total}] 正在生成: {question}", "info") # 流式生成 full_content = "" for chunk in call_bailian_stream(self.api_key, self.model, question): if self._is_stopped: self.finished_signal.emit() return while self._is_paused and not self._is_stopped: time.sleep(0.2) full_content += chunk self.chunk_signal.emit(chunk) if not full_content.strip(): self.log_signal.emit(f"问题「{question}」生成内容为空，跳过", "error") continue # 保存文件 file_path = save_generated_content(question, full_content) self.log_signal.emit(f"已保存: {file_path}", "success") # 自动发布 if self.auto_publish and self.access_token: self.log_signal.emit("正在发布到头条...", "info") result = publish_to_toutiao(full_content, self.access_token) if result.get("error_code", -1) == 0: self.log_signal.emit("✅ 发布成功", "success") else: msg = result.get("message", "未知错误") self.log_signal.emit(f"❌ 发布失败: {msg}", "error") elif self.auto_publish and not self.access_token: self.log_signal.emit("未配置头条Token，跳过发布", "error") # 每条之间间隔避免频率限制 if idx < total - 1 and not self._is_stopped: self.log_signal.emit("等待3秒后处理下一条...", "info") for _ in range(30): if self._is_stopped: break while self._is_paused and not self._is_stopped: time.sleep(0.2) time.sleep(0.1) self.log_signal.emit(f"批量任务完成（共{total}条）", "success") self.finished_signal.emit()def pause(self): self._is_paused = Truedef resume(self): self._is_paused = Falsedef stop(self): self._is_stopped = True self._is_paused = False============================================================主窗口============================================================class MainWindow(QMainWindow):"""主窗口，多Tab页布局"""def __init__(self): super().__init__() self.batch_worker = None self.generate_worker = None self.timer = None # 定时任务定时器 self._init_ui() self._load_saved_config()def _init_ui(self): """初始化界面""" self.setWindowTitle(WINDOW_TITLE) self.setMinimumSize(WINDOW_WIDTH, WINDOW_HEIGHT) self.resize(WINDOW_WIDTH, WINDOW_HEIGHT) self.setFont(QFont("Microsoft YaHei", 9)) central = QWidget() self.setCentralWidget(central) layout = QVBoxLayout(central) layout.setContentsMargins(10, 10, 10, 10) # Tab页 self.tabs = QTabWidget() layout.addWidget(self.tabs) # 各Tab self.tabs.addTab(self._create_settings_tab(), "⚙️ 设置") self.tabs.addTab(self._create_questions_tab(), "📝 问题输入") self.tabs.addTab(self._create_timer_tab(), "⏰ 定时运行") self.tabs.addTab(self._create_log_tab(), "📋 日志输出") self._apply_styles()# ========================================================# Tab 1: 设置页面# ========================================================def _create_settings_tab(self) -> QWidget: """设置Tab：API Key配置""" widget = QWidget() layout = QVBoxLayout(widget) layout.setSpacing(12) # 阿里百炼配置 group1 = QGroupBox("阿里百炼 API 配置") g1_layout = QVBoxLayout(group1) row1 = QHBoxLayout() row1.addWidget(QLabel("API Key (SK):")) self.input_bailian_key = QLineEdit() self.input_bailian_key.setPlaceholderText("请输入阿里百炼 API Key（sk-xxx）") self.input_bailian_key.setEchoMode(QLineEdit.EchoMode.Password) row1.addWidget(self.input_bailian_key) g1_layout.addLayout(row1) row2 = QHBoxLayout() row2.addWidget(QLabel("模型选择:")) self.combo_model = QComboBox() self.combo_model.addItems([ "qwen-turbo", "qwen-plus", "qwen-max", "qwen-long", "qwen2.5-72b-instruct" ]) self.combo_model.setEditable(True) row2.addWidget(self.combo_model) g1_layout.addLayout(row2) layout.addWidget(group1) # 头条发布配置 group2 = QGroupBox("今日头条发布配置") g2_layout = QVBoxLayout(group2) row3 = QHBoxLayout() row3.addWidget(QLabel("Access Token:")) self.input_toutiao_token = QLineEdit() self.input_toutiao_token.setPlaceholderText("头条 Access Token（不填则只生成不发布）") row3.addWidget(self.input_toutiao_token) g2_layout.addLayout(row3) layout.addWidget(group2) # 按钮 btn_row = QHBoxLayout() btn_save = QPushButton("保存配置") btn_load = QPushButton("读取配置") btn_clear = QPushButton("清空配置") btn_save.clicked.connect(self._on_save_config) btn_load.clicked.connect(self._load_saved_config) btn_clear.clicked.connect(self._on_clear_config) btn_row.addWidget(btn_save) btn_row.addWidget(btn_load) btn_row.addWidget(btn_clear) btn_row.addStretch() layout.addLayout(btn_row) layout.addStretch() return widget# ========================================================# Tab 2: 问题输入页面# ========================================================def _create_questions_tab(self) -> QWidget: """问题输入Tab""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("输入问题列表（每行一个问题）:")) self.input_questions = QPlainTextEdit() self.input_questions.setPlaceholderText( "每行输入一个问题，例如：\n" "写一篇关于春天的微头条\n" "分享一个Python编程技巧\n" "推荐3本值得阅读的科技书籍") layout.addWidget(self.input_questions) # 问题数统计 info_row = QHBoxLayout() self.label_question_count = QLabel("当前问题数: 0") info_row.addWidget(self.label_question_count) info_row.addStretch() layout.addLayout(info_row) self.input_questions.textChanged.connect(self._update_question_count) # 操作按钮 btn_row = QHBoxLayout() self.btn_test_single = QPushButton("测试生成（第一条）") self.btn_test_publish = QPushButton("测试发布") self.btn_generate_all = QPushButton("一键生成发布") self.btn_pause = QPushButton("暂停") self.btn_resume = QPushButton("继续") self.btn_stop = QPushButton("停止") self.btn_pause.setEnabled(False) self.btn_resume.setEnabled(False) self.btn_stop.setEnabled(False) self.btn_test_single.clicked.connect(self._on_test_single) self.btn_test_publish.clicked.connect(self._on_test_publish) self.btn_generate_all.clicked.connect(self._on_generate_all) self.btn_pause.clicked.connect(self._on_pause) self.btn_resume.clicked.connect(self._on_resume) self.btn_stop.clicked.connect(self._on_stop) btn_row.addWidget(self.btn_test_single) btn_row.addWidget(self.btn_test_publish) btn_row.addWidget(self.btn_generate_all) btn_row.addWidget(self.btn_pause) btn_row.addWidget(self.btn_resume) btn_row.addWidget(self.btn_stop) layout.addLayout(btn_row) return widget# ========================================================# Tab 3: 定时运行页面# ========================================================def _create_timer_tab(self) -> QWidget: """定时运行Tab""" widget = QWidget() layout = QVBoxLayout(widget) group = QGroupBox("定时任务设置") g_layout = QVBoxLayout(group) row1 = QHBoxLayout() row1.addWidget(QLabel("执行间隔（秒）:")) self.spin_interval = QSpinBox() self.spin_interval.setRange(10, 86400) self.spin_interval.setValue(60) self.spin_interval.setSuffix(" 秒") row1.addWidget(self.spin_interval) row1.addStretch() g_layout.addLayout(row1) row2 = QHBoxLayout() row2.addWidget(QLabel("说明: 定时任务将按间隔逐条处理问题列表中的问题，自动生成内容并发布")) g_layout.addLayout(row2) layout.addWidget(group) # 定时任务控制按钮 btn_row = QHBoxLayout() self.btn_start_timer = QPushButton("启动定时任务") self.btn_stop_timer = QPushButton("停止定时任务") self.btn_stop_timer.setEnabled(False) self.btn_start_timer.clicked.connect(self._on_start_timer) self.btn_stop_timer.clicked.connect(self._on_stop_timer) btn_row.addWidget(self.btn_start_timer) btn_row.addWidget(self.btn_stop_timer) btn_row.addStretch() layout.addLayout(btn_row) # 定时任务状态 self.label_timer_status = QLabel("状态: 未启动") self.label_timer_status.setStyleSheet("font-size: 13px; color: #666;") layout.addWidget(self.label_timer_status) self.label_timer_progress = QLabel("") layout.addWidget(self.label_timer_progress) layout.addStretch() return widget# ========================================================# Tab 4: 日志输出页面# ========================================================def _create_log_tab(self) -> QWidget: """日志输出Tab""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("实时输出日志:")) self.log_browser = QTextBrowser() self.log_browser.setReadOnly(True) self.log_browser.setOpenExternalLinks(False) layout.addWidget(self.log_browser) # 流式输出预览 layout.addWidget(QLabel("流式生成预览:")) self.stream_preview = QTextBrowser() self.stream_preview.setReadOnly(True) self.stream_preview.setMaximumHeight(180) layout.addWidget(self.stream_preview) # 清空日志按钮 btn_row = QHBoxLayout() btn_clear_log = QPushButton("清空日志") btn_clear_log.clicked.connect(lambda: self.log_browser.clear()) btn_clear_stream = QPushButton("清空预览") btn_clear_stream.clicked.connect(lambda: self.stream_preview.clear()) btn_row.addWidget(btn_clear_log) btn_row.addWidget(btn_clear_stream) btn_row.addStretch() layout.addLayout(btn_row) return widget# ========================================================# 样式# ========================================================def _apply_styles(self): self.setStyleSheet(""" QMainWindow { background-color: #f5f5f5; } QTabWidget::pane { border: 1px solid #ddd; border-radius: 4px; } QTabBar::tab { padding: 8px 16px; margin-right: 2px; border: 1px solid #ddd; border-bottom: none; border-radius: 4px 4px 0 0; background: #eee; } QTabBar::tab:selected { background: white; font-weight: bold; } QGroupBox { font-weight: bold; border: 1px solid #ddd; border-radius: 4px; margin-top: 8px; padding-top: 16px; } QGroupBox::title { subcontrol-origin: margin; left: 10px; padding: 0 4px; } QLineEdit, QPlainTextEdit, QTextEdit, QSpinBox, QComboBox { border: 1px solid #ccc; border-radius: 3px; padding: 4px 6px; background: white; } QLineEdit:focus, QPlainTextEdit:focus { border-color: #4a90d9; } QPushButton { background-color: #4a90d9; color: white; border: none; border-radius: 3px; padding: 7px 14px; min-width: 70px; } QPushButton:hover { background-color: #357abd; } QPushButton:pressed { background-color: #2a5f9e; } QPushButton:disabled { background-color: #aaa; color: #ddd; } QTextBrowser { border: 1px solid #ccc; border-radius: 3px; background: #1e1e1e; color: #aaa; font-family: Consolas, monospace; font-size: 12px; } """)# ========================================================# 配置操作# ========================================================def _load_saved_config(self): config = load_config() self.input_bailian_key.setText(config.get("bailian_api_key", "")) self.combo_model.setCurrentText(config.get("bailian_model", "qwen-turbo")) self.input_toutiao_token.setText(config.get("toutiao_access_token", "")) self.spin_interval.setValue(config.get("interval_seconds", 60)) self._append_log("配置已加载", "info")def _on_save_config(self): config = { "bailian_api_key": self.input_bailian_key.text().strip(), "bailian_model": self.combo_model.currentText().strip(), "toutiao_access_token": self.input_toutiao_token.text().strip(), "interval_seconds": self.spin_interval.value() } if save_config(config): self._append_log("配置已保存", "success") else: self._append_log("配置保存失败", "error")def _on_clear_config(self): self.input_bailian_key.clear() self.input_toutiao_token.clear() self.combo_model.setCurrentIndex(0) self._append_log("配置已清空", "info")# ========================================================# 问题列表操作# ========================================================def _get_questions(self) -> list: """获取问题列表（过滤空行）""" text = self.input_questions.toPlainText() return [line.strip() for line in text.split("\n") if line.strip()]def _update_question_count(self): count = len(self._get_questions()) self.label_question_count.setText(f"当前问题数: {count}")# ========================================================# 测试生成（单条）# ========================================================def _on_test_single(self): """测试生成第一条问题""" api_key = self.input_bailian_key.text().strip() if not api_key: QMessageBox.warning(self, "提示", "请先在设置页填写阿里百炼 API Key") return questions = self._get_questions() if not questions: QMessageBox.warning(self, "提示", "请先输入至少一个问题") return model = self.combo_model.currentText().strip() question = questions[0] self.stream_preview.clear() self._append_log(f"测试生成: {question}", "info") self.btn_test_single.setEnabled(False) self.generate_worker = GenerateWorker(api_key, model, question) self.generate_worker.chunk_signal.connect(self._on_stream_chunk) self.generate_worker.finished_signal.connect(self._on_test_generate_done) self.generate_worker.error_signal.connect( lambda msg: self._append_log(msg, "error")) self.generate_worker.finished.connect( lambda: self.btn_test_single.setEnabled(True)) self.generate_worker.start()def _on_stream_chunk(self, text: str): """流式文本片段到达""" self.stream_preview.moveCursor(QTextCursor.MoveOperation.End) self.stream_preview.insertPlainText(text) self.stream_preview.moveCursor(QTextCursor.MoveOperation.End)def _on_test_generate_done(self, question: str, content: str): """测试生成完成""" file_path = save_generated_content(question, content) self._append_log(f"✅ 测试生成完成，已保存: {file_path}", "success")# ========================================================# 测试发布# ========================================================def _on_test_publish(self): """测试发布功能""" token = self.input_toutiao_token.text().strip() if not token: QMessageBox.warning(self, "提示", "请先在设置页填写头条 Access Token") return self._append_log("测试发布: 发送测试内容到头条...", "info") test_content = f"[测试] 这是一条自动发布测试 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" result = publish_to_toutiao(test_content, token) if result.get("error_code", -1) == 0: self._append_log("✅ 测试发布成功", "success") else: msg = result.get("message", "未知错误") self._append_log(f"❌ 测试发布失败: {msg}", "error")# ========================================================# 一键生成发布# ========================================================def _on_generate_all(self): """一键批量生成+发布""" api_key = self.input_bailian_key.text().strip() if not api_key: QMessageBox.warning(self, "提示", "请先在设置页填写阿里百炼 API Key") return questions = self._get_questions() if not questions: QMessageBox.warning(self, "提示", "请先输入至少一个问题") return model = self.combo_model.currentText().strip() token = self.input_toutiao_token.text().strip() self.stream_preview.clear() self._set_task_buttons(running=True) self.batch_worker = BatchWorker(api_key, model, questions, token, auto_publish=True) self.batch_worker.log_signal.connect(self._append_log) self.batch_worker.chunk_signal.connect(self._on_stream_chunk) self.batch_worker.progress_signal.connect(self._on_progress) self.batch_worker.finished_signal.connect( lambda: self._set_task_buttons(running=False)) self.batch_worker.start()# ========================================================# 暂停/继续/停止# ========================================================def _on_pause(self): if self.batch_worker: self.batch_worker.pause() self._append_log("任务已暂停", "info") self.btn_pause.setEnabled(False) self.btn_resume.setEnabled(True)def _on_resume(self): if self.batch_worker: self.batch_worker.resume() self._append_log("任务已继续", "info") self.btn_pause.setEnabled(True) self.btn_resume.setEnabled(False)def _on_stop(self): if self.batch_worker: self.batch_worker.stop() self._append_log("正在停止任务...", "info") self._set_task_buttons(running=False)def _set_task_buttons(self, running: bool): """切换按钮状态""" self.btn_test_single.setEnabled(not running) self.btn_test_publish.setEnabled(not running) self.btn_generate_all.setEnabled(not running) self.btn_pause.setEnabled(running) self.btn_resume.setEnabled(False) self.btn_stop.setEnabled(running)def _on_progress(self, current: int, total: int): """进度更新""" self._append_log(f"进度: {current}/{total}", "info")# ========================================================# 定时任务# ========================================================def _on_start_timer(self): """启动定时任务""" api_key = self.input_bailian_key.text().strip() if not api_key: QMessageBox.warning(self, "提示", "请先在设置页填写阿里百炼 API Key") return questions = self._get_questions() if not questions: QMessageBox.warning(self, "提示", "请先在问题页输入问题") return interval_ms = self.spin_interval.value() * 1000 self._timer_questions = list(questions) self._timer_index = 0 self.timer = QTimer(self) self.timer.timeout.connect(self._timer_tick) self.timer.start(interval_ms) # 立即执行第一条 self._timer_tick() self.btn_start_timer.setEnabled(False) self.btn_stop_timer.setEnabled(True) self.label_timer_status.setText( f"状态: 运行中（间隔 {self.spin_interval.value()} 秒）") self.label_timer_status.setStyleSheet("font-size: 13px; color: #4caf50;") self._append_log(f"定时任务已启动，共{len(questions)}个问题，间隔{self.spin_interval.value()}秒", "success")def _on_stop_timer(self): """停止定时任务""" if self.timer: self.timer.stop() self.timer = None self.btn_start_timer.setEnabled(True) self.btn_stop_timer.setEnabled(False) self.label_timer_status.setText("状态: 已停止") self.label_timer_status.setStyleSheet("font-size: 13px; color: #666;") self._append_log("定时任务已停止", "info")def _timer_tick(self): """定时器触发：处理一条问题""" if self._timer_index >= len(self._timer_questions): self._append_log("所有问题已处理完毕，定时任务结束", "success") self._on_stop_timer() return question = self._timer_questions[self._timer_index] self._timer_index += 1 self.label_timer_progress.setText( f"进度: {self._timer_index}/{len(self._timer_questions)} - 当前: {question[:30]}") api_key = self.input_bailian_key.text().strip() model = self.combo_model.currentText().strip() token = self.input_toutiao_token.text().strip() self._append_log(f"[定时] 处理第{self._timer_index}条: {question}", "info") # 使用单条批量worker处理（只包含1条） worker = BatchWorker(api_key, model, [question], token, auto_publish=True) worker.log_signal.connect(self._append_log) worker.chunk_signal.connect(self._on_stream_chunk) worker.start() # 保存引用防止GC self._current_timer_worker = worker# ========================================================# 日志# ========================================================def _append_log(self, message: str, level: str = "info"): """追加日志""" timestamp = datetime.now().strftime("%H:%M:%S") color_map = {"success": "#4caf50", "error": "#f44336", "info": "#aaaaaa"} color = color_map.get(level, "#aaaaaa") html = f'<span style="color:{color};">[{timestamp}] {message}</span>' self.log_browser.append(html) cursor = self.log_browser.textCursor() cursor.movePosition(QTextCursor.MoveOperation.End) self.log_browser.setTextCursor(cursor)============================================================全局异常捕获============================================================def global_exception_handler(exc_type, exc_value, exc_tb):import tracebackerror_msg = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))try:QMessageBox.critical(None, "程序异常",f"发生未预期的错误:\n{str(exc_value)}")except Exception:passsys.stderr.write(f"[FATAL]\n{error_msg}\n")============================================================入口============================================================def main():sys.excepthook = global_exception_handlerapp = QApplication(sys.argv)app.setStyle("Fusion")window = MainWindow()window.show()sys.exit(app.exec())if name == "main":main()
