#!/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))