OpenClaw Agent
679cca3b2c
- 无论content步骤有没有字幕，都执行ocr步骤 - ocr超时设置为30分钟（1800秒） - 如果没有字幕文件但有视频文件，使用视频作为ocr源
713 lines
25 KiB
Python
713 lines
25 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
KB Workflow - Knowledge Base Workflow Orchestrator
|
||
|
||
知识库工作流编排技能。
|
||
|
||
核心职责：
|
||
1. 接收用户消息，识别触发词和链接
|
||
2. 查重（根据URL/文件名）
|
||
3. 调用子agent处理各阶段
|
||
4. 文件流转（mv）
|
||
5. Git版本管理
|
||
6. 归档和清理
|
||
|
||
设计原则：
|
||
- workflow不读文件内容
|
||
- 从文件名提取平台（第一个下划线之前）
|
||
- 子技能自己定义输入输出格式
|
||
- workflow只传递提示词 + 需要的路径信息
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import argparse
|
||
import subprocess
|
||
import shutil
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Any
|
||
|
||
# ==================== 子技能映射 ====================
|
||
|
||
LINK_PARSER_AGENT = "kb-link-parser"
|
||
METADATA_AGENTS = {
|
||
"bilibili": "kb-bilibili-metadata",
|
||
"zhihu": "kb-zhihu-metadata",
|
||
}
|
||
CONTENT_AGENTS = {
|
||
"bilibili": "kb-bilibili-content",
|
||
}
|
||
OCR_AGENT = "kb-ocr-processor"
|
||
ANALYZER_AGENT = "kb-deep-analyzer"
|
||
|
||
|
||
def extract_platform(filename: str) -> str:
|
||
"""从文件名提取平台（第一个下划线之前）"""
|
||
match = re.match(r'^([a-zA-Z0-9]+)_(.+)$', filename)
|
||
if match:
|
||
return match.group(1)
|
||
return "unknown"
|
||
|
||
|
||
def extract_base_name(filename: str) -> str:
|
||
"""从文件名提取基础名称（去掉扩展名）"""
|
||
return Path(filename).stem
|
||
|
||
|
||
def ensure_directories(base_path: Path) -> Dict[str, Path]:
|
||
"""确保目录存在"""
|
||
dirs = {
|
||
"pending": base_path / "pending",
|
||
"metadata": base_path / "metadata",
|
||
"content": base_path / "content",
|
||
"content_downloads": base_path / "content" / "downloads",
|
||
"content_subtitles": base_path / "content" / "subtitles",
|
||
"analyzed": base_path / "analyzed",
|
||
"archive": base_path / "archive",
|
||
"archive_subscriptions": base_path / "archive" / "subscriptions",
|
||
}
|
||
|
||
for name, dir_path in dirs.items():
|
||
dir_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
return dirs
|
||
|
||
|
||
def check_duplicate(base_path: Path, url: str) -> Dict[str, Any]:
|
||
"""查重检查"""
|
||
metadata_dir = base_path / "metadata"
|
||
archive_dir = base_path / "archive"
|
||
|
||
# 从URL提取ID
|
||
bv_match = re.search(r'BV[a-zA-Z0-9]+', url)
|
||
content_id = bv_match.group() if bv_match else None
|
||
|
||
result = {"duplicate": False, "status": None, "files": []}
|
||
|
||
if metadata_dir.exists():
|
||
for f in metadata_dir.iterdir():
|
||
if f.is_file() and content_id and content_id in f.name:
|
||
result["duplicate"] = True
|
||
result["status"] = "metadata"
|
||
result["files"].append(str(f))
|
||
|
||
if archive_dir.exists():
|
||
for date_dir in archive_dir.iterdir():
|
||
if date_dir.is_dir():
|
||
for f in date_dir.iterdir():
|
||
if f.is_file() and content_id and content_id in f.name:
|
||
result["duplicate"] = True
|
||
result["status"] = "archived"
|
||
result["files"].append(str(f))
|
||
|
||
return result
|
||
|
||
|
||
def call_subagent(agent_id: str, task: str, timeout: int = 120) -> Dict[str, Any]:
|
||
"""调用子agent（通过 OpenClaw sessions_spawn 或命令行）"""
|
||
print(f"\n{'='*50}")
|
||
print(f"调用子技能: {agent_id}")
|
||
print(f"{'='*50}")
|
||
|
||
# 尝试使用 sessions_spawn（OpenClaw 环境）
|
||
try:
|
||
from openclaw_tool import sessions_spawn
|
||
result = sessions_spawn(
|
||
agentId=agent_id,
|
||
task=task,
|
||
timeoutSeconds=timeout
|
||
)
|
||
print(f"\n子技能返回结果:")
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
return result
|
||
except ImportError:
|
||
# 回退到命令行调用
|
||
pass
|
||
|
||
# 命令行回退：调用子技能的脚本
|
||
print(f"使用命令行方式调用: {agent_id}")
|
||
|
||
# 技能目录路径
|
||
skills_path = Path(__file__).resolve().parent.parent
|
||
|
||
# 从任务中提取关键信息
|
||
import re
|
||
|
||
# 提取URL
|
||
url_match = re.search(r'URL: (\S+)', task)
|
||
url = url_match.group(1) if url_match else ""
|
||
|
||
# 提取输出目录
|
||
output_match = re.search(r'输出目录: (\S+)', task)
|
||
output_dir = output_match.group(1) if output_match else ""
|
||
|
||
# 提取输出文件名格式
|
||
filename_match = re.search(r'输出文件名格式: (\S+)', task)
|
||
filename_format = filename_match.group(1) if filename_match else "{platform}_{id}.json"
|
||
|
||
# 提取链接文件路径
|
||
link_file_match = re.search(r'链接文件: (\S+)', task)
|
||
link_file = link_file_match.group(1) if link_file_match else ""
|
||
|
||
# 提取元数据文件路径
|
||
meta_file_match = re.search(r'元数据文件: (\S+)', task)
|
||
meta_file = meta_file_match.group(1) if meta_file_match else ""
|
||
|
||
# kb-link-parser
|
||
if agent_id == "kb-link-parser":
|
||
cmd = ['python3', str(skills_path / "kb-link-parser" / "link-parser.py"), '--url', url]
|
||
try:
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
try:
|
||
data = json.loads(proc.stdout)
|
||
if data.get('success'):
|
||
platform = data.get('platform', 'unknown')
|
||
content_id = data.get('id', 'unknown')
|
||
ext = '.link.json'
|
||
filename = f"{platform}_{content_id}{ext}"
|
||
output_file = Path(output_dir) / filename
|
||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
print(f"保存文件: {output_file}")
|
||
return {"success": True, "output_file": str(output_file)}
|
||
else:
|
||
return {"success": False, "error": data.get('error', 'unknown')}
|
||
except json.JSONDecodeError:
|
||
return {"success": False, "error": f"解析失败: {proc.stdout[:200]}"}
|
||
else:
|
||
return {"success": False, "error": proc.stderr}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# kb-bilibili-metadata
|
||
elif agent_id == "kb-bilibili-metadata":
|
||
# 从链接文件或URL中提取BV号
|
||
bv = None
|
||
if link_file and Path(link_file).exists():
|
||
with open(link_file, 'r', encoding='utf-8') as f:
|
||
link_data = json.load(f)
|
||
bv = link_data.get('id')
|
||
if not bv:
|
||
bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or link_file)
|
||
bv = bv_match.group(1) if bv_match else None
|
||
|
||
if not bv:
|
||
return {"success": False, "error": "无法提取BV号"}
|
||
|
||
# 生成输出文件名
|
||
filename = f"bilibili_{bv}.meta.json"
|
||
output_file = Path(output_dir) / filename
|
||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
cmd = [
|
||
'python3', str(skills_path / "kb-bilibili-metadata" / "bilibili_metadata.py"),
|
||
'--id', bv,
|
||
'--type', 'video',
|
||
'--output', str(output_file)
|
||
]
|
||
|
||
try:
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
print(f"保存文件: {output_file}")
|
||
return {"success": True, "output_file": str(output_file)}
|
||
else:
|
||
return {"success": False, "error": proc.stderr}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# kb-bilibili-content
|
||
elif agent_id == "kb-bilibili-content":
|
||
bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or meta_file)
|
||
bv = bv_match.group(1) if bv_match else None
|
||
|
||
if not bv:
|
||
return {"success": False, "error": "无法提取BV号"}
|
||
|
||
downloads_dir = Path(output_dir) / "downloads"
|
||
subtitles_dir = Path(output_dir) / "subtitles"
|
||
downloads_dir.mkdir(parents=True, exist_ok=True)
|
||
subtitles_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
video_file = None
|
||
subtitle_file = None
|
||
|
||
# 下载视频
|
||
cmd_video = [
|
||
'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"),
|
||
'--id', bv,
|
||
'--type', 'video',
|
||
'--output', output_dir
|
||
]
|
||
|
||
try:
|
||
proc = subprocess.run(cmd_video, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
# 查找生成的视频文件（可能包含BV号）
|
||
for f in Path(output_dir).iterdir():
|
||
if f.is_file() and bv in f.name and f.suffix in ['.flv', '.mp4']:
|
||
# 重命名为标准格式
|
||
new_name = f"bilibili_{bv}{f.suffix}"
|
||
new_path = downloads_dir / new_name
|
||
f.rename(new_path)
|
||
video_file = str(new_path)
|
||
break
|
||
except Exception as e:
|
||
print(f"视频下载警告: {e}")
|
||
|
||
# 下载字幕
|
||
cmd_subtitle = [
|
||
'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"),
|
||
'--id', bv,
|
||
'--type', 'subtitle',
|
||
'--output', output_dir
|
||
]
|
||
|
||
try:
|
||
proc = subprocess.run(cmd_subtitle, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
# 查找生成的字幕文件
|
||
for f in Path(output_dir).iterdir():
|
||
if f.is_file() and bv in f.name and f.suffix in ['.srt', '.lrc', '.ass']:
|
||
# 重命名为标准格式
|
||
new_name = f"bilibili_{bv}{f.suffix}"
|
||
new_path = subtitles_dir / new_name
|
||
f.rename(new_path)
|
||
subtitle_file = str(new_path)
|
||
break
|
||
except Exception as e:
|
||
print(f"字幕下载警告: {e}")
|
||
|
||
if video_file or subtitle_file:
|
||
return {"success": True, "output_file": video_file, "subtitles": subtitle_file}
|
||
else:
|
||
return {"success": False, "error": "下载未生成任何文件"}
|
||
|
||
# kb-ocr-processor
|
||
elif agent_id == "kb-ocr-processor":
|
||
# 查找字幕文件
|
||
subtitles_dir = Path(output_dir) / "subtitles"
|
||
subtitle_file = None
|
||
if subtitles_dir.exists():
|
||
for f in subtitles_dir.iterdir():
|
||
if f.suffix in ['.srt', '.lrc']:
|
||
subtitle_file = str(f)
|
||
break
|
||
|
||
if not subtitle_file:
|
||
return {"success": False, "error": "未找到字幕文件"}
|
||
|
||
# 生成输出文件名
|
||
text_file = Path(output_dir) / f"{Path(subtitle_file).stem}.txt"
|
||
|
||
cmd = [
|
||
'python3', str(skills_path / "kb-ocr-processor" / "ocr_processor.py"),
|
||
'--input', subtitle_file,
|
||
'--output', str(text_file)
|
||
]
|
||
|
||
try:
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
print(f"保存文件: {text_file}")
|
||
return {"success": True, "output_file": str(text_file)}
|
||
else:
|
||
return {"success": False, "error": proc.stderr}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# kb-deep-analyzer
|
||
elif agent_id == "kb-deep-analyzer":
|
||
# 生成输出文件名
|
||
meta_name = Path(meta_file).stem if meta_file else "unknown"
|
||
filename = f"{meta_name}.md"
|
||
output_file = Path(output_dir) / filename
|
||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 从任务中提取文字内容文件路径
|
||
text_file_match = re.search(r'文字内容文件: (\S+)', task)
|
||
text_content_file = text_file_match.group(1) if text_file_match else ""
|
||
|
||
# 如果没有文字内容文件，使用 metadata 作为 content
|
||
content_file = meta_file
|
||
if text_content_file and Path(text_content_file).exists():
|
||
# 从文字文件创建 content JSON
|
||
content_json = {
|
||
"type": "transcript",
|
||
"source": text_content_file,
|
||
"text": Path(text_content_file).read_text(encoding='utf-8')[:10000] if Path(text_content_file).exists() else ""
|
||
}
|
||
content_file = Path(output_dir) / f"{meta_name}.content.json"
|
||
with open(content_file, 'w', encoding='utf-8') as f:
|
||
json.dump(content_json, f, ensure_ascii=False, indent=2)
|
||
|
||
cmd = [
|
||
'python3', str(skills_path / "kb-deep-analyzer" / "deep_analyzer.py"),
|
||
'--metadata', meta_file or output_dir,
|
||
'--content', str(content_file),
|
||
'--output', str(output_file)
|
||
]
|
||
|
||
try:
|
||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||
if proc.returncode == 0:
|
||
print(f"保存文件: {output_file}")
|
||
return {"success": True, "output_file": str(output_file)}
|
||
else:
|
||
return {"success": False, "error": proc.stderr}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
return {"success": False, "error": f"暂未实现命令行回退: {agent_id}"}
|
||
|
||
|
||
def call_link_parser(url: str, pending_dir: Path) -> Dict[str, Any]:
|
||
"""调用 kb-link-parser"""
|
||
task = f"""解析以下URL，将结果保存到指定目录：
|
||
|
||
URL: {url}
|
||
|
||
输出目录: {pending_dir}
|
||
输出文件名格式: {{platform}}_{{id}}.link.json
|
||
|
||
请解析URL中的平台和内容ID，将解析结果保存为JSON文件。
|
||
返回结果时告诉我输出文件的完整路径。
|
||
|
||
只需要解析URL并保存结果，不需要下载内容。"""
|
||
|
||
return call_subagent(LINK_PARSER_AGENT, task, timeout=60)
|
||
|
||
|
||
def call_metadata_agent(platform: str, link_file: Path, metadata_dir: Path) -> Dict[str, Any]:
|
||
"""调用对应的元数据获取技能"""
|
||
agent_id = METADATA_AGENTS.get(platform)
|
||
if not agent_id:
|
||
return {"success": False, "error": f"不支持的平台: {platform}"}
|
||
|
||
task = f"""获取链接的元数据：
|
||
|
||
链接文件: {link_file}
|
||
输出目录: {metadata_dir}
|
||
|
||
请读取链接文件，提取URL，然后获取该链接的元数据（标题、作者、播放量等）。
|
||
将元数据保存为JSON文件，文件名格式: {{platform}}_{{id}}.meta.json
|
||
|
||
返回结果时告诉我输出文件的完整路径。"""
|
||
|
||
return call_subagent(agent_id, task, timeout=120)
|
||
|
||
|
||
def call_content_agent(platform: str, meta_file: Path, content_dir: Path) -> Dict[str, Any]:
|
||
"""调用对应的内容获取技能"""
|
||
agent_id = CONTENT_AGENTS.get(platform)
|
||
if not agent_id:
|
||
return {"success": False, "error": f"不支持的平台: {platform}"}
|
||
|
||
task = f"""下载视频内容：
|
||
|
||
元数据文件: {meta_file}
|
||
下载目录: {content_dir}
|
||
|
||
请读取元数据文件，获取视频信息，然后下载视频和字幕。
|
||
视频保存到: content/downloads/
|
||
字幕保存到: content/subtitles/
|
||
|
||
文件名格式: {{platform}}_{{id}}.mp4 和 {{platform}}_{{id}}.srt
|
||
|
||
返回结果时告诉我下载的视频文件和字幕文件的完整路径。"""
|
||
|
||
return call_subagent(agent_id, task, timeout=600)
|
||
|
||
|
||
def call_ocr_processor(source_file: Path, content_dir: Path, timeout: int = 1800) -> Dict[str, Any]:
|
||
"""调用 OCR 处理"""
|
||
task = f"""处理文件，提取文字内容：
|
||
|
||
源文件: {source_file}
|
||
输出目录: {content_dir}
|
||
|
||
请读取文件，提取文字内容。
|
||
输出文件名格式: {{platform}}_{{id}}.txt
|
||
|
||
返回结果时告诉我输出文件的完整路径。"""
|
||
|
||
return call_subagent(OCR_AGENT, task, timeout=timeout)
|
||
|
||
|
||
def call_deep_analyzer(meta_file: Path, text_file: Path, analyzed_dir: Path) -> Dict[str, Any]:
|
||
"""调用深度分析"""
|
||
task = f"""深度分析内容：
|
||
|
||
元数据文件: {meta_file}
|
||
文字内容文件: {text_file}
|
||
输出目录: {analyzed_dir}
|
||
|
||
请读取元数据和文字内容，进行深度分析。
|
||
生成分析报告，保存为Markdown文件。
|
||
文件名格式: {{platform}}_{{id}}.md
|
||
|
||
分析报告应包含：
|
||
- 一句话总结
|
||
- 核心要点
|
||
- 时间轴（如适用）
|
||
- 推荐标签
|
||
|
||
返回结果时告诉我输出文件的完整路径。"""
|
||
|
||
return call_subagent(ANALYZER_AGENT, task, timeout=180)
|
||
|
||
|
||
def archive_and_git(base_path: Path, analyzed_file: Path, git_root: Path, git_user: str, git_email: str, git_remote: str) -> Dict[str, Any]:
|
||
"""归档并提交Git"""
|
||
result = {
|
||
"success": False,
|
||
"archive_path": "",
|
||
"git_commit": False,
|
||
"git_push": None,
|
||
"error": None
|
||
}
|
||
|
||
# 1. 创建归档目录
|
||
date_str = datetime.now().strftime('%Y-%m-%d')
|
||
date_dir = base_path / "archive" / date_str
|
||
date_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 2. 复制文件到归档目录
|
||
archive_file = date_dir / analyzed_file.name
|
||
shutil.copy(analyzed_file, archive_file)
|
||
result["archive_path"] = str(archive_file)
|
||
|
||
# 3. 复制元数据
|
||
meta_file = base_path / "metadata" / f"{extract_base_name(analyzed_file.name)}.meta.json"
|
||
if meta_file.exists():
|
||
shutil.copy(meta_file, date_dir / meta_file.name)
|
||
|
||
# 4. Git操作
|
||
git_dir = git_root / ".git"
|
||
if not git_dir.exists():
|
||
subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True)
|
||
if git_user:
|
||
subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root))
|
||
if git_email:
|
||
subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root))
|
||
|
||
if (git_root / ".git").exists():
|
||
subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True)
|
||
commit_msg = f"Archive: {analyzed_file.name} - {datetime.now().isoformat()}"
|
||
commit_result = subprocess.run(
|
||
['git', 'commit', '-m', commit_msg],
|
||
cwd=str(git_root), capture_output=True, text=True
|
||
)
|
||
result["git_commit"] = commit_result.returncode == 0
|
||
|
||
if git_remote and commit_result.returncode == 0:
|
||
push_result = subprocess.run(
|
||
['git', 'push', 'origin', 'master'],
|
||
cwd=str(git_root), capture_output=True, text=True, timeout=60
|
||
)
|
||
result["git_push"] = {"success": push_result.returncode == 0}
|
||
|
||
result["success"] = True
|
||
return result
|
||
|
||
|
||
def process_url(url: str, base_path: Path = None) -> Dict[str, Any]:
|
||
"""处理单个URL的完整流程"""
|
||
if not base_path:
|
||
base_path = Path.home() / "work" / "knowledge-base"
|
||
|
||
git_root = os.environ.get('KB_GIT_ROOT', str(base_path / 'archive'))
|
||
git_user = os.environ.get('KB_GIT_USER', '')
|
||
git_email = os.environ.get('KB_GIT_EMAIL', '')
|
||
git_remote = os.environ.get('KB_GIT_REMOTE', '')
|
||
|
||
dirs = ensure_directories(base_path)
|
||
|
||
result = {
|
||
"url": url,
|
||
"success": False,
|
||
"stages": [],
|
||
"final_file": None,
|
||
"archive": None,
|
||
"error": None
|
||
}
|
||
|
||
try:
|
||
# 查重
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤1: 查重")
|
||
print(f"{'='*50}")
|
||
|
||
dup_result = check_duplicate(base_path, url)
|
||
if dup_result["duplicate"]:
|
||
result["error"] = f"已存在: {dup_result['status']}"
|
||
result["duplicate"] = True
|
||
result["existing_files"] = dup_result["files"]
|
||
return result
|
||
|
||
# kb-link-parser
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤2: 链接解析 (kb-link-parser)")
|
||
print(f"{'='*50}")
|
||
|
||
link_result = call_link_parser(url, dirs["pending"])
|
||
if not link_result.get("success"):
|
||
result["error"] = f"链接解析失败: {link_result.get('error')}"
|
||
return result
|
||
|
||
link_file = link_result.get("output_file")
|
||
if not link_file:
|
||
result["error"] = "链接解析未返回输出文件"
|
||
return result
|
||
|
||
link_file = Path(link_file)
|
||
result["stages"].append({"stage": "link_parsing", "file": str(link_file)})
|
||
|
||
# 提取平台
|
||
filename = link_file.name
|
||
platform = extract_platform(filename)
|
||
content_id = extract_base_name(filename)
|
||
|
||
print(f"\n平台: {platform}")
|
||
print(f"内容ID: {content_id}")
|
||
|
||
# 元数据
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤3: 元数据提取 ({METADATA_AGENTS.get(platform, 'unknown')})")
|
||
print(f"{'='*50}")
|
||
|
||
meta_result = call_metadata_agent(platform, link_file, dirs["metadata"])
|
||
if not meta_result.get("success"):
|
||
result["error"] = f"元数据获取失败: {meta_result.get('error')}"
|
||
return result
|
||
|
||
meta_file = meta_result.get("output_file")
|
||
if not meta_file:
|
||
result["error"] = "元数据获取未返回输出文件"
|
||
return result
|
||
|
||
meta_file = Path(meta_file)
|
||
result["stages"].append({"stage": "metadata", "file": str(meta_file)})
|
||
|
||
# 内容下载
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤4: 内容下载 ({CONTENT_AGENTS.get(platform, 'unknown')})")
|
||
print(f"{'='*50}")
|
||
|
||
content_result = call_content_agent(platform, meta_file, dirs["content"])
|
||
if not content_result.get("success"):
|
||
result["error"] = f"内容下载失败: {content_result.get('error')}"
|
||
return result
|
||
|
||
video_file = content_result.get("output_file")
|
||
subtitle_file = content_result.get("subtitle") or content_result.get("subtitles")
|
||
|
||
result["stages"].append({"stage": "content", "video": video_file, "subtitles": subtitle_file})
|
||
|
||
# OCR
|
||
text_file = None
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤5: OCR处理 (kb-ocr-processor)")
|
||
print(f"{'='*50}")
|
||
|
||
# 无论有没有字幕，都尝试OCR处理
|
||
ocr_source = subtitle_file
|
||
if not ocr_source and video_file:
|
||
ocr_source = video_file
|
||
|
||
if ocr_source:
|
||
# OCR超时设置为30分钟
|
||
ocr_result = call_ocr_processor(Path(ocr_source), dirs["content"], timeout=1800)
|
||
text_file = ocr_result.get("output_file") if ocr_result.get("success") else None
|
||
|
||
if text_file:
|
||
result["stages"].append({"stage": "ocr", "file": text_file})
|
||
else:
|
||
print(f"OCR处理未生成文件")
|
||
else:
|
||
print(f"无字幕文件，跳过OCR处理")
|
||
|
||
# 深度分析
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤6: 深度分析 (kb-deep-analyzer)")
|
||
print(f"{'='*50}")
|
||
|
||
analyze_result = call_deep_analyzer(
|
||
meta_file,
|
||
Path(text_file) if text_file else meta_file,
|
||
dirs["analyzed"]
|
||
)
|
||
if not analyze_result.get("success"):
|
||
result["error"] = f"深度分析失败: {analyze_result.get('error')}"
|
||
return result
|
||
|
||
analyzed_file = analyze_result.get("output_file")
|
||
if not analyzed_file:
|
||
result["error"] = "深度分析未返回输出文件"
|
||
return result
|
||
|
||
analyzed_file = Path(analyzed_file)
|
||
result["stages"].append({"stage": "analyzer", "file": str(analyzed_file)})
|
||
result["final_file"] = str(analyzed_file)
|
||
|
||
# 归档和Git
|
||
print(f"\n{'='*50}")
|
||
print(f"步骤7: 归档和Git")
|
||
print(f"{'='*50}")
|
||
|
||
archive_result = archive_and_git(
|
||
base_path, analyzed_file,
|
||
Path(git_root), git_user, git_email, git_remote
|
||
)
|
||
|
||
result["archive"] = archive_result
|
||
result["success"] = True
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"处理完成！")
|
||
print(f"{'='*50}")
|
||
|
||
except Exception as e:
|
||
result["error"] = str(e)
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
return result
|
||
|
||
|
||
def main():
|
||
"""CLI入口"""
|
||
parser = argparse.ArgumentParser(
|
||
description='KB Workflow - Knowledge Base Workflow Orchestrator',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter
|
||
)
|
||
|
||
parser.add_argument('--url', type=str, help='要处理的URL')
|
||
parser.add_argument('--base-path', type=str, help='知识库根目录')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.url:
|
||
print("请提供URL: --url https://...")
|
||
sys.exit(1)
|
||
|
||
base_path = Path(args.base_path) if args.base_path else None
|
||
result = process_url(args.url, base_path)
|
||
|
||
print(f"\n{'='*50}")
|
||
print("KB Workflow Result")
|
||
print(f"{'='*50}")
|
||
print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
|
||
|
||
sys.exit(0 if result.get('success') else 1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|