scripts/finalize_step.py - main - kochenov/universal-modular

scripts/finalize_step.py
332 строки14 KB
Kochenov Dmitry
feat(foundation): Шаг 1.5 — Создать pyproject.toml + ruff.toml + pre-commit конфиг
10 июл 2026, 16:06
10 июл 2026, 16:06e52daa8
#!/usr/bin/env python3
"""
Автоматический финализатор шага (аудит v11).

Заменяет чтение step-finalizer.md (46KB) одной командой.

Выполняет:
1. Проверка docs_target (Phase 2, VERIFIED_CHECKLIST, история ≥ 5)
2. Многосторонняя проверка через verify_step_completion.py
3. Git: add, commit (squash), checkout main, merge --squash, push
4. Обновление STEP_STATE.md (CURRENT_PHASE=awaiting-continue, COMPLETED_STEPS)
5. Удаление feature-ветки
6. Обновление снимка файлов (detect_manual_changes.py --update-snapshot)
7. Логирование всех действий

Запуск:
    uv run python scripts/finalize_step.py <step_id>
    uv run python scripts/finalize_step.py 1.0

Exit codes:
    0 — шаг финализирован успешно
    1 — ошибка (проверки не пройдены, git конфликт, и т.д.)
    2 — шаг не найден
"""
import argparse
import os
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path

import yaml

BASE = Path(__file__).parent.parent
PLANS_DIR = BASE / 'plans'

# Импорт функций из agent_log
sys.path.insert(0, str(BASE / 'scripts'))
from agent_log import add_entry


def find_step(step_id: str) -> tuple[dict, str, str] | None:
    """Найти шаг в YAML. Вернуть (step_dict, yaml_filename, next_step_dict)."""
    for yfile in sorted(PLANS_DIR.glob('*.yaml')):
        if yfile.name.startswith('99'):
            continue
        try:
            data = yaml.safe_load(yfile.read_text(encoding='utf-8'))
        except Exception:
            continue
        steps = data.get('steps', [])
        for i, step in enumerate(steps):
            if step.get('id') == step_id:
                next_step = steps[i + 1] if i + 1 < len(steps) else None
                return step, yfile.name, next_step
    return None


def run(cmd: str, check: bool = True, cwd: Path | None = None) -> subprocess.CompletedProcess:
    """Запустить shell-команду."""
    if cwd is None:
        cwd = BASE
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, cwd=cwd, timeout=120)
    if check and result.returncode != 0:
        print(f'ERROR: command failed: {cmd}', file=sys.stderr)
        print(f'stdout: {result.stdout[:500]}', file=sys.stderr)
        print(f'stderr: {result.stderr[:500]}', file=sys.stderr)
    return result


def log(msg: str, step: str, entry_type: str = 'action', details: dict | None = None):
    """Логировать действие."""
    print(f'[{entry_type}] {msg}')
    add_entry(msg, entry_type, step=step, details=details)


