Files
Ubuntu 32b43f85bd a
2026-07-03 16:58:28 +08:00

507 lines
17 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
内容格式转换器
将技术文章转换为小红书格式
"""
import re
import json
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
@dataclass
class Article:
"""文章数据结构"""
title: str
content: str
source: str
url: str
images: List[str]
publish_date: Optional[str] = None
author: Optional[str] = None
category: Optional[str] = None
class ContentConverter:
"""内容格式转换器"""
# 表情符号映射
EMOJI_MAPPING = {
'ai': '🤖',
'人工智能': '🧠',
'技术': '💻',
'编程': '⌨️',
'开发': '🔧',
'数据': '📊',
'安全': '🔒',
'网络': '🌐',
'开源': '📦',
'更新': '🔄',
'发布': '🚀',
'教程': '📚',
'新闻': '📰',
'突破': '⚡',
'趋势': '📈',
'分析': '🔍',
'工具': '🛠️',
'框架': '🏗️',
'算法': '🧮',
'测试': '🧪',
'部署': '🚢',
'云': '☁️',
'移动': '📱',
'web': '🕸️',
'设计': '🎨',
'性能': '⚡',
'优化': '✨',
'bug': '🐛',
'修复': '🔧',
'文档': '📄',
'社区': '👥',
'会议': '🎤',
'学习': '🎓',
'项目': '📁',
'代码': '💾',
'API': '🔌',
'数据库': '🗄️',
'服务器': '🖥️',
'前端': '🎯',
'后端': '⚙️',
'全栈': '🔗',
'DevOps': '🔀',
'容器': '📦',
'微服务': '🧩',
'区块链': '⛓️',
'物联网': '📡',
'5G': '📶',
'AR/VR': '👓',
'量子': '⚛️',
'芯片': '🔋',
'硬件': '💾',
'软件': '💿',
'企业': '🏢',
'创业': '💡',
'投资': '💰',
'市场': '📈',
'用户': '👤',
'体验': '❤️',
'隐私': '🕵️',
'合规': '⚖️',
'标准': '📏',
'未来': '🔮',
'创新': '💫',
'变革': '🌪️',
'挑战': '🏔️',
'机遇': '🎯',
'总结': '📝',
'建议': '💡',
'资源': '📚',
'链接': '🔗',
'关注': '👀',
'分享': '📤',
'评论': '💬',
'点赞': '👍',
'收藏': '⭐'
}
# 技术关键词标签映射
TAG_MAPPING = {
'ai': ['#AI', '#人工智能', '#机器学习'],
'python': ['#Python', '#编程'],
'javascript': ['#JavaScript', '#前端'],
'java': ['#Java', '#后端'],
'go': ['#Go语言', '#后端'],
'rust': ['#Rust', '#系统编程'],
'react': ['#React', '#前端框架'],
'vue': ['#Vue', '#前端框架'],
'node': ['#Nodejs', '#后端'],
'docker': ['#Docker', '#容器化'],
'kubernetes': ['#Kubernetes', '#容器编排'],
'aws': ['#AWS', '#云计算'],
'azure': ['#Azure', '#云计算'],
'gcp': ['#GCP', '#云计算'],
'linux': ['#Linux', '#操作系统'],
'windows': ['#Windows', '#操作系统'],
'macos': ['#macOS', '#操作系统'],
'ios': ['#iOS', '#移动开发'],
'android': ['#Android', '#移动开发'],
'flutter': ['#Flutter', '#跨平台'],
'react native': ['#ReactNative', '#跨平台'],
'数据库': ['#数据库', '#SQL'],
'mysql': ['#MySQL', '#数据库'],
'postgresql': ['#PostgreSQL', '#数据库'],
'mongodb': ['#MongoDB', '#NoSQL'],
'redis': ['#Redis', '#缓存'],
'elasticsearch': ['#Elasticsearch', '#搜索'],
'kafka': ['#Kafka', '#消息队列'],
'rabbitmq': ['#RabbitMQ', '#消息队列'],
'nginx': ['#Nginx', '#Web服务器'],
'apache': ['#Apache', '#Web服务器'],
'git': ['#Git', '#版本控制'],
'github': ['#GitHub', '#开源'],
'gitlab': ['#GitLab', '#DevOps'],
'jenkins': ['#Jenkins', '#CI/CD'],
'terraform': ['#Terraform', '#基础设施即代码'],
'ansible': ['#Ansible', '#自动化'],
'prometheus': ['#Prometheus', '#监控'],
'grafana': ['#Grafana', '#可视化'],
'区块链': ['#区块链', '#Web3'],
'比特币': ['#比特币', '#加密货币'],
'以太坊': ['#以太坊', '#智能合约'],
'nft': ['#NFT', '#数字资产'],
'元宇宙': ['#元宇宙', '#虚拟现实'],
'ar': ['#AR', '#增强现实'],
'vr': ['#VR', '#虚拟现实'],
'iot': ['#物联网', '#智能设备'],
'5g': ['#5G', '#通信技术'],
'6g': ['#6G', '#未来通信'],
'量子计算': ['#量子计算', '#未来科技'],
'芯片': ['#芯片', '#半导体'],
'自动驾驶': ['#自动驾驶', '#智能交通'],
'机器人': ['#机器人', '#自动化'],
'无人机': ['#无人机', '#航空科技'],
'生物识别': ['#生物识别', '#安全技术'],
'边缘计算': ['#边缘计算', '#分布式计算'],
'serverless': ['#Serverless', '#无服务器'],
'微服务': ['#微服务', '#架构设计'],
'api': ['#API', '#接口设计'],
'rest': ['#REST', '#API设计'],
'graphql': ['#GraphQL', '#API设计'],
'websocket': ['#WebSocket', '#实时通信'],
'http': ['#HTTP', '#网络协议'],
'https': ['#HTTPS', '#安全协议'],
'ssl': ['#SSL', '#加密'],
'tls': ['#TLS', '#加密'],
'oauth': ['#OAuth', '#认证授权'],
'jwt': ['#JWT', '#令牌'],
'oauth2': ['#OAuth2', '#认证协议'],
'openid': ['#OpenID', '#身份认证'],
'单点登录': ['#单点登录', '#SSO'],
'多因素认证': ['#多因素认证', '#MFA'],
'零信任': ['#零信任', '#安全架构'],
'隐私计算': ['#隐私计算', '#数据安全'],
'同态加密': ['#同态加密', '#加密技术'],
'联邦学习': ['#联邦学习', '#分布式AI'],
'迁移学习': ['#迁移学习', '#机器学习'],
'强化学习': ['#强化学习', '#机器学习'],
'深度学习': ['#深度学习', '#神经网络'],
'自然语言处理': ['#自然语言处理', '#NLP'],
'计算机视觉': ['#计算机视觉', '#CV'],
'语音识别': ['#语音识别', '#ASR'],
'语音合成': ['#语音合成', '#TTS'],
'推荐系统': ['#推荐系统', '#个性化'],
'搜索引擎': ['#搜索引擎', '#信息检索'],
'知识图谱': ['#知识图谱', '#语义网络'],
'大数据': ['#大数据', '#数据处理'],
'数据挖掘': ['#数据挖掘', '#数据分析'],
'数据可视化': ['#数据可视化', '#图表'],
'数据仓库': ['#数据仓库', '#BI'],
'数据湖': ['#数据湖', '#数据存储'],
'数据治理': ['#数据治理', '#数据管理'],
'数据安全': ['#数据安全', '#隐私保护'],
'数据合规': ['#数据合规', '#GDPR'],
'数据伦理': ['#数据伦理', '#AI伦理']
}
def __init__(self, config: Optional[Dict] = None):
"""
初始化转换器
Args:
config: 配置字典
"""
self.config = config or {}
self.max_title_length = self.config.get('max_title_length', 30)
self.max_content_length = self.config.get('max_content_length', 1000)
self.auto_generate_tags = self.config.get('auto_generate_tags', True)
self.add_source_info = self.config.get('add_source_info', True)
self.add_emoji = self.config.get('add_emoji', True)
logger.info("内容转换器初始化完成")
def to_xiaohongshu(self, article: Article) -> str:
"""
将文章转换为小红书格式(使用oh-my-writing-skill优化)
Args:
article: 文章对象
Returns:
小红书格式的内容
"""
logger.info(f"开始转换文章: {article.title}")
try:
# 尝试使用oh-my-writing-skill转换器
from oh_my_writing_converter import OhMyWritingConverter
converter = OhMyWritingConverter()
# 构建完整内容
full_content = f"# {article.title}\n\n"
if article.source:
full_content += f"来源: {article.source}\n\n"
full_content += article.content
# 转换为小红书格式
result = converter.convert_to_xiaohongshu(full_content, article.title)
# 添加来源信息(如果转换器没有包含)
xhs_content = result.content
if article.source and f"来源: {article.source}" not in xhs_content:
xhs_content = xhs_content.replace("往下看 ⬇️", f"往下看 ⬇️\n\n📰 来源: {article.source}", 1)
# 添加原文链接
if article.url:
xhs_content += f"\n\n🔗 原文链接: {article.url}"
logger.info(f"oh-my-writing-skill转换完成,标题: {result.title}")
return xhs_content
except Exception as e:
logger.warning(f"oh-my-writing-skill转换失败,使用基础转换: {e}")
# 回退到基础转换
return self._to_xiaohongshu_basic(article)
def _to_xiaohongshu_basic(self, article: Article) -> str:
"""基础小红书格式转换(回退方案)"""
# 处理标题
title = self._process_title(article.title)
# 构建小红书内容
lines = []
# 添加标题(带emoji)
if self.add_emoji:
title_with_emoji = self._add_title_emoji(title)
lines.append(f"🔥 {title_with_emoji}")
else:
lines.append(f"🔥 {title}")
lines.append("") # 空行
# 添加来源信息
if self.add_source_info and article.source:
lines.append(f"📰 来源: {article.source}")
if article.publish_date:
lines.append(f"📅 时间: {article.publish_date}")
if article.author:
lines.append(f"✍️ 作者: {article.author}")
lines.append("") # 空行
# 处理正文内容
content_lines = self._process_content(article.content)
lines.extend(content_lines)
# 添加原文链接
if article.url:
lines.append("")
lines.append(f"🔗 原文链接: {article.url}")
# 添加标签
if self.auto_generate_tags:
tags = self._generate_tags(article)
if tags:
lines.append("")
lines.append(tags)
# 合并所有行
xhs_content = "\n".join(lines)
# 限制内容长度
if len(xhs_content) > self.max_content_length:
xhs_content = xhs_content[:self.max_content_length - 3] + "..."
logger.warning(f"内容过长,已截断至 {self.max_content_length} 字符")
logger.info(f"基础转换完成,内容长度: {len(xhs_content)} 字符")
return xhs_content
def _process_title(self, title: str) -> str:
"""处理标题"""
# 移除多余空格
title = re.sub(r'\s+', ' ', title.strip())
# 限制长度
if len(title) > self.max_title_length:
title = title[:self.max_title_length - 3] + "..."
return title
def _add_title_emoji(self, title: str) -> str:
"""为标题添加emoji"""
title_lower = title.lower()
# 查找匹配的emoji
for keyword, emoji in self.EMOJI_MAPPING.items():
if keyword in title_lower:
# 在标题开头添加emoji
return f"{emoji} {title}"
# 如果没有匹配,使用默认emoji
return f"📝 {title}"
def _process_content(self, content: str) -> List[str]:
"""处理正文内容"""
lines = []
# 分割段落
paragraphs = content.split('\n')
for para in paragraphs:
para = para.strip()
if not para:
continue
# 处理列表项
if re.match(r'^[0-9]+[\.\)]', para) or para.startswith('- ') or para.startswith('• '):
lines.append(f"👉 {para}")
# 处理标题
elif para.endswith(':') or len(para) < 50 and not para.endswith('。'):
lines.append(f"📌 {para}")
# 处理长段落
elif len(para) > 80:
# 拆分长段落
sub_lines = self._split_long_paragraph(para)
for sub_line in sub_lines:
lines.append(sub_line)
else:
lines.append(para)
# 段落间添加空行(除了列表项)
if not (para.startswith('- ') or para.startswith('• ') or re.match(r'^[0-9]+[\.\)]', para)):
lines.append("")
return lines
def _split_long_paragraph(self, paragraph: str, max_length: int = 80) -> List[str]:
"""拆分长段落"""
words = paragraph.split()
lines = []
current_line = []
current_length = 0
for word in words:
word_length = len(word)
# 如果当前行加上这个词会超过限制,且当前行不为空
if current_length + word_length + (1 if current_line else 0) > max_length and current_line:
lines.append(' '.join(current_line))
current_line = [word]
current_length = word_length
else:
current_line.append(word)
current_length += word_length + (1 if current_line else 0)
if current_line:
lines.append(' '.join(current_line))
return lines
def _generate_tags(self, article: Article) -> str:
"""生成话题标签"""
# 基础标签
tags = ['#技术资讯', '#科技新闻']
# 添加分类标签
if article.category:
tags.append(f'#{article.category}')
# 分析内容生成标签
content_lower = (article.title + ' ' + article.content).lower()
# 查找匹配的技术关键词
matched_tags = set()
for keyword, tag_list in self.TAG_MAPPING.items():
if keyword in content_lower:
matched_tags.update(tag_list)
# 添加匹配的标签
tags.extend(list(matched_tags))
# 限制标签数量
max_tags = self.config.get('max_tags', 10)
tags = tags[:max_tags]
return ' '.join(tags)
def analyze_article(self, article: Article) -> Dict:
"""分析文章特征"""
content_lower = (article.title + ' ' + article.content).lower()
# 检测技术关键词
detected_keywords = []
for keyword in self.TAG_MAPPING.keys():
if keyword in content_lower:
detected_keywords.append(keyword)
# 计算内容特征
features = {
'title_length': len(article.title),
'content_length': len(article.content),
'image_count': len(article.images),
'detected_keywords': detected_keywords,
'recommended_tags': self._generate_tags(article).split(),
'suitable_for_xhs': len(article.content) > 100 and len(article.content) < 2000
}
return features
# 使用示例
if __name__ == "__main__":
# 测试文章
test_article = Article(
title="OpenAI发布GPT-5:AI技术的新里程碑",
content="""OpenAI近日宣布正在开发GPT-5,这是其语言模型系列的最新版本。
主要特点:
1. 更强的逻辑推理能力
2. 多模态理解(文本、图像、音频)
3. 更低的计算成本
4. 更好的中文支持
技术突破:
- 参数规模达到10万亿
- 训练数据涵盖100种语言
- 推理速度提升50%
应用场景:
• 智能助手
• 内容创作
• 代码生成
• 教育辅导
预计GPT-5将在2025年发布,这将推动AI技术进入新的发展阶段。""",
source="TechCrunch",
url="https://techcrunch.com/gpt5",
images=["ai_image1.jpg", "ai_image2.jpg"],
publish_date="2026-03-15",
author="John Doe",
category="AI"
)
# 创建转换器
converter = ContentConverter()
# 转换文章
xhs_content = converter.to_xiaohongshu(test_article)
print("=" * 60)
print("小红书格式内容:")
print("=" * 60)
print(xhs_content)
print("=" * 60)
# 分析文章
analysis = converter.analyze_article(test_article)
print("\n文章分析:")
print(json.dumps(analysis, indent=2, ensure_ascii=False))