scripts/dashboard.py - dev/test - ump-team/ump-infra

scripts/dashboard.py
486 строк17 KB
Dmitry Kochenov
fix: исправить 46 ошибок basedpyright и добавить type hint stubs
09 авг 2026, 15:25
09 авг 2026, 15:258a46eb7
#!/usr/bin/env python3
"""
UMP Dashboard — локальный веб-интерфейс для визуализации прогресса проекта.

Этап 1: UMP Dashboard (локальный веб-интерфейс)
  - 1.1 ump dashboard — поднимает локальный веб-сервер (FastAPI)
  - 1.2 Live-лента agent.log (WebSocket/polling)
  - 1.3 Просмотр и one-click restore чекпоинтов
  - 1.4 Просмотр/редактирование Memory Bank (5 markdown-файлов)

Запуск:
  uv run python scripts/dashboard.py

Открыть в браузере:
  http://localhost:8000
"""

from __future__ import annotations

import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any

import uvicorn  # type: ignore[import-not-found, unused-ignore]
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

try:
    from scripts.agent_log import LOG_FILE as AGENT_LOG_FILE
    from scripts.ump.status_api import get_project_status, status_to_dict
except ImportError:
    import sys
    _here = Path(__file__).resolve().parent
    sys.path.insert(0, str(_here.parent))
    from agent_log import LOG_FILE as AGENT_LOG_FILE
    from ump.status_api import get_project_status, status_to_dict

# UMP Phase 3 + 4: Import analytics and budget APIs
try:
    from scripts.ump.config import UmpConfig
except ImportError:
    try:
        _here_path = Path(__file__).resolve().parent
        sys.path.insert(0, str(_here_path.parent))
        from ump.config import UmpConfig
    except Exception:
        UmpConfig = None

BASE = Path.cwd()
STATIC_DIR = BASE / 'scripts' / 'ump' / 'static'
DASHBOARD_DIR = BASE / 'scripts' / 'ump' / 'templates'


