pipeline-app/bin/init_data.py

#!/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),
]


async def init_data(demo=False):
    config_file = os.path.join(ROOT_DIR, 'conf', 'config.json')
    if not os.path.exists(config_file):
        config_file = os.path.join(ROOT_DIR, '..', 'conf', 'config.json')
    config = getConfig(config_file, 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. 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))