735 lines
25 KiB
Python
735 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
KB Workflow - Knowledge Base Workflow Orchestrator
|
|
|
|
知识库工作流辅助工具函数。
|
|
|
|
此模块提供查重、文件操作、Git 管理、状态管理等工具函数，
|
|
供主代理在执行 SKILL.md 中定义的 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, Tuple
|
|
|
|
|
|
# ==================== 状态机定义 ====================
|
|
|
|
# 工作流步骤顺序
|
|
WORKFLOW_STEPS = [
|
|
"init",
|
|
"link_parse",
|
|
"metadata",
|
|
"content",
|
|
"whisper", # 可选步骤
|
|
"analyze",
|
|
"archive",
|
|
"done"
|
|
]
|
|
|
|
# 每个步骤的必需输出字段
|
|
STEP_OUTPUT_FIELDS = {
|
|
"init": [],
|
|
"link_parse": ["platform", "content_type", "content_id"],
|
|
"metadata": ["title"],
|
|
"content": ["filepath"],
|
|
"whisper": ["output_files"],
|
|
"analyze": ["output_file"],
|
|
"archive": ["archive_path"],
|
|
"done": []
|
|
}
|
|
|
|
# 步骤描述（用于汇报）
|
|
STEP_DESCRIPTIONS = {
|
|
"init": "初始化",
|
|
"link_parse": "链接解析",
|
|
"metadata": "元数据提取",
|
|
"content": "内容下载",
|
|
"whisper": "语音识别",
|
|
"analyze": "深度分析",
|
|
"archive": "归档",
|
|
"done": "完成"
|
|
}
|
|
|
|
|
|
# ==================== 子技能映射 ====================
|
|
|
|
METADATA_AGENTS = {
|
|
"bilibili": "kb-bilibili-metadata",
|
|
"zhihu": "kb-zhihu-metadata",
|
|
}
|
|
|
|
CONTENT_AGENTS = {
|
|
"bilibili": "kb-bilibili-content",
|
|
"zhihu": "kb-zhihu-content",
|
|
}
|
|
|
|
|
|
# ==================== 内容类型判断 ====================
|
|
|
|
# 需要下载视频/字幕的内容类型
|
|
VIDEO_CONTENT_TYPES = {"video", "zvideo"}
|
|
|
|
# 需要提取文字的内容类型
|
|
TEXT_CONTENT_TYPES = {"article", "zhuanlan", "question", "answer"}
|
|
|
|
# 需要下载图片的内容类型
|
|
IMAGE_CONTENT_TYPES = {"note"}
|
|
|
|
|
|
def is_video_content(content_type: str) -> bool:
|
|
"""判断是否为视频内容类型"""
|
|
return content_type in VIDEO_CONTENT_TYPES
|
|
|
|
|
|
def is_text_content(content_type: str) -> bool:
|
|
"""判断是否为文字内容类型（专栏、文章、问答等）"""
|
|
return content_type in TEXT_CONTENT_TYPES
|
|
|
|
|
|
def is_image_content(content_type: str) -> bool:
|
|
"""判断是否为图片内容类型（小红书等）"""
|
|
return content_type in IMAGE_CONTENT_TYPES
|
|
|
|
|
|
def get_content_handler(content_type: str) -> str:
|
|
"""
|
|
获取内容处理方式
|
|
|
|
Returns:
|
|
"video" - 需要下载视频/字幕
|
|
"text" - 需要提取文字
|
|
"image" - 需要下载图片
|
|
"unknown" - 未知类型
|
|
"""
|
|
if is_video_content(content_type):
|
|
return "video"
|
|
elif is_text_content(content_type):
|
|
return "text"
|
|
elif is_image_content(content_type):
|
|
return "image"
|
|
else:
|
|
return "unknown"
|
|
|
|
|
|
# ==================== 工具函数 ====================
|
|
|
|
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 get_knowledge_base_path() -> Path:
|
|
"""获取知识库根目录"""
|
|
base_path_str = os.environ.get('KB_BASE_PATH', '')
|
|
if base_path_str:
|
|
return Path(base_path_str).expanduser()
|
|
return Path.home() / "work" / "knowledge-base"
|
|
|
|
|
|
def get_archive_path() -> Path:
|
|
"""获取归档目录（AI分析文档存放位置）"""
|
|
archive_path_str = os.environ.get('KB_ARCHIVE_PATH', '')
|
|
if archive_path_str:
|
|
return Path(archive_path_str).expanduser()
|
|
# 默认：~/work/zq/30.资源/AI分析
|
|
return Path.home() / "work" / "zq" / "30.资源" / "AI分析"
|
|
|
|
|
|
def get_git_root() -> Path:
|
|
"""获取 Git 仓库根目录（用于归档提交）"""
|
|
git_root_str = os.environ.get('KB_GIT_ROOT', '')
|
|
if git_root_str:
|
|
return Path(git_root_str).expanduser()
|
|
# 默认：~/work/zq
|
|
return Path.home() / "work" / "zq"
|
|
|
|
|
|
def ensure_directories(base_path: Path = None) -> Dict[str, Path]:
|
|
"""确保目录存在"""
|
|
if base_path is None:
|
|
base_path = get_knowledge_base_path()
|
|
|
|
dirs = {
|
|
"base": base_path,
|
|
"pending": base_path / "pending",
|
|
"metadata": base_path / "metadata",
|
|
"content": base_path / "content",
|
|
"content_downloads": base_path / "content" / "downloads",
|
|
"analyzed": base_path / "analyzed",
|
|
"archive": base_path / "archive",
|
|
"archive_subscriptions": base_path / "archive" / "subscriptions",
|
|
}
|
|
|
|
for name, dir_path in dirs.items():
|
|
if isinstance(dir_path, Path):
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
return dirs
|
|
|
|
|
|
def check_duplicate(base_path: Path = None, url: str = None, platform: str = None, content_id: str = None) -> Dict[str, Any]:
|
|
"""查重检查"""
|
|
if base_path is None:
|
|
base_path = get_knowledge_base_path()
|
|
|
|
metadata_dir = base_path / "metadata"
|
|
archive_dir = base_path / "archive"
|
|
|
|
# 从URL提取ID
|
|
if url and not content_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 archive_and_git(
|
|
analyzed_file: str,
|
|
base_path: Path = None,
|
|
git_root: Path = None,
|
|
git_user: str = None,
|
|
git_email: str = None,
|
|
git_remote: str = None
|
|
) -> Dict[str, Any]:
|
|
"""归档并提交Git"""
|
|
if base_path is None:
|
|
base_path = get_knowledge_base_path()
|
|
if git_root is None:
|
|
git_root = get_git_root()
|
|
if git_user is None:
|
|
git_user = os.environ.get('KB_GIT_USER', '')
|
|
if git_email is None:
|
|
git_email = os.environ.get('KB_GIT_EMAIL', '')
|
|
if git_remote is None:
|
|
git_remote = os.environ.get('KB_GIT_REMOTE', '')
|
|
|
|
result = {
|
|
"success": False,
|
|
"archive_path": "",
|
|
"git_commit": False,
|
|
"git_push": None,
|
|
"error": None
|
|
}
|
|
|
|
try:
|
|
# 1. 检查并初始化 Git 仓库
|
|
git_dir = git_root / ".git"
|
|
if not git_dir.exists():
|
|
# 初始化仓库
|
|
subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True)
|
|
subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root))
|
|
subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root))
|
|
|
|
# 关联远端仓库
|
|
if git_remote:
|
|
subprocess.run(['git', 'remote', 'add', 'origin', git_remote], cwd=str(git_root), capture_output=True)
|
|
# 拉取远端分支
|
|
subprocess.run(['git', 'fetch', 'origin'], cwd=str(git_root), capture_output=True)
|
|
# 设置 main 分支并拉取
|
|
subprocess.run(['git', 'branch', '--set-upstream-to=origin/main', 'main'], cwd=str(git_root), capture_output=True)
|
|
subprocess.run(['git', 'pull', 'origin', 'main', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120)
|
|
else:
|
|
# 已有仓库，检查远端是否配置
|
|
if git_remote:
|
|
remote_check = subprocess.run(['git', 'remote', '-v'], cwd=str(git_root), capture_output=True, text=True)
|
|
if 'origin' not in remote_check.stdout:
|
|
subprocess.run(['git', 'remote', 'add', 'origin', git_remote], cwd=str(git_root), capture_output=True)
|
|
|
|
# 拉取远端最新代码
|
|
subprocess.run(['git', 'fetch', 'origin'], cwd=str(git_root), capture_output=True)
|
|
# 尝试拉取 main 分支，如果失败则尝试 master
|
|
pull_result = subprocess.run(['git', 'pull', 'origin', 'main', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120)
|
|
if pull_result.returncode != 0:
|
|
subprocess.run(['git', 'pull', 'origin', 'master', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120)
|
|
|
|
# 2. 创建归档目录
|
|
date_str = datetime.now().strftime('%Y-%m-%d')
|
|
archive_base = get_archive_path()
|
|
date_dir = archive_base / date_str
|
|
date_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 3. 复制分析报告
|
|
src_file = Path(analyzed_file)
|
|
archive_file = date_dir / src_file.name
|
|
shutil.copy(src_file, archive_file)
|
|
result["archive_path"] = str(archive_file)
|
|
|
|
# 4. 复制元数据
|
|
meta_file = base_path / "metadata" / f"{src_file.stem}.meta.json"
|
|
if meta_file.exists():
|
|
shutil.copy(meta_file, date_dir / meta_file.name)
|
|
|
|
# 5. Git 提交并推送
|
|
if (git_root / ".git").exists():
|
|
# 添加所有变更
|
|
subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True)
|
|
|
|
# 检查是否有变更需要提交
|
|
status_result = subprocess.run(['git', 'status', '--porcelain'], cwd=str(git_root), capture_output=True, text=True)
|
|
|
|
if status_result.stdout.strip():
|
|
commit_msg = f"Archive: {src_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', 'main'],
|
|
cwd=str(git_root), capture_output=True, text=True, timeout=60
|
|
)
|
|
if push_result.returncode != 0:
|
|
# 如果 main 分支推送失败，尝试 master
|
|
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, "output": push_result.stdout}
|
|
|
|
result["success"] = True
|
|
|
|
except Exception as e:
|
|
result["error"] = str(e)
|
|
|
|
return result
|
|
|
|
|
|
# ==================== 状态管理 ====================
|
|
|
|
def get_state_file_path(url: str, base_path: Path = None) -> Path:
|
|
"""获取状态文件路径"""
|
|
if base_path is None:
|
|
base_path = get_knowledge_base_path()
|
|
pending_dir = base_path / "pending"
|
|
pending_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 从 URL 提取标识符
|
|
# 尝试提取 BV 号或其他 ID
|
|
bv_match = re.search(r'BV[a-zA-Z0-9]+', url)
|
|
if bv_match:
|
|
content_id = bv_match.group()
|
|
platform = "bilibili"
|
|
else:
|
|
# 尝试提取知乎 ID
|
|
zhihu_match = re.search(r'zhihu\.com/(?:question|zvideo|p)/(\d+)', url)
|
|
if zhihu_match:
|
|
content_id = zhihu_match.group(1)
|
|
platform = "zhihu"
|
|
else:
|
|
# 使用 URL hash 作为标识
|
|
import hashlib
|
|
content_id = hashlib.md5(url.encode()).hexdigest()[:12]
|
|
platform = "unknown"
|
|
|
|
return pending_dir / f"{platform}_{content_id}.state.json"
|
|
|
|
|
|
def init_state(url: str, base_path: Path = None) -> Dict[str, Any]:
|
|
"""初始化工作流状态"""
|
|
state_file = get_state_file_path(url, base_path)
|
|
|
|
# 如果状态文件已存在，返回现有状态
|
|
if state_file.exists():
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
# 创建新状态
|
|
now = datetime.now().isoformat()
|
|
state = {
|
|
"url": url,
|
|
"platform": None,
|
|
"content_id": None,
|
|
"content_type": None,
|
|
"current_step": "init",
|
|
"completed_steps": [],
|
|
"step_results": {},
|
|
"errors": [],
|
|
"created_at": now,
|
|
"updated_at": now
|
|
}
|
|
|
|
with open(state_file, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
return state
|
|
|
|
|
|
def get_state(url: str, base_path: Path = None) -> Dict[str, Any]:
|
|
"""获取当前工作流状态"""
|
|
state_file = get_state_file_path(url, base_path)
|
|
|
|
if not state_file.exists():
|
|
return {
|
|
"success": False,
|
|
"error": "状态文件不存在",
|
|
"url": url,
|
|
"state_file": str(state_file)
|
|
}
|
|
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
return {
|
|
"success": True,
|
|
"state": state,
|
|
"state_file": str(state_file)
|
|
}
|
|
|
|
|
|
def advance_step(
|
|
url: str,
|
|
step_name: str,
|
|
result: Dict[str, Any] = None,
|
|
base_path: Path = None
|
|
) -> Dict[str, Any]:
|
|
"""推进到下一步并记录结果"""
|
|
state_file = get_state_file_path(url, base_path)
|
|
|
|
if not state_file.exists():
|
|
return {
|
|
"success": False,
|
|
"error": "状态文件不存在，请先执行 --init-state"
|
|
}
|
|
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
# 验证步骤名称
|
|
if step_name not in WORKFLOW_STEPS:
|
|
return {
|
|
"success": False,
|
|
"error": f"无效的步骤名称: {step_name}",
|
|
"valid_steps": WORKFLOW_STEPS
|
|
}
|
|
|
|
# 获取当前步骤索引
|
|
try:
|
|
current_idx = WORKFLOW_STEPS.index(state["current_step"])
|
|
new_idx = WORKFLOW_STEPS.index(step_name)
|
|
except ValueError:
|
|
return {
|
|
"success": False,
|
|
"error": f"步骤索引错误"
|
|
}
|
|
|
|
# 更新状态
|
|
if step_name != state["current_step"]:
|
|
# 标记当前步骤为完成
|
|
if state["current_step"] not in state["completed_steps"]:
|
|
state["completed_steps"].append(state["current_step"])
|
|
|
|
state["current_step"] = step_name
|
|
state["updated_at"] = datetime.now().isoformat()
|
|
|
|
# 记录步骤结果
|
|
if result:
|
|
state["step_results"][step_name] = result
|
|
|
|
# 提取平台和 ID 信息（从 link_parse 步骤）
|
|
if step_name == "link_parse" and result.get("success"):
|
|
state["platform"] = result.get("platform")
|
|
state["content_id"] = result.get("content_id")
|
|
state["content_type"] = result.get("type")
|
|
|
|
# 保存状态
|
|
with open(state_file, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
return {
|
|
"success": True,
|
|
"current_step": step_name,
|
|
"completed_steps": state["completed_steps"],
|
|
"state_file": str(state_file)
|
|
}
|
|
|
|
|
|
def record_error(
|
|
url: str,
|
|
error_msg: str,
|
|
step: str = None,
|
|
base_path: Path = None
|
|
) -> Dict[str, Any]:
|
|
"""记录错误"""
|
|
state_file = get_state_file_path(url, base_path)
|
|
|
|
if not state_file.exists():
|
|
return {
|
|
"success": False,
|
|
"error": "状态文件不存在，请先执行 --init-state"
|
|
}
|
|
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
# 记录错误
|
|
error_entry = {
|
|
"step": step or state["current_step"],
|
|
"error": error_msg,
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
state["errors"].append(error_entry)
|
|
state["updated_at"] = datetime.now().isoformat()
|
|
|
|
# 保存状态
|
|
with open(state_file, 'w', encoding='utf-8') as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
|
|
return {
|
|
"success": True,
|
|
"error_recorded": error_entry,
|
|
"total_errors": len(state["errors"])
|
|
}
|
|
|
|
|
|
def get_resume_info(url: str, base_path: Path = None) -> Dict[str, Any]:
|
|
"""获取恢复执行的指令"""
|
|
state_file = get_state_file_path(url, base_path)
|
|
|
|
if not state_file.exists():
|
|
return {
|
|
"success": False,
|
|
"error": "状态文件不存在",
|
|
"suggestion": "请使用 --init-state 初始化工作流"
|
|
}
|
|
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
state = json.load(f)
|
|
|
|
current_step = state["current_step"]
|
|
step_desc = STEP_DESCRIPTIONS.get(current_step, current_step)
|
|
|
|
# 检查是否有错误
|
|
if state["errors"]:
|
|
last_error = state["errors"][-1]
|
|
return {
|
|
"success": True,
|
|
"status": "error",
|
|
"current_step": current_step,
|
|
"step_description": step_desc,
|
|
"last_error": last_error,
|
|
"completed_steps": state["completed_steps"],
|
|
"suggestion": f"请先解决 {last_error['step']} 步骤的错误: {last_error['error']}"
|
|
}
|
|
|
|
# 检查是否已完成
|
|
if current_step == "done":
|
|
return {
|
|
"success": True,
|
|
"status": "completed",
|
|
"current_step": "done",
|
|
"step_description": "已完成",
|
|
"completed_steps": state["completed_steps"],
|
|
"step_results": state["step_results"]
|
|
}
|
|
|
|
# 返回下一步指令
|
|
return {
|
|
"success": True,
|
|
"status": "in_progress",
|
|
"current_step": current_step,
|
|
"step_description": step_desc,
|
|
"completed_steps": state["completed_steps"],
|
|
"next_action": f"继续执行步骤: {step_desc}",
|
|
"platform": state.get("platform"),
|
|
"content_id": state.get("content_id"),
|
|
"content_type": state.get("content_type"),
|
|
"step_results": state.get("step_results", {})
|
|
}
|
|
|
|
|
|
def generate_summary(url: str, stages: List[Dict], final_file: str, archive_result: Dict = None) -> str:
|
|
"""生成处理完成汇报"""
|
|
# 提取视频标题（如果有）
|
|
title_match = re.search(r'com/video/([^?]+)', url)
|
|
title = title_match.group(1) if title_match else "未知"
|
|
|
|
summary = f"""
|
|
✅ **处理完成！**
|
|
|
|
**视频:** {title}
|
|
|
|
**处理阶段:**
|
|
"""
|
|
for stage in stages:
|
|
stage_name = stage.get('stage', 'unknown')
|
|
emoji = "✓" if stage.get('success') else "○"
|
|
summary += f"{emoji} {stage_name}\n"
|
|
|
|
if final_file:
|
|
summary += f"\n**最终文件:** `{final_file}`"
|
|
|
|
if archive_result:
|
|
if archive_result.get('git_commit'):
|
|
summary += f"\n**Git:** 已提交"
|
|
if archive_result.get('git_push', {}).get('success'):
|
|
summary += " ✓"
|
|
elif archive_result.get('git_push'):
|
|
summary += " ✗"
|
|
|
|
return summary
|
|
|
|
|
|
# ==================== CLI ====================
|
|
|
|
def main():
|
|
"""CLI入口 - 用于命令行测试"""
|
|
parser = argparse.ArgumentParser(
|
|
description='KB Workflow - Knowledge Base Workflow Orchestrator',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
使用示例:
|
|
# 查重
|
|
python kb_workflow.py --check --url "https://www.bilibili.com/video/BVxxx"
|
|
|
|
# 归档
|
|
python kb_workflow.py --archive-file "/path/to/analyzed.md"
|
|
|
|
# 状态管理
|
|
python kb_workflow.py --init-state --url "https://..."
|
|
python kb_workflow.py --get-state --url "https://..."
|
|
python kb_workflow.py --advance-step --url "https://..." --step "metadata" --result '{"success":true}'
|
|
python kb_workflow.py --record-error --url "https://..." --error-msg "403 Forbidden"
|
|
python kb_workflow.py --resume --url "https://..."
|
|
"""
|
|
)
|
|
|
|
# 基础参数
|
|
parser.add_argument('--url', type=str, help='URL 地址')
|
|
parser.add_argument('--base-path', type=str, help='知识库根目录')
|
|
|
|
# 查重和归档
|
|
parser.add_argument('--check', action='store_true', help='查重')
|
|
parser.add_argument('--archive-file', type=str, help='归档的文件路径')
|
|
|
|
# 目录路径查询
|
|
parser.add_argument('--pending', action='store_true', help='获取 URL 解析目录')
|
|
parser.add_argument('--metadata', action='store_true', help='获取元数据目录')
|
|
parser.add_argument('--content', action='store_true', help='获取内容目录')
|
|
parser.add_argument('--content_downloads', action='store_true', help='获取内容下载目录')
|
|
parser.add_argument('--analyzed', action='store_true', help='获取深度分析目录')
|
|
parser.add_argument('--archive_dir', action='store_true', help='获取归档目录')
|
|
parser.add_argument('--archive_subscriptions', action='store_true', help='获取订阅信息目录')
|
|
parser.add_argument('--extract_platform', type=str, help='从文件名提取平台名称')
|
|
|
|
# 状态管理
|
|
parser.add_argument('--init-state', action='store_true', help='初始化工作流状态')
|
|
parser.add_argument('--get-state', action='store_true', help='获取当前状态')
|
|
parser.add_argument('--advance-step', action='store_true', help='推进到下一步')
|
|
parser.add_argument('--step', type=str, help='步骤名称')
|
|
parser.add_argument('--result', type=str, help='步骤结果（JSON 字符串）')
|
|
parser.add_argument('--record-error', action='store_true', help='记录错误')
|
|
parser.add_argument('--error-msg', type=str, help='错误信息')
|
|
parser.add_argument('--resume', action='store_true', help='获取恢复执行信息')
|
|
|
|
args = parser.parse_args()
|
|
|
|
base_path = Path(args.base_path) if args.base_path else None
|
|
dirs = ensure_directories(base_path)
|
|
|
|
# 查重
|
|
if args.check and args.url:
|
|
result = check_duplicate(base_path, args.url)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
# 归档
|
|
elif args.archive_file:
|
|
result = archive_and_git(args.archive_file, base_path)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
# 目录路径查询
|
|
elif args.pending:
|
|
print(json.dumps({"path": str(dirs["pending"])}, ensure_ascii=False, indent=2))
|
|
elif args.metadata:
|
|
print(json.dumps({"path": str(dirs["metadata"])}, ensure_ascii=False, indent=2))
|
|
elif args.content:
|
|
print(json.dumps({"path": str(dirs["content"])}, ensure_ascii=False, indent=2))
|
|
elif args.content_downloads:
|
|
print(json.dumps({"path": str(dirs["content_downloads"])}, ensure_ascii=False, indent=2))
|
|
elif args.analyzed:
|
|
print(json.dumps({"path": str(dirs["analyzed"])}, ensure_ascii=False, indent=2))
|
|
elif args.archive_dir:
|
|
print(json.dumps({"path": str(get_archive_path())}, ensure_ascii=False, indent=2))
|
|
elif args.archive_subscriptions:
|
|
print(json.dumps({"path": str(dirs["archive_subscriptions"])}, ensure_ascii=False, indent=2))
|
|
elif args.extract_platform:
|
|
platform = extract_platform(args.extract_platform)
|
|
print(json.dumps({"platform": platform}, ensure_ascii=False, indent=2))
|
|
elif args.base_path:
|
|
print(json.dumps({"path": str(dirs["base"])}, ensure_ascii=False, indent=2))
|
|
|
|
# 状态管理
|
|
elif args.init_state and args.url:
|
|
result = init_state(args.url, base_path)
|
|
print(json.dumps({"success": True, "state": result}, ensure_ascii=False, indent=2))
|
|
|
|
elif args.get_state and args.url:
|
|
result = get_state(args.url, base_path)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
elif args.advance_step and args.url and args.step:
|
|
result_data = None
|
|
if args.result:
|
|
try:
|
|
result_data = json.loads(args.result)
|
|
except json.JSONDecodeError:
|
|
result_data = {"raw": args.result}
|
|
result = advance_step(args.url, args.step, result_data, base_path)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
elif args.record_error and args.url and args.error_msg:
|
|
result = record_error(args.url, args.error_msg, args.step, base_path)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
elif args.resume and args.url:
|
|
result = get_resume_info(args.url, base_path)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|