starmen1265/openclaw-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update: KB workflow git integration + todo items

Browse Source 
- Add git remote check and pull before archive
- Support main/master branch auto-fallback
- Add todo: SenseVoice test, git ops, session routing
- Memory: BV1x3ctzkEdf knowledge base workflow
This commit is contained in:
starmen1265
2026-02-13 19:09:57 +08:00
parent 3bd9cb7793
commit 6093a0a09d
4 changed files with 167 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/kb-workflow/kb_workflow.py
+58 -23
View File
@@ -142,46 +142,81 @@ def archive_and_git(
}
try:
# 1. 创建归档目录
# 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')
date_dir = base_path / "archive" / date_str
date_dir.mkdir(parents=True, exist_ok=True)
# 2. 复制分析报告
# 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)
# 3. 复制元数据
# 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)
# 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))
# 5. Git 提交并推送
if (git_root / ".git").exists():
# 添加所有变更
subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True)
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', 'master'],
cwd=str(git_root), capture_output=True, text=True, timeout=60
# 检查是否有变更需要提交
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_push"] = {"success": push_result.returncode == 0}
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