基于Python的AI图文自动生成与小红书发布系统实现

1次阅读

共计 3998 个字符，预计需要花费 10 分钟才能阅读完成。

作为内容创作者，每天手动生产大量图文内容并发布到小红书上是一项极其耗时的工作。特别是在需要保持内容更新频率的情况下，传统的手工操作方式效率低下，难以满足快速内容生产的需求。本文介绍一套基于 Python 的自动化解决方案，帮助开发者构建一个端到端的自动化内容生产流水线。

对于内容创作者来说，手动生产内容存在以下主要问题：

时间成本高：从构思文案、寻找配图到最终发布，每个环节都需要人工参与
创作瓶颈：持续产出高质量创意内容对创作者是巨大挑战
发布效率低：手动操作发布流程无法实现批量处理

这些痛点催生了自动化内容生产的需求，通过技术手段实现 ’ 内容冷启动 ’，帮助创作者快速生成大量优质内容。

GPT-3：生成质量高但 API 调用成本较高
ChatGLM：中文处理能力强，本地部署成本低
其他开源模型：如 BLOOM、GPT- J 等，需要自行调整参数

CLIP 模型：由 OpenAI 开发，图文匹配效果优秀
其他 CV 模型：如 ResNet 等传统视觉模型，匹配精度相对较低

经过对比，我们选择 ChatGLM+CLIP 的组合，兼顾中文处理能力和图文匹配效果。

import requests

def generate_text(prompt):
    """调用 ChatGLM API 生成文案"""
    url = "https://api.chatglm.cn/v1/completions"
    headers = {
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json"
    }
    data = {
        "prompt": prompt,
        "max_tokens": 200,
        "temperature": 0.7
    }

    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()['choices'][0]['text']
    except Exception as e:
        print(f"文案生成失败: {str(e)}")
        return None

import clip
import torch
from PIL import Image

def load_clip_model():
    """加载 CLIP 模型"""
    device = "cuda" if torch.cuda.is_available() else "cpu"
    model, preprocess = clip.load("ViT-B/32", device=device)
    return model, preprocess, device

def match_image_text(text, image_paths, top_k=1):
    """匹配文本与图片"""
    model, preprocess, device = load_clip_model()

    # 处理文本输入
    text_input = clip.tokenize([text]).to(device)

    # 处理图片输入
    images = [preprocess(Image.open(img_path)) for img_path in image_paths]
    image_input = torch.stack(images).to(device)

    # 计算相似度
    with torch.no_grad():
        text_features = model.encode_text(text_input)
        image_features = model.encode_image(image_input)

        # 计算余弦相似度
        similarity = (text_features @ image_features.T).softmax(dim=-1)
        values, indices = similarity[0].topk(top_k)

    return [image_paths[i] for i in indices]

import datetime

def post_to_xiaohongshu(content, image_path):
    """发布内容到小红书"""
    url = "https://open.xiaohongshu.com/api/content/publish"
    headers = {
        "Authorization": "Bearer YOUR_ACCESS_TOKEN",
        "Content-Type": "multipart/form-data"
    }

    files = {'image': open(image_path, 'rb')}
    data = {
        'content': content,
        'publish_time': datetime.datetime.now().isoformat()
    }

    try:
        response = requests.post(url, headers=headers, data=data, files=files)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"发布失败: {str(e)}")
        return None

import logging
from typing import List, Optional

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class XiaohongshuAutoPublisher:
    """小红书自动发布系统"""

    def __init__(self):
        self.text_model = "chatglm"
        self.image_model = "clip"

    def generate_content(self, prompt: str) -> Optional[str]:
        """生成文案内容"""
        try:
            content = generate_text(prompt)
            if content:
                logger.info("文案生成成功")
                return content
            logger.warning("文案生成为空")
            return None
        except Exception as e:
            logger.error(f"文案生成异常: {str(e)}")
            return None

    def select_image(self, text: str, image_pool: List[str]) -> Optional[str]:
        """选择匹配的图片"""
        try:
            matched_images = match_image_text(text, image_pool)
            if matched_images:
                logger.info(f"匹配到 {len(matched_images)} 张图片")
                return matched_images[0]
            logger.warning("未匹配到合适图片")
            return None
        except Exception as e:
            logger.error(f"图片匹配异常: {str(e)}")
            return None

    def publish(self, content: str, image_path: str) -> bool:
        """发布内容"""
        try:
            result = post_to_xiaohongshu(content, image_path)
            if result and result.get('success'):
                logger.info("发布成功")
                return True
            logger.warning("发布失败")
            return False
        except Exception as e:
            logger.error(f"发布异常: {str(e)}")
            return False

    def run(self, prompt: str, image_pool: List[str]) -> bool:
        """执行完整流程"""
        logger.info("开始执行自动发布流程")

        # 生成文案
        content = self.generate_content(prompt)
        if not content:
            return False

        # 匹配图片
        image_path = self.select_image(content, image_pool)
        if not image_path:
            return False

        # 发布内容
        return self.publish(content, image_path)

小红书 API 有调用频率限制(通常每分钟 60 次)
实现请求队列和速率控制

from ratelimit import limits, sleep_and_retry

# 限制每分钟 50 次调用
@sleep_and_retry
@limits(calls=50, period=60)
def call_api_safely():
    # API 调用代码
    pass