def check_docs_target(step: dict, step_id: str) -> tuple[bool, str]:
    """Проверить docs_target."""
    docs_target = step.get('docs_target', '')
    if not docs_target:
        return False, 'docs_target не указан'
    full = BASE / docs_target
    if not full.exists():
        return False, f'файл {docs_target} не существует'
    content = full.read_text(encoding='utf-8')

    # Phase 2 маркер
    if '> Phase 1 — инструкция для AI-агента' in content and '> Обучающая документация' not in content:
        return False, 'Phase 2 не применена (остался Phase 1 маркер)'

    # VERIFIED_CHECKLIST — все [x]
    match = re.search(r'## VERIFIED_CHECKLIST\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if not match:
        return False, 'секция VERIFIED_CHECKLIST не найдена'
    open_items = re.findall(r'^- \[ \]\s+(.+)$', match.group(1), re.MULTILINE)
    if open_items:
        return False, f'незакрытые пункты: {open_items[:3]}'

    # История ≥ 5
    match = re.search(r'## История действий агента\s*\n(.*?)(?=\n## |\Z)', content, re.DOTALL)
    if match:
        rows = re.findall(r'^\|.*\|$', match.group(1), re.MULTILINE)
        data_rows = [r for r in rows if not re.match(r'^\|[\s\-:|]+\|$', r) and 'Время' not in r]
        if len(data_rows) < 5:
            return False, f'история действий: {len(data_rows)} записей (нужно ≥ 5)'

    return True, 'OK'


def git_operations(step: dict, step_id: str, step_branch: str) -> tuple[bool, str]:
    """Выполнить git-операции: add, commit, squash-merge, push."""
    use_pr = step.get('use_pr', False)

    # 1. Проверить, что мы на feature-ветке
    branch = run('git branch --show-current').stdout.strip()
    if branch != step_branch:
        return False, f'не на feature-ветке (current={branch}, expected={step_branch})'

    # 2. git add .
    log('git add .', step_id, 'action')
    r = run('git add .')
    if r.returncode != 0:
        return False, f'git add failed: {r.stderr}'

    # 3. git commit (в feature-ветке)
    step_title = step.get('title', '')
    commit_msg = f'feat(foundation): Шаг {step_id} — {step_title}\n\nДокументация: {step.get("docs_target", "")}'
    log('git commit (squash в feature-ветке)', step_id, 'commit')
    r = run(f'git commit --no-verify -m "{commit_msg}"')
    if r.returncode != 0 and 'nothing to commit' not in r.stdout:
        # Может быть уже закоммичено — проверим
        log(f'git commit returned {r.returncode}: {r.stdout[:200]}', step_id, 'warning')

    # 4. Если use_pr — создать PR
    if use_pr:
        log('use_pr=true — push feature и создание PR', step_id, 'action')
        r = run(f'git push origin {step_branch}')
        if r.returncode != 0:
            log(f'git push failed: {r.stderr}', step_id, 'error')
            return False, f'git push failed: {r.stderr[:200]}'
        # Попытка создать PR через gh-cli (если установлен)
        r = run(f'gh pr create --title "Шаг {step_id}: {step_title}" --body "Автоматический PR" --base main --head {step_branch}', check=False)
        if r.returncode == 0:
            log('PR создан', step_id, 'action', {'pr_url': r.stdout.strip()})
            return True, f'PR создан: {r.stdout.strip()}'
        else:
            log(f'gh-cli не установлен или PR не создан: {r.stderr[:200]}', step_id, 'warning')
            log(f'Создайте PR вручную: git push origin {step_branch} → GitHub/GitVerse UI', step_id, 'warning')
            return True, 'feature-ветка запушена, создайте PR вручную'

    # 5. checkout main
    log('git checkout main', step_id, 'action')
    r = run('git checkout main')
    if r.returncode != 0:
        # Возможно main не существует — создадим
        r = run('git checkout -b main')
        if r.returncode != 0:
            return False, f'git checkout main failed: {r.stderr}'

    # 6. pull main (если есть remote)
    run('git pull origin main 2>/dev/null || true', check=False)

    # 7. Проверить, что main чист
    r = run('git status --porcelain')
    if r.stdout.strip():
        log(f'main не чист: {r.stdout.strip()[:200]}', step_id, 'warning')

    # 8. squash-merge
    log(f'git merge --squash {step_branch}', step_id, 'action')
    r = run(f'git merge --squash {step_branch}')
    if r.returncode != 0:
        # Конфликт?
        r2 = run('git status --porcelain')
        if 'UU' in r2.stdout or 'AA' in r2.stdout:
            return False, f'MERGE CONFLICT: {r2.stdout[:200]}. Требуется ручное разрешение.'
        return False, f'git merge --squash failed: {r.stderr}'

    # 9. commit в main
    os.environ['STEP_FINALIZER_ACTIVE'] = '1'
    log('git commit в main (squash)', step_id, 'commit')
    r = run(f'git commit --no-verify -m "{commit_msg}"')
    if r.returncode != 0 and 'nothing to commit' not in r.stdout:
        return False, f'git commit in main failed: {r.stderr}'

    # 10. push main
    r = run('git remote -v')
    if r.stdout.strip():
        log('git push origin main', step_id, 'action')
        r = run('git push origin main 2>&1', check=False)
        if r.returncode != 0:
            log(f'git push failed (возможно нет remote): {r.stderr[:200]}', step_id, 'warning')
        else:
            log('git push OK', step_id, 'action')

    # 11. Удалить feature-ветку
    log(f'git branch -D {step_branch}', step_id, 'action')
    run(f'git branch -D {step_branch}', check=False)
    run(f'git push origin --delete {step_branch} 2>/dev/null || true', check=False)

    return True, 'squash-merge выполнен, feature-ветка удалена'


def update_step_state(step: dict, step_id: str, next_step: dict | None):
    """Обновить STEP_STATE.md."""
    step_state = BASE / 'STEP_STATE.md'
    if not step_state.exists():
        log('STEP_STATE.md не существует — пропускаю обновление', step_id, 'warning')
        return

    now = datetime.now().strftime('%Y-%m-%d')
    content = step_state.read_text(encoding='utf-8')

    # CURRENT_STEP → next_step_id (или остаётся, если next нет)
    next_step_id = next_step.get('id', step_id) if next_step else step_id
    next_step_docs = next_step.get('docs_target', '') if next_step else ''

    # LAST_COMPLETED_STEP → step_id
    content = re.sub(r'LAST_COMPLETED_STEP:.*', f'LAST_COMPLETED_STEP: {step_id}', content)

    # CURRENT_PHASE → awaiting-continue
    content = re.sub(r'CURRENT_PHASE:.*', 'CURRENT_PHASE: awaiting-continue', content)

    # CURRENT_STEP → next
    if next_step:
        content = re.sub(r'CURRENT_STEP:.*', f'CURRENT_STEP: {next_step_id}', content)
        content = re.sub(r'CURRENT_STEP_DOCS:.*', f'CURRENT_STEP_DOCS: {next_step_docs}', content)

    # Добавить в COMPLETED_STEPS
    step_title = step.get('title', '')
    docs_target = step.get('docs_target', '')
    new_row = f'| {step_id} | {step_title} | {now} | [{docs_target}]({docs_target}) |'

    # Найти секцию COMPLETED_STEPS и добавить строку после заголовка таблицы
    content = re.sub(
        r'(\| Шаг \| Название \| Дата \| Документация \|)\n(\|-----\|----------\|------\|--------------\|)\n',
        rf'\1\n\2\n{new_row}\n',
        content,
        count=1
    )

    step_state.write_text(content, encoding='utf-8')
    log(f'STEP_STATE.md обновлён: CURRENT_STEP={next_step_id}, LAST_COMPLETED={step_id}', step_id, 'action')


def main():
    parser = argparse.ArgumentParser(description='Автоматический финализатор шага')
    parser.add_argument('step_id', help='ID шага, например 1.0')
    parser.add_argument('--skip-verify', action='store_true', help='Пропустить verify_step_completion.py')
    args = parser.parse_args()

    step_info = find_step(args.step_id)
    if step_info is None:
        print(f'ERROR: шаг {args.step_id} не найден', file=sys.stderr)
        sys.exit(2)

    step, yfile, next_step = step_info
    step_branch = step.get('branch', f'feature/step-{args.step_id}')
    step_type = step.get('step_type', '')

    print(f'\n{"=" * 70}')
    print(f'ФИНАЛИЗАЦИЯ ШАГА {args.step_id} — {step.get("title", "")}')
    print(f'Ветка: {step_branch}')
    print(f'Тип: {step_type}')
    print(f'{"=" * 70}\n')

    # 1. Проверка docs_target
    print('--- Проверка docs_target ---')
    ok, msg = check_docs_target(step, args.step_id)
    print(f'  [{"✓" if ok else "✗"}] {msg}')
    if not ok:
        log(f'Финализация FAIL: docs_target — {msg}', args.step_id, 'error')
        sys.exit(1)
    log(f'docs_target проверен: {msg}', args.step_id, 'test')

    # 2. Многосторонняя проверка (если не --skip-verify)
    if not args.skip_verify:
        print('\n--- Многосторонняя проверка (verify_step_completion.py) ---')
        r = run(f'python3 scripts/verify_step_completion.py {args.step_id}', check=False)
        print(r.stdout)
        if r.returncode != 0:
            print(f'\n❌ Многосторонняя проверка НЕ пройдена (exit {r.returncode})')
            print('Используйте --skip-verify для пропуска (не рекомендуется).')
            log(f'Финализация FAIL: verify_step_completion exit {r.returncode}', args.step_id, 'error')
            sys.exit(1)
        log('verify_step_completion PASSED', args.step_id, 'test')

    # 3. Git-операции
    print('\n--- Git-операции ---')
    ok, msg = git_operations(step, args.step_id, step_branch)
    print(f'  [{"✓" if ok else "✗"}] {msg}')
    if not ok:
        log(f'Финализация FAIL: git — {msg}', args.step_id, 'error')
        sys.exit(1)

    # 4. Обновление STEP_STATE.md
    print('\n--- Обновление STEP_STATE.md ---')
    update_step_state(step, args.step_id, next_step)
    print('  ✓ STEP_STATE.md обновлён')

    # 5. Обновить снимок файлов
    print('\n--- Обновление снимка файлов ---')
    run('python3 scripts/detect_manual_changes.py --update-snapshot --quiet', check=False)
    print('  ✓ Снимок обновлён')

    # 6. Финальный лог
    next_step_id = next_step.get('id', '?') if next_step else '?'
    next_step_title = next_step.get('title', '') if next_step else ''
    log(f'Шаг {args.step_id} финализирован. Следующий: {next_step_id}', args.step_id, 'commit',
        details={'next_step': next_step_id, 'next_title': next_step_title})

    print(f'\n{"=" * 70}')
    print(f'✓ ШАГ {args.step_id} ФИНАЛИЗИРОВАН')
    print(f'{"=" * 70}')
    print('\nGit: squash-merge в main выполнен, feature-ветка удалена.')
    print(f'Документация: {step.get("docs_target", "")}')
    print(f'\nСледующий шаг: {next_step_id} — {next_step_title}')
    print('\nСкажите "продолжить" для запуска следующего шага.')
    print()

    # Переименовать progress-файл
    progress_file = BASE / '_meta' / 'step-progress.yaml'
    done_file = BASE / '_meta' / f'step-progress-{args.step_id}.done.yaml'
    if progress_file.exists():
        progress_file.rename(done_file)
        print(f'Progress-файл: {done_file.name}')

    sys.exit(0)


if __name__ == '__main__':
    main()