Automating WeChat Public Account Publishing with AI (DeepSeek & Qwen)

The guide demonstrates how to automate the creation and drafting of WeChat public‑account articles using large‑language‑model APIs from DeepSeek and Alibaba Qwen.

Environment variables

Create a .env file containing the required keys, for example:

# Alibaba Qwen configuration
DASHSCOPE_API_KEY=sk-5...
QWEN_MODEL=qwen-plus

# WeChat configuration
WECHAT_APP_ID=wx...
WECHAT_APP_SECRET=6...

# System configuration
ARTICLE_TOPIC=AI软件测试
PUBLISH_TIME=8:00  # daily publish time
SAVE_TO_DRAFT=true  # true saves as draft, false publishes directly

Obtain WECHAT_APP_ID from the WeChat official account settings page and WECHAT_APP_SECRET from the WeChat developer console.

Qwen client (qwen_client.py)

import os, json
from openai import OpenAI
from loguru import logger
from dotenv import load_dotenv

load_dotenv()

class QwenClient:
    """Alibaba Qwen API client wrapper"""
    def __init__(self):
        self.api_key = os.getenv("DASHSCOPE_API_KEY")
        self.model = os.getenv("QWEN_MODEL", "qwen-plus")
        if not self.api_key:
            logger.error("Missing DASHSCOPE_API_KEY")
        self.client = OpenAI(api_key=self.api_key, base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")
        logger.info(f"Qwen client initialized with model: {self.model}")

    def generate_article(self, topic: str = None) -> dict:
        """Generate an AI‑written article about the given topic"""
        system_prompt = """You are a professional AI software‑testing expert ..."""
        user_prompt = f"请撰写一篇关于「{topic or 'AI软件测试'}」的技术文章"
        try:
            completion = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "system", "content": system_prompt},
                          {"role": "user", "content": user_prompt}],
                response_format={"type": "json_object"},
                temperature=0.8,
            )
            article = json.loads(completion.choices[0].message.content)
            if "content" in article:
                article["content"] = self._format_content(article["content"])
            if "title" in article and len(article["title"]) > 64:
                article["title"] = article["title"][:61] + "..."
            return article
        except json.JSONDecodeError as e:
            logger.error(f"JSON decode error: {e}")
            return self._get_fallback_article(topic)
        except Exception as e:
            logger.error(f"Article generation failed: {e}")
            return self._get_fallback_article(topic)

    def _format_content(self, content: str) -> str:
        """Convert plain text to HTML for WeChat"""
        paragraphs = content.split('

')
        html_parts = []
        for p in paragraphs:
            p = p.strip()
            if not p:
                continue
            if p.startswith('# '):
                html_parts.append(f'<h2>{p[2:]}</h2>')
            elif p.startswith('## '):
                html_parts.append(f'<h3>{p[3:]}</h3>')
            else:
                html_parts.append(f'<p>{p}</p>')
        return '
'.join(html_parts)

    def _get_fallback_article(self, topic: str) -> dict:
        default_title = f"今日AI测试洞察：{topic or 'AI软件测试'}"
        if len(default_title) > 64:
            default_title = default_title[:61] + "..."
        return {
            "title": default_title,
            "content": "<p>内容生成服务暂时不可用，请稍后查看。</p>",
            "summary": "内容生成服务暂时不可用",
            "image_prompt": "AI software testing, futuristic, blue tone",
        }

WeChat client (wechat_client.py)

import os, time, requests, json, urllib3
from loguru import logger
from dotenv import load_dotenv

load_dotenv()
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class WeChatClient:
    """WeChat public‑account API client"""
    def __init__(self):
        self.app_id = os.getenv("WECHAT_APP_ID")
        self.app_secret = os.getenv("WECHAT_APP_SECRET")
        logger.debug(f"Init - app_id: {self.app_id}")
        self.access_token = None
        self.token_expires = 0

    def _get_access_token(self):
        if self.access_token and time.time() < self.token_expires:
            return self.access_token
        url = "https://api.weixin.qq.com/cgi-bin/token"
        params = {"grant_type": "client_credential", "appid": self.app_id, "secret": self.app_secret}
        try:
            resp = requests.get(url, params=params, timeout=10)
            data = resp.json()
            if "access_token" in data:
                self.access_token = data["access_token"]
                self.token_expires = time.time() + data["expires_in"] - 200
                logger.info("access_token obtained")
                return self.access_token
            logger.error(f"Token fetch failed: {data}")
            return None
        except Exception as e:
            logger.error(f"Token fetch exception: {e}")
            return None

    # Methods for uploading permanent/temporary images, adding news, creating drafts, etc.
    # (code omitted for brevity – identical to the source)

Image generation module (image_gen.py)

import os, requests
from loguru import logger
from dotenv import load_dotenv

load_dotenv()

class ImageGenerator:
    """Tongyi Wanxiang image generation"""
    def __init__(self):
        self.api_key = os.getenv("DASHSCOPE_API_KEY")
        self.base_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"
        logger.info(f"ImageGenerator init, API key set: {'yes' if self.api_key else 'no'}")

    def generate(self, prompt: str):
        if not self.api_key:
            logger.error("API key missing, cannot generate image")
            return None
        headers = {"Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}"}
        data = {"model": "wanx-v1", "input": {"prompt": prompt}, "parameters": {"size": "1024*1024", "n": 1}}
        try:
            logger.info(f"Generating image for prompt: {prompt[:50]}...")
            resp = requests.post(self.base_url, headers=headers, json=data, timeout=60)
            if resp.status_code != 200:
                logger.error(f"Image generation HTTP error: {resp.status_code}")
                return None
            result = resp.json()
            # Handle several possible return formats (results, task_status, choices)
            if "output" in result:
                output = result["output"]
                if "results" in output and output["results"]:
                    url = output["results"][0].get("url")
                    if url:
                        logger.success(f"Image generated: {url}")
                        return url
                if "task_status" in output and output["task_status"] == "SUCCEEDED" and output.get("results"):
                    url = output["results"][0].get("url")
                    if url:
                        logger.success(f"Image generated: {url}")
                        return url
                if "choices" in output:
                    for choice in output["choices"]:
                        msg = choice.get("message", {})
                        content = msg.get("content", [])
                        if isinstance(content, list):
                            for item in content:
                                if isinstance(item, dict) and "image" in item:
                                    url = item["image"]
                                    logger.success(f"Image generated: {url}")
                                    return url
            logger.error(f"Unrecognized response format: {result}")
            return None
        except requests.exceptions.Timeout:
            logger.error("Image generation timeout")
            return None
        except requests.exceptions.RequestException as e:
            logger.error(f"Image generation request error: {e}")
            return None
        except Exception as e:
            logger.error(f"Unknown image generation error: {e}")
            return None

Topic generator (topic_generator.py)

import random
from datetime import datetime

class TopicGenerator:
    """AI software‑testing topic generator"""
    TOPICS = [
        "大模型测试", "AI驱动测试", "测试用例自动生成", "智能体测试",
        "LLM测试实践", "提示词测试", "RAG系统测试", "AI安全测试",
        "测试数据生成", "自愈测试脚本", "测试左移", "AI测试工具",
        "模型评估", "对抗测试", "AI在CI/CD中的应用", "测试覆盖率优化",
        "测试预测分析", "智能回归测试", "多模态测试", "A/B测试自动化"
    ]
    ANGLES = [
        "2026年最新趋势", "实战案例", "技术深度解析", "工具对比",
        "常见误区", "性能优化", "落地实践", "未来展望",
        "与传统的对比", "团队转型", "成本效益分析", "开源方案"
    ]
    @classmethod
    def generate(cls, base_topic: str = "AI软件测试") -> str:
        topic = random.choice(cls.TOPICS)
        angle = random.choice(cls.ANGLES)
        templates = [
            f"{topic}：{angle}",
            f"{angle}：{topic}实战",
            f"2026年{topic}的{angle}",
            f"深度解读：{topic}{angle}",
            f"测试专家必看：{topic}{angle}"
        ]
        return random.choice(templates)

Publisher script (run_publisher.py)

import os, sys, time, json
from datetime import datetime
from dotenv import load_dotenv
from loguru import logger
import requests

from qwen_client import QwenClient
from wechat_client import WeChatClient
from image_gen import ImageGenerator
from topic_generator import TopicGenerator

# Logging configuration omitted for brevity

def get_local_fallback_image() -> str:
    base_dir = os.path.dirname(os.path.abspath(__file__))
    path = os.path.join(base_dir, "default_cover.jpg")
    return path if os.path.exists(path) else None

# Helper functions for extracting URLs, uploading local images, saving drafts, etc.
# (code identical to source, kept inside <pre><code>)

def run_publish_task():
    logger.info("="*30)
    logger.info("Starting AI article publishing task")
    logger.info(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    logger.info("="*30)
    try:
        qwen = QwenClient()
        wechat = WeChatClient()
        image_gen = ImageGenerator()
    except Exception as e:
        logger.error(f"Client init failed: {e}")
        return False

    # 1. Generate topic
    logger.info("Step 1: Generate article topic")
    topic = TopicGenerator.generate("AI软件测试")
    logger.info(f"Today's topic: {topic}")

    # 2. Generate article content
    logger.info("Step 2: Generate article content")
    article = qwen.generate_article(topic)
    if not article or not article.get("title"):
        logger.error("Article generation failed")
        return False
    logger.success(f"Article generated: {article['title']}")

    # 3. Prepare cover image
    logger.info("Step 3: Prepare cover image")
    image_prompt = article.get("image_prompt", "AI software testing, futuristic technology, blue tone, 4k")
    image_result = image_gen.generate(image_prompt)
    image_url = extract_image_url_from_result(image_result)
    if not image_url:
        # fallback to random online image
        fallback_url = f"https://picsum.photos/1024/1024?random={int(time.time())}"
        try:
            resp = requests.get(fallback_url, timeout=5, headers={'User-Agent': 'Mozilla/5.0'}, stream=True)
            if resp.status_code == 200:
                image_url = fallback_url
                logger.info(f"Using fallback image: {fallback_url}")
        except Exception:
            pass
    if not image_url:
        local_path = get_local_fallback_image()
        if local_path:
            image_path = local_path
            logger.success(f"Loaded local fallback image: {image_path}")
        else:
            logger.warning("No image available")

    # 4. Upload image to WeChat
    logger.info("Step 4: Upload image to WeChat")
    media_id = None
    if image_url:
        media_id = wechat.upload_permanent_image(image_url)
        if not media_id:
            # download then upload as local file
            try:
                img_resp = requests.get(image_url, timeout=30, headers={'User-Agent': 'Mozilla/5.0'})
                if img_resp.status_code == 200:
                    temp_path = f"/tmp/temp_cover_{int(time.time())}.jpg"
                    with open(temp_path, 'wb') as f:
                        f.write(img_resp.content)
                    media_id = upload_local_image(wechat, temp_path)
                    os.remove(temp_path)
            except Exception as e:
                logger.error(f"Download/upload image failed: {e}")
    elif image_path:
        media_id = upload_local_image(wechat, image_path)

    if not media_id:
        logger.error("Image upload failed, saving article locally as draft")
        save_article_locally(article, image_url, image_path)
        return False

    # 5. Create WeChat draft
    logger.info("Step 5: Create WeChat draft")
    success = wechat.add_draft(article, media_id)
    if success:
        logger.success("Task completed – article saved to draft box")
        return True
    else:
        logger.error("Draft save failed, saving local backup")
        save_article_locally(article, image_url, image_path)
        return False

if __name__ == "__main__":
    load_dotenv()
    os.makedirs("logs", exist_ok=True)
    os.makedirs("drafts", exist_ok=True)
    is_success = run_publish_task()
    sys.exit(0 if is_success else 1)

Dependencies

fastapi==0.115.8
uvicorn==0.34.0
apscheduler==3.11.0
python-dotenv==1.0.1
loguru==0.7.3
requests==2.32.3
openai==1.59.6
urllib3==2.3.0

Install them with pip install -r requirements.txt.

Deployment notes

After the script runs, check the error logs for the outbound IP address and add it to the WeChat developer console under "IP whitelist". The article also outlines how to deploy the pipeline to Alibaba Function Compute (FC) for serverless execution.

Quote

Gu Xiangfan says: "In the AI era, mastering the known unknowns is no longer hard; the key is discovering the unknown unknowns, which often hide within the exploration of the known unknowns."