backlog-miner/src/fbminer/cli.py - master - sweetebin/hackaton_26

backlog-miner/src/fbminer/cli.py
282 строки13 KB
sweetebin
Трассировка до первоисточника в UI + CLI-инструментарий решений для агента
28 июл 2026, 17:22
28 июл 2026, 17:22b4de763
"""CLI — интерфейс, который описывается в skill.md. Каждая стадия — команда,
статус машиночитаем (fbminer status --json), агент по нему решает, что дальше."""
import argparse
import json
import sys
from pathlib import Path


def main():
    p = argparse.ArgumentParser(prog="fbminer",
                                description="Фидбек → JTBD → приоритизированный бэклог")
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("ingest", help="загрузить исходники в базу")
    sp.add_argument("--interviews", type=Path, help="папка с docx-транскриптами")
    sp.add_argument("--tracker", type=Path, help="csv тасктрекера (тикеты и/или бэклог)")
    sp.add_argument("--tickets-range", default="", help="напр. 153-202")
    sp.add_argument("--backlog-range", default="", help="напр. 203-232")
    sp.add_argument("--context", type=Path, help="xlsx с контекстом продукта")

    sp = sub.add_parser("extract", help="извлечь JTBD-сигналы (LLM)")
    sp.add_argument("--redo", action="store_true", help="снести сигналы и извлечь заново")
    sp.add_argument("--limit", type=int, default=0, help="ограничить число источников (отладка)")

    sp = sub.add_parser("match", help="эмбеддинги + кластеризация в потребности")
    sp.add_argument("--threshold", type=float, default=0.55,
                    help="косинусная дистанция склейки (больше = крупнее кластеры)")

    sub.add_parser("score", help="приоритизация: частота × эмоция × охват")

    sp = sub.add_parser("stories", help="User Stories + критерии приёмки (LLM)")
    sp.add_argument("--top", type=int, default=20)

    sub.add_parser("audit", help="аудит бэклога против потребностей (LLM)")

    sp = sub.add_parser("report", help="итоговый список 10-15 задач + report.md + карта")
    sp.add_argument("--out", default="report.md")

    sp = sub.add_parser("run", help="весь конвейер: extract→match→score→stories→audit→report")
    sp.add_argument("--threshold", type=float, default=0.55)
    sp.add_argument("--out", default="report.md")

    sp = sub.add_parser("status", help="прогресс стадий")
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("serve", help="дашборд (realtime)")
    sp.add_argument("--host", default="0.0.0.0")
    sp.add_argument("--port", type=int, default=8400)

    sp = sub.add_parser("embed-server", help="сервер эмбеддингов (машина с GPU)")
    sp.add_argument("--host", default="0.0.0.0")
    sp.add_argument("--port", type=int, default=8399)

    sp = sub.add_parser("sbertrack",
                        help="фасад MCP SberTrack: check | tools | export | call")
    sp.add_argument("rest", nargs=argparse.REMAINDER)

    # --- инструментарий решений (агент работает с результатами конвейера) ---
    sp = sub.add_parser("tasks", help="итоговые задачи: список с судьбой каждой")
    sp.add_argument("--status", choices=["pending", "approved", "rejected"], default="")
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("task", help="карточка задачи: потребность, сигналы, цитаты, пересечение с бэклогом")
    sp.add_argument("id", type=int)
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("approve", help="принять задачу; после заведения в СберТрек допиши --ref")
    sp.add_argument("id", type=int)
    sp.add_argument("--ref", default="", help="ключ заведённой задачи в СберТреке")

    sp = sub.add_parser("reject", help="отклонить задачу")
    sp.add_argument("id", type=int)
    sp.add_argument("--reason", default="", help="почему отклонена (для защиты решения)")

    sp = sub.add_parser("needs", help="потребности: score и все его множители")
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("signals", help="сигналы; фильтры --need N / --source REF")
    sp.add_argument("--need", type=int, default=0)
    sp.add_argument("--source", default="", help="ref источника (INT-T-001 / HD2026-171)")
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("source", help="первичный документ целиком + его сигналы")
    sp.add_argument("ref")
    sp.add_argument("--json", action="store_true")

    sp = sub.add_parser("verdicts", help="аудит бэклога: confirmed / orphan с обоснованиями")
    sp.add_argument("--json", action="store_true")

    args = p.parse_args()

    if args.cmd == "sbertrack":  # до connect(): фасаду база не нужна
        from .sbertrack import main as sbertrack_main
        return sbertrack_main(args.rest)

    from .db import connect, get_progress
    con = connect()

    if args.cmd == "ingest":
        from . import ingest
        if args.interviews:
            print("интервью:", ingest.ingest_interviews(con, args.interviews))
        if args.tracker:
            print("трекер:", ingest.ingest_tracker(con, args.tracker,
                                                   args.tickets_range, args.backlog_range))
        if args.context:
            print("контекст: строк", ingest.ingest_context(con, args.context))
        for k in ("interview", "ticket", "backlog", "context"):
            n = con.execute("SELECT COUNT(*) c FROM sources WHERE kind=?", (k,)).fetchone()["c"]
            print(f"  в базе {k}: {n}")

    elif args.cmd == "extract":
        from .stages import extract
        print("сигналов:", extract.run(con, redo=args.redo, limit=args.limit))

    elif args.cmd == "match":
        from .stages import match
        print("потребностей:", match.run(con, threshold=args.threshold))

    elif args.cmd == "score":
        from .stages import score
        print("оценено:", score.run(con))

    elif args.cmd == "stories":
        from .stages import stories
        print("stories:", stories.run(con, top=args.top))

    elif args.cmd == "audit":
        from .stages import audit
        print("проверено задач:", audit.run(con))

    elif args.cmd == "report":
        from .stages import report
        print("отчёт:", report.run(con, out_path=args.out))

    elif args.cmd == "run":
        from .stages import audit, extract, match, report, score, stories
        extract.run(con)
        match.run(con, threshold=args.threshold)
        score.run(con)
        stories.run(con)
        audit.run(con)
        print("отчёт:", report.run(con, out_path=args.out))

    elif args.cmd == "status":
        prog = get_progress(con)
        if args.json:
            print(json.dumps(prog, ensure_ascii=False, indent=2))
        else:
            for stage in ("extract", "embed", "match", "score", "stories", "audit", "report"):
                s = prog.get(stage)
                line = (f"{s['status']:>16} {s['done']}/{s['total']} {s['message']}"
                        if s else "—")
                print(f"{stage:>8}: {line}")

    elif args.cmd == "tasks":
        from . import review
        rows = review.list_tasks(con, status=args.status)
        if args.json:
            print(json.dumps(rows, ensure_ascii=False, indent=2))
        else:
            for r in rows:
                over = f" ⇄ {r['backlogRef']}" if r["backlogRef"] else ""
                trk = f" → {r['trackerRef']}" if r["trackerRef"] else ""
                print(f"[{r['id']:>2}] #{r['rank']:<2} {r['status']:<8} {r['action']:<6}"
                      f" {r['title']}{over}{trk}")
                print(f"      потребность {r['needId']} «{r['need']}»"
                      f" (score {r['score']}, сигналов {r['signals']})")

    elif args.cmd == "task":
        from . import review
        t = review.task_detail(con, args.id)
        if not t:
            print(f"задачи id={args.id} нет (см. fbminer tasks)", file=sys.stderr)
            return 1
        if args.json:
            print(json.dumps(t, ensure_ascii=False, indent=2))
        else:
            print(f"#{t['rank']} [{t['status']}] {t['title']}  (действие: {t['action']})")
            print(t["description"], "\n")
            nd = t["need"]
            if nd:
                print(f"Потребность {nd['id']} «{nd['name']}»: {nd['jtbd']}")
                print(f"  score {nd['score']} = частота {nd['freq']} × эмоция "
                      f"{nd['emotion']} × охват {nd['coverage']}"
                      + ("; подтверждена обоими источниками" if nd["confirmed"] else ""))
                if nd["contradictions"]:
                    print(f"  ⚠ {nd['contradictions']}")
            if t["story"]:
                print(f"User Story: {t['story']}")
                for a in t["acceptance"]:
                    print(f"  - {a}")
            if t["backlogOverlap"]:
                o = t["backlogOverlap"]
                print(f"Пересечение с бэклогом: {o['ref']} «{o['title']}»"
                      f" [{o['verdict']}] {o['reason']}")
            print(f"\nСигналы ({len(t['signals'])}):")
            for s in t["signals"]:
                print(f"  [{s['source']}, {s['kind']}, эмоция {s['emotion']}] {s['jtbd']}")
                if s["quote"]:
                    print(f"    «{s['quote']}»")

    elif args.cmd in ("approve", "reject"):
        from . import review
        try:
            res = review.decide(con, args.id,
                                "approved" if args.cmd == "approve" else "rejected",
                                tracker_ref=getattr(args, "ref", ""),
                                reason=getattr(args, "reason", ""))
        except ValueError as e:
            print(e, file=sys.stderr)
            return 1
        print(json.dumps(res, ensure_ascii=False))

    elif args.cmd == "needs":
        from . import review
        rows = review.list_needs(con)
        if args.json:
            print(json.dumps(rows, ensure_ascii=False, indent=2))
        else:
            for n in rows:
                marks = ("✓оба" if n["confirmed"] else "") + \
                        (" ⚠" if n["contradictions"] else "")
                print(f"[{n['id']:>2}] score {str(n['score']):>5}  "
                      f"{n['freq']}={n['fromInterviews']}и+{n['fromTickets']}о "
                      f"эм{n['emotion']} охв{n['coverage']} {marks:<5} {n['name']}")

    elif args.cmd == "signals":
        from . import review
        rows = review.list_signals(con, need_id=args.need, source_ref=args.source)
        if args.json:
            print(json.dumps(rows, ensure_ascii=False, indent=2))
        else:
            for s in rows:
                need = f" → need {s['needId']} «{s['need']}»" if s["needId"] else ""
                print(f"[{s['id']:>3}] [{s['source']}, эмоция {s['emotion']}]"
                      f" {s['jtbd'] or 'N/A'}{need}")

    elif args.cmd == "source":
        from . import review
        s = review.get_source(con, args.ref)
        if not s:
            print(f"источника ref={args.ref} нет", file=sys.stderr)
            return 1
        if args.json:
            print(json.dumps(s, ensure_ascii=False, indent=2))
        else:
            print(f"=== {s['ref']} ({s['kind']}) {s['title']} ===\n")
            print(s["body"])
            if s["signals"]:
                print(f"\n--- сигналы ({len(s['signals'])}) ---")
                for g in s["signals"]:
                    print(f"[{g['id']}] {g['jtbd'] or 'N/A'}"
                          + (f" → need {g['needId']} «{g['need']}»" if g["needId"] else ""))

    elif args.cmd == "verdicts":
        from . import review
        rows = review.list_verdicts(con)
        if args.json:
            print(json.dumps(rows, ensure_ascii=False, indent=2))
        else:
            for v in rows:
                need = f" → need {v['needId']} «{v['need']}»" if v["needId"] else ""
                print(f"{v['verdict']:>9}  {v['ref']} «{v['title']}»{need}")
                if v["reason"]:
                    print(f"           {v['reason']}")

    elif args.cmd == "serve":
        import uvicorn
        from .dashboard import app
        uvicorn.run(app, host=args.host, port=args.port)

    elif args.cmd == "embed-server":
        import uvicorn
        from .embed_server import app
        uvicorn.run(app, host=args.host, port=args.port)

    return 0


if __name__ == "__main__":
    sys.exit(main())