Ubuntu
176 lines
5.9 KiB
Python
176 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI内容生成模块
|
|
负责基于话题创作原创小红书内容
|
|
"""
|
|
|
|
import os
|
|
import requests
|
|
import re
|
|
from typing import Dict, Optional
|
|
|
|
class ContentGenerator:
|
|
"""AI内容生成器"""
|
|
|
|
def __init__(self):
|
|
self.api_key = os.environ.get('MOONSHOT_API_KEY')
|
|
self.api_base = os.environ.get('MOONSHOT_API_BASE', 'https://api.moonshot.cn/v1')
|
|
self.model = "kimi-k2.5"
|
|
|
|
def generate(self, topic: str, style: str = "tech") -> Dict[str, str]:
|
|
"""
|
|
基于话题生成原创内容
|
|
|
|
Args:
|
|
topic: 话题关键词
|
|
style: 内容风格 (tech/warm/business)
|
|
|
|
Returns:
|
|
包含title, body, tags的字典
|
|
"""
|
|
# 根据风格选择系统提示
|
|
style_prompts = {
|
|
"tech": "你是科技内容创作专家，擅长分析技术趋势。",
|
|
"warm": "你是生活方式博主，擅长温暖有趣的科技分享。",
|
|
"business": "你是商业分析师，擅长解读科技商业逻辑。"
|
|
}
|
|
|
|
system_prompt = style_prompts.get(style, style_prompts["tech"])
|
|
|
|
# 构建创作提示
|
|
user_prompt = f"""基于"{topic}"这个话题，创作一篇全新的小红书科技笔记。
|
|
|
|
创作要求：
|
|
1. 标题吸引人，20-30字，不用任何markdown符号
|
|
2. 正文600-800字，结构清晰
|
|
3. 使用-作为列表符号，不要用**加粗
|
|
4. 自然融入emoji
|
|
5. 包含独到的技术趋势洞察
|
|
6. 纯原创，不要复制任何现有文章
|
|
7. 结尾带5个话题标签，格式为#标签
|
|
|
|
输出格式：
|
|
标题：[你的标题]
|
|
|
|
正文：[你的正文]
|
|
|
|
标签：[标签1] [标签2] [标签3] [标签4] [标签5]"""
|
|
|
|
# 调用API
|
|
try:
|
|
response = requests.post(
|
|
f"{self.api_base}/chat/completions",
|
|
headers={
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json"
|
|
},
|
|
json={
|
|
"model": self.model,
|
|
"messages": [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt}
|
|
],
|
|
"temperature": 1,
|
|
"max_tokens": 1500
|
|
},
|
|
timeout=60
|
|
)
|
|
|
|
result = response.json()
|
|
|
|
# 检查API响应是否包含错误
|
|
if 'error' in result:
|
|
print(f"API错误: {result['error']}")
|
|
return self._fallback_content(topic)
|
|
|
|
# 检查choices是否存在
|
|
if 'choices' not in result or not result['choices']:
|
|
print(f"API响应异常: {result}")
|
|
return self._fallback_content(topic)
|
|
|
|
content = result['choices'][0]['message']['content']
|
|
|
|
# 解析内容
|
|
return self._parse_content(content)
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("API请求超时")
|
|
return self._fallback_content(topic)
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"API请求失败: {e}")
|
|
return self._fallback_content(topic)
|
|
except (KeyError, IndexError) as e:
|
|
print(f"API响应解析失败: {e}")
|
|
return self._fallback_content(topic)
|
|
except Exception as e:
|
|
print(f"内容生成失败: {e}")
|
|
return self._fallback_content(topic)
|
|
|
|
def _parse_content(self, content: str) -> Dict[str, str]:
|
|
"""解析AI生成的内容"""
|
|
# 提取标题
|
|
title_match = re.search(r'标题：[\s]*(.+?)(?:\n|$)', content)
|
|
title = title_match.group(1).strip() if title_match else "科技趋势分析"
|
|
|
|
# 提取正文（标题和标签之间的内容）
|
|
body_match = re.search(r'正文：[\s]*(.+?)(?:标签：|$)', content, re.DOTALL)
|
|
body = body_match.group(1).strip() if body_match else content
|
|
|
|
# 清理正文中的markdown
|
|
body = re.sub(r'\*\*(.+?)\*\*', r'\1', body) # 去除加粗
|
|
body = re.sub(r'##?\s*', '', body) # 去除标题标记
|
|
|
|
# 提取标签
|
|
tags_match = re.search(r'标签：[\s]*(.+?)(?:\n|$)', content)
|
|
tags_str = tags_match.group(1) if tags_match else "#科技趋势 #人工智能"
|
|
tags = [t.strip().replace('#', '') for t in tags_str.split() if t.strip()]
|
|
|
|
return {
|
|
"title": title,
|
|
"body": body,
|
|
"tags": tags[:5], # 最多5个标签
|
|
"raw": content
|
|
}
|
|
|
|
def _fallback_content(self, topic: str) -> Dict[str, str]:
|
|
"""备用内容（API失败时使用）"""
|
|
return {
|
|
"title": f"深度解析：{topic}的前沿趋势",
|
|
"body": f"今天来聊聊{topic}的最新发展...",
|
|
"tags": ["科技趋势", "人工智能", "AI产业"],
|
|
"raw": ""
|
|
}
|
|
|
|
def humanize(self, content: Dict[str, str]) -> Dict[str, str]:
|
|
"""
|
|
对内容进行去AI化处理
|
|
|
|
Args:
|
|
content: 原始内容
|
|
|
|
Returns:
|
|
去AI化后的内容
|
|
"""
|
|
try:
|
|
from shared_utils import get_anti_ai_humanizer
|
|
humanizer = get_anti_ai_humanizer()
|
|
|
|
if humanizer:
|
|
content['body'] = humanizer(content['body'], mode='xiaohongshu')
|
|
|
|
except Exception as e:
|
|
print(f"去AI化处理失败: {e}")
|
|
|
|
return content
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试
|
|
generator = ContentGenerator()
|
|
result = generator.generate("AI芯片竞争", style="tech")
|
|
|
|
print("生成结果:")
|
|
print(f"标题: {result['title']}")
|
|
print(f"\n正文预览: {result['body'][:150]}...")
|
|
print(f"\n标签: {result['tags']}")
|