#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ dev_workflow.py - Spec-Driven 开发流程辅助工具 用法: python3 dev_workflow.py --create-spec --title "标题" --description "描述" python3 dev_workflow.py --create-branch --spec .planning/spec.md python3 dev_workflow.py --create-pr --branch feature/xxx --title "PR标题" python3 dev_workflow.py --status """ import os import sys import json import argparse import subprocess import re from datetime import datetime from pathlib import Path # 配置 GITEA_URL = os.environ.get('GITEA_URL', 'http://192.168.1.22:8418') GITEA_TOKEN = os.environ.get('GITEA_TOKEN', '') GITEA_REPO = os.environ.get('GITEA_REPO', 'starmen1265/kb-skills') REPO_PATH = Path.home() / 'work' / 'kb-skills' PLANNING_DIR = REPO_PATH / '.planning' def run_cmd(cmd, cwd=None, capture=True): """执行 shell 命令""" result = subprocess.run( cmd, shell=True, cwd=cwd or REPO_PATH, capture_output=capture, text=True ) return result def create_spec(title: str, description: str) -> dict: """创建 spec 文件""" PLANNING_DIR.mkdir(parents=True, exist_ok=True) # 生成分支名 branch_name = re.sub(r'[^a-z0-9]+', '-', title.lower()).strip('-') # 读取模板 template_path = Path(__file__).parent / 'templates' / 'spec-template.md' if template_path.exists(): content = template_path.read_text(encoding='utf-8') else: content = """# Spec: {{TITLE}} **创建时间:** {{DATE}} **状态:** draft **分支:** feature/{{BRANCH_NAME}} --- ## 概述 {{DESCRIPTION}} ## 需求 ### 功能需求 - [ ] **REQ-01**: 待补充 ## 验收标准 1. 待补充 --- ## 变更日志 | 日期 | 变更 | |------|------| | {{DATE}} | 初始创建 | """ # 替换变量 now = datetime.now().strftime('%Y-%m-%d %H:%M') content = content.replace('{{TITLE}}', title) content = content.replace('{{DATE}}', now) content = content.replace('{{BRANCH_NAME}}', branch_name) content = content.replace('{{DESCRIPTION}}', description) # 写入文件 spec_file = PLANNING_DIR / 'spec.md' spec_file.write_text(content, encoding='utf-8') return { "success": True, "spec_file": str(spec_file), "branch_name": branch_name, "title": title } def create_branch(spec_file: str) -> dict: """从 spec 创建功能分支""" spec_path = Path(spec_file) if not spec_path.exists(): return {"success": False, "error": f"Spec 文件不存在: {spec_file}"} # 从 spec 提取分支名 content = spec_path.read_text(encoding='utf-8') match = re.search(r'\*\*分支:\*\*\s*feature/([^\n]+)', content) if match: branch_suffix = match.group(1).strip() else: branch_suffix = f"feature-{datetime.now().strftime('%Y%m%d%H%M%S')}" branch_name = f"feature/{branch_suffix}" # 确保在 main 分支 run_cmd('git checkout main') run_cmd('git pull origin main') # 创建功能分支 result = run_cmd(f'git checkout -b {branch_name}') if result.returncode != 0: # 分支可能已存在，尝试切换 result = run_cmd(f'git checkout {branch_name}') if result.returncode != 0: return {"success": False, "error": f"无法创建分支: {result.stderr}"} return { "success": True, "branch": branch_name, "spec_file": spec_file } def create_pr(branch: str, title: str, body: str = "") -> dict: """调用 Gitea API 创建 PR""" if not GITEA_TOKEN: return {"success": False, "error": "GITEA_TOKEN 未配置"} # 获取最新提交信息作为 body（如果未提供） if not body: result = run_cmd('git log -1 --pretty=%B') body = result.stdout.strip() if result.returncode == 0 else "" # 调用 Gitea API import urllib.request import urllib.parse import json as json_module url = f"{GITEA_URL}/api/v1/repos/{GITEA_REPO}/pulls" data = { "title": title, "head": branch, "base": "main", "body": body } req = urllib.request.Request( url, data=json_module.dumps(data).encode('utf-8'), headers={ 'Authorization': f'token {GITEA_TOKEN}', 'Content-Type': 'application/json' }, method='POST' ) try: with urllib.request.urlopen(req, timeout=30) as response: result = json_module.loads(response.read().decode('utf-8')) return { "success": True, "pr_number": result.get("number"), "pr_url": result.get("html_url"), "branch": branch } except urllib.error.HTTPError as e: error_body = e.read().decode('utf-8') if e.fp else str(e) return {"success": False, "error": f"API 错误: {e.code} - {error_body}"} except Exception as e: return {"success": False, "error": f"请求失败: {str(e)}"} def get_status() -> dict: """获取当前开发状态""" result = { "current_branch": None, "spec_exists": False, "planning_exists": False, "uncommitted_changes": False } # 当前分支 branch_result = run_cmd('git branch --show-current') if branch_result.returncode == 0: result["current_branch"] = branch_result.stdout.strip() # spec 文件 result["spec_exists"] = (PLANNING_DIR / 'spec.md').exists() # planning 目录 result["planning_exists"] = PLANNING_DIR.exists() # 未提交的更改 status_result = run_cmd('git status --porcelain') result["uncommitted_changes"] = bool(status_result.stdout.strip()) return result def main(): parser = argparse.ArgumentParser(description='dev-workflow 辅助工具') subparsers = parser.add_subparsers(dest='command', help='命令') # create-spec spec_parser = subparsers.add_parser('create-spec', help='创建 spec 文件') spec_parser.add_argument('--title', required=True, help='功能标题') spec_parser.add_argument('--description', default='', help='功能描述') # create-branch branch_parser = subparsers.add_parser('create-branch', help='创建功能分支') branch_parser.add_argument('--spec', required=True, help='spec 文件路径') # create-pr pr_parser = subparsers.add_parser('create-pr', help='创建 PR') pr_parser.add_argument('--branch', required=True, help='分支名') pr_parser.add_argument('--title', required=True, help='PR 标题') pr_parser.add_argument('--body', default='', help='PR 描述') # status subparsers.add_parser('status', help='查看状态') args = parser.parse_args() if args.command == 'create-spec': result = create_spec(args.title, args.description) elif args.command == 'create-branch': result = create_branch(args.spec) elif args.command == 'create-pr': result = create_pr(args.branch, args.title, args.body) elif args.command == 'status': result = get_status() else: parser.print_help() return print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()