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

874 lines
31 KiB
Python
Executable File
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
"""
oh-my-writing-skill 集成管理器
将下载的技能集成到IT资讯自动化系统中
"""
import os
import sys
import json
import shutil
from pathlib import Path
class OhMyWritingIntegration:
"""oh-my-writing-skill 集成管理器"""
def __init__(self):
self.source_dir = Path("/home/ubuntu/oh-my-writing-skill")
self.target_dir = Path("/home/ubuntu/.nanobot/workspace/skills/xiaohongshu_integration")
self.skills_config = {}
# 技能映射表
self.skill_mapping = {
'xiaohongshu-converter': {
'name': '小红书转换器',
'description': '将通用文章转换为小红书格式',
'source': 'skills/xiaohongshu-converter/SKILL.md',
'target': 'oh_my_writing_converter.py',
'already_implemented': True
},
'humanizer-cn': {
'name': '中文去AI化',
'description': '消除AI写作痕迹,让文章更自然',
'source': 'skills/humanizer-cn/SKILL.md',
'target': 'humanizer_cn.py',
'already_implemented': False
},
'general-writing': {
'name': '通用写作',
'description': '整合素材创作图文并茂的文章',
'source': 'skills/general-writing/SKILL.md',
'target': 'general_writing.py',
'already_implemented': False
},
'deep-research': {
'name': '深度研究',
'description': '联网搜索收集资料',
'source': 'skills/deep-research/SKILL.md',
'target': 'deep_research.py',
'already_implemented': False
},
'image-search': {
'name': '图片搜索',
'description': '搜索相关配图',
'source': 'skills/image-search/SKILL.md',
'target': 'image_search.py',
'already_implemented': False
},
'image-processing': {
'name': '图片处理',
'description': '处理图片添加配文',
'source': 'skills/image-processing/SKILL.md',
'target': 'image_processing.py',
'already_implemented': False
},
'zhihu-converter': {
'name': '知乎转换器',
'description': '转换为知乎格式',
'source': 'skills/zhihu-converter/SKILL.md',
'target': 'zhihu_converter.py',
'already_implemented': False
},
'wechat-converter': {
'name': '公众号转换器',
'description': '转换为微信公众号格式',
'source': 'skills/wechat-converter/SKILL.md',
'target': 'wechat_converter.py',
'already_implemented': False
}
}
def analyze_skills(self):
"""分析可用技能"""
print("🔍 分析 oh-my-writing-skill 技能包...")
print(f"源目录: {self.source_dir}")
print(f"目标目录: {self.target_dir}")
print()
available_skills = []
missing_skills = []
for skill_id, skill_info in self.skill_mapping.items():
source_path = self.source_dir / skill_info['source']
if source_path.exists():
available_skills.append({
'id': skill_id,
'name': skill_info['name'],
'description': skill_info['description'],
'source': str(source_path),
'status': '已下载'
})
else:
missing_skills.append({
'id': skill_id,
'name': skill_info['name'],
'description': skill_info['description'],
'status': '未找到'
})
print(f"📊 技能分析结果:")
print(f" 已下载: {len(available_skills)} 个技能")
print(f" 未找到: {len(missing_skills)} 个技能")
print()
if available_skills:
print("✅ 可用技能:")
for skill in available_skills:
print(f" • {skill['name']} ({skill['id']})")
print(f" 描述: {skill['description']}")
if missing_skills:
print("⚠️ 缺失技能:")
for skill in missing_skills:
print(f" • {skill['name']} ({skill['id']})")
return available_skills, missing_skills
def extract_skill_knowledge(self, skill_id):
"""提取技能知识"""
if skill_id not in self.skill_mapping:
print(f"❌ 未知技能: {skill_id}")
return None
skill_info = self.skill_mapping[skill_id]
source_path = self.source_dir / skill_info['source']
if not source_path.exists():
print(f"❌ 技能文件不存在: {source_path}")
return None
print(f"📖 提取技能知识: {skill_info['name']}")
print(f" 文件: {source_path}")
try:
with open(source_path, 'r', encoding='utf-8') as f:
content = f.read()
# 解析技能内容
knowledge = {
'id': skill_id,
'name': skill_info['name'],
'description': skill_info['description'],
'content': content,
'summary': self._summarize_skill_content(content),
'key_points': self._extract_key_points(content),
'examples': self._extract_examples(content)
}
print(f"✅ 知识提取完成")
print(f" 内容长度: {len(content)} 字符")
print(f" 关键点: {len(knowledge['key_points'])} 个")
return knowledge
except Exception as e:
print(f"❌ 提取失败: {e}")
return None
def _summarize_skill_content(self, content):
"""总结技能内容"""
lines = content.split('\n')
summary = []
# 提取标题和主要章节
for line in lines:
if line.startswith('# '):
summary.append(line)
elif line.startswith('## '):
summary.append(line)
elif line.startswith('### '):
summary.append(line)
return '\n'.join(summary[:10]) # 限制长度
def _extract_key_points(self, content):
"""提取关键点"""
key_points = []
# 查找列表项
lines = content.split('\n')
for line in lines:
if line.strip().startswith('- ') or line.strip().startswith('* '):
key_points.append(line.strip())
elif line.strip().startswith('1. ') or line.strip().startswith('• '):
key_points.append(line.strip())
return key_points[:20] # 限制数量
def _extract_examples(self, content):
"""提取示例"""
examples = []
# 查找代码块
in_code_block = False
code_block = []
for line in content.split('\n'):
if line.strip().startswith('```'):
if in_code_block:
examples.append('\n'.join(code_block))
code_block = []
in_code_block = False
else:
in_code_block = True
elif in_code_block:
code_block.append(line)
return examples
def implement_humanizer_cn(self):
"""实现中文去AI化技能"""
print("🛠️ 实现中文去AI化技能...")
# 提取知识
knowledge = self.extract_skill_knowledge('humanizer-cn')
if not knowledge:
return False
# 创建实现文件
target_file = self.target_dir / "humanizer_cn.py"
implementation = f'''#!/usr/bin/env python3
"""
中文去AI化工具
基于 oh-my-writing-skill 的 humanizer-cn 技能
消除AI写作痕迹,让文章更自然
"""
import re
import json
from typing import Dict, List, Optional
class HumanizerCN:
"""中文去AI化处理器"""
def __init__(self):
# AI写作特征模式
self.ai_patterns = {{
# 过度强调意义和历史地位
'overemphasis': [
r'彰显[了着]', r'见证[了着]', r'标志着', r'具有里程碑意义',
r'揭示了', r'体现了', r'承载着', r'奠定了基础',
r'树立了标杆', r'开创了先河', r'具有划时代意义', r'承前启后'
],
# 过度强调知名度和媒体报道
'overhype': [
r'广受关注', r'引发热议', r'备受瞩目', r'获得广泛报道',
r'在业界引起强烈反响', r'各大媒体纷纷报道'
],
# "-ing 式"假深度分析
'fake_depth': [
r'致力于', r'旨在', r'聚焦于', r'着眼于',
r'立足于', r'依托于', r'围绕着', r'基于', r'借助于'
],
# 广告式推销语言
'advertising': [
r'匠心独运', r'精心打造', r'倾力呈现', r'独具匠心',
r'别具一格', r'独树一帜', r'引领潮流', r'开启新篇章',
r'重新定义'
],
# 模糊归因和万金油说法
'vague_attribution': [
r'业内人士指出', r'专家表示', r'有观点认为',
r'据了解', r'据悉', r'相关人士透露', r'多方消息显示'
],
# 套路化的"挑战与展望"段落
'cliche_ending': [
r'尽管如此', r'虽然面临挑战', r'但我们有理由相信',
r'未来可期', r'前景广阔', r'任重道远', r'砥砺前行'
],
# 高频AI词汇
'ai_vocabulary': [
r'此外', r'值得一提的是', r'值得注意的是',
r'不言而喻', r'毋庸置疑', r'显而易见',
r'与此同时', r'综上所述', r'总而言之',
r'一言以蔽之', r'鉴于此', r'基于此',
r'有鉴于此', r'就此而言', r'从某种意义上说'
],
# 回避简单动词
'avoid_simple_verbs': [
r'作为[^,。]*而存在', r'扮演着[^,。]*的角色',
r'承担着[^,。]*的使命', r'发挥着[^,。]*的作用',
r'被视为', r'被认为是'
],
# 否定式并列句
'negative_parallel': [
r'不仅[^,。]*更是[^,。]*',
r'不只是[^,。]*而是[^,。]*'
],
# 三段式堆砌
'triple_structure': [
r'创新、品质与服务的完美结合',
r'高效、稳定、易用三大特点'
]
}}
# 替换映射
self.replacement_map = {{
# 语气转换
'本文将介绍': '今天来聊聊',
'建议用户': '建议大家',
'该产品性能优秀': '这个真的很好用',
'综上所述': '总之就是',
'值得注意的是': '划重点',
'需要指出的是': '这里要注意',
'根据研究显示': '我发现',
'一般来说': '通常来说',
'具体来说': '具体来说',
'换句话说': '简单说就是',
'例如': '比如',
'此外': '另外',
'然而': '不过',
'因此': '所以',
'最终': '最后',
# 简化表达
'为了能够实现这一目标': '为了',
'由于这一事实的存在': '因为',
'在当前这个时间点上': '现在',
'在如果需要帮助的情况下': '如果需要帮助',
'具有能够处理的能力': '能处理',
'需要特别指出的是': '',
# 具体化
'该项目一经发布便引发业界广泛关注': '发布后反响不错',
'业内专家普遍认为': '有专家认为',
'具有广阔的应用前景': '应用前景不错',
'对行业产生深远影响': '对行业影响挺大'
}}
def detect_ai_patterns(self, text: str) -> Dict:
"""检测AI写作模式"""
detection = {{
'total_patterns': 0,
'patterns_by_type': {{}},
'suggestions': []
}}
for pattern_type, patterns in self.ai_patterns.items():
count = 0
found_patterns = []
for pattern in patterns:
matches = re.findall(pattern, text)
if matches:
count += len(matches)
found_patterns.extend(matches)
if count > 0:
detection['patterns_by_type'][pattern_type] = {{
'count': count,
'examples': list(set(found_patterns))[:3] # 最多显示3个例子
}}
detection['total_patterns'] += count
# 生成建议
if detection['total_patterns'] > 10:
detection['suggestions'].append('AI痕迹较重,建议全面重写')
elif detection['total_patterns'] > 5:
detection['suggestions'].append('有较多AI痕迹,建议重点修改')
elif detection['total_patterns'] > 0:
detection['suggestions'].append('有少量AI痕迹,建议微调')
else:
detection['suggestions'].append('AI痕迹较少,质量不错')
return detection
def humanize_text(self, text: str, aggressive: bool = False) -> str:
"""去AI化处理"""
if not text:
return text
# 第一步:替换模式化表达
humanized = text
for pattern, replacement in self.replacement_map.items():
humanized = humanized.replace(pattern, replacement)
# 第二步:处理段落结构
paragraphs = humanized.split('\\n\\n')
processed_paragraphs = []
for para in paragraphs:
if not para.strip():
processed_paragraphs.append(para)
continue
# 缩短长段落
if len(para) > 200:
sentences = re.split(r'[。!?]', para)
if len(sentences) > 3:
para = '。'.join(sentences[:3]) + '。'
# 添加一点"人味"
if aggressive and len(para) > 50:
# 随机添加一些口语化表达
import random
human_touches = [
'说实话',
'我个人觉得',
'其实吧',
'怎么说呢',
'你猜怎么着'
]
if random.random() < 0.3: # 30%概率添加
touch = random.choice(human_touches)
para = f"{{touch}},{{para}}"
processed_paragraphs.append(para)
humanized = '\\n\\n'.join(processed_paragraphs)
# 第三步:修复过度修饰
# 移除过多的修饰词
modifiers = ['非常', '极其', '十分', '特别', '相当', '颇为']
for modifier in modifiers:
pattern = rf'{{modifier}}[的得地]?[^,。]*?的'
humanized = re.sub(pattern, '', humanized)
# 第四步:简化列表结构
lines = humanized.split('\\n')
simplified_lines = []
for line in lines:
# 简化加粗列表项
if re.search(r'\\*\\*.*?\\*\\*:', line):
line = re.sub(r'\\*\\*(.*?)\\*\\*:', r'\\1:', line)
simplified_lines.append(line)
humanized = '\\n'.join(simplified_lines)
return humanized
def add_human_touch(self, text: str) -> str:
"""添加人味"""
if not text:
return text
# 添加一些个人化的表达
touches = [
'我试过',
'我的经验是',
'说实话',
'个人觉得',
'其实吧',
'你猜怎么着',
'没想到的是'
]
import random
# 在开头添加
if random.random() < 0.4:
touch = random.choice(touches)
text = f"{{touch}},{{text}}"
# 在中间随机添加
sentences = re.split(r'[。!?]', text)
if len(sentences) > 2:
insert_pos = random.randint(1, len(sentences)-1)
touch = random.choice(touches)
sentences.insert(insert_pos, f"{{touch}},")
text = '。'.join(sentences)
return text
def analyze_and_humanize(self, text: str) -> Dict:
"""分析并去AI化"""
print("🔍 分析文本中的AI痕迹...")
# 检测AI模式
detection = self.detect_ai_patterns(text)
print(f"📊 检测结果:")
print(f" 总AI模式数: {{detection['total_patterns']}}")
for pattern_type, info in detection['patterns_by_type'].items():
print(f" {{pattern_type}}: {{info['count']}} 处")
if info['examples']:
print(f" 示例: {{', '.join(info['examples'][:2])}}")
# 去AI化处理
print("\\n🛠️ 进行去AI化处理...")
humanized = self.humanize_text(text, aggressive=detection['total_patterns'] > 5)
# 添加人味(如果需要)
if detection['total_patterns'] > 3:
print("💬 添加人味...")
humanized = self.add_human_touch(humanized)
# 计算改进程度
original_length = len(text)
humanized_length = len(humanized)
length_change = ((humanized_length - original_length) / original_length) * 100
result = {{
'original_text': text,
'humanized_text': humanized,
'detection': detection,
'stats': {{
'original_length': original_length,
'humanized_length': humanized_length,
'length_change_percent': round(length_change, 2),
'ai_patterns_removed': detection['total_patterns']
}},
'suggestions': detection['suggestions']
}}
print(f"✅ 处理完成")
print(f" 原文长度: {{original_length}} 字符")
print(f" 处理后长度: {{humanized_length}} 字符")
print(f" 长度变化: {{length_change:.1f}}%")
print(f" 建议: {{', '.join(detection['suggestions'])}}")
return result
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='中文去AI化工具')
subparsers = parser.add_subparsers(dest='command', help='命令')
# analyze命令
analyze_parser = subparsers.add_parser('analyze', help='分析文本')
analyze_parser.add_argument('--text', help='要分析的文本')
analyze_parser.add_argument('--file', help='文本文件路径')
# humanize命令
humanize_parser = subparsers.add_parser('humanize', help='去AI化处理')
humanize_parser.add_argument('--text', help='要处理的文本')
humanize_parser.add_argument('--file', help='文本文件路径')
humanize_parser.add_argument('--output', help='输出文件路径')
humanize_parser.add_argument('--aggressive', action='store_true', help='激进模式')
# test命令
subparsers.add_parser('test', help='测试功能')
args = parser.parse_args()
humanizer = HumanizerCN()
if args.command == 'analyze':
text = args.text
if args.file:
with open(args.file, 'r', encoding='utf-8') as f:
text = f.read()
if not text:
print("❌ 请提供文本")
return
result = humanizer.analyze_and_humanize(text)
print("\\n📊 分析结果:")
print(json.dumps(result['detection'], indent=2, ensure_ascii=False))
elif args.command == 'humanize':
text = args.text
if args.file:
with open(args.file, 'r', encoding='utf-8') as f:
text = f.read()
if not text:
print("❌ 请提供文本")
return
result = humanizer.analyze_and_humanize(text)
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(result['humanized_text'])
print(f"✅ 已保存到: {{args.output}}")
else:
print("\\n📝 处理后的文本:")
print("=" * 60)
print(result['humanized_text'])
print("=" * 60)
elif args.command == 'test':
test_text = """该软件的发布标志着人工智能领域的一个重要里程碑,彰显了公司在技术创新方面的不懈追求。此外,值得一提的是,该产品具有高性能、低功耗、易扩展三大优势。业内专家普遍认为,这将对整个行业产生深远影响。尽管面临诸多挑战,但我们有理由相信,未来可期。"""
print("🧪 测试去AI化功能")
print("=" * 60)
print("测试文本:")
print(test_text)
print("=" * 60)
result = humanizer.analyze_and_humanize(test_text)
print("\\n📝 处理后的文本:")
print("=" * 60)
print(result['humanized_text'])
print("=" * 60)
else:
parser.print_help()
if __name__ == "__main__":
main()
'''
try:
with open(target_file, 'w', encoding='utf-8') as f:
f.write(implementation)
# 设置执行权限
os.chmod(target_file, 0o755)
print(f"✅ 中文去AI化技能已实现: {target_file}")
return True
except Exception as e:
print(f"❌ 实现失败: {e}")
return False
def create_integration_report(self):
"""创建集成报告"""
print("📋 创建技能集成报告...")
available_skills, missing_skills = self.analyze_skills()
report = f"""# oh-my-writing-skill 集成报告
## 概览
- 源目录: {self.source_dir}
- 目标目录: {self.target_dir}
- 分析时间: {self._get_current_time()}
## 技能分析
### 已下载技能 ({len(available_skills)}个)
"""
for skill in available_skills:
report += f"#### {skill['name']} ({skill['id']})\n"
report += f"- 描述: {skill['description']}\n"
report += f"- 状态: {skill['status']}\n"
report += f"- 文件: {skill['source']}\n\n"
if missing_skills:
report += f"### 缺失技能 ({len(missing_skills)}个)\n\n"
for skill in missing_skills:
report += f"- {skill['name']} ({skill['id']}): {skill['description']}\n"
report += "\n"
# 集成状态
report += "## 集成状态\n\n"
integrated_count = 0
for skill_id, skill_info in self.skill_mapping.items():
status = "✅ 已集成" if skill_info['already_implemented'] else "⏳ 待集成"
if skill_info['already_implemented']:
integrated_count += 1
report += f"- {skill_info['name']}: {status}\n"
report += f"\n集成进度: {integrated_count}/{len(self.skill_mapping)} ({integrated_count/len(self.skill_mapping)*100:.1f}%)\n\n"
# 下一步建议
report += "## 下一步建议\n\n"
if integrated_count < len(self.skill_mapping):
report += "1. **优先集成中文去AI化技能** - 提升内容自然度\n"
report += "2. **集成深度研究技能** - 增强内容质量\n"
report += "3. **集成图片搜索技能** - 丰富内容配图\n"
report += "4. **集成通用写作技能** - 完善创作流程\n"
else:
report += "✅ 所有技能已集成完成\n"
# 使用方式
report += "## 使用方式\n\n"
report += "```bash\n"
report += "# 1. 小红书内容转换\n"
report += "python3 oh_my_writing_converter.py convert --input article.md --output xiaohongshu.md\n\n"
report += "# 2. 中文去AI化\n"
report += "python3 humanizer_cn.py humanize --file article.md --output humanized.md\n\n"
report += "# 3. 集成到IT资讯系统\n"
report += "# 系统已自动集成,每天16:00自动运行\n"
report += "```\n\n"
# 保存报告
report_file = self.target_dir / "integration_report.md"
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
print(f"✅ 集成报告已保存: {report_file}")
return report
def _get_current_time(self):
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def integrate_with_it_system(self):
"""集成到IT资讯系统"""
print("🔗 集成到IT资讯自动化系统...")
# 1. 更新内容转换器
print("1. 更新内容转换器...")
# 我们已经更新了content_converter.py,现在确认一下
converter_file = self.target_dir / "content_converter.py"
if converter_file.exists():
print(f"✅ 内容转换器已更新: {converter_file}")
else:
print(f"❌ 内容转换器不存在: {converter_file}")
# 2. 创建集成配置文件
print("2. 创建集成配置文件...")
config = {
'oh_my_writing_integration': {
'enabled': True,
'integration_date': self._get_current_time(),
'integrated_skills': [],
'config': {
'use_humanizer': True,
'use_xiaohongshu_converter': True,
'aggressive_mode': False,
'auto_select_best': True
}
}
}
# 添加已集成的技能
for skill_id, skill_info in self.skill_mapping.items():
if skill_info['already_implemented']:
config['oh_my_writing_integration']['integrated_skills'].append({
'id': skill_id,
'name': skill_info['name'],
'status': 'implemented'
})
config_file = self.target_dir / "oh_my_writing_config.json"
with open(config_file, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
print(f"✅ 集成配置文件已创建: {config_file}")
# 3. 更新IT资讯系统脚本
print("3. 更新IT资讯系统脚本...")
# 检查IT资讯系统目录
it_system_dir = Path("/home/ubuntu/.nanobot/workspace/InformationCrawler")
if it_system_dir.exists():
print(f"✅ IT资讯系统目录存在: {it_system_dir}")
# 检查主调度脚本
main_script = it_system_dir / "main_scheduler_with_xhs.sh"
if main_script.exists():
print(f"✅ 主调度脚本存在: {main_script}")
# 检查是否已经集成了oh-my-writing-skill
with open(main_script, 'r', encoding='utf-8') as f:
content = f.read()
if 'oh_my_writing' in content or 'oh-my-writing' in content:
print("✅ oh-my-writing-skill 已集成到IT资讯系统")
else:
print("⚠️ oh-my-writing-skill 未集成到IT资讯系统,需要手动集成")
else:
print(f"❌ 主调度脚本不存在: {main_script}")
else:
print(f"❌ IT资讯系统目录不存在: {it_system_dir}")
print("✅ 集成完成")
return True
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='oh-my-writing-skill 集成管理器')
subparsers = parser.add_subparsers(dest='command', help='命令')
# analyze命令
subparsers.add_parser('analyze', help='分析技能')
# extract命令
extract_parser = subparsers.add_parser('extract', help='提取技能知识')
extract_parser.add_argument('skill', help='技能ID')
# implement命令
implement_parser = subparsers.add_parser('implement', help='实现技能')
implement_parser.add_argument('skill', help='技能ID')
# report命令
subparsers.add_parser('report', help='创建集成报告')
# integrate命令
subparsers.add_parser('integrate', help='集成到IT系统')
# all命令
subparsers.add_parser('all', help='执行完整集成流程')
args = parser.parse_args()
integrator = OhMyWritingIntegration()
if args.command == 'analyze':
integrator.analyze_skills()
elif args.command == 'extract':
knowledge = integrator.extract_skill_knowledge(args.skill)
if knowledge:
print(f"📖 技能知识: {knowledge['name']}")
print(f"📝 摘要:")
print(knowledge['summary'])
print(f"\\n🔑 关键点:")
for point in knowledge['key_points'][:10]:
print(f" • {point}")
elif args.command == 'implement':
if args.skill == 'humanizer-cn':
integrator.implement_humanizer_cn()
else:
print(f"❌ 暂不支持实现技能: {args.skill}")
print(f"✅ 已支持的技能: humanizer-cn")
elif args.command == 'report':
integrator.create_integration_report()
elif args.command == 'integrate':
integrator.integrate_with_it_system()
elif args.command == 'all':
print("🚀 执行完整集成流程...")
print()
# 1. 分析技能
integrator.analyze_skills()
print()
# 2. 实现中文去AI化技能
integrator.implement_humanizer_cn()
print()
# 3. 创建集成报告
integrator.create_integration_report()
print()
# 4. 集成到IT系统
integrator.integrate_with_it_system()
print()
print("🎉 完整集成流程完成!")
else:
parser.print_help()
if __name__ == "__main__":
main()