reuse-surface/reuse_surface/reports.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import yaml

from reuse_surface.registry import ROOT, level_at_least, load_index, parse_vector


def _availability_at_most(current: str, maximum: str) -> bool:
    from reuse_surface.registry import LEVEL_ORDERS

    order = LEVEL_ORDERS["availability"]
    return order.index(current) <= order.index(maximum)


def cohort_filters_from_args(args: Any) -> dict[str, str | None]:
    filters: dict[str, str | None] = {
        "discovery_min": getattr(args, "discovery_min", None),
        "availability_min": getattr(args, "availability_min", None),
        "availability_max": getattr(args, "availability_max", None),
        "domain": getattr(args, "domain", None),
    }
    if getattr(args, "planning_min", None):
        filters["discovery_min"] = args.planning_min
        filters["availability_max"] = filters["availability_max"] or "A1"
    if getattr(args, "implementation_min", None):
        filters["availability_min"] = args.implementation_min
    return filters


def select_cohort(
    filters: dict[str, str | None],
    index: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    data = index or load_index()
    matches: list[dict[str, Any]] = []
    for item in data.get("capabilities", []):
        vector = parse_vector(item["vector"])
        if filters.get("discovery_min") and not level_at_least(
            "discovery", vector["discovery"], filters["discovery_min"]
        ):
            continue
        if filters.get("availability_min") and not level_at_least(
            "availability", vector["availability"], filters["availability_min"]
        ):
            continue
        if filters.get("availability_max") and not _availability_at_most(
            vector["availability"], filters["availability_max"]
        ):
            continue
        if filters.get("domain") and item.get("domain") != filters["domain"]:
            continue
        matches.append(item)
    return matches


def format_cohort_markdown(
    matches: list[dict[str, Any]],
    filters: dict[str, str | None],
) -> str:
    lines = ["# Capability cohort report", ""]
    active = {key: value for key, value in filters.items() if value}
    if active:
        lines.append("Filters:")
        for key, value in sorted(active.items()):
            lines.append(f"- `{key}`: `{value}`")
        lines.append("")
    if not matches:
        lines.append("_No capabilities matched._")
        return "\n".join(lines) + "\n"
    lines.append("| ID | Vector | Consumption modes |")
    lines.append("|---|---|---|")
    for item in matches:
        modes = ", ".join(item.get("consumption_modes", []))
        lines.append(f"| `{item['id']}` | {item['vector']} | {modes} |")
    lines.append("")
    lines.append(f"**{len(matches)}** capabilit{'y' if len(matches) == 1 else 'ies'}.")
    return "\n".join(lines) + "\n"


def format_cohort_json(matches: list[dict[str, Any]], filters: dict[str, str | None]) -> str:
    payload = {
        "count": len(matches),
        "filters": {key: value for key, value in filters.items() if value},
        "capabilities": matches,
    }
    return json.dumps(payload, indent=2, sort_keys=True)


def collect_gap_report(
    roster_path: Path,
    *,
    index: dict[str, Any] | None = None,
) -> dict[str, Any]:
    roster = yaml.safe_load(roster_path.read_text(encoding="utf-8"))
    repos = roster.get("repos", [])
    summary = roster.get("summary", {})
    local_index = index or load_index()

    local_by_owner: dict[str, list[str]] = {}
    for row in local_index.get("capabilities", []):
        owner = row.get("owner") or "unknown"
        local_by_owner.setdefault(owner, []).append(row["id"])

    publish_fail = [r for r in repos if r.get("publish_check") == "fail"]
    empty_scaffolds = [
        r for r in repos
        if r.get("status") == "established" and r.get("capability_count", 0) == 0
    ]
    unclassified = [r for r in empty_scaffolds if r.get("capability_status", "pending") == "pending"]
    explicit_none = [r for r in empty_scaffolds if r.get("capability_status") == "none"]
    covered = [r for r in repos if r.get("capability_status") in ("has", "none")]
    seeded = [r for r in repos if r.get("seed_from_reuse_surface")]
    dedup_pending = [
        {
            "slug": owner,
            "local_ids": ids,
        }
        for owner, ids in sorted(local_by_owner.items())
        if owner not in {"reuse-surface", "unknown"}
    ]

    return {
        "roster_path": str(roster_path),
        "summary": summary,
        "publish_fail": [
            {
                "slug": r["slug"],
                "hub_registered": r.get("hub_registered"),
                "publish_note": r.get("publish_note"),
            }
            for r in publish_fail
        ],
        "empty_scaffold_count": len(empty_scaffolds),
        "empty_scaffolds": [r["slug"] for r in empty_scaffolds],
        "unclassified_count": len(unclassified),
        "unclassified": [r["slug"] for r in unclassified],
        "explicit_none_count": len(explicit_none),
        "explicit_none": [r["slug"] for r in explicit_none],
        "coverage_ratio": f"{len(covered)}/{len(repos)}" if repos else "0/0",
        "seeded_repos": [
            {
                "slug": r["slug"],
                "seed_capability_ids": r.get("seed_capability_ids", []),
                "publish_check": r.get("publish_check"),
            }
            for r in seeded
        ],
        "dedup_pending_local_owners": dedup_pending,
        "local_capability_count": len(local_index.get("capabilities", [])),
    }


def format_gap_markdown(report: dict[str, Any]) -> str:
    lines = ["# Registry gap report", ""]
    lines.append(f"**Roster:** `{report['roster_path']}`")
    summary = report.get("summary", {})
    if summary:
        lines.append(
            f"**Workstation:** {summary.get('established', '?')}/"
            f"{summary.get('total', '?')} established; "
            f"publish pass {summary.get('publish_pass', '?')}/"
            f"{summary.get('total', '?')}"
        )
    lines.append("")

    fails = report.get("publish_fail", [])
    lines.append(f"## Publish blocked ({len(fails)})")
    if fails:
        for item in fails:
            note = item.get("publish_note") or ""
            suffix = f" — {note}" if note else ""
            lines.append(f"- `{item['slug']}`{suffix}")
    else:
        lines.append("- none")
    lines.append("")

    dedup = report.get("dedup_pending_local_owners", [])
    lines.append(f"## Local index owner stubs ({len(dedup)})")
    if dedup:
        for item in dedup:
            ids = ", ".join(f"`{cap_id}`" for cap_id in item["local_ids"])
            lines.append(f"- **{item['slug']}:** {ids}")
    else:
        lines.append("- none (owner rows migrated to canonical repos)")
    lines.append("")

    lines.append(f"**Capability coverage:** {report.get('coverage_ratio', '?')} "
                 "(repos with a capability or an explicit no-capability marker)")
    lines.append("")

    unclassified = report.get("unclassified", [])
    lines.append(f"## Unclassified scaffolds ({report.get('unclassified_count', len(unclassified))})")
    if unclassified:
        for slug in unclassified:
            lines.append(f"- `{slug}`")
    else:
        lines.append("- none")
    lines.append("")

    explicit_none = report.get("explicit_none", [])
    lines.append(f"## Explicitly no-capability ({report.get('explicit_none_count', len(explicit_none))})")
    if explicit_none:
        for slug in explicit_none:
            lines.append(f"- `{slug}`")
    else:
        lines.append("- none")
    lines.append("")

    seeded = report.get("seeded_repos", [])
    lines.append(f"## Seed-ready repos ({len(seeded)})")
    for item in seeded:
        publish = item.get("publish_check", "?")
        lines.append(f"- `{item['slug']}` (publish: {publish})")
    lines.append("")

    lines.append(
        f"**Local reuse-surface capabilities:** {report.get('local_capability_count', 0)}"
    )
    return "\n".join(lines) + "\n"


def format_gap_json(report: dict[str, Any]) -> str:
    return json.dumps(report, indent=2, sort_keys=True)


def default_roster_path() -> Path:
    return ROOT / "registry/federation/local-repo-roster.yaml"