夜雨聆风今日头条微头条 AI 批量发文桌面工具｜完整开源教程（PyQt6 可视化 + 多AI模型 + 定时批量发布）
一、项目前言（300+字）
在自媒体运营、内容矩阵搭建、流量账号运维的实际场景中，人工手动撰写、编辑、发布微头条内容存在效率极低、重复性高、耗时耗力、发布时间不精准等诸多痛点。传统运营模式下，创作者需要每日花费大量时间撰写文案、手动排版、逐篇发布，同时很难保证内容产出数量充足、风格多样、发布时段稳定，极大限制了账号起号、流量沉淀与矩阵规模化运营的效率。为彻底解决自媒体批量发文的行业痛点，本项目基于 Python + PyQt6 开发全可视化、无命令行、全自动今日头条微头条批量发文桌面工具，整合主流AI大模型自动文案生成、Excel批量导入发文、毫秒级批量发布、自定义定时任务、全流程日志监控、参数持久化配置等全套核心能力。
该工具摒弃了传统脚本晦涩难用、需要改代码配置、无界面、易报错、无法定时的缺陷，打造出精致美观的桌面客户端，支持 Windows 系统直接运行，所有参数可视化配置、一键保存、永久生效。同时深度对接阿里百炼、Kimi、DeepSeek 三大主流AI接口，随机匹配提示词生成差异化原创文案，规避内容同质化问题。发布逻辑完全复刻今日头条官网抓包协议，模拟真实浏览器请求，稳定性强、通过率高。全程异步线程运行，不会出现界面卡顿、程序卡死等问题，是自媒体个人、工作室、矩阵团队实现无人值守、批量量产、定时稳更的专业级自动化工具。
二、项目核心架构与功能详解
本项目采用模块化分层架构，代码解耦度高、维护性强、拓展性极佳，整体分为 5 大核心模块，各司其职、互不干扰，完美实现功能闭环。下面对核心代码模块进行精细化拆解讲解，剔除所有密钥、隐私配置，保留纯净可用逻辑。
模块一：全局配置与日志系统（核心基础）
配置模块是整个工具的基石，负责全局参数持久化存储、默认参数初始化、配置读写容错处理。所有用户设置的发布接口、AI参数、提示词、定时时间都会自动保存至 config.json，重启软件自动加载，无需重复配置。同时内置全局日志系统，同步输出控制台日志与本地日志文件，精准记录每一次操作、AI生成状态、发布结果、报错信息，方便用户排查问题、复盘运行记录。模块内置容错机制，配置文件损坏、缺失时自动生成默认配置，保证程序永不崩溃。
模块二：多AI模型统一调用模块（内容生成核心）
工具内置阿里百炼、Kimi、DeepSeek 三大主流大模型接口，封装统一调用入口，支持自由切换默认AI模型、自定义生成数量、自定义创作提示词。所有AI请求采用标准化POST协议，固定返回 JSON 结构化数据，自动解析标题+正文，无需人工整理文案。通过多模型轮换、随机提示词机制，彻底解决批量发文内容重复、同质化严重、被判低质的问题。所有接口采用超时保护、异常捕获机制，网络波动、接口报错时会精准日志提示，不影响程序整体运行。
模块三：可视化UI + 异步任务调度模块（交互与运行核心）
基于 PyQt6 搭建高颜值桌面客户端，分区设计三大标签页：发布管理、参数设置、定时日志。支持 Excel 批量导入文案、AI一键批量生成、文章预览、状态实时刷新、立即批量发布、定时无人值守发布。所有耗时操作（AI生成、批量发文）全部开启子线程异步运行，彻底解决UI卡顿、程序假死问题。内置定时轮询机制，30秒检测一次系统时间，到达预设时间自动触发批量发布，实现全自动运维。同时搭配精美样式美化，界面简洁高级、操作零门槛，新手可直接上手使用。
三、完整纯净可运行源码（无密钥、可直接复制使用）
#!/usr/bin/env python3# -*- coding: utf-8 -*-"""今日头条微头条API批量发文桌面工具===================================基于 PyQt6 + 多AI接口 + 定时任务 + 批量发布开源纯净版：无内置密钥、可直接配置使用"""import sysimport osimport jsonimport timeimport randomimport loggingfrom datetime import datetimefrom typing import List, Dictimport requestsfrom PyQt6.QtWidgets import ( QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QTextEdit, QPushButton, QFileDialog, QGroupBox, QMessageBox, QTextBrowser, QTabWidget, QTableWidget, QTableWidgetItem, QHeaderView, QSpinBox, QComboBox, QCheckBox, QSplitter, QTimeEdit, QListWidget, QListWidgetItem, QDialog, QDialogButtonBox)from PyQt6.QtCore import Qt, QThread, pyqtSignal, QTimerfrom PyQt6.QtGui import QFont, QTextCursor# ============================================================# 常量与日志系统# ============================================================CONFIG_FILE = "config.json"LOG_FILE = "runtime.log"WINDOW_TITLE = "今日头条微头条API批量发文工具"WINDOW_WIDTH = 1200WINDOW_HEIGHT = 800logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE, encoding="utf-8"), logging.StreamHandler(sys.stdout) ])logger = logging.getLogger(__name__)# ============================================================# 配置管理模块# ============================================================def load_config() -> dict: default = { "publish_url": "https://mp.toutiao.com/mp/agw/article/wtt?mp_publish_ab_val=0&aid=1231", "cookie_str": "", "ai_keys": {"aliyun_bailian": "", "kimi": "", "deepseek": ""}, "prompts": ["请根据以下主题生成一篇今日头条微头条文章，标题简短有吸引力，内容200-500字，通俗易懂"], "default_ai": "aliyun_bailian", "generate_count": 50, "schedule_times": [], "schedule_enabled": False } if not os.path.exists(CONFIG_FILE): return default try: 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] = v return data except (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) logger.info("配置已保存到 config.json") return True except IOError as e: logger.error(f"保存配置失败: {e}") return False# ============================================================# 多AI接口统一调用模块# ============================================================def call_deepseek_api(api_key: str, prompt: str, count: int) -> List[Dict[str, str]]: url = "https://api.deepseek.com/chat/completions" headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} full_prompt = f"{prompt}\n\n请生成{count}篇文章，每篇包含标题和内容。严格返回JSON数组：[{{\"title\":\"标题\",\"content\":\"内容\"}}]" payload = {"model": "deepseek-chat", "messages": [{"role": "user", "content": full_prompt}], "temperature": 0.8, "max_tokens": 4096} try: resp = requests.post(url, headers=headers, json=payload, timeout=120) resp.raise_for_status() text = resp.json()["choices"][0]["message"]["content"] start, end = text.find("["), text.rfind("]") + 1 if start >= 0 and end > start: return json.loads(text[start:end]) except Exception as e: logger.error(f"DeepSeek API调用失败: {e}") return []def call_kimi_api(api_key: str, prompt: str, count: int) -> List[Dict[str, str]]: url = "https://api.moonshot.cn/v1/chat/completions" headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} full_prompt = f"{prompt}\n\n请生成{count}篇文章，每篇包含标题和内容。严格返回JSON数组：[{{\"title\":\"标题\",\"content\":\"内容\"}}]" payload = {"model": "moonshot-v1-8k", "messages": [{"role": "user", "content": full_prompt}], "temperature": 0.8} try: resp = requests.post(url, headers=headers, json=payload, timeout=120) resp.raise_for_status() text = resp.json()["choices"][0]["message"]["content"] start, end = text.find("["), text.rfind("]") + 1 if start >= 0 and end > start: return json.loads(text[start:end]) except Exception as e: logger.error(f"Kimi API调用失败: {e}") return []def call_aliyun_bailian_api(api_key: str, prompt: str, count: int) -> List[Dict[str, str]]: url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} full_prompt = f"{prompt}\n\n请生成{count}篇文章，每篇包含标题和内容。严格返回JSON数组：[{{\"title\":\"标题\",\"content\":\"内容\"}}]" payload = {"model": "qwen-plus", "messages": [{"role": "user", "content": full_prompt}], "temperature": 0.8} try: resp = requests.post(url, headers=headers, json=payload, timeout=120) resp.raise_for_status() text = resp.json()["choices"][0]["message"]["content"] start, end = text.find("["), text.rfind("]") + 1 if start >= 0 and end > start: return json.loads(text[start:end]) except Exception as e: logger.error(f"阿里百炼API调用失败: {e}") return []def call_ai_generate(ai_type: str, api_key: str, prompt: str, count: int) -> List[Dict[str, str]]: if ai_type == "deepseek": return call_deepseek_api(api_key, prompt, count) elif ai_type == "kimi": return call_kimi_api(api_key, prompt, count) elif ai_type == "aliyun_bailian": return call_aliyun_bailian_api(api_key, prompt, count) return []# ============================================================# 头条发布接口模块# ============================================================def publish_article(publish_url: str, cookie_str: str, title: str, content: str) -> dict: if "msToken" not in publish_url or "a_bogus" not in publish_url: return {"error": "发布URL缺少msToken或a_bogus参数，请在设置中配置完整的抓包URL"} headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Cookie": cookie_str, "Content-Type": "application/json", "Referer": "https://mp.toutiao.com/" } extra_param = "{\"claim_exclusive\":\"1\",\"add_music_to_article_flag\":\"0\",\"tuwen_wtt_trans_flag\":\"1\",\"info_source\":\"{\\\"source_type\\\":5,\\\"source_author_uid\\\":\\\"\\\",\\\"time_format\\\":\\\"\\\",\\\"position\\\":{}}\"}" post_json = { "content": content, "image_list": [], "extra": extra_param, "is_fans_article": 2, "pre_upload": 1, "welfare_card": "", "entrance": "main" } try: logger.info(f"[publish] URL({len(publish_url)}): {publish_url[:120]}...") logger.info(f"[publish] Cookie长度: {len(cookie_str)}") logger.info(f"[publish] Content长度: {len(content)}") resp = requests.post(publish_url, headers=headers, json=post_json, timeout=35) resp_text = resp.text.strip() if not resp_text: return {"error": f"空响应, HTTP {resp.status_code}"} try: result = resp.json() result["_http_status"] = resp.status_code return result except json.JSONDecodeError: return {"error": f"非JSON响应(HTTP {resp.status_code}): {resp_text[:200]}"} except requests.exceptions.Timeout: return {"error": "请求超时"} except requests.exceptions.ConnectionError: return {"error": "网络连接失败"} except Exception as e: logger.error(f"发布失败: {e}") return {"error": str(e)}# ============================================================# 异步任务线程模块# ============================================================class AIGenerateWorker(QThread): log_signal = pyqtSignal(str, str) result_signal = pyqtSignal(list) finished_signal = pyqtSignal() def __init__(self, ai_type, api_key, prompt, count): super().__init__() self.ai_type = ai_type self.api_key = api_key self.prompt = prompt self.count = count def run(self): self.log_signal.emit(f"正在调用 {self.ai_type} API 生成 {self.count} 篇内容...", "info") results = call_ai_generate(self.ai_type, self.api_key, self.prompt, self.count) if results: self.log_signal.emit(f"AI生成完成，共获得 {len(results)} 篇内容", "success") else: self.log_signal.emit("AI生成失败或返回为空，请检查API Key和网络", "error") self.result_signal.emit(results) self.finished_signal.emit()class BatchPublishWorker(QThread): log_signal = pyqtSignal(str, str) progress_signal = pyqtSignal(int, str) finished_signal = pyqtSignal() def __init__(self, articles, publish_url, cookie_str, interval=1.0): super().__init__() self.articles = articles self.publish_url = publish_url self.cookie_str = cookie_str self.interval = interval self._stop_flag = False def stop(self): self._stop_flag = True def run(self): total = len(self.articles) self.log_signal.emit(f"开始批量发布，共 {total} 篇，间隔 {self.interval} 秒", "info") for i, article in enumerate(self.articles): if self._stop_flag: self.log_signal.emit("发布已手动停止", "info") break title = article.get("title", "") content = article.get("content", "") self.log_signal.emit(f"[{i+1}/{total}] 发布: {title}", "info") result = publish_article(self.publish_url, self.cookie_str, title, content) if "error" in result: self.log_signal.emit(f"[{i+1}/{total}] ❌ 失败: {result['error']}", "error") self.progress_signal.emit(i, "失败") logger.error(f"发布第{i+1}篇失败: {title} - {result}") else: http_status = result.get("_http_status", 0) if http_status == 200: self.log_signal.emit(f"[{i+1}/{total}] ✅ 发布成功 (HTTP 200)", "success") self.progress_signal.emit(i, "成功") else: msg = result.get("err_tips", result.get("message", f"HTTP {http_status}")) self.log_signal.emit(f"[{i+1}/{total}] ❌ 失败: {msg}", "error") self.progress_signal.emit(i, "失败") logger.info(f"发布第{i+1}篇: {title} - HTTP:{http_status} - {json.dumps(result, ensure_ascii=False)[:200]}") if i < total - 1: time.sleep(self.interval) self.log_signal.emit("批量发布任务结束", "info") self.finished_signal.emit()# ============================================================# 内容预览弹窗# ============================================================class ContentPreviewDialog(QDialog): def __init__(self, title: str, content: str, parent=None): super().__init__(parent) self.setWindowTitle(f"内容预览 - {title}") self.setMinimumSize(500, 400) layout = QVBoxLayout(self) lbl = QLabel(f"标题: {title}") lbl.setStyleSheet("font-weight:bold;font-size:14px;margin-bottom:8px;") layout.addWidget(lbl) tv = QTextEdit() tv.setPlainText(content) tv.setReadOnly(True) layout.addWidget(tv) btn = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) btn.rejected.connect(self.close) layout.addWidget(btn)# ============================================================# 设置页面模块# ============================================================class SettingsTab(QWidget): def __init__(self, config: dict, save_callback): super().__init__() self.config = config self.save_callback = save_callback self._init_ui() def _init_ui(self): layout = QVBoxLayout(self) layout.setSpacing(12) grp_pub = QGroupBox("发布接口配置") gl = QVBoxLayout(grp_pub) row = QHBoxLayout() row.addWidget(QLabel("发布URL:")) self.input_url = QLineEdit(self.config.get("publish_url", "")) self.input_url.setPlaceholderText("https://mp.toutiao.com/mp/agw/article/wtt?...") row.addWidget(self.input_url) gl.addLayout(row) gl.addWidget(QLabel("Cookie:")) self.input_cookie = QTextEdit() self.input_cookie.setPlainText(self.config.get("cookie_str", "")) self.input_cookie.setPlaceholderText("请输入Cookie字符串") self.input_cookie.setMaximumHeight(80) gl.addWidget(self.input_cookie) layout.addWidget(grp_pub) grp_ai = QGroupBox("AI API Keys") al = QVBoxLayout(grp_ai) ai_keys = self.config.get("ai_keys", {}) for label_text, key_name, attr_name in [ ("阿里百炼 Key:", "aliyun_bailian", "input_aliyun_key"), ("Kimi Key:", "kimi", "input_kimi_key"), ("DeepSeek Key:", "deepseek", "input_deepseek_key"), ]: r = QHBoxLayout() r.addWidget(QLabel(label_text)) inp = QLineEdit(ai_keys.get(key_name, "")) inp.setEchoMode(QLineEdit.EchoMode.Password) r.addWidget(inp) al.addLayout(r) setattr(self, attr_name, inp) row_def = QHBoxLayout() row_def.addWidget(QLabel("默认AI:")) self.combo_default_ai = QComboBox() self.combo_default_ai.addItem("阿里百炼", "aliyun_bailian") self.combo_default_ai.addItem("Kimi", "kimi") self.combo_default_ai.addItem("DeepSeek", "deepseek") idx = self.combo_default_ai.findData(self.config.get("default_ai", "aliyun_bailian")) if idx >= 0: self.combo_default_ai.setCurrentIndex(idx) self.combo_default_ai.setMinimumWidth(160) row_def.addWidget(self.combo_default_ai) row_def.addStretch() al.addLayout(row_def) layout.addWidget(grp_ai) grp_prompt = QGroupBox("提示词管理（每行一条，可直接编辑增删）") pl = QVBoxLayout(grp_prompt) self.prompt_edit = QTextEdit() self.prompt_edit.setMinimumHeight(120) self.prompt_edit.setPlaceholderText("每行一条提示词，AI生成时随机选取一条...") self.prompt_edit.setPlainText("\n".join(self.config.get("prompts", []))) pl.addWidget(self.prompt_edit) layout.addWidget(grp_prompt) row_cnt = QHBoxLayout() row_cnt.addWidget(QLabel("AI默认生成数量:")) self.spin_count = QSpinBox() self.spin_count.setRange(1, 200) self.spin_count.setValue(self.config.get("generate_count", 50)) row_cnt.addWidget(self.spin_count) row_cnt.addStretch() layout.addLayout(row_cnt) btn_save = QPushButton("保存所有设置") btn_save.setMinimumHeight(36) btn_save.clicked.connect(self._on_save) layout.addWidget(btn_save) layout.addStretch() def _on_save(self): self.config["publish_url"] = self.input_url.text().strip() self.config["cookie_str"] = self.input_cookie.toPlainText().strip() self.config["ai_keys"] = { "deepseek": self.input_deepseek_key.text().strip(), "kimi": self.input_kimi_key.text().strip(), "aliyun_bailian": self.input_aliyun_key.text().strip() } self.config["default_ai"] = self.combo_default_ai.currentData() self.config["generate_count"] = self.spin_count.value() prompts = [l.strip() for l in self.prompt_edit.toPlainText().split("\n") if l.strip()] self.config["prompts"] = prompts self.save_callback(self.config) QMessageBox.information(self, "提示", "设置已保存")# ============================================================# 定时任务与日志模块# ============================================================class ScheduleTab(QWidget): def __init__(self, config: dict, save_callback): super().__init__() self.config = config self.save_callback = save_callback self._init_ui() def _init_ui(self): layout = QVBoxLayout(self) grp_sch = QGroupBox("定时任务设置") sl = QVBoxLayout(grp_sch) row_en = QHBoxLayout() self.chk_enable = QCheckBox("启用定时发布") self.chk_enable.setChecked(self.config.get("schedule_enabled", False)) row_en.addWidget(self.chk_enable) row_en.addStretch() sl.addLayout(row_en) row_t = QHBoxLayout() row_t.addWidget(QLabel("添加时间:")) self.time_edit = QTimeEdit() self.time_edit.setDisplayFormat("HH:mm") row_t.addWidget(self.time_edit) btn_add = QPushButton("添加") btn_add.clicked.connect(self._add_time) row_t.addWidget(btn_add) btn_del = QPushButton("删除选中") btn_del.clicked.connect(self._del_time) row_t.addWidget(btn_del) row_t.addStretch() sl.addLayout(row_t) self.time_list = QListWidget() self.time_list.setMaximumHeight(100) for t in self.config.get("schedule_times", []): self.time_list.addItem(t) sl.addWidget(self.time_list) btn_save = QPushButton("保存定时设置") btn_save.clicked.connect(self._on_save) sl.addWidget(btn_save) layout.addWidget(grp_sch) grp_log = QGroupBox("本地运行日志监控") ll = QVBoxLayout(grp_log) self.log_view = QTextBrowser() self.log_view.setStyleSheet("background:#1e1e1e;color:#aaa;font-family:Consolas;font-size:12px;") ll.addWidget(self.log_view) br = QHBoxLayout() btn_ref = QPushButton("刷新日志") btn_ref.clicked.connect(self._refresh_log) btn_clr = QPushButton("清空显示") btn_clr.clicked.connect(self.log_view.clear) br.addWidget(btn_ref) br.addWidget(btn_clr) br.addStretch() ll.addLayout(br) layout.addWidget(grp_log) def _add_time(self): t = self.time_edit.time().toString("HH:mm") for i in range(self.time_list.count()): if self.time_list.item(i).text() == t: return self.time_list.addItem(t) def _del_time(self): row = self.time_list.currentRow() if row >= 0: self.time_list.takeItem(row) def _on_save(self): times = [self.time_list.item(i).text() for i in range(self.time_list.count())] self.config["schedule_times"] = times self.config["schedule_enabled"] = self.chk_enable.isChecked() self.save_callback(self.config) QMessageBox.information(self, "提示", "定时设置已保存") def _refresh_log(self): if os.path.exists(LOG_FILE): try: with open(LOG_FILE, "r", encoding="utf-8") as f: lines = f.readlines() self.log_view.setPlainText("".join(lines[-200:])) c = self.log_view.textCursor() c.movePosition(QTextCursor.MoveOperation.End) self.log_view.setTextCursor(c) except Exception: self.log_view.setPlainText("读取日志失败") else: self.log_view.setPlainText("日志文件不存在") def append_log(self, msg: str, level: str = "info"): color_map = {"success": "#4caf50", "error": "#f44336", "info": "#aaaaaa"} color = color_map.get(level, "#aaaaaa") ts = datetime.now().strftime("%H:%M:%S") self.log_view.append(f'<span style="color:{color};">[{ts}] {msg}</span>')# ============================================================# 核心发布管理页面# ============================================================class MainTab(QWidget): def __init__(self, config: dict, log_callback): super().__init__() self.config = config self.log_callback = log_callback self.articles = [] self.ai_worker = None self.publish_worker = None self._init_ui() def _init_ui(self): splitter = QSplitter(Qt.Orientation.Horizontal) left = QWidget() ll = QVBoxLayout(left) ll.setContentsMargins(0, 0, 0, 0) br = QHBoxLayout() btn_imp = QPushButton("导入XLS文件") btn_imp.clicked.connect(self._import_xls) br.addWidget(btn_imp) br.addWidget(QLabel("AI生成数量:")) self.spin_ai_count = QSpinBox() self.spin_ai_count.setRange(1, 200) self.spin_ai_count.setValue(self.config.get("generate_count", 50)) br.addWidget(self.spin_ai_count) btn_ai = QPushButton("AI生成") btn_ai.clicked.connect(self._ai_generate) br.addWidget(btn_ai) br.addStretch() ll.addLayout(br) self.table = QTableWidget(0, 3) self.table.setHorizontalHeaderLabels(["标题", "内容(双击预览)", "状态"]) self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeMode.Interactive) self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) self.table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents) self.table.setColumnWidth(0, 220) self.table.cellDoubleClicked.connect(self._on_cell_double_clicked) ll.addWidget(self.table) pr = QHBoxLayout() self.btn_send_now = QPushButton("立即发送（每隔1秒）") self.btn_send_now.clicked.connect(self._send_now) pr.addWidget(self.btn_send_now) self.btn_send_schedule = QPushButton("定时发送") self.btn_send_schedule.clicked.connect(self._send_schedule) pr.addWidget(self.btn_send_schedule) self.btn_stop = QPushButton("停止发送") self.btn_stop.clicked.connect(self._stop_send) self.btn_stop.setEnabled(False) pr.addWidget(self.btn_stop) btn_clr = QPushButton("清空列表") btn_clr.clicked.connect(self._clear_table) pr.addWidget(btn_clr) pr.addStretch() ll.addLayout(pr) splitter.addWidget(left) right = QWidget() rl = QVBoxLayout(right) rl.setContentsMargins(0, 0, 0, 0) rl.addWidget(QLabel("运行日志")) self.log_browser = QTextBrowser() self.log_browser.setStyleSheet("background:#1e1e1e;color:#aaa;font-family:Consolas;font-size:12px;") rl.addWidget(self.log_browser) splitter.addWidget(right) splitter.setSizes([700, 400]) main_layout = QVBoxLayout(self) main_layout.addWidget(splitter) def _on_cell_double_clicked(self, row: int, col: int): if row < len(self.articles): art = self.articles[row] dlg = ContentPreviewDialog(art.get("title", ""), art.get("content", ""), self) dlg.exec() def _append_log(self, msg: str, level: str = "info"): color_map = {"success": "#4caf50", "error": "#f44336", "info": "#aaaaaa"} color = color_map.get(level, "#aaaaaa") ts = datetime.now().strftime("%H:%M:%S") self.log_browser.append(f'<span style="color:{color};">[{ts}] {msg}</span>') c = self.log_browser.textCursor() c.movePosition(QTextCursor.MoveOperation.End) self.log_browser.setTextCursor(c) logger.info(msg) if self.log_callback: self.log_callback(msg, level) def _import_xls(self): fp, _ = QFileDialog.getOpenFileName(self, "选择XLS文件", "", "Excel文件 (*.xls *.xlsx);;所有文件 (*)") if not fp: return try: import openpyxl wb = openpyxl.load_workbook(fp, read_only=True) ws = wb.active count = 0 for row in ws.iter_rows(min_row=2, values_only=True): if row and len(row) >= 2 and row[0]: self.articles.append({"title": str(row[0]).strip(), "content": str(row[1] or "").strip(), "status": "待发送"}) count += 1 wb.close() self._refresh_table() self._append_log(f"成功导入 {count} 篇文章", "success") except ImportError: self._append_log("请安装 openpyxl: pip install openpyxl", "error") QMessageBox.warning(self, "缺少依赖", "请安装openpyxl:\npip install openpyxl") except Exception as e: self._append_log(f"导入XLS失败: {e}", "error") def _ai_generate(self): ai_type = self.config.get("default_ai", "aliyun_bailian") api_key = self.config.get("ai_keys", {}).get(ai_type, "") if not api_key: self._append_log(f"未配置 {ai_type} 的API Key，请到设置页面配置", "error") QMessageBox.warning(self, "提示", f"请先在设置页面配置 {ai_type} 的API Key") return prompts = self.config.get("prompts", []) if not prompts: self._append_log("未配置提示词，请到设置页面添加", "error") return prompt = random.choice(prompts) count = self.spin_ai_count.value() self.ai_worker = AIGenerateWorker(ai_type, api_key, prompt, count) self.ai_worker.log_signal.connect(self._append_log) self.ai_worker.result_signal.connect(self._on_ai_result) self.ai_worker.start() def _on_ai_result(self, results: list): for item in results: if isinstance(item, dict) and "title" in item and "content" in item: self.articles.append({"title": item["title"], "content": item["content"], "status": "待发送"}) self._refresh_table() def _refresh_table(self): self.table.setRowCount(len(self.articles)) for i, art in enumerate(self.articles): self.table.setItem(i, 0, QTableWidgetItem(art.get("title", ""))) ct = art.get("content", "") self.table.setItem(i, 1, QTableWidgetItem(ct[:80] + ("..." if len(ct) > 80 else ""))) self.table.setItem(i, 2, QTableWidgetItem(art.get("status", "待发送"))) def _send_now(self): if not self.articles: self._append_log("没有可发送的文章", "error") return pub_url = self.config.get("publish_url", "") cookie = self.config.get("cookie_str", "") if not pub_url or not cookie: self._append_log("请先在设置页面配置发布URL和Cookie", "error") QMessageBox.warning(self, "提示", "请先配置发布URL和Cookie") return self.btn_send_now.setEnabled(False) self.btn_send_schedule.setEnabled(False) self.btn_stop.setEnabled(True) self.publish_worker = BatchPublishWorker(self.articles, pub_url, cookie, interval=1.0) self.publish_worker.log_signal.connect(self._append_log) self.publish_worker.progress_signal.connect(self._on_publish_progress) self.publish_worker.finished_signal.connect(self._on_publish_done) self.publish_worker.start() def _send_schedule(self): if not self.articles: self._append_log("没有可发送的文章", "error") return self._append_log("文章已加入定时发送队列，将在设定时间自动发送", "info") def _stop_send(self): if self.publish_worker: self.publish_worker.stop() self.btn_stop.setEnabled(False) def _on_publish_progress(self, index: int, status: str): if index < len(self.articles): self.articles[index]["status"] = status self.table.setItem(index, 2, QTableWidgetItem(status)) def _on_publish_done(self): self.btn_send_now.setEnabled(True) self.btn_send_schedule.setEnabled(True) self.btn_stop.setEnabled(False) self._append_log("所有发布任务已完成", "success") def _clear_table(self): self.articles.clear() self.table.setRowCount(0) self._append_log("已清空文章列表", "info") def update_config(self, config: dict): self.config = config self.spin_ai_count.setValue(config.get("generate_count", 50))# ============================================================# 主窗口程序# ============================================================class MainWindow(QMainWindow): def __init__(self): super().__init__() self.config = load_config() self._init_ui() self._start_schedule_timer() 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(8, 8, 8, 8) self.tabs = QTabWidget() self.main_tab = MainTab(self.config, self._on_log_from_main) self.tabs.addTab(self.main_tab, "发布管理") self.settings_tab = SettingsTab(self.config, self._on_config_saved) self.tabs.addTab(self.settings_tab, "设置") self.schedule_tab = ScheduleTab(self.config, self._on_config_saved) self.tabs.addTab(self.schedule_tab, "定时任务 & 日志") layout.addWidget(self.tabs) self._apply_styles() def _apply_styles(self): self.setStyleSheet(""" QMainWindow { background-color: #f5f5f5; } 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, QTextEdit, QSpinBox, QTimeEdit { border: 1px solid #ccc; border-radius: 3px; padding: 4px 6px; background: white; } QLineEdit:focus, QTextEdit:focus { border-color: #4a90d9; } QComboBox { border: 1px solid #ccc; border-radius: 3px; padding: 6px 12px; background: white; min-height: 26px; font-size: 13px; } QComboBox::drop-down { subcontrol-origin: padding; subcontrol-position: top right; width: 28px; border-left: 1px solid #ddd; background: #fafafa; } QComboBox::down-arrow { width: 0; height: 0; border-left: 5px solid transparent; border-right: 5px solid transparent; border-top: 6px solid #666; } QComboBox QAbstractItemView { border: 1px solid #ccc; background: white; selection-background-color: #4a90d9; selection-color: white; padding: 2px; outline: none; } QComboBox QAbstractItemView::item { min-height: 30px; padding: 4px 10px; } QPushButton { background-color: #4a90d9; color: white; border: none; border-radius: 3px; padding: 6px 14px; min-width: 80px; } QPushButton:hover { background-color: #357abd; } QPushButton:pressed { background-color: #2a5f9e; } QPushButton:disabled { background-color: #aaa; color: #ddd; } QTableWidget { border: 1px solid #ccc; gridline-color: #eee; } QTabWidget::pane { border: 1px solid #ccc; } QTextBrowser { border: 1px solid #ccc; border-radius: 3px; } """) def _on_config_saved(self, config: dict): self.config = config save_config(config) self.main_tab.update_config(config) def _on_log_from_main(self, msg: str, level: str): self.schedule_tab.append_log(msg, level) def _start_schedule_timer(self): self.schedule_timer = QTimer() self.schedule_timer.timeout.connect(self._check_schedule) self.schedule_timer.start(30000) def _check_schedule(self): if not self.config.get("schedule_enabled", False): return now = datetime.now().strftime("%H:%M") times = self.config.get("schedule_times", []) if now in times and self.main_tab.articles: if not hasattr(self, '_last_trigger') or self._last_trigger != now: self._last_trigger = now logger.info(f"定时任务触发: {now}") self.schedule_tab.append_log(f"定时任务触发: {now}，开始自动发送", "info") self.main_tab._send_now()# ============================================================# 程序入口# ============================================================def main(): import traceback def exc_handler(et, ev, tb): logger.error("".join(traceback.format_exception(et, ev, tb))) sys.excepthook = exc_handler app = QApplication(sys.argv) app.setStyle("Fusion") window = MainWindow() window.show() sys.exit(app.exec())if __name__ == "__main__": main()四、核心技术知识点总结
1. PyQt6 桌面GUI开发：掌握Qt布局管理、标签页组件、表格控件、弹窗组件、样式美化等桌面开发核心技术，实现高颜值、交互友好的桌面客户端，同时通过异步子线程解决UI卡顿问题。
2. 多AI大模型API对接：熟练使用HTTP请求对接主流大模型接口，实现标准化参数封装、JSON数据解析、异常捕获、超时处理，掌握多模型兼容调用、随机提示词优化文案质量的实战技巧。
3. 网页抓包与接口逆向：复刻今日头条官方发文接口协议，模拟真实浏览器请求头、Cookie身份认证、固定报文参数，理解Web接口请求逻辑、参数校验机制、HTTP状态码异常排查。
4. Python异步多线程开发：通过QThread实现耗时任务异步执行，区分UI主线程与工作子线程，避免程序卡死，掌握信号与槽机制实现跨线程数据、日志、状态传输。
5. 配置持久化与日志体系：基于JSON文件实现参数永久保存，搭建分级日志系统，同时输出本地文件日志与前端可视化日志，掌握程序容错、异常处理、运行监控的开发思维。
6. 定时任务调度机制：利用QTimer定时器实现轮询检测，精准触发定时任务，实现无人值守自动化执行，掌握轻量级桌面定时程序开发逻辑。
五、测试步骤与拓展应用场景
✅ 完整测试步骤（新手零基础可跑通）
1. 环境部署：安装Python3.8及以上版本，执行依赖安装命令：pip install pyqt6 requests openpyxl。
2. 源码运行：将纯净源码保存为 main.py，终端运行 python main.py，启动可视化客户端。
3. 参数配置：进入设置页面，填入抓包获取的头条发布URL、账号Cookie，填写任意一款AI模型的API密钥，自定义创作提示词并保存。
4. 功能测试：测试AI一键生成文案、Excel批量导入、文章预览、立即批量发布，查看日志判断运行状态。
5. 定时测试：添加1分钟后的测试时间，开启定时开关，等待程序自动触发发布任务，验证定时功能有效性。
6. 异常测试：故意清空密钥、错误URL、断开网络，查看程序是否正常报错、不崩溃、日志精准记录问题。
✅ 拓展应用场景
1. 自媒体矩阵运营：多账号批量轮询发文，每日定时定量更新，快速养号、提升账号活跃度与平台权重。
2. 垂直领域内容量产：适配情感、职场、三农、科普、美食等多个领域，通过自定义提示词批量生成垂直原创内容，打造垂直账号。
3. 工作室批量运维：多设备部署工具，分工生成、审核、发布内容，实现团队标准化内容量产作业。
4. 自动化流量变现：稳定日更产出内容，积累自然流量，适配头条流量分成、内容带货、品牌接单变现模式。
5. 二次开发拓展：可基于本框架拓展图文混排、视频发文、多平台同步（西瓜、百家号）、AI自动改写、内容查重等高级功能。