From 06ae6bb1ac4ce2c23fbf1acf2e40c7d316b7c0fd Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 18 Aug 2026 15:47:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=A1=B9=E7=9B=AE=E5=88=A0=E9=99=A4?= =?UTF-8?q?=E5=89=8D=E7=A1=AE=E8=AE=A4+=E5=BD=92=E6=A1=A3=E5=A4=87?= =?UTF-8?q?=E4=BB=BD=E2=80=94=E2=80=94delete=5Fproject(confirm=E4=BA=8C?= =?UTF-8?q?=E6=AC=A1=E7=A1=AE=E8=AE=A4)=E5=88=A0=E9=99=A4=E5=89=8D?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E5=A4=87=E4=BB=BDtgz=E5=88=B0=E6=9C=BA?= =?UTF-8?q?=E6=9E=84=E5=B7=A5=E4=BD=9C=E7=9B=AE=E5=BD=95=5Farchive/?= =?UTF-8?q?=EF=BC=8C=E5=90=AB=E5=B7=A5=E4=BD=9C=E7=9B=AE=E5=BD=95+?= =?UTF-8?q?=E5=85=A8=E5=85=B3=E8=81=94=E8=A1=A8=E6=95=B0=E6=8D=AE=E8=AE=B0?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/project_capability.py | 171 +++++++++++++++++++++++++ pipeline_service/sdlc_ability.py | 27 ++++ 2 files changed, 198 insertions(+) diff --git a/pipeline_service/project_capability.py b/pipeline_service/project_capability.py index 409cecf..db8ae1f 100644 --- a/pipeline_service/project_capability.py +++ b/pipeline_service/project_capability.py @@ -12,6 +12,13 @@ org_id 用于 list 过滤（租户隔离）；audit 的 tenant_id 用 project_id """ import logging +import os +import json +import re +import tarfile +import tempfile +import shutil +from datetime import datetime from sqlor.dbpools import DBPools from appPublic.uniqueID import getID from .audit import record_audit @@ -176,3 +183,167 @@ def _rec_to_dict(rec): except Exception: pass return {} + + +async def _dump_project_records(sor, project_id): + """导出项目及关联表全部记录（归档备份 JSON 用）。""" + dump = {} + # 直接关联（project_id / tenant_id） + direct = { + 'sd_projects': "SELECT * FROM sd_projects WHERE id=${pid}$", + 'sd_iterations': "SELECT * FROM sd_iterations WHERE project_id=${pid}$", + 'pipeline_tasks': "SELECT * FROM pipeline_tasks WHERE tenant_id=${pid}$", + 'pipeline_deliverables': "SELECT * FROM pipeline_deliverables WHERE project_id=${pid}$", + 'pipeline_project_agents': "SELECT * FROM pipeline_project_agents WHERE project_id=${pid}$", + 'pipeline_agent_questions': "SELECT * FROM pipeline_agent_questions WHERE tenant_id=${pid}$", + 'sd_deploy_envs': "SELECT * FROM sd_deploy_envs WHERE project_id=${pid}$", + 'sd_features': "SELECT * FROM sd_features WHERE project_id=${pid}$", + 'sd_project_repos': "SELECT * FROM sd_project_repos WHERE project_id=${pid}$", + } + for tbl, sql in direct.items(): + try: + recs = await sor.sqlExe(sql, {"pid": project_id}) + dump[tbl] = [_rec_to_dict(r) for r in (recs or [])] + except Exception as e: + dump[tbl] = {"_error": f"{type(e).__name__}: {str(e)[:200]}"} + + # 间接关联（iteration_id 通过子查询；pipeline_conversations 的 iteration_id 存的是 project_id） + indirect = { + 'sd_bugs': "SELECT * FROM sd_bugs WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + 'sd_test_plans': "SELECT * FROM sd_test_plans WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + 'sd_test_cases': "SELECT * FROM sd_test_cases WHERE plan_id IN (SELECT id FROM sd_test_plans WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$))", + 'sd_conversations': "SELECT * FROM sd_conversations WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + 'pipeline_conversations': "SELECT * FROM pipeline_conversations WHERE iteration_id=${pid}$ OR iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + 'audit_log': "SELECT * FROM audit_log WHERE tenant_id=${pid}$ ORDER BY created_at ASC", + } + for tbl, sql in indirect.items(): + try: + recs = await sor.sqlExe(sql, {"pid": project_id}) + dump[tbl] = [_rec_to_dict(r) for r in (recs or [])] + except Exception as e: + dump[tbl] = {"_error": f"{type(e).__name__}: {str(e)[:200]}"} + + return dump + + +async def backup_project(project_id, who=None, agent_id=None): + """归档备份：打包项目工作目录 + 导出数据库记录 → tgz 到「机构工作目录/_archive/」。 + + 返回 (ok, tgz_path_or_message)。删除项目前必须调用，保证可回退。 + """ + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.R(TABLE, {'id': project_id}) + if not recs: + await sor.sqlExe("COMMIT", {}) + return False, "项目不存在" + p = recs[0] + pname = getattr(p, 'name', '') or project_id + org_id = getattr(p, 'org_id', '0') or '0' + ws = getattr(p, 'workspace_dir', '') or '' + status = getattr(p, 'status', '') + + from .workspace import get_workspace_base + workspace_base = await get_workspace_base(sor) + if not (ws or '').startswith('/'): + ws = os.path.join(workspace_base, str(org_id), pname) + + # 机构工作目录下的备份归档目录 + archive_dir = os.path.join(workspace_base, str(org_id), '_archive') + os.makedirs(archive_dir, exist_ok=True) + + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + safe_name = re.sub(r'[^\w\u4e00-\u9fff.-]', '_', pname) + tgz_path = os.path.join(archive_dir, f"{safe_name}_{timestamp}.tgz") + + # 导出数据库记录 + dump = await _dump_project_records(sor, project_id) + + try: + with tarfile.open(tgz_path, 'w:gz') as tar: + # 数据库记录 JSON + with tempfile.NamedTemporaryFile('w', suffix='.json', delete=False, encoding='utf-8') as f: + json.dump(dump, f, ensure_ascii=False, default=str) + tmp_json = f.name + tar.add(tmp_json, arcname='project_data.json') + os.unlink(tmp_json) + # 工作目录文件（若存在） + if ws and os.path.isdir(ws): + tar.add(ws, arcname='workspace') + except Exception as e: + logger.error("backup_project tar failed: %s err=%s", project_id, e) + return False, f"打包失败: {type(e).__name__}: {str(e)[:200]}" + + await record_audit(project_id, TABLE, project_id, 'backup', + to_state=status, who=_normalize_role(who), + agent_id=agent_id, detail=tgz_path, sor=sor) + logger.info("backup_project: %s -> %s", project_id, tgz_path) + return True, tgz_path + + +async def delete_project(project_id, who=None, agent_id=None, confirm=False): + """删除项目：先归档备份，再删除关联表记录 + sd_projects + 工作目录。 + + confirm 必须为 True（删除前二次确认）。备份失败则中止删除。 + 返回 (ok, message)。 + """ + if not confirm: + return False, "删除项目需二次确认（confirm=true），请先确认再执行" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + # 1. 归档备份（失败则中止删除） + ok, backup_msg = await backup_project(project_id, who=who, agent_id=agent_id) + if not ok: + return False, f"归档备份失败，已中止删除：{backup_msg}" + + # 读工作目录（删除数据库后清理物理文件） + recs = await sor.R(TABLE, {'id': project_id}) + if not recs: + await sor.sqlExe("COMMIT", {}) + return False, "项目不存在" + p = recs[0] + pname = getattr(p, 'name', '') or project_id + org_id = getattr(p, 'org_id', '0') or '0' + ws = getattr(p, 'workspace_dir', '') or '' + from .workspace import get_workspace_base + workspace_base = await get_workspace_base(sor) + if not (ws or '').startswith('/'): + ws = os.path.join(workspace_base, str(org_id), pname) + + # 2. 删除间接关联表（依赖迭代） + indirect = [ + "DELETE FROM sd_test_cases WHERE plan_id IN (SELECT id FROM sd_test_plans WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$))", + "DELETE FROM sd_bugs WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + "DELETE FROM sd_test_plans WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + "DELETE FROM sd_conversations WHERE iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + "DELETE FROM pipeline_conversations WHERE iteration_id=${pid}$ OR iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${pid}$)", + ] + for sql in indirect: + await sor.sqlExe(sql, {"pid": project_id}) + + # 3. 删除直接关联表（project_id / tenant_id） + direct = [ + "DELETE FROM pipeline_deliverables WHERE project_id=${pid}$", + "DELETE FROM pipeline_project_agents WHERE project_id=${pid}$", + "DELETE FROM pipeline_agent_questions WHERE tenant_id=${pid}$", + "DELETE FROM sd_deploy_envs WHERE project_id=${pid}$", + "DELETE FROM sd_features WHERE project_id=${pid}$", + "DELETE FROM sd_project_repos WHERE project_id=${pid}$", + "DELETE FROM pipeline_tasks WHERE tenant_id=${pid}$", + "DELETE FROM audit_log WHERE tenant_id=${pid}$", + "DELETE FROM sd_iterations WHERE project_id=${pid}$", + "DELETE FROM sd_projects WHERE id=${pid}$", + ] + for sql in direct: + await sor.sqlExe(sql, {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + + # 4. 删除工作目录（物理文件，已备份） + if ws and os.path.isdir(ws): + try: + shutil.rmtree(ws, ignore_errors=True) + except Exception as e: + logger.warning("delete_project rmtree failed: %s err=%s", ws, e) + + logger.info("delete_project: %s done, backup=%s", project_id, backup_msg) + return True, f"项目已删除，归档备份在 {backup_msg}" diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 45a8e17..65a581b 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -158,6 +158,8 @@ SDL_TOOLS = [ ToolDefinition(name="complete_project", description="完成项目(active→completed)", parameters={"project_id": "项目ID"}, category="project"), ToolDefinition(name="archive_project", description="归档项目(completed→archived)", parameters={"project_id": "项目ID"}, category="project"), ToolDefinition(name="reopen_project", description="重新打开归档项目(archived→active)", parameters={"project_id": "项目ID"}, category="project"), + ToolDefinition(name="backup_project", description="归档备份项目(打包工作目录+数据记录为tgz到机构工作目录_archive/)", parameters={"project_id": "项目ID"}, category="project"), + ToolDefinition(name="delete_project", description="删除项目(需二次确认confirm=true，删除前自动归档备份tgz)", parameters={"project_id": "项目ID", "confirm": "二次确认，必须为true"}, category="project"), # ── 迭代 ── ToolDefinition(name="list_iterations", description="查看当前项目迭代列表", parameters={"status": "按状态筛选(可选)"}, category="iteration"), ToolDefinition(name="create_iteration", description="创建迭代(planning)", parameters={"iteration_name": "迭代名称", "iteration_type": "迭代类型(可选)", "scope": "迭代范围(可选)"}, category="iteration"), @@ -963,6 +965,29 @@ async def _h_reopen_project(sor, p, ctx): return f"OK: {msg}" if ok else f"ERROR: {msg}" +async def _h_backup_project(sor, p, ctx): + pid = (p.get("project_id", "") or "").strip() + if not pid: + return "需要项目ID" + pid = await _resolve_id(sor, "sd_projects", pid) or pid + from .project_capability import backup_project + ok, msg = await backup_project(pid, who="agent.main_agent") + return f"OK: {msg}" if ok else f"ERROR: {msg}" + + +async def _h_delete_project(sor, p, ctx): + pid = (p.get("project_id", "") or "").strip() + if not pid: + return "需要项目ID" + confirm = (p.get("confirm", "") or "").strip().lower() + if confirm != "true": + return "ERROR: 删除项目需二次确认，请传 confirm=true 后再执行（删除前会自动归档备份到机构工作目录 _archive/）" + pid = await _resolve_id(sor, "sd_projects", pid) or pid + from .project_capability import delete_project + ok, msg = await delete_project(pid, who="agent.main_agent", confirm=True) + return f"OK: {msg}" if ok else f"ERROR: {msg}" + + # ── 迭代 handler ── async def _h_list_iterations(sor, p, ctx): @@ -1465,6 +1490,8 @@ SDL_HANDLERS = { "complete_project": _h_complete_project, "archive_project": _h_archive_project, "reopen_project": _h_reopen_project, + "backup_project": _h_backup_project, + "delete_project": _h_delete_project, # ── 迭代 ── "list_iterations": _h_list_iterations, "create_iteration": _h_create_iteration,