# -*- coding: utf-8 -*- """Агрегатор журнала опыта kit-dev-flow (общий git-репозиторий статистики). Читает карточки прогонов runs/**/*.md (YAML-шапка между первыми двумя строками '---'), считает агрегаты по всем проектам команды и печатает markdown-отчёт. Точка опыта: --dir > env KIT_FLOW_EXP_DIR > ~/.zcode/kit-knowledge.json (flow_exp_dir) > I:\\Kit\\DB.FlowExp_WorkStatistics. Примеры: python flow_stats.py # всё python flow_stats.py --kind этап-модуля # класс текущего шага python flow_stats.py --days 30 --last 40 python flow_stats.py --write-summary # обновить _summary.md """ import argparse import datetime as dt import glob import json import os import sys DEFAULT_DIR = r"I:\Kit\DB.FlowExp_WorkStatistics" NUM_FIELDS = ("agents", "revisions", "escalations", "duration_min") STR_FIELDS = ("date", "project", "step", "kind", "form", "preset", "model", "result", "run_id") def resolve_dir(cli_dir): if cli_dir: return cli_dir env = os.environ.get("KIT_FLOW_EXP_DIR") if env: return env cfg_path = os.path.join(os.path.expanduser("~"), ".zcode", "kit-knowledge.json") try: with open(cfg_path, encoding="utf-8") as fh: cfg = json.load(fh) if cfg.get("flow_exp_dir"): return cfg["flow_exp_dir"] except (OSError, ValueError): pass return DEFAULT_DIR def parse_card(path): """YAML-шапка карточки: ключ: значение (одна строка на ключ).""" with open(path, encoding="utf-8") as fh: lines = fh.read().splitlines() if not lines or lines[0].strip() != "---": return None meta, end = {}, None for i, line in enumerate(lines[1:], start=1): s = line.strip() if s == "---": end = i break if ":" in s and not s.startswith("#"): key, _, val = s.partition(":") val = val.split(" #")[0].strip().strip('"').strip("'") if key.strip() in STR_FIELDS or key.strip() in NUM_FIELDS: meta[key.strip()] = val if end is None or not meta.get("date"): return None lessons, in_lessons = [], False for line in lines[end + 1:]: bs = line.strip() if bs.startswith("## Уроки"): in_lessons = True continue if in_lessons and bs.startswith("## "): in_lessons = False elif in_lessons and bs.startswith("- "): lessons.append(bs[2:].strip()) meta["_lessons"] = lessons return meta def num(meta, field): try: return float(meta.get(field, "")) except (TypeError, ValueError): return None def load_cards(root, kind, project, days): cards = [] for path in glob.glob(os.path.join(root, "runs", "**", "*.md"), recursive=True): meta = parse_card(path) if not meta: continue if kind and meta.get("kind") != kind: continue if project and meta.get("project") != project: continue try: d = dt.date.fromisoformat(meta["date"][:10]) except ValueError: d = None if days and d and (dt.date.today() - d).days > days: continue cards.append(meta) cards.sort(key=lambda m: m.get("date", "")) return cards def avg(values): vals = [v for v in values if v is not None] return (sum(vals) / len(vals)) if vals else None def fmt(v, suffix=""): return "%.1f%s" % (v, suffix) if v is not None else "—" def group_stats(cards, field): """{значение поля: [всего, success]} по карточкам с непустым полем.""" stats = {} for m in cards: key = m.get(field) or "—" s = stats.setdefault(key, [0, 0]) s[0] += 1 if m.get("result") == "success": s[1] += 1 return stats def render(cards, title): out = ["# Статистика kit-dev-flow — " + title, ""] out.append("Карточек: %d (проектов: %d)" % ( len(cards), len({m.get("project") for m in cards if m.get("project")}))) out.append("") for field, label in [("form", "форма"), ("preset", "пресет"), ("kind", "класс шага")]: groups = group_stats(cards, field) if len(groups) < 2 and field != "form": continue out.append("| %s | прогонов | success |" % label) out.append("|---|---|---|") for key, (total, ok) in sorted(groups.items(), key=lambda kv: -kv[1][0]): out.append("| %s | %d | %d%% |" % (key, total, ok * 100 // total)) out.append("") out.append("Средние: агентов %s, ревизий %s, эскалаций %s, " "длительность %s мин." % ( fmt(avg([num(m, "agents") for m in cards])), fmt(avg([num(m, "revisions") for m in cards])), fmt(avg([num(m, "escalations") for m in cards])), fmt(avg([num(m, "duration_min") for m in cards])))) out.append("") fails = [m for m in cards if m.get("result") in ("fail", "partial")] out.append("Провальных/частичных: %d. Уроки из последних:" % len(fails)) shown = 0 for m in reversed(fails): if shown >= 8: break for lesson in m.get("_lessons", []): out.append("- (%s, %s, %s) %s" % ( m.get("project", "?"), m.get("step", "?"), m.get("form", "?"), lesson)) shown += 1 if shown >= 8: break if not shown: out.append("- (нет провальных карточек в выборке)") return "\n".join(out) + "\n" def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--dir", help="локальный клон репозитория опыта") ap.add_argument("--kind", help="фильтр: класс шага (kind карточки)") ap.add_argument("--project", help="фильтр: проект (по умолчанию все)") ap.add_argument("--days", type=int, help="только последние N дней") ap.add_argument("--last", type=int, help="только последние N карточек") ap.add_argument("--write-summary", action="store_true", help="записать отчёт в _summary.md репозитория") args = ap.parse_args() root = resolve_dir(args.dir) if not os.path.isdir(root): print("ОШИБКА: каталог журнала не найден: %s" % root, file=sys.stderr) print("Настрой flow_exp_dir в ~/.zcode/kit-knowledge.json " "или склонируй репозиторий опыта.", file=sys.stderr) sys.exit(1) cards = load_cards(root, args.kind, args.project, args.days) if args.last: cards = cards[-args.last:] title = args.kind or ("проект " + args.project if args.project else "все проекты") report = render(cards, title) if args.write_summary: path = os.path.join(root, "_summary.md") with open(path, "w", encoding="utf-8", newline="\n") as fh: fh.write(report) print("Записан отчёт: %s" % path) print(report) if __name__ == "__main__": main()