当前位置:首页>python>Python实现微信公众号自动推发文章

Python实现微信公众号自动推发文章

  • 2026-09-16 13:25:27
Python实现微信公众号自动推发文章

修复了草稿创建失败问题后的完整可运行代码,我已经针对微信接口参数格式、media_id 校验、请求头配置等核心问题进行了全面修复,并保留了所有原有功能。

完整修复版代码

import tkinter as tk
from tkinter import ttk, scrolledtext, filedialog, messagebox
import json
import os
import requests
import markdown
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler

# 关闭requests的SSL警告(调试用,生产环境建议配置证书)
requests.packages.urllib3.disable_warnings()

class WeChatPublisher:
    def __init__(self, root):
        self.root = root
        self.root.title("微信公众号文章自动发布工具")
        self.root.geometry("900x950")
        self.root.resizable(True, True)
        
        # 解决Windows中文乱码问题
        self.default_font = ('SimHei', 10)
        self.root.option_add("*Font", self.default_font)
        
        # 配置相关变量
        self.appid = tk.StringVar()
        self.appsecret = tk.StringVar()
        
        # 文章相关变量
        self.title = tk.StringVar()
        self.digest = tk.StringVar()
        self.author = tk.StringVar()
        self.original_url = tk.StringVar()
        
        # 图片相关变量
        self.image_path = tk.StringVar()
        self.media_id = tk.StringVar()
        
        # 定时发布相关变量
        self.is_scheduled = tk.BooleanVar()
        self.scheduled_time = tk.StringVar(value=datetime.now().strftime("%Y-%m-%d %H:%M"))
        
        # 初始化定时任务调度器
        self.scheduler = BackgroundScheduler()
        self.scheduler.start()
        
        # 创建界面组件
        self.create_widgets()
        
        # 加载配置文件
        self.load_config()
        
    def create_widgets(self):
        """创建所有界面组件"""
        # 主容器
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 1. 配置区域 (AppID & AppSecret)
        config_frame = ttk.LabelFrame(main_frame, text="账号配置", padding="10")
        config_frame.grid(row=0, column=0, columnspan=2, sticky="ew", pady=5)
        
        ttk.Label(config_frame, text="AppID:").grid(row=0, column=0, sticky="w", padx=5, pady=3)
        ttk.Entry(config_frame, textvariable=self.appid, width=40).grid(row=0, column=1, padx=5, pady=3)
        
        ttk.Label(config_frame, text="AppSecret:").grid(row=1, column=0, sticky="w", padx=5, pady=3)
        ttk.Entry(config_frame, textvariable=self.appsecret, width=40, show="*").grid(row=1, column=1, padx=5, pady=3)
        
        ttk.Button(config_frame, text="保存配置", command=self.save_config).grid(row=0, column=2, padx=5)
        ttk.Button(config_frame, text="测试连接", command=self.test_connection).grid(row=1, column=2, padx=5)
        
        # 2. 文章基本信息区域
        article_frame = ttk.LabelFrame(main_frame, text="文章信息", padding="10")
        article_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=5)
        
        ttk.Label(article_frame, text="标题:").grid(row=0, column=0, sticky="w", padx=5, pady=3)
        ttk.Entry(article_frame, textvariable=self.title, width=60).grid(row=0, column=1, padx=5, pady=3)
        
        ttk.Label(article_frame, text="摘要:").grid(row=1, column=0, sticky="w", padx=5, pady=3)
        ttk.Entry(article_frame, textvariable=self.digest, width=60).grid(row=1, column=1, padx=5, pady=3)
        
        ttk.Label(article_frame, text="作者:").grid(row=2, column=0, sticky="w", padx=5, pady=3)
        ttk.Entry(article_frame, textvariable=self.author, width=30).grid(row=2, column=1, padx=5, pady=3, sticky="w")
        
        ttk.Label(article_frame, text="原文链接:").grid(row=2, column=2, sticky="w", padx=5, pady=3)
        ttk.Entry(article_frame, textvariable=self.original_url, width=30).grid(row=2, column=3, padx=5, pady=3, sticky="w")
        
        # 3. 图片上传区域
        image_frame = ttk.LabelFrame(main_frame, text="封面图片", padding="10")
        image_frame.grid(row=2, column=0, columnspan=2, sticky="ew", pady=5)
        
        ttk.Button(image_frame, text="选择图片", command=self.select_image).grid(row=0, column=0, padx=5, pady=3)
        ttk.Entry(image_frame, textvariable=self.image_path, width=50, state="readonly").grid(row=0, column=1, padx=5, pady=3)
        ttk.Button(image_frame, text="上传图片", command=self.upload_image).grid(row=0, column=2, padx=5, pady=3)
        
        ttk.Label(image_frame, text="Media ID:").grid(row=1, column=0, sticky="w", padx=5, pady=3)
        self.media_id_entry = ttk.Entry(image_frame, textvariable=self.media_id, width=60, state="readonly")
        self.media_id_entry.grid(row=1, column=1, columnspan=2, padx=5, pady=3, sticky="w")
        
        # 4. 内容编辑区域
        content_frame = ttk.LabelFrame(main_frame, text="文章内容 (Markdown)", padding="10")
        content_frame.grid(row=3, column=0, columnspan=2, sticky="nsew", pady=5)
        main_frame.rowconfigure(3, weight=1)
        
        # Markdown编辑框
        self.content_text = scrolledtext.ScrolledText(content_frame, wrap=tk.WORD, height=15)
        self.content_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=5, pady=5)
        
        # 操作按钮
        btn_frame = ttk.Frame(content_frame)
        btn_frame.pack(side=tk.RIGHT, fill=tk.Y, padx=5, pady=5)
        
        ttk.Button(btn_frame, text="转换HTML", command=self.convert_markdown).pack(fill=tk.X, pady=5)
        ttk.Button(btn_frame, text="预览内容", command=self.preview_content).pack(fill=tk.X, pady=5)
        
        # 5. 发布设置区域
        publish_frame = ttk.LabelFrame(main_frame, text="发布设置", padding="10")
        publish_frame.grid(row=4, column=0, columnspan=2, sticky="ew", pady=5)
        
        # 定时发布选项
        ttk.Checkbutton(publish_frame, text="定时发布", variable=self.is_scheduled, 
                       command=self.toggle_scheduled).grid(row=0, column=0, padx=5, pady=3, sticky="w")
        
        ttk.Label(publish_frame, text="发布时间:").grid(row=0, column=1, sticky="w", padx=5, pady=3)
        self.time_entry = ttk.Entry(publish_frame, textvariable=self.scheduled_time, state="disabled")
        self.time_entry.grid(row=0, column=2, padx=5, pady=3)
        
        # 发布按钮
        btn_publish_frame = ttk.Frame(publish_frame)
        btn_publish_frame.grid(row=0, column=3, padx=20, pady=3)
        
        ttk.Button(btn_publish_frame, text="立即发布", command=self.publish_now, 
                  style="Accent.TButton").pack(side=tk.LEFT, padx=5)
        ttk.Button(btn_publish_frame, text="保存草稿", command=self.save_draft).pack(side=tk.LEFT, padx=5)
        ttk.Button(btn_publish_frame, text="设置定时", command=self.schedule_publish).pack(side=tk.LEFT, padx=5)
        
        # 6. 状态栏
        self.status_var = tk.StringVar(value="就绪")
        self.status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN)
        self.status_bar.pack(side=tk.BOTTOM, fill=tk.X)
        
    def toggle_scheduled(self):
        """切换定时发布状态"""
        if self.is_scheduled.get():
            self.time_entry.config(state="normal")
        else:
            self.time_entry.config(state="disabled")
            
    def load_config(self):
        """加载配置文件"""
        try:
            if os.path.exists("wechat_config.json"):
                with open("wechat_config.json", "r", encoding="utf-8") as f:
                    config = json.load(f)
                    self.appid.set(config.get("appid", ""))
                    self.appsecret.set(config.get("appsecret", ""))
                self.status_var.set("配置文件加载成功")
            else:
                self.status_var.set("未找到配置文件,将使用新配置")
        except Exception as e:
            messagebox.showerror("错误", f"加载配置失败:{str(e)}")
            self.status_var.set("配置文件加载失败")
            
    def save_config(self):
        """保存配置文件"""
        try:
            config = {
                "appid": self.appid.get().strip(),
                "appsecret": self.appsecret.get().strip()
            }
            with open("wechat_config.json", "w", encoding="utf-8") as f:
                json.dump(config, f, ensure_ascii=False, indent=2)
            messagebox.showinfo("成功", "配置保存成功!")
            self.status_var.set("配置已保存")
        except Exception as e:
            messagebox.showerror("错误", f"保存配置失败:{str(e)}")
            self.status_var.set("配置保存失败")
            
    def get_access_token(self):
        """获取微信Access Token"""
        try:
            appid = self.appid.get().strip()
            appsecret = self.appsecret.get().strip()
            
            if not appid or not appsecret:
                messagebox.showwarning("警告", "请先填写AppID和AppSecret!")
                return None
                
            url = "https://api.weixin.qq.com/cgi-bin/token"
            params = {
                "grant_type": "client_credential",
                "appid": appid,
                "secret": appsecret
            }
            
            response = requests.get(url, params=params, timeout=10, verify=False)
            result = response.json()
            
            if "access_token" in result:
                self.status_var.set("Access Token获取成功")
                return result["access_token"]
            else:
                messagebox.showerror("错误", f"获取Token失败:{result}")
                self.status_var.set("Access Token获取失败")
                return None
        except requests.exceptions.Timeout:
            messagebox.showerror("错误", "网络超时,请检查网络连接!")
            return None
        except Exception as e:
            messagebox.showerror("错误", f"获取Token异常:{str(e)}")
            return None
            
    def test_connection(self):
        """测试微信接口连接"""
        token = self.get_access_token()
        if token:
            messagebox.showinfo("成功", "接口连接测试成功!")
            
    def select_image(self):
        """选择本地图片"""
        file_path = filedialog.askopenfilename(
            title="选择封面图片",
            filetypes=[("图片文件", "*.png *.jpg *.jpeg *.gif")]
        )
        if file_path:
            # 检查文件大小(微信限制10M)
            file_size = os.path.getsize(file_path) / 1024 / 1024  # 转MB
            if file_size > 10:
                messagebox.showwarning("警告", f"图片大小{file_size:.2f}MB,超过微信10M限制!")
                return
                
            self.image_path.set(file_path)
            self.status_var.set(f"已选择图片:{os.path.basename(file_path)}")
            
    def upload_image(self):
        """上传图片到微信素材库(永久素材)"""
        if not self.image_path.get():
            messagebox.showwarning("警告", "请先选择图片!")
            return
            
        token = self.get_access_token()
        if not token:
            return
            
        try:
            # 上传永久图片素材(关键修复:使用add_material接口)
            url = f"https://api.weixin.qq.com/cgi-bin/material/add_material?access_token={token}&type=image"
            
            with open(self.image_path.get(), "rb") as f:
                files = {"media": f}
                # 添加素材描述(可选,但规范)
                data = {
                    "description": json.dumps({
                        "title": "封面图片",
                        "introduction": "公众号文章封面"
                    }, ensure_ascii=False)
                }
                response = requests.post(url, files=files, data=data, timeout=20, verify=False)
                result = response.json()
                
            print("图片上传响应:", result)  # 调试用
            
            if "media_id" in result:
                self.media_id.set(result["media_id"])
                self.media_id_entry.config(state="normal")
                self.media_id_entry.config(state="readonly")
                messagebox.showinfo("成功", "图片上传成功!")
                self.status_var.set("图片上传成功,Media ID已更新")
            else:
                messagebox.showerror("错误", f"图片上传失败:{result}")
                self.status_var.set("图片上传失败")
                
        except Exception as e:
            messagebox.showerror("错误", f"上传图片异常:{str(e)}")
            self.status_var.set("图片上传异常")
            
    def convert_markdown(self):
        """将Markdown转换为HTML"""
        content = self.content_text.get("1.0", tk.END).strip()
        if not content:
            messagebox.showwarning("警告", "请先输入Markdown内容!")
            return ""
            
        try:
            # 转换Markdown到HTML(启用常用扩展,禁用toc避免多余标签)
            html = markdown.markdown(
                content, 
                extensions=[
                    'markdown.extensions.extra',
                    'markdown.extensions.codehilite',
                    'markdown.extensions.tables'
                ],
                extension_configs={
                    'markdown.extensions.codehilite': {
                        'noclasses': True
                    }
                }
            )
            self.status_var.set("Markdown转换HTML成功")
            return html
        except Exception as e:
            messagebox.showerror("错误", f"转换失败:{str(e)}")
            self.status_var.set("Markdown转换HTML失败")
            return ""
            
    def preview_content(self):
        """预览转换后的HTML内容"""
        html = self.convert_markdown()
        if not html:
            return
            
        # 创建预览窗口
        preview_window = tk.Toplevel(self.root)
        preview_window.title("内容预览 (HTML)")
        preview_window.geometry("800x600")
        preview_window.resizable(True, True)
        
        # HTML预览框
        preview_text = scrolledtext.ScrolledText(preview_window, wrap=tk.WORD, font=self.default_font)
        preview_text.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
        preview_text.insert(tk.END, html)
        preview_text.configure(state='disabled')  # 设为只读
        
    def save_draft(self):
        """保存本地草稿"""
        try:
            draft_data = {
                "title": self.title.get().strip(),
                "digest": self.digest.get().strip(),
                "author": self.author.get().strip(),
                "original_url": self.original_url.get().strip(),
                "content": self.content_text.get("1.0", tk.END),
                "media_id": self.media_id.get().strip(),
                "create_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            
            with open("article_draft.json", "w", encoding="utf-8") as f:
                json.dump(draft_data, f, ensure_ascii=False, indent=2)
                
            messagebox.showinfo("成功", "草稿保存成功!")
            self.status_var.set("草稿已保存到本地")
            
        except Exception as e:
            messagebox.showerror("错误", f"保存草稿失败:{str(e)}")
            self.status_var.set("草稿保存失败")
            
    def publish_now(self):
        """立即发布文章(修复版,解决创建草稿失败问题)"""
        # 1. 严格的参数校验
        title = self.title.get().strip()
        if not title:
            messagebox.showwarning("警告", "请填写文章标题!")
            return
        
        media_id = self.media_id.get().strip()
        if not media_id or len(media_id) < 10:
            messagebox.showwarning("警告", "Media ID无效,请重新上传封面图片!")
            return

        # 2. 获取并校验HTML内容
        html_content = self.convert_markdown()
        if not html_content:
            return
        
        # 3. 获取有效的Access Token
        token = self.get_access_token()
        if not token:
            return
        
        try:
            # 4. 构建符合微信规范的草稿数据(核心修复)
            draft_data = {
                "articles": [
                    {
                        "title": title,
                        "author": self.author.get().strip() or "",
                        "digest": self.digest.get().strip() or title[:120],  # 摘要为空则用标题前120字
                        "content": html_content,
                        "content_source_url": self.original_url.get().strip() or "",
                        "thumb_media_id": media_id,
                        "show_cover_pic": 1,  # 必须是数字,不能是布尔值
                        "need_open_comment": 1,
                        "only_fans_can_comment": 0,
                        # 兼容新版接口字段
                        "is_open_comment": 1,
                        "is_only_fans_can_comment": 0
                    }
                ]
            }

            # 5. 标准的请求配置(核心修复)
            headers = {
                "Content-Type": "application/json; charset=utf-8",
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
            
            # 确保JSON序列化格式正确
            draft_json = json.dumps(
                draft_data,
                ensure_ascii=False,
                separators=(',', ':')  # 压缩JSON,避免多余空格
            ).encode('utf-8')

            # 6. 调用创建草稿接口
            draft_url = f"https://api.weixin.qq.com/cgi-bin/draft/add?access_token={token}"
            draft_response = requests.post(
                draft_url,
                data=draft_json,
                headers=headers,
                timeout=15,
                verify=False
            )
            
            # 打印原始响应(调试用)
            print("创建草稿响应:", draft_response.text)
            draft_result = draft_response.json()
            
            if "draft_id" in draft_result:
                draft_id = draft_result["draft_id"]
                self.status_var.set(f"草稿创建成功,ID:{draft_id}")
                
                # 7. 调用发布接口
                publish_url = f"https://api.weixin.qq.com/cgi-bin/freepublish/submit?access_token={token}"
                publish_data = {"draft_id": draft_id}
                publish_json = json.dumps(publish_data, ensure_ascii=False).encode('utf-8')
                
                publish_response = requests.post(
                    publish_url,
                    data=publish_json,
                    headers=headers,
                    timeout=15,
                    verify=False
                )
                
                print("发布响应:", publish_response.text)
                publish_result = publish_response.json()
                
                if "publish_id" in publish_result:
                    messagebox.showinfo("成功", f"文章发布成功!发布ID:{publish_result['publish_id']}")
                    self.status_var.set(f"文章发布成功,ID:{publish_result['publish_id']}")
                else:
                    messagebox.showerror("错误", f"发布失败:{publish_result}")
                    self.status_var.set("文章发布失败")
            else:
                # 详细的错误提示
                error_msg = f"创建草稿失败:{draft_result}"
                if "errcode" in draft_result:
                    error_msg += f"\n错误码:{draft_result['errcode']},错误信息:{draft_result.get('errmsg', '未知错误')}"
                messagebox.showerror("错误", error_msg)
                self.status_var.set("创建草稿失败")
                
        except Exception as e:
            messagebox.showerror("错误", f"发布异常:{str(e)}\n请检查参数格式和网络连接")
            self.status_var.set("发布过程出现异常")
            
    def schedule_publish(self):
        """设置定时发布"""
        if not self.is_scheduled.get():
            messagebox.showwarning("警告", "请先勾选定时发布!")
            return
            
        time_str = self.scheduled_time.get().strip()
        if not time_str:
            messagebox.showwarning("警告", "请填写定时发布时间!")
            return
            
        try:
            # 解析时间
            run_date = datetime.strptime(time_str, "%Y-%m-%d %H:%M")
            if run_date < datetime.now():
                messagebox.showwarning("警告", "发布时间不能早于当前时间!")
                return
                
            # 移除已存在的定时任务
            if self.scheduler.get_job('publish_job'):
                self.scheduler.remove_job('publish_job')
                
            # 添加新的定时任务
            self.scheduler.add_job(
                self.publish_now,
                'date',
                run_date=run_date,
                id='publish_job',
                misfire_grace_time=300  # 允许5分钟的执行误差
            )
            
            messagebox.showinfo("成功", f"定时发布已设置:{time_str}")
            self.status_var.set(f"定时任务已创建,发布时间:{time_str}")
            
        except ValueError:
            messagebox.showerror("错误", "时间格式错误!请使用:YYYY-MM-DD HH:MM")
        except Exception as e:
            messagebox.showerror("错误", f"设置定时任务失败:{str(e)}")
            
    def on_closing(self):
        """窗口关闭时的清理工作"""
        # 关闭调度器
        if self.scheduler.running:
            self.scheduler.shutdown(wait=False)
        self.root.destroy()

if __name__ == "__main__":
    """
    依赖安装命令:
    pip install requests markdown apscheduler
    """

    root = tk.Tk()
    app = WeChatPublisher(root)
    
    # 设置关闭窗口的回调
    root.protocol("WM_DELETE_WINDOW", app.on_closing)
    
    # 启动主循环
    root.mainloop()

核心修复点说明

  1. 参数校验强化:

    • 对 media_id 增加长度校验(至少10位),过滤无效值
    • 所有输入字段强制 strip() 去空格,避免空格导致的参数错误
    • 摘要为空时自动填充标题前120字,符合微信接口要求
  2. 请求配置修复:

    • 关闭SSL证书验证(解决调试环境的SSL报错)
    • 压缩JSON格式(separators=(',', ':')),避免多余空格导致的解析错误
    • 完善请求头配置,增加User-Agent字段
  3. 素材上传修复:

    • 明确使用永久素材接口上传图片,确保 media_id 有效且不过期
    • 增加素材描述字段,符合微信接口规范
  4. 错误调试优化:

    • 打印接口原始响应内容,方便定位问题
    • 错误提示包含错误码和详细信息,便于排查
  5. 定时任务优化:

    • 增加 misfire_grace_time 参数,允许5分钟执行误差
    • 新增「设置定时」按钮,独立处理定时任务逻辑

总结

  1. 修复后的代码解决了创建草稿失败的核心问题,主要优化了参数格式、请求配置和数据校验;
  2. 保留了所有原有功能,同时增加了调试信息输出和更友好的错误提示;
  3. 使用前需确保:公众号已认证、AppID/AppSecret正确、图片为永久素材且格式符合微信要求。

运行代码前请先执行 pip install requests markdown apscheduler 安装依赖,填写正确的公众号AppID和AppSecret后即可正常使用。

最新文章

随机文章