#!/usr/bin/env python3 """ Import initialization data for pipeline-app. Creates default step types, pipeline templates, and demo data. Run: source py3/bin/activate && python bin/init_data.py [--demo] """ import sys, os, asyncio, json SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR = os.path.dirname(SCRIPT_DIR) sys.path.insert(0, os.path.join(ROOT_DIR, 'py3', 'lib', 'python3.10', 'site-packages')) sys.path.insert(0, ROOT_DIR) from sqlor.dbpools import DBPools from appPublic.jsonConfig import getConfig from appPublic.folderUtils import ProgramPath from appPublic.uniqueID import getID from ahserver.serverenv import ServerEnv from ahserver.globalEnv import initEnv from appPublic.timeUtils import curDateString, timestampstr # ---- Default step types ---- STEP_TYPES = [ # (step_type, display_name, category, is_interactive, description, form_schema) ('ai_generate', 'AI代码生成', 'AI生成', '0', '由AI自动生成代码', '{}'), ('ai_review', 'AI代码审查', 'AI审查', '0', '由AI自动审查代码质量', '{}'), ('ai_test', 'AI测试生成', 'AI测试', '0', '由AI生成测试用例', '{}'), ('human_review', '人工代码审查', '人工审核', '1', '人工审查代码变更', json.dumps({"fields":[{"name":"approved","label":"审批通过","uitype":"switch"},{"name":"comment","label":"审批意见","uitype":"text"}]})), ('human_approval', '发布审批', '人工审批', '1', '人工审批发布请求', json.dumps({"fields":[{"name":"approved","label":"批准发布","uitype":"switch"},{"name":"version","label":"发布版本","uitype":"text"}]})), ('deploy_staging', '部署到测试环境', '部署', '0', '自动部署到测试环境', '{}'), ('deploy_production', '部署到生产环境', '部署', '1', '审批后部署到生产环境', json.dumps({"fields":[{"name":"confirm","label":"确认部署","uitype":"switch"}]})), ('run_tests', '运行测试套件', '测试', '0', '自动运行测试套件', '{}'), ('build_artifact', '构建制品', '构建', '0', '构建发布制品', '{}'), ('notify_webhook', 'Webhook通知', '通知', '0', '发送Webhook通知', '{}'), ] # ---- Default pipeline (SDLC template) ---- SDLC_PIPELINE = { 'id': 'pipeline_sdlc_default', 'name': '标准SDLC产线', 'description': '软件开发生命周期标准流程：AI生成→审查→测试→部署', 'pipeline_type': 'sdlc', 'status': 'published', 'version': '1.0', } SDLC_STEPS = [ ('ai_generate', 'AI代码生成', 1), ('ai_review', 'AI代码审查', 2), ('run_tests', '自动测试', 3), ('human_review', '人工代码审查', 4), ('build_artifact', '构建制品', 5), ('human_approval', '发布审批', 6), ('deploy_production', '生产部署', 7), ] # ---- 团队沟通（通用问题冒泡）：sdlc_general 产线的团队角色 + 问题类型 ---- TEAM_ROLE_TYPE_CODES = [ # (k, v) ('agent', 'Agent'), ('human', '人'), ] SDLC_TEAM_ROLES = [ # (role_name, agentid, role_type, display_name, description) ('main_agent', '', 'agent', '主Agent', '用户面对话agent，回答问题、路由到客户'), ('pm', '', 'agent', '项目经理', '审核交付件、退回意见'), ('requirement', '', 'agent', '需求分析师', '产出需求规格文档'), ('design', '', 'agent', '系统设计师', '产出架构/DB/API设计'), ('develop', '', 'agent', '开发工程师', '产出可运行源码'), ('test', '', 'agent', '测试工程师', '产出测试用例+报告'), ('deploy', '', 'agent', '部署运维', '产出部署文档+配置'), ('customer', '', 'human', '客户', '人类客户，最终兜底'), ] SDLC_PROBLEM_TYPES = [ # (name, title, description, escalation_path) ('need_info', '缺信息提问', '角色agent缺关键信息向上提问', '["main_agent", "customer"]'), ('review_reject', '审核退回', 'PM审核退回意见，被退角色响应后复审', '["pm"]'), ('fault_report', '故障报告', '任务失败报障，人工介入', '["main_agent", "customer"]'), ] async def init_data(demo=False): # getConfig 接收应用根目录（内部拼接 conf/config.json），非 config 文件路径 config = getConfig(ROOT_DIR, NS={'workdir': ROOT_DIR, 'ProgramPath': ProgramPath()}) DBPools(config.databases) initEnv() env = ServerEnv() env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage' async with DBPools().sqlorContext('pipeline') as sor: # 1. Create step types print('Step types:') for stype, display, cat, interactive, desc, schema in STEP_TYPES: existing = await sor.sqlExe( "SELECT step_type FROM pipeline_step_types WHERE step_type=${st}$", {'st': stype}) if existing: print(f' SKIP {stype} ({display})') continue await sor.C('pipeline_step_types', { 'step_type': stype, 'display_name': display, 'category': cat, 'is_interactive': interactive, 'description': desc, 'form_schema': schema, 'created_at': timestampstr() }) print(f' ADD {stype} ({display})') # 2. Create default SDLC pipeline print('\nPipeline:') existing = await sor.sqlExe( "SELECT id FROM pipelines WHERE id=${pid}$", {'pid': SDLC_PIPELINE['id']}) if not existing: await sor.C('pipelines', { **SDLC_PIPELINE, 'created_by': 'system', 'created_at': timestampstr(), 'updated_at': timestampstr() }) print(f' ADD {SDLC_PIPELINE["name"]}') # Create steps for stype, sname, order in SDLC_STEPS: await sor.C('pipeline_steps', { 'id': getID(), 'pipeline_id': SDLC_PIPELINE['id'], 'step_order': order, 'step_name': sname, 'step_type': stype, 'step_config': '{}', 'timeout_seconds': 3600, 'retry_count': 1, 'created_at': timestampstr() }) print(f' step {order}: {sname}') else: print(f' SKIP {SDLC_PIPELINE["name"]} (exists)') # 3. 团队沟通：角色类型代码表 + sdlc_general 团队角色 + 问题类型 print('\nTeam communication:') for k, v in TEAM_ROLE_TYPE_CODES: existing = await sor.sqlExe( "SELECT id FROM appcodes_kv WHERE parentid='team_role_type' AND k=${k}$", {'k': k}) if not existing: await sor.C('appcodes_kv', {'id': getID(), 'parentid': 'team_role_type', 'k': k, 'v': v}) print(f' ADD code team_role_type.{k}') # 确保 sdlc_general 产线存在（能力包注册的产线 id） existing = await sor.sqlExe("SELECT id FROM pipelines WHERE id='sdlc_general'", {}) if not existing: await sor.C('pipelines', { 'id': 'sdlc_general', 'name': '通用软件开发产线', 'description': 'SDLC 开发产线（能力包注册）', 'pipeline_type': 'dev', 'status': 'published', 'version': '1.0', 'org_id': '0', 'created_by': 'system', 'created_at': timestampstr(), 'updated_at': timestampstr(), }) print(' ADD pipeline sdlc_general') for rname, aid, rtype, dname, desc in SDLC_TEAM_ROLES: existing = await sor.sqlExe( "SELECT id FROM pipeline_team_roles WHERE pipeline_id='sdlc_general' AND role_name=${r}$", {'r': rname}) if existing: continue await sor.C('pipeline_team_roles', { 'id': getID(), 'pipeline_id': 'sdlc_general', 'role_name': rname, 'agentid': aid, 'role_type': rtype, 'display_name': dname, 'description': desc, 'created_at': timestampstr(), }) print(f' ADD role {rname}') for pname, ptitle, pdesc, ppath in SDLC_PROBLEM_TYPES: existing = await sor.sqlExe( "SELECT id FROM pipeline_problem_types WHERE pipeline_id='sdlc_general' AND name=${n}$", {'n': pname}) if existing: continue await sor.C('pipeline_problem_types', { 'id': getID(), 'pipeline_id': 'sdlc_general', 'name': pname, 'title': ptitle, 'description': pdesc, 'escalation_path': ppath, 'created_at': timestampstr(), }) print(f' ADD problem_type {pname}') # 4. Demo data if demo: print('\nDemo data:') await _create_demo(sor) print('\nInitialization complete.') async def _create_demo(sor): """Create demo project and iteration.""" pid = getID() existing = await sor.sqlExe("SELECT id FROM sd_projects LIMIT 1", {}) if existing: print(' SKIP (data exists)') return await sor.C('sd_projects', { 'id': pid, 'name': 'Pipeline App Demo', 'description': '产线平台演示项目', 'project_type': 'web', 'status': 'active', 'created_by': 'system', 'created_at': timestampstr() }) print(f' ADD project: Pipeline App Demo') iid = getID() await sor.C('sd_iterations', { 'id': iid, 'name': 'Sprint 1 - 基础框架', 'project_id': pid, 'status': 'in_progress', 'start_date': curDateString(), 'created_at': timestampstr() }) print(f' ADD iteration: Sprint 1') if __name__ == '__main__': demo_mode = '--demo' in sys.argv asyncio.run(init_data(demo=demo_mode))