app = FastAPI(
    title='UMP Dashboard',
    description='Локальный веб-интерфейс для UMP проектов',
    version='0.0.1',
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

if STATIC_DIR.exists():
    app.mount('/static', StaticFiles(directory=str(STATIC_DIR)), name='static')


@app.get('/api/status')
async def api_status() -> dict[str, Any]:
    """Единый статус проекта для всех клиентов (dashboard, notifications, analytics)."""
    try:
        status = get_project_status()
        return status_to_dict(status)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Status API error: {e}') from e


@app.get('/api/logs')
async def api_logs(tail: int = 20) -> dict[str, Any]:
    """Чтение последних N записей из agent.log."""
    if not AGENT_LOG_FILE.exists():
        return {'logs': [], 'total': 0}

    try:
        lines = AGENT_LOG_FILE.read_text(encoding='utf-8').strip().split('\n')
        entries = []
        for line in lines[-tail:]:
            try:
                entry = json.loads(line)
                entries.append(entry)
            except json.JSONDecodeError:
                continue
        return {'logs': entries, 'total': len(lines)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read logs: {e}') from e


@app.get('/api/checkpoints')
async def api_checkpoints() -> dict[str, Any]:
    """Список всех checkpoint'ов проекта."""
    checkpoints_dir = BASE / '_meta' / 'checkpoints'
    if not checkpoints_dir.exists():
        return {'checkpoints': [], 'latest': None}

    try:
        checkpoints = []
        for cp_file in sorted(checkpoints_dir.glob('step-*.json')):
            try:
                data = json.loads(cp_file.read_text(encoding='utf-8'))
                if '_signature' in data:
                    checkpoints.append({
                        'step_id': data.get('step_id', '?'),
                        'phase': data.get('phase', '?'),
                        'saved_at': data.get('saved_at', ''),
                        'path': str(cp_file),
                        'data_keys': list(data.keys()),
                        'can_restore': 'state' in data,
                    })
            except Exception:
                continue

        latest = checkpoints[-1] if checkpoints else None
        return {'checkpoints': checkpoints, 'latest': latest}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read checkpoints: {e}') from e


@app.get('/api/memory-bank')
async def api_memory_bank() -> dict[str, str]:
    """Содержимое 5 файлов Memory Bank."""
    memory_dir = BASE / 'memory-bank'
    if not memory_dir.exists():
        return {
            'projectbrief.md': '',
            'activeContext.md': '',
            'progress.md': '',
            'systemPatterns.md': '',
            'techContext.md': '',
        }

    files = ['projectbrief.md', 'activeContext.md', 'progress.md', 'systemPatterns.md', 'techContext.md']
    result = {}
    for fname in files:
        fpath = memory_dir / fname
        if fpath.exists():
            try:
                result[fname] = fpath.read_text(encoding='utf-8')
            except Exception:
                result[fname] = ''
        else:
            result[fname] = ''
    return result


@app.get('/api/memory-bank/{filename}')
async def api_memory_file(filename: str) -> dict[str, str]:
    """Один файл Memory Bank."""
    valid_files = ['projectbrief.md', 'activeContext.md', 'progress.md', 'systemPatterns.md', 'techContext.md']
    if filename not in valid_files:
        raise HTTPException(status_code=404, detail=f'Invalid filename. Must be one of: {valid_files}')

    fpath = BASE / 'memory-bank' / filename
    if not fpath.exists():
        return {'content': ''}

    try:
        content = fpath.read_text(encoding='utf-8')
        return {'content': content}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read file: {e}') from e


@app.post('/api/memory-bank/{filename}')
async def api_memory_update(filename: str, data: dict[str, str]) -> dict[str, str]:
    """Обновление одного файла Memory Bank."""
    valid_files = ['projectbrief.md', 'activeContext.md', 'progress.md', 'systemPatterns.md', 'techContext.md']
    if filename not in valid_files:
        raise HTTPException(status_code=404, detail=f'Invalid filename. Must be one of: {valid_files}')

    if 'content' not in data:
        raise HTTPException(status_code=400, detail='Missing "content" field')

    fpath = BASE / 'memory-bank' / filename
    fpath.parent.mkdir(parents=True, exist_ok=True)

    try:
        fpath.write_text(data['content'], encoding='utf-8')
        return {'status': 'ok', 'filename': filename}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to write file: {e}') from e


@app.get('/api/gates')
async def api_gates() -> dict[str, Any]:
    """Список гейтов с их статусами."""
    try:
        config_path = BASE / 'ump-ui-config.yaml'
        if not config_path.exists():
            return {'gates': [], 'total': 0, 'passed': 0}

        import yaml
        config = yaml.safe_load(config_path.read_text(encoding='utf-8'))

        if not config or 'agent' not in config or 'interactive' not in config['agent']:
            return {'gates': [], 'total': 0, 'passed': 0}

        gates_config = config['agent']['interactive'].get('gates', {})
        gates_list = ['stage_analysis']
        if gates_config.get('settings'):
            gates_list.append('settings')
        if gates_config.get('deps'):
            gates_list.append('deps')
        if gates_config.get('adr'):
            gates_list.append('adr')
        if gates_config.get('verify_done'):
            gates_list.append('verify_done')
        if gates_config.get('ready_to_continue'):
            gates_list.append('ready_to_continue')
        if gates_config.get('schema_change'):
            gates_list.append('schema_change')
        gates_list.append('protect')

        gates = []
        for gate_name in gates_list:
            gates.append({
                'name': gate_name,
                'status': 'passing' if gate_name != 'stage_analysis' else 'required',
                'message': None,
            })

        return {'gates': gates, 'total': len(gates_list), 'passed': len(gates_list)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read gates: {e}') from e


@app.get('/api/plans')
async def api_plans() -> dict[str, Any]:
    """Список всех этапов и шагов из YAML-планов."""
    plans_dir = BASE / 'plans'
    if not plans_dir.exists():
        return {'plans': [], 'total_steps': 0}

    try:
        import yaml
        plans = []
        total_steps = 0
        for f in sorted(plans_dir.glob('*.yaml')):
            if f.name.startswith('99'):
                continue
            try:
                data = yaml.safe_load(f.read_text(encoding='utf-8'))
                if data and 'steps' in data:
                    step_count = len(data.get('steps', []))
                    total_steps += step_count
                    plans.append({
                        'stage': data.get('stage', '?'),
                        'name': data.get('name', '?'),
                        'file': f.name,
                        'step_count': step_count,
                    })
            except Exception:
                continue

        return {'plans': plans, 'total_steps': total_steps}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read plans: {e}') from e


@app.get('/api/version')
async def api_version() -> dict[str, str]:
    """Версия UMP Dashboard."""
    return {'version': '0.0.1', 'ump_version': 'v0.0.6', 'features': 'analytics,budget'}

@app.get('/api/analytics')
async def api_analytics(step_id: str | None = None, phase: str | None = None) -> dict[str, Any]:
    """Аналитический отчет по проекту (Этап 3)."""
    log_file = BASE / '_meta' / 'agent.log'

    if not log_file.exists():
        return {
            'error': 'agent.log not found',
            'steps': 0,
            'total_time': 0.0,
            'avg_time_per_step': 0.0,
            'phases': [],
            'gates': {'total': 0, 'failed': 0, 'pass_rate': 100.0},
        }

    entries = []
    try:
        with open(log_file, encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                    entries.append(entry)
                except json.JSONDecodeError:
                    continue
    except Exception as e:
        raise HTTPException(status_code=500, detail=f'Failed to read logs: {e}') from e

    if step_id:
        entries = [e for e in entries if e.get('step') == step_id]
    if phase:
        entries = [e for e in entries if e.get('phase') == phase]

    phase_times = {}
    phase_entries = {}
    gate_failures = 0
    total_gates = 0
    step_start_times = {}
    step_end_times = {}

    for entry in entries:
        step = entry.get('step')
        phase_name = entry.get('phase', 'unknown')
        entry_type = entry.get('type')
        timestamp = entry.get('timestamp')

        if entry_type == 'step_start' and step:
            step_start_times[step] = timestamp
        elif entry_type == 'step_complete' and step:
            step_end_times[step] = timestamp
        elif entry_type == 'phase' and phase_name:
            if phase_name not in phase_times:
                phase_times[phase_name] = []
                phase_entries[phase_name] = 0
            phase_times[phase_name].append(timestamp)
            phase_entries[phase_name] += 1
        elif entry_type == 'gate':
            total_gates += 1
            if entry.get('details', {}).get('status') == 'failed':
                gate_failures += 1

    total_steps = len(step_end_times)
    total_time = 0
    if step_start_times and step_end_times:
        for step, start in step_start_times.items():
            if step in step_end_times:
                try:
                    start_dt = datetime.fromisoformat(start)
                    end_dt = datetime.fromisoformat(step_end_times[step])
                    total_time += (end_dt - start_dt).total_seconds()
                except Exception:
                    pass

    avg_time_per_step = total_time / total_steps if total_steps > 0 else 0

    phases = []
    for phase_name in sorted(phase_times.keys()):
        avg = 0
        if len(phase_times[phase_name]) > 1:
            try:
                times = [datetime.fromisoformat(ts) for ts in phase_times[phase_name]]
                if len(times) >= 2:
                    avg = (max(times) - min(times)).total_seconds() / len(times)
            except Exception:
                pass
        phases.append({
            'name': phase_name,
            'count': phase_entries[phase_name],
            'avg_time': round(avg, 2),
        })

    pass_rate = (total_gates - gate_failures) / total_gates * 100 if total_gates > 0 else 100

    return {
        'steps': total_steps,
        'total_time': round(total_time, 2),
        'avg_time_per_step': round(avg_time_per_step, 2),
        'total_entries': len(entries),
        'phases': phases,
        'gates': {
            'total': total_gates,
            'failed': gate_failures,
            'pass_rate': round(pass_rate, 1),
        },
    }


@app.get('/api/budget')
async def api_budget(step_id: str | None = None, threshold: float = 80.0) -> dict[str, Any]:
    """Статус бюджета AI-агентов (Этап 4)."""
    config_path = BASE / 'ump-ui-config.yaml'
    budget_limit = 100.0

    if config_path.exists() and UmpConfig:
        try:
            cfg = UmpConfig.load(BASE)
            _analytics = cfg.extra_fields.get('analytics', {})
            budget_limit = float(_analytics.get('budget_limit_usd', 100.0))
        except Exception:
            try:
                import yaml
                config = yaml.safe_load(config_path.read_text(encoding='utf-8'))
                budget_limit = float(config.get('analytics', {}).get('budget_limit_usd', 100.0))
            except Exception:
                pass

    log_file = BASE / '_meta' / 'agent.log'
    step_costs = {}
    current_spent = 0.0

    if log_file.exists():
        try:
            with open(log_file, encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        entry = json.loads(line)
                        step = entry.get('step')
                        tokens = entry.get('tokens_used', 0)
                        cost = entry.get('cost_usd', 0.0)

                        if step:
                            if step not in step_costs:
                                step_costs[step] = {'tokens': 0, 'cost': 0.0}
                            step_costs[step]['tokens'] += tokens
                            step_costs[step]['cost'] += cost
                            current_spent += cost
                    except json.JSONDecodeError:
                        continue
        except Exception:
            pass

    usage_percent = (current_spent / budget_limit * 100) if budget_limit > 0 else 0
    status = 'normal' if usage_percent < threshold else 'warning'

    steps_list = []
    for step, costs in step_costs.items():
        if step_id and step != step_id:
            continue
        steps_list.append({
            'step_id': step,
            'tokens': costs['tokens'],
            'cost': round(costs['cost'], 4),
        })

    return {
        'limit_usd': round(budget_limit, 2),
        'spent_usd': round(current_spent, 2),
        'usage_percent': round(usage_percent, 1),
        'status': status,
        'threshold': threshold,
        'steps': steps_list,
    }


@app.get("/")
async def index() -> FileResponse:
    """Index.html для SPA."""
    index_path = DASHBOARD_DIR / 'index.html'
    if index_path.exists():
        return FileResponse(index_path)
    raise HTTPException(status_code=404, detail='Dashboard not found')


def main() -> None:
    """Запуск FastAPI сервера."""
    host = os.environ.get('DASHBOARD_HOST', '127.0.0.1')
    port = int(os.environ.get('DASHBOARD_PORT', '8000'))

    print(f'UMP Dashboard v0.0.1 running on http://{host}:{port}')
    uvicorn.run(
        'dashboard:app',
        host=host,
        port=port,
        reload=False,
        log_level='warning',
    )


if __name__ == '__main__':
    main()