别再手动调API了！用Python搭建自己的AI工作流，5分钟自动化小红书/知乎多平台发布

pip install playwright deepseek-api loguru
playwright install

import os
from loguru import logger
from playwright.sync_api import sync_playwright
from deepseek_api import DeepSeek

# 初始化日志和AI
logger.add("publish.log", rotation="1 day")
deepseek = DeepSeek(api_key=os.getenv("DEEPSEEK_KEY"))

def adapt_content(platform, original_content):
    """使用AI适配不同平台的内容风格"""
    prompt = f"将以下内容转换为适合{platform}风格的文案，保持核心信息不变:\n{original_content}"
    return deepseek.generate(prompt)

def publish_xiaohongshu(title, content, images=None):
    """小红书发布逻辑"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        try:
            page.goto("https://creator.xiaohongshu.com")
            # 这里简化了登录流程，实际应使用存储的cookies
            page.click("button.publish-btn")
            page.fill("textarea.title", title)
            page.fill("div.editor", content)

            if images:
                for img in images:
                    page.click("input.upload-btn")
                    page.set_input_files("input.upload-btn", img)

            page.click("button.submit-btn")
            logger.success("小红书发布成功")
            return True
        except Exception as e:
            logger.error(f"小红书发布失败: {str(e)}")
            return False
        finally:
            browser.close()

def publish_zhihu(question, answer):
    """知乎发布逻辑"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        try:
            page.goto("https://www.zhihu.com")
            page.click("button.WriteButton")
            page.fill("input.QuestionInput", question)
            page.fill("div.RichText-editor", answer)
            page.click("button.SubmitButton")
            logger.success("知乎发布成功")
            return True
        except Exception as e:
            logger.error(f"知乎发布失败: {str(e)}")
            return False
        finally:
            browser.close()

if __name__ == "__main__":
    original_content = """Python自动化可以大幅提升工作效率。
    本文介绍了5个实用技巧，包括文件处理、邮件自动发送等..."""

    # 平台特定内容适配
    xhs_content = adapt_content("小红书", original_content)
    zh_content = adapt_content("知乎", original_content)

    # 执行发布
    publish_xiaohongshu("Python自动化5大技巧", xhs_content, ["pic1.png"])
    publish_zhihu("如何用Python提高工作效率？", zh_content)

context = browser.new_context(storage_state="auth.json")
page = context.new_page()

prompt += "请使用更多emoji和分段，添加1-2个相关话题标签"

import asyncio

async def publish_all():
    await asyncio.gather(
        async_publish_xiaohongshu(),
        async_publish_zhihu()
    )

2024-03-20 09:00:12 | SUCCESS | 小红书发布成功
2024-03-20 09:01:05 | SUCCESS | 知乎发布成功
2024-03-20 09:01:23 | WARNING | 微博发布超时，已重试

logger.add(
    lambda msg: send_lark_alert(msg) if "ERROR" in msg else None,
    level="ERROR"
)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=4, max=10))
def publish_xiaohongshu(title, content):
    # 原有逻辑不变

ai_publisher/
├── platforms/
│   ├── xiaohongshu.py
│   ├── zhihu.py
│   └── __init__.py
├── adapters/
│   ├── content_ai.py
│   └── image_processor.py
├── logs/
│   └── publish.log
├── config.py
└── main.py

from time import sleep
import random

def human_delay():
    sleep(random.uniform(0.5, 2.0))

# 在关键操作前调用
page.click("button")
human_delay()