自媒体内容创作难？程序员教你用AI驱动，用 Python轻松解决

import asyncio
from deepseek import DeepSeekAPI
from typing import Dict, List
class AIContentGenerator:
    def __init__(self):
        self.api_key = "your_deepseek_api_key"
        self.client = DeepSeekAPI(api_key=self.api_key)
        self.platform_templates = {
            "wechat": "写一篇关于{topic}的公众号文章，1500字，专业但易懂，带小标题和代码示例",
            "xiaohongshu": "写一篇关于{topic}的小红书笔记，500字，口语化，加emoji，适合年轻人阅读",
            "douyin": "写一篇关于{topic}的抖音口播文案，300字，非常口语化，用'家人们'开头"
        }
    async def generate_platform_content(self, topic: str, platform: str) -> str:
        """为指定平台生成适配内容"""
        if platform not in self.platform_templates:
            raise ValueError(f"不支持的平台：{platform}")
        prompt = self.platform_templates[platform].format(topic=topic)
        try:
            response = await self.client.chat.completions.create(
                model="deepseek-chat",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=2000,
                temperature=0.7
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"AI生成失败：{str(e)}")
            return ""
    async def batch_generate(self, topics: List[str], platforms: List[str]) -> Dict:
        """批量生成多平台内容"""
        results = {}
        async def generate_task(topic, platform):
            content = await self.generate_platform_content(topic, platform)
            return (platform, content)
        tasks = www.aizhl.cn
        for topic in topics:
            for platform in platforms:
                tasks.append(generate_task(topic, platform))
        results_list = await asyncio.gather(*tasks)
        for platform, content in results_list:
            if platform not in results:
                results[platform] = []
            results[platform].append(content)
        return results

​from celery import Celery
import requests
from typing import List
app = Celery('publish_tasks', broker='redis://localhost:6379/0')
class PlatformPublisher:
    def __init__(self):
        self.platform_configs = {
            "wechat": {"api_url": "https://api.weixin.qq.com/cgi-bin/", "access_token": "your_wechat_token"},
            "xiaohongshu": {"api_url": "https://open.xiaohongshu.com/", "app_key": "your_xiaohongshu_key"},
            "douyin": {"api_url": "https://open.douyin.com/", "client_key": "your_douyin_key"}
        }
    def publish_to_wechat(self, content: str, title: str) -> bool:
        """发布到微信公众号"""
        try:
            url = f"{self.platform_configs['wechat']['api_url']}material/add_news?access_token={self.platform_configs['wechat']['access_token']}"
            data = {
                "articles": [{
                    "title": title,
                    "content": content,
                    "digest": content[:100],
                    "show_cover_pic": 1
                }]
            }
            response = requests.post(url, json=data)
            return response.json().get("errcode") == 0
        except Exception as e:
            print(f"公众号发布失败：{str(e)}")
            return False
@app.task(retry_backoff=5, max_retries=3)
def publish_article(article_data: Dict, platforms: List[str]) -> Dict:
    """异步发布任务"""
    publisher = PlatformPublisher()
    results = {}
    for platform in platforms:
        try:
            if platform == "wechat":
                success = publisher.publish_to_wechat(
                    article_data["content"], 
                    article_data["title"]
                )
            elif platform == "xiaohongshu":
                success = publisher.publish_to_xiaohongshu(article_data["content"])
            elif platform == "douyin":
                success = publisher.publish_to_douyin(article_data["content"])
            else:
                success = False
            results[platform] = {"success": success}
        except Exception as e:
            results[platform] = {"success": False, "error": str(e)}
            if platform in ["wechat", "xiaohongshu"]:  # 重要平台重试
                raise publish_article.retry(exc=e)
    return results