interface:
  display_name: "Security Ownership Map"
  short_description: "Map maintainers, bus factor, and sensitive code ownership"
  default_prompt: "Build a security ownership map for this repository and identify bus-factor risks in sensitive code."

# Neo4j Import Notes

Use these steps when persisting the ownership graph to Neo4j.

## Quick import (LOAD CSV)

1. Copy `people.csv`, `files.csv`, and `edges.csv` into the Neo4j import directory.
2. Run the following Cypher from Neo4j Browser or `cypher-shell`:

```cypher
CREATE CONSTRAINT person_id IF NOT EXISTS FOR (p:Person) REQUIRE p.id IS UNIQUE;
CREATE CONSTRAINT file_id IF NOT EXISTS FOR (f:File) REQUIRE f.id IS UNIQUE;

LOAD CSV WITH HEADERS FROM 'file:///people.csv' AS row
MERGE (p:Person {id: row.person_id})
SET p.name = row.name,
    p.email = row.email,
    p.first_seen = row.first_seen,
    p.last_seen = row.last_seen,
    p.commit_count = toInteger(row.commit_count),
    p.touches = toInteger(row.touches),
    p.sensitive_touches = toFloat(row.sensitive_touches),
    p.primary_tz_offset = CASE row.primary_tz_offset WHEN '' THEN null ELSE row.primary_tz_offset END,
    p.primary_tz_minutes = CASE row.primary_tz_minutes WHEN '' THEN null ELSE toInteger(row.primary_tz_minutes) END,
    p.timezone_offsets = CASE row.timezone_offsets WHEN '' THEN null ELSE row.timezone_offsets END;

LOAD CSV WITH HEADERS FROM 'file:///files.csv' AS row
MERGE (f:File {id: row.file_id})
SET f.path = row.path,
    f.first_seen = row.first_seen,
    f.last_seen = row.last_seen,
    f.commit_count = toInteger(row.commit_count),
    f.touches = toInteger(row.touches),
    f.bus_factor = toInteger(row.bus_factor),
    f.sensitivity_score = toFloat(row.sensitivity_score),
    f.sensitivity_tags = row.sensitivity_tags;

LOAD CSV WITH HEADERS FROM 'file:///edges.csv' AS row
MATCH (p:Person {id: row.person_id})
MATCH (f:File {id: row.file_id})
MERGE (p)-[r:TOUCHES]->(f)
SET r.touches = toInteger(row.touches),
    r.recency_weight = toFloat(row.recency_weight),
    r.first_seen = row.first_seen,
    r.last_seen = row.last_seen,
    r.sensitive_weight = toFloat(row.sensitive_weight);

LOAD CSV WITH HEADERS FROM 'file:///cochange_edges.csv' AS row
MATCH (f1:File {id: row.file_a})
MATCH (f2:File {id: row.file_b})
MERGE (f1)-[r:COCHANGES]->(f2)
SET r.cochange_count = toInteger(row.cochange_count),
    r.jaccard = toFloat(row.jaccard);
```

## Visualization tips

- Use Neo4j Bloom or Browser with `MATCH (p:Person)-[r:TOUCHES]->(f:File) RETURN p,r,f`.
- Filter by `f.sensitivity_score > 0` to highlight security-relevant clusters.
- For Gephi, import `edges.csv` as edges and `files.csv` / `people.csv` as nodes.

#!/usr/bin/env python3
"""Build a security ownership map from git history."""

from __future__ import annotations

import argparse
import csv
import datetime as dt
import fnmatch
import json
import math
import os
import re
import subprocess
import sys
from collections import defaultdict
from pathlib import Path
from typing import Iterable

DEFAULT_SENSITIVE_RULES: list[tuple[str, str, float]] = [
    ("**/auth/**", "auth", 1.0),
    ("**/oauth/**", "auth", 1.0),
    ("**/rbac/**", "auth", 1.0),
    ("**/session/**", "auth", 1.0),
    ("**/token/**", "auth", 1.0),
    ("**/crypto/**", "crypto", 1.0),
    ("**/tls/**", "crypto", 1.0),
    ("**/ssl/**", "crypto", 1.0),
    ("**/secrets/**", "secrets", 1.0),
    ("**/keys/**", "secrets", 1.0),
    ("**/*.pem", "secrets", 1.0),
    ("**/*.key", "secrets", 1.0),
    ("**/*.p12", "secrets", 1.0),
    ("**/*.pfx", "secrets", 1.0),
    ("**/iam/**", "auth", 1.0),
    ("**/sso/**", "auth", 1.0),
]

DEFAULT_AUTHOR_EXCLUDE_REGEXES = [
    "dependabot",
]

DEFAULT_COCHANGE_EXCLUDES = [
    "**/Cargo.lock",
    "**/Cargo.toml",
    "**/package-lock.json",
    "**/yarn.lock",
    "**/pnpm-lock.yaml",
    "**/go.sum",
    "**/go.mod",
    "**/Gemfile.lock",
    "**/Pipfile.lock",
    "**/poetry.lock",
    "**/composer.lock",
    "**/.github/**",
    "**/.gitignore",
    "**/.gitattributes",
    "**/.gitmodules",
    "**/.editorconfig",
    "**/.vscode/**",
    "**/.idea/**",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Build ownership graphs and security ownership summaries from git history."
    )
    parser.add_argument("--repo", default=".", help="Path to the git repo (default: .)")
    parser.add_argument(
        "--out",
        default="ownership-map-out",
        help="Output directory for graph artifacts",
    )
    parser.add_argument("--since", default=None, help="Limit git log to commits since date")
    parser.add_argument("--until", default=None, help="Limit git log to commits until date")
    parser.add_argument(
        "--identity",
        choices=("author", "committer"),
        default="author",
        help="Identity to attribute touches to",
    )
    parser.add_argument(
        "--date-field",
        choices=("author", "committer"),
        default="author",
        help="Date field to use for recency and bucketing",
    )
    parser.add_argument(
        "--include-merges",
        action="store_true",
        help="Include merge commits (excluded by default)",
    )
    parser.add_argument(
        "--half-life-days",
        type=float,
        default=180.0,
        help="Half life for recency weighting",
    )
    parser.add_argument(
        "--sensitive-config",
        default=None,
        help="CSV file with pattern,tag,weight for sensitive paths",
    )
    parser.add_argument(
        "--owner-threshold",
        type=float,
        default=0.5,
        help="Share threshold for hidden owner detection",
    )
    parser.add_argument(
        "--bus-factor-threshold",
        type=int,
        default=1,
        help="Bus factor threshold for hotspots",
    )
    parser.add_argument(
        "--stale-days",
        type=int,
        default=365,
        help="Days since last touch to consider stale",
    )
    parser.add_argument(
        "--min-touches",
        type=int,
        default=1,
        help="Minimum touches to keep an edge",
    )
    parser.add_argument(
        "--emit-commits",
        action="store_true",
        help="Write commit list to commits.jsonl",
    )
    parser.add_argument(
        "--author-exclude-regex",
        action="append",
        default=[],
        help="Regex for author name/email to exclude (repeatable)",
    )
    parser.add_argument(
        "--no-default-author-excludes",
        action="store_true",
        help="Disable default author excludes (dependabot)",
    )
    parser.add_argument(
        "--no-cochange",
        action="store_true",
        help="Disable co-change graph output",
    )
    parser.add_argument(
        "--cochange-max-files",
        type=int,
        default=50,
        help="Ignore commits touching more than this many files for co-change graph",
    )
    parser.add_argument(
        "--cochange-min-count",
        type=int,
        default=2,
        help="Minimum co-change count to keep file-file edge",
    )
    parser.add_argument(
        "--cochange-min-jaccard",
        type=float,
        default=0.05,
        help="Minimum Jaccard similarity to keep file-file edge",
    )
    parser.add_argument(
        "--cochange-exclude",
        action="append",
        default=[],
        help="Glob to exclude from co-change graph (repeatable)",
    )
    parser.add_argument(
        "--no-default-cochange-excludes",
        action="store_true",
        help="Disable default co-change excludes (lockfiles, .github, editor config)",
    )
    parser.add_argument(
        "--no-communities",
        dest="communities",
        action="store_false",
        help="Disable community detection (enabled by default, requires networkx)",
    )
    parser.add_argument(
        "--graphml",
        action="store_true",
        help="Emit ownership.graphml (requires networkx)",
    )
    parser.add_argument(
        "--max-community-files",
        type=int,
        default=50,
        help="Max files listed per community",
    )
    parser.add_argument(
        "--community-top-owners",
        type=int,
        default=5,
        help="Top maintainers saved per community",
    )
    parser.set_defaults(communities=True)
    return parser.parse_args()


def load_sensitive_rules(path: str | None) -> list[tuple[str, str, float]]:
    if not path:
        return list(DEFAULT_SENSITIVE_RULES)
    rules: list[tuple[str, str, float]] = []
    with open(path, "r", encoding="utf-8") as handle:
        for raw in handle:
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            parts = [part.strip() for part in line.split(",")]
            if not parts:
                continue
            pattern = parts[0]
            tag = parts[1] if len(parts) > 1 and parts[1] else "sensitive"
            weight = float(parts[2]) if len(parts) > 2 and parts[2] else 1.0
            rules.append((pattern, tag, weight))
    return rules


def parse_date(value: str) -> dt.datetime:
    parsed = dt.datetime.fromisoformat(value)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=dt.timezone.utc)
    return parsed


def offset_minutes(timestamp: dt.datetime) -> int | None:
    offset = timestamp.utcoffset()
    if offset is None:
        return None
    return int(offset.total_seconds() / 60)


def format_offset(minutes: int) -> str:
    sign = "+" if minutes >= 0 else "-"
    minutes = abs(minutes)
    return f"{sign}{minutes // 60:02d}:{minutes % 60:02d}"


def recency_weighted(now: dt.datetime, when: dt.datetime, half_life_days: float) -> float:
    if half_life_days <= 0:
        return 1.0
    age_days = max(0.0, (now - when).total_seconds() / 86400.0)
    return math.exp(-math.log(2) * age_days / half_life_days)


def match_sensitive(path: str, rules: Iterable[tuple[str, str, float]]) -> dict[str, float]:
    tags: dict[str, float] = defaultdict(float)
    posix = path.replace("\\", "/")
    for pattern, tag, weight in rules:
        patterns = [pattern]
        if pattern.startswith("**/"):
            patterns.append(pattern[3:])
        for candidate in patterns:
            if fnmatch.fnmatchcase(posix, candidate):
                tags[tag] += weight
                break
    return tags


def matches_glob(path: str, pattern: str) -> bool:
    posix = path.replace("\\", "/")
    patterns = [pattern]
    if pattern.startswith("**/"):
        patterns.append(pattern[3:])
    return any(fnmatch.fnmatchcase(posix, candidate) for candidate in patterns)


def is_excluded(path: str, patterns: Iterable[str]) -> bool:
    return any(matches_glob(path, pattern) for pattern in patterns)


def author_excluded(name: str, email: str, patterns: Iterable[re.Pattern[str]]) -> bool:
    if not patterns:
        return False
    haystack = f"{name} {email}".strip()
    return any(pattern.search(haystack) for pattern in patterns)


def compute_community_owners(
    community_files: Iterable[str],
    people: dict[str, dict[str, object]],
    file_people_touches: dict[str, dict[str, int]],
    file_people_recency: dict[str, dict[str, float]],
    file_people_sensitive: dict[str, dict[str, float]],
    top_n: int,
) -> dict[str, object]:
    touches_by_person: dict[str, int] = defaultdict(int)
    recency_by_person: dict[str, float] = defaultdict(float)
    sensitive_by_person: dict[str, float] = defaultdict(float)

    for path in community_files:
        for person, touches in file_people_touches.get(path, {}).items():
            touches_by_person[person] += touches
        for person, recency in file_people_recency.get(path, {}).items():
            recency_by_person[person] += recency
        for person, weight in file_people_sensitive.get(path, {}).items():
            sensitive_by_person[person] += weight

    total_touches = sum(touches_by_person.values())
    total_recency = sum(recency_by_person.values())
    total_sensitive = sum(sensitive_by_person.values())

    ranked = sorted(touches_by_person.items(), key=lambda item: item[1], reverse=True)
    owners = []
    for person_id, touches in ranked[:top_n]:
        recency = recency_by_person.get(person_id, 0.0)
        sensitive = sensitive_by_person.get(person_id, 0.0)
        owners.append(
            {
                "person_id": person_id,
                "name": people.get(person_id, {}).get("name", person_id),
                "touches": touches,
                "touch_share": round(touches / total_touches, 4) if total_touches else 0.0,
                "recency_share": round(recency / total_recency, 4) if total_recency else 0.0,
                "sensitive_share": round(sensitive / total_sensitive, 4)
                if total_sensitive
                else 0.0,
                "primary_tz_offset": people.get(person_id, {}).get("primary_tz_offset", ""),
            }
        )

    return {
        "bus_factor": len(touches_by_person),
        "owner_count": len(touches_by_person),
        "totals": {
            "touches": total_touches,
            "recency_weight": round(total_recency, 6),
            "sensitive_weight": round(total_sensitive, 2),
        },
        "top_maintainers": owners,
    }


def run_git_log(
    repo: str, since: str | None, until: str | None, include_merges: bool
) -> Iterable[list[str]]:
    cmd = [
        "git",
        "-C",
        repo,
        "log",
        "--name-only",
        "--no-renames",
        "--date=iso-strict",
        "--format=---%n%H%n%P%n%an%n%ae%n%ad%n%cn%n%ce%n%cd",
    ]
    if not include_merges:
        cmd.append("--no-merges")
    if since:
        cmd.extend(["--since", since])
    if until:
        cmd.extend(["--until", until])

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    assert proc.stdout is not None

    batch: list[str] = []
    for line in proc.stdout:
        batch.append(line.rstrip("\n"))
        if line.rstrip("\n") == "---" and len(batch) > 1:
            yield batch[:-1]
            batch = ["---"]

    if batch:
        yield batch

    stderr = proc.stderr.read() if proc.stderr else ""
    exit_code = proc.wait()
    if exit_code != 0:
        raise RuntimeError(stderr.strip() or "git log failed")


def iter_commits(lines: Iterable[list[str]]) -> Iterable[tuple[dict[str, object], list[str]]]:
    for chunk in lines:
        if not chunk or chunk[0] != "---":
            continue
        header = chunk[1:9]
        if len(header) < 8:
            continue
        parents = [entry for entry in header[1].split(" ") if entry]
        commit = {
            "hash": header[0],
            "parents": parents,
            "is_merge": len(parents) > 1,
            "author_name": header[2],
            "author_email": header[3],
            "author_date": header[4],
            "committer_name": header[5],
            "committer_email": header[6],
            "committer_date": header[7],
        }
        files = [line for line in chunk[9:] if line.strip()]
        yield commit, files


def ensure_out_dir(path: str) -> Path:
    out_dir = Path(path)
    out_dir.mkdir(parents=True, exist_ok=True)
    return out_dir


def write_csv(path: Path, header: list[str], rows: Iterable[list[str]]) -> None:
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(header)
        for row in rows:
            writer.writerow(row)


def build_ownership_map(args: argparse.Namespace) -> Path:
    now = dt.datetime.now(dt.timezone.utc)
    rules = load_sensitive_rules(args.sensitive_config)
    out_dir = ensure_out_dir(args.out)

    people: dict[str, dict[str, object]] = {}
    files: dict[str, dict[str, object]] = {}
    edges: dict[tuple[str, str], dict[str, object]] = {}
    file_people_touches: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    file_people_recency: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    file_people_sensitive: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    tag_totals: dict[str, float] = defaultdict(float)
    tag_person_totals: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))
    person_timezone_counts: dict[str, dict[int, int]] = defaultdict(lambda: defaultdict(int))
    cochange_counts: dict[tuple[str, str], int] = defaultdict(int)
    cochange_file_commits: dict[str, int] = defaultdict(int)
    cochange_commits_used = 0
    cochange_commits_skipped = 0
    cochange_commits_filtered = 0
    cochange_files_excluded = 0

    commits_path = out_dir / "commits.jsonl"
    commit_handle = None
    if args.emit_commits:
        commit_handle = commits_path.open("w", encoding="utf-8")

    total_commits_seen = 0
    total_commits_included = 0
    commits_excluded_identities = 0
    commits_excluded_merges = 0
    total_edges = 0

    author_exclude_regexes = []
    if not args.no_default_author_excludes:
        author_exclude_regexes.extend(DEFAULT_AUTHOR_EXCLUDE_REGEXES)
    author_exclude_regexes.extend(args.author_exclude_regex)
    author_exclude_patterns = [
        re.compile(pattern, re.IGNORECASE) for pattern in author_exclude_regexes
    ]

    cochange_excludes = []
    if not args.no_default_cochange_excludes:
        cochange_excludes.extend(DEFAULT_COCHANGE_EXCLUDES)
    cochange_excludes.extend(args.cochange_exclude)

    log_lines = run_git_log(args.repo, args.since, args.until, args.include_merges)
    for commit, touched_files in iter_commits(log_lines):
        total_commits_seen += 1

        if commit.get("is_merge") and not args.include_merges:
            commits_excluded_merges += 1
            continue

        identity_name = commit.get(f"{args.identity}_name", "")
        identity_email = commit.get(f"{args.identity}_email", "")
        if author_excluded(
            identity_name,
            identity_email,
            author_exclude_patterns,
        ):
            commits_excluded_identities += 1
            continue

        if not touched_files:
            continue

        total_commits_included += 1
        if commit_handle:
            commit_handle.write(json.dumps({**commit, "files": touched_files}) + "\n")

        identity_name = commit.get(f"{args.identity}_name", "")
        identity_email = commit.get(f"{args.identity}_email", "") or identity_name
        commit_date = parse_date(commit.get(f"{args.date_field}_date", ""))
        recency = recency_weighted(now, commit_date, args.half_life_days)
        tz_minutes = offset_minutes(commit_date)
        if tz_minutes is not None:
            person_timezone_counts[identity_email][tz_minutes] += 1
        unique_files = sorted(set(touched_files))
        if not args.no_cochange and len(unique_files) > 1:
            if len(unique_files) > args.cochange_max_files:
                cochange_commits_skipped += 1
            else:
                filtered_files = [
                    path for path in unique_files if not is_excluded(path, cochange_excludes)
                ]
                excluded = len(unique_files) - len(filtered_files)
                if excluded:
                    cochange_files_excluded += excluded
                if len(filtered_files) < 2:
                    cochange_commits_filtered += 1
                if filtered_files:
                    for path in filtered_files:
                        cochange_file_commits[path] += 1
                if len(filtered_files) >= 2:
                    cochange_commits_used += 1
                    for idx, path in enumerate(filtered_files):
                        for other in filtered_files[idx + 1 :]:
                            cochange_counts[(path, other)] += 1

        person = people.setdefault(
            identity_email,
            {
                "name": identity_name,
                "email": identity_email,
                "first_seen": commit_date,
                "last_seen": commit_date,
                "commit_count": 0,
                "touches": 0,
                "sensitive_touches": 0.0,
            },
        )
        person["commit_count"] = int(person["commit_count"]) + 1
        person["first_seen"] = min(person["first_seen"], commit_date)
        person["last_seen"] = max(person["last_seen"], commit_date)

        for path in touched_files:
            file_entry = files.setdefault(
                path,
                {
                    "path": path,
                    "first_seen": commit_date,
                    "last_seen": commit_date,
                    "commit_count": 0,
                    "touches": 0,
                    "authors": set(),
                    "sensitive_tags": {},
                },
            )
            file_entry["commit_count"] = int(file_entry["commit_count"]) + 1
            file_entry["first_seen"] = min(file_entry["first_seen"], commit_date)
            file_entry["last_seen"] = max(file_entry["last_seen"], commit_date)
            file_entry["touches"] = int(file_entry["touches"]) + 1
            file_entry["authors"].add(identity_email)

            edge = edges.setdefault(
                (identity_email, path),
                {
                    "touches": 0,
                    "first_seen": commit_date,
                    "last_seen": commit_date,
                    "recency_weight": 0.0,
                    "sensitive_weight": 0.0,
                },
            )
            edge["touches"] = int(edge["touches"]) + 1
            edge["first_seen"] = min(edge["first_seen"], commit_date)
            edge["last_seen"] = max(edge["last_seen"], commit_date)
            edge["recency_weight"] = float(edge["recency_weight"]) + recency

            tags = match_sensitive(path, rules)
            if tags:
                file_entry["sensitive_tags"] = tags
                sensitive_weight = sum(tags.values())
                edge["sensitive_weight"] = float(edge["sensitive_weight"]) + sensitive_weight
                person["sensitive_touches"] = float(person["sensitive_touches"]) + sensitive_weight
                file_people_sensitive[path][identity_email] += sensitive_weight
                for tag, weight in tags.items():
                    tag_totals[tag] += weight
                    tag_person_totals[tag][identity_email] += weight

            person["touches"] = int(person["touches"]) + 1
            file_people_touches[path][identity_email] += 1
            file_people_recency[path][identity_email] += recency
            total_edges += 1

    if commit_handle:
        commit_handle.close()

    people_rows = []
    for email, person in sorted(people.items()):
        tz_counts = person_timezone_counts.get(email, {})
        primary_tz_offset = ""
        primary_tz_minutes = ""
        timezone_offsets = ""
        if tz_counts:
            primary_tz_minutes_value = max(tz_counts.items(), key=lambda item: (item[1], item[0]))[
                0
            ]
            primary_tz_offset = format_offset(primary_tz_minutes_value)
            primary_tz_minutes = str(primary_tz_minutes_value)
            timezone_offsets = ";".join(
                f"{format_offset(minutes)}:{count}"
                for minutes, count in sorted(tz_counts.items(), key=lambda item: item[0])
            )
            person["primary_tz_offset"] = primary_tz_offset
        people_rows.append(
            [
                email,
                str(person["name"]),
                email,
                person["first_seen"].isoformat(),
                person["last_seen"].isoformat(),
                str(person["commit_count"]),
                str(person["touches"]),
                f"{person['sensitive_touches']:.2f}",
                primary_tz_offset,
                primary_tz_minutes,
                timezone_offsets,
            ]
        )

    file_rows = []
    for path, file_entry in sorted(files.items()):
        authors = file_entry["authors"]
        bus_factor = len(authors)
        tags = file_entry["sensitive_tags"]
        tag_list = ";".join(sorted(tags.keys()))
        sensitivity_score = sum(tags.values()) if tags else 0.0
        file_rows.append(
            [
                path,
                path,
                file_entry["first_seen"].isoformat(),
                file_entry["last_seen"].isoformat(),
                str(file_entry["commit_count"]),
                str(file_entry["touches"]),
                str(bus_factor),
                f"{sensitivity_score:.2f}",
                tag_list,
            ]
        )

    edge_rows = []
    for (email, path), edge in edges.items():
        if int(edge["touches"]) < args.min_touches:
            continue
        edge_rows.append(
            [
                email,
                path,
                str(edge["touches"]),
                f"{edge['recency_weight']:.6f}",
                edge["first_seen"].isoformat(),
                edge["last_seen"].isoformat(),
                f"{edge['sensitive_weight']:.2f}",
            ]
        )

    cochange_rows: list[list[str]] = []
    if not args.no_cochange:
        for (file_a, file_b), count in cochange_counts.items():
            if count < args.cochange_min_count:
                continue
            commits_a = cochange_file_commits.get(file_a, 0)
            commits_b = cochange_file_commits.get(file_b, 0)
            denom = commits_a + commits_b - count
            if denom <= 0:
                continue
            jaccard = count / denom
            if jaccard < args.cochange_min_jaccard:
                continue
            cochange_rows.append([file_a, file_b, str(count), f"{jaccard:.6f}"])

    write_csv(
        out_dir / "people.csv",
        [
            "person_id",
            "name",
            "email",
            "first_seen",
            "last_seen",
            "commit_count",
            "touches",
            "sensitive_touches",
            "primary_tz_offset",
            "primary_tz_minutes",
            "timezone_offsets",
        ],
        people_rows,
    )
    write_csv(
        out_dir / "files.csv",
        [
            "file_id",
            "path",
            "first_seen",
            "last_seen",
            "commit_count",
            "touches",
            "bus_factor",
            "sensitivity_score",
            "sensitivity_tags",
        ],
        file_rows,
    )
    write_csv(
        out_dir / "edges.csv",
        [
            "person_id",
            "file_id",
            "touches",
            "recency_weight",
            "first_seen",
            "last_seen",
            "sensitive_weight",
        ],
        edge_rows,
    )
    if not args.no_cochange:
        write_csv(
            out_dir / "cochange_edges.csv",
            [
                "file_a",
                "file_b",
                "cochange_count",
                "jaccard",
            ],
            cochange_rows,
        )

    orphaned_sensitive_code = []
    bus_factor_hotspots = []
    for path, file_entry in files.items():
        tags = file_entry["sensitive_tags"]
        if not tags:
            continue
        bus_factor = len(file_entry["authors"])
        last_seen = file_entry["last_seen"]
        age_days = (now - last_seen).days
        top_owner = None
        if path in file_people_touches:
            top_owner = max(file_people_touches[path].items(), key=lambda item: item[1])[0]
        hotspot = {
            "path": path,
            "bus_factor": bus_factor,
            "last_touch": last_seen.isoformat(),
            "sensitivity_tags": sorted(tags.keys()),
            "top_owner": top_owner,
        }
        if bus_factor <= args.bus_factor_threshold:
            bus_factor_hotspots.append(hotspot)
            if age_days >= args.stale_days:
                orphaned_sensitive_code.append(
                    {
                        **hotspot,
                        "last_security_touch": last_seen.isoformat(),
                    }
                )

    hidden_owners = []
    for tag, total in tag_totals.items():
        if total <= 0:
            continue
        person_totals = tag_person_totals[tag]
        if not person_totals:
            continue
        top_email, top_value = max(person_totals.items(), key=lambda item: item[1])
        share = top_value / total
        if share >= args.owner_threshold:
            person_name = people.get(top_email, {}).get("name", top_email)
            hidden_owners.append(
                {
                    "person": top_email,
                    "name": person_name,
                    "controls": f"{share * 100:.0f}% of {tag} code",
                    "category": tag,
                    "share": round(share, 4),
                }
            )

    summary = {
        "generated_at": now.isoformat(),
        "repo": os.path.abspath(args.repo),
        "parameters": {
            "since": args.since,
            "until": args.until,
            "half_life_days": args.half_life_days,
            "bus_factor_threshold": args.bus_factor_threshold,
            "stale_days": args.stale_days,
            "owner_threshold": args.owner_threshold,
            "sensitive_config": args.sensitive_config,
            "identity": args.identity,
            "date_field": args.date_field,
            "include_merges": args.include_merges,
            "cochange_enabled": not args.no_cochange,
            "cochange_max_files": args.cochange_max_files,
            "cochange_min_count": args.cochange_min_count,
            "cochange_min_jaccard": args.cochange_min_jaccard,
            "cochange_default_excludes": not args.no_default_cochange_excludes,
            "cochange_excludes": cochange_excludes,
            "author_default_excludes": not args.no_default_author_excludes,
            "author_exclude_regexes": author_exclude_regexes,
            "community_top_owners": args.community_top_owners,
        },
        "orphaned_sensitive_code": orphaned_sensitive_code,
        "hidden_owners": hidden_owners,
        "bus_factor_hotspots": bus_factor_hotspots,
        "stats": {
            "commits": total_commits_included,
            "commits_seen": total_commits_seen,
            "commits_excluded_identities": commits_excluded_identities,
            "commits_excluded_merges": commits_excluded_merges,
            "edges": total_edges,
            "people": len(people),
            "files": len(files),
            "cochange_pairs_total": len(cochange_counts) if not args.no_cochange else 0,
            "cochange_edges": len(cochange_rows) if not args.no_cochange else 0,
            "cochange_commits_used": cochange_commits_used if not args.no_cochange else 0,
            "cochange_commits_skipped": cochange_commits_skipped if not args.no_cochange else 0,
            "cochange_commits_filtered": cochange_commits_filtered if not args.no_cochange else 0,
            "cochange_files_excluded": cochange_files_excluded if not args.no_cochange else 0,
        },
    }

    with (out_dir / "summary.json").open("w", encoding="utf-8") as handle:
        json.dump(summary, handle, indent=2)

    if args.communities or args.graphml:
        try:
            import networkx as nx
            from networkx.algorithms import bipartite
        except ImportError:
            raise RuntimeError(
                "networkx is required for communities/graphml output. Install with: pip install networkx"
            )
        else:
            graph_bipartite = None
            graph_cochange = None
            person_nodes = set()
            file_nodes = set()
            community_index: dict[str, int] = {}
            community_metadata: list[dict[str, object]] = []

            if args.graphml or (args.communities and (args.no_cochange or not cochange_rows)):
                graph_bipartite = nx.Graph()
                for (email, path), edge in edges.items():
                    if int(edge["touches"]) < args.min_touches:
                        continue
                    graph_bipartite.add_node(email, node_type="person")
                    graph_bipartite.add_node(path, node_type="file")
                    graph_bipartite.add_edge(email, path, weight=float(edge["touches"]))
                    person_nodes.add(email)
                    file_nodes.add(path)

            if not args.no_cochange and cochange_rows:
                graph_cochange = nx.Graph()
                for file_a, file_b, count, jaccard in cochange_rows:
                    graph_cochange.add_edge(
                        file_a,
                        file_b,
                        weight=float(jaccard),
                        count=int(count),
                    )

            if args.communities:
                communities_result = None
                if graph_cochange is not None:
                    communities_result = list(
                        nx.algorithms.community.greedy_modularity_communities(
                            graph_cochange, weight="weight"
                        )
                    )
                elif graph_bipartite is not None and file_nodes:
                    projected = bipartite.weighted_projected_graph(graph_bipartite, file_nodes)
                    communities_result = list(
                        nx.algorithms.community.greedy_modularity_communities(projected)
                    )

                if communities_result is not None:
                    serialized = []
                    for idx, community in enumerate(communities_result, start=1):
                        files_list = sorted(community)
                        owners = compute_community_owners(
                            files_list,
                            people,
                            file_people_touches,
                            file_people_recency,
                            file_people_sensitive,
                            args.community_top_owners,
                        )
                        for path in files_list:
                            community_index[path] = idx
                        entry = {
                            "id": idx,
                            "size": len(files_list),
                            "files": files_list[: args.max_community_files],
                            "maintainers": owners["top_maintainers"],
                            "bus_factor": owners["bus_factor"],
                            "owner_count": owners["owner_count"],
                            "totals": owners["totals"],
                        }
                        serialized.append(entry)
                        metadata = dict(entry)
                        metadata.pop("files", None)
                        community_metadata.append(metadata)
                    with (out_dir / "communities.json").open("w", encoding="utf-8") as handle:
                        json.dump(serialized, handle, indent=2)

            if args.communities:
                for node, community_id in community_index.items():
                    if graph_cochange is not None and node in graph_cochange:
                        graph_cochange.nodes[node]["community_id"] = community_id
                    if graph_bipartite is not None and node in graph_bipartite:
                        graph_bipartite.nodes[node]["community_id"] = community_id

                graph_for_json = graph_cochange or graph_bipartite
                if graph_for_json is not None:
                    try:
                        from networkx.readwrite import json_graph
                    except ImportError:
                        pass
                    else:
                        data = json_graph.node_link_data(graph_for_json, edges="edges")
                        data.setdefault("graph", {})
                        data["graph"]["community_maintainers"] = community_metadata
                        json_name = (
                            "cochange.graph.json"
                            if graph_for_json is graph_cochange
                            else "ownership.graph.json"
                        )
                        with (out_dir / json_name).open("w", encoding="utf-8") as handle:
                            json.dump(data, handle, indent=2)

            if args.graphml:
                if graph_bipartite is not None:
                    nx.write_graphml(graph_bipartite, out_dir / "ownership.graphml")
                if graph_cochange is not None:
                    nx.write_graphml(graph_cochange, out_dir / "cochange.graphml")

    return out_dir


def main() -> int:
    args = parse_args()
    try:
        out_dir = build_ownership_map(args)
    except RuntimeError as exc:
        print(str(exc), file=sys.stderr)
        return 1

    print(f"Ownership map written to {out_dir}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Report monthly maintainers for a file's community."""

from __future__ import annotations

import argparse
import csv
import datetime as dt
import json
import math
import re
import subprocess
import sys
from collections import Counter, defaultdict
from pathlib import Path
from typing import Iterable


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Compute maintainers for a file's community over time."
    )
    parser.add_argument(
        "--data-dir",
        default="ownership-map-out",
        help="Directory containing graph outputs",
    )
    parser.add_argument(
        "--repo",
        default=None,
        help="Git repo path (required if commits.jsonl is missing)",
    )
    parser.add_argument(
        "--file",
        default=None,
        help="File path (exact or substring) to locate community",
    )
    parser.add_argument(
        "--community-id",
        type=int,
        default=None,
        help="Community id to analyze",
    )
    parser.add_argument(
        "--since",
        default=None,
        help="Filter commits since date (ISO or 'YYYY-MM-DD')",
    )
    parser.add_argument(
        "--until",
        default=None,
        help="Filter commits until date (ISO or 'YYYY-MM-DD')",
    )
    parser.add_argument(
        "--identity",
        choices=("author", "committer"),
        default="author",
        help="Identity to attribute touches to",
    )
    parser.add_argument(
        "--date-field",
        choices=("author", "committer"),
        default="author",
        help="Date field to use for bucketing",
    )
    parser.add_argument(
        "--include-merges",
        action="store_true",
        help="Include merge commits (excluded by default)",
    )
    parser.add_argument(
        "--top",
        type=int,
        default=5,
        help="Top maintainers per month",
    )
    parser.add_argument(
        "--bucket",
        choices=("month", "quarter"),
        default="month",
        help="Time bucket for grouping",
    )
    parser.add_argument(
        "--touch-mode",
        choices=("commit", "file"),
        default="commit",
        help="Count one touch per commit or one per file touched",
    )
    parser.add_argument(
        "--window-days",
        type=int,
        default=0,
        help="Use a rolling window of N days ending each month (0 = calendar month only)",
    )
    parser.add_argument(
        "--weight",
        choices=("touches", "recency"),
        default="touches",
        help="Weight touches by recency using exponential decay",
    )
    parser.add_argument(
        "--half-life-days",
        type=float,
        default=180.0,
        help="Half-life days for recency weighting",
    )
    parser.add_argument(
        "--min-share",
        type=float,
        default=0.0,
        help="Minimum share within a month to include a maintainer",
    )
    parser.add_argument(
        "--ignore-author-regex",
        default=None,
        help="Regex to skip authors by name or email (e.g., '(bot|dependabot)')",
    )
    parser.add_argument(
        "--min-touches",
        type=int,
        default=1,
        help="Minimum touches per month to include a maintainer",
    )
    return parser.parse_args()


def parse_date(value: str) -> dt.datetime:
    try:
        parsed = dt.datetime.fromisoformat(value)
    except ValueError:
        parsed = dt.datetime.fromisoformat(value + "T00:00:00")
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=dt.timezone.utc)
    return parsed


def month_key(timestamp: dt.datetime) -> str:
    return timestamp.strftime("%Y-%m")


def quarter_key(timestamp: dt.datetime) -> str:
    quarter = (timestamp.month - 1) // 3 + 1
    return f"{timestamp.year}-Q{quarter}"


def month_end(timestamp: dt.datetime) -> dt.datetime:
    year = timestamp.year
    month = timestamp.month
    if month == 12:
        next_month = dt.datetime(year + 1, 1, 1, tzinfo=dt.timezone.utc)
    else:
        next_month = dt.datetime(year, month + 1, 1, tzinfo=dt.timezone.utc)
    return next_month - dt.timedelta(seconds=1)


def quarter_start(timestamp: dt.datetime) -> dt.datetime:
    quarter = (timestamp.month - 1) // 3
    start_month = quarter * 3 + 1
    return dt.datetime(timestamp.year, start_month, 1, tzinfo=dt.timezone.utc)


def quarter_end(timestamp: dt.datetime) -> dt.datetime:
    start = quarter_start(timestamp)
    end_month = start.month + 2
    end_year = start.year
    if end_month > 12:
        end_month -= 12
        end_year += 1
    end_anchor = dt.datetime(end_year, end_month, 1, tzinfo=dt.timezone.utc)
    return month_end(end_anchor)


def add_months(timestamp: dt.datetime, months: int) -> dt.datetime:
    year = timestamp.year + (timestamp.month - 1 + months) // 12
    month = (timestamp.month - 1 + months) % 12 + 1
    return dt.datetime(year, month, 1, tzinfo=dt.timezone.utc)


def recency_weight(age_days: float, half_life_days: float) -> float:
    if half_life_days <= 0:
        return 1.0
    return math.exp(-age_days / half_life_days)


def read_csv(path: Path) -> Iterable[dict[str, str]]:
    with path.open("r", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        yield from reader


def load_people(data_dir: Path) -> dict[str, dict[str, str]]:
    people_path = data_dir / "people.csv"
    people = {}
    for row in read_csv(people_path):
        people[row.get("person_id", "")] = {
            "name": row.get("name", ""),
            "email": row.get("email", ""),
            "primary_tz_offset": row.get("primary_tz_offset", ""),
        }
    return people


def load_graph_json(data_dir: Path) -> dict[str, object] | None:
    cochange_path = data_dir / "cochange.graph.json"
    ownership_path = data_dir / "ownership.graph.json"
    if cochange_path.exists():
        return json.loads(cochange_path.read_text(encoding="utf-8"))
    if ownership_path.exists():
        return json.loads(ownership_path.read_text(encoding="utf-8"))
    return None


def find_file_node(nodes: list[dict[str, object]], query: str) -> dict[str, object]:
    exact = [node for node in nodes if node.get("id") == query]
    if exact:
        return exact[0]
    contains = [node for node in nodes if query in str(node.get("id", ""))]
    if len(contains) == 1:
        return contains[0]
    if not contains:
        raise ValueError(f"File not found in graph: {query}")
    candidates = ", ".join(str(node.get("id")) for node in contains[:10])
    raise ValueError(f"Multiple matches for file {query}: {candidates}")


def load_community_files(
    data_dir: Path, file_query: str | None, community_id: int | None
) -> tuple[int, list[str]]:
    graph = load_graph_json(data_dir)
    if graph:
        nodes = graph.get("nodes", [])
        if file_query:
            node = find_file_node(nodes, file_query)
            community_id = int(node.get("community_id", -1))
        if community_id is None:
            raise ValueError("Provide --file or --community-id")
        files = [node.get("id") for node in nodes if node.get("community_id") == community_id]
        files = [entry for entry in files if entry]
        if not files:
            raise ValueError(f"No files found for community {community_id}")
        return community_id, files

    communities_path = data_dir / "communities.json"
    if not communities_path.exists():
        raise FileNotFoundError("Missing graph json and communities.json")
    communities = json.loads(communities_path.read_text(encoding="utf-8"))
    if file_query:
        for entry in communities:
            files = entry.get("files", [])
            if any(file_query == f or file_query in f for f in files):
                return int(entry.get("id", -1)), list(files)
        raise ValueError("File not found in communities.json (list may be truncated)")
    if community_id is None:
        raise ValueError("Provide --file or --community-id")
    for entry in communities:
        if int(entry.get("id", -1)) == community_id:
            return community_id, list(entry.get("files", []))
    raise ValueError(f"Community id not found: {community_id}")


def iter_commits_from_json(
    commits_path: Path,
    since: dt.datetime | None,
    until: dt.datetime | None,
    date_field: str,
) -> Iterable[dict[str, object]]:
    with commits_path.open("r", encoding="utf-8") as handle:
        for line in handle:
            entry = json.loads(line)
            author_date = entry.get("author_date") or entry.get("date")
            committer_date = entry.get("committer_date")
            if author_date:
                author_dt = parse_date(author_date)
            else:
                author_dt = None
            if committer_date:
                committer_dt = parse_date(committer_date)
            else:
                committer_dt = None
            if date_field == "committer":
                commit_date = committer_dt or author_dt
            else:
                commit_date = author_dt or committer_dt
            if commit_date is None:
                continue
            if since and commit_date < since:
                continue
            if until and commit_date > until:
                continue
            yield {
                "hash": entry.get("hash", ""),
                "parents": entry.get("parents", []),
                "is_merge": entry.get("is_merge", False),
                "author_name": entry.get("author_name", ""),
                "author_email": entry.get("author_email", ""),
                "author_date": author_date,
                "committer_name": entry.get("committer_name", ""),
                "committer_email": entry.get("committer_email", ""),
                "committer_date": committer_date,
                "files": entry.get("files", []),
            }


def iter_commits_from_git(
    repo: str, since: str | None, until: str | None, include_merges: bool
) -> Iterable[dict[str, object]]:
    cmd = [
        "git",
        "-C",
        repo,
        "log",
        "--name-only",
        "--no-renames",
        "--date=iso-strict",
        "--format=---%n%H%n%P%n%an%n%ae%n%ad%n%cn%n%ce%n%cd",
    ]
    if not include_merges:
        cmd.append("--no-merges")
    if since:
        cmd.extend(["--since", since])
    if until:
        cmd.extend(["--until", until])

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    assert proc.stdout is not None

    block: list[str] = []
    for line in proc.stdout:
        line = line.rstrip("\n")
        if line == "---":
            if block:
                yield from parse_git_block(block)
                block = []
        else:
            block.append(line)
    if block:
        yield from parse_git_block(block)

    stderr = proc.stderr.read() if proc.stderr else ""
    exit_code = proc.wait()
    if exit_code != 0:
        raise RuntimeError(stderr.strip() or "git log failed")


def parse_git_block(block: list[str]) -> Iterable[dict[str, object]]:
    if len(block) < 8:
        return []
    commit_hash = block[0]
    parents = [entry for entry in block[1].split(" ") if entry]
    author_name = block[2]
    author_email = block[3]
    author_date = block[4]
    committer_name = block[5]
    committer_email = block[6]
    committer_date = block[7]
    files = [line for line in block[8:] if line]
    return [
        {
            "hash": commit_hash,
            "parents": parents,
            "is_merge": len(parents) > 1,
            "author_name": author_name,
            "author_email": author_email,
            "author_date": author_date,
            "committer_name": committer_name,
            "committer_email": committer_email,
            "committer_date": committer_date,
            "files": files,
        }
    ]


def main() -> int:
    args = parse_args()
    data_dir = Path(args.data_dir)
    if not data_dir.exists():
        print(f"Data directory not found: {data_dir}", file=sys.stderr)
        return 1

    since = parse_date(args.since) if args.since else None
    until = parse_date(args.until) if args.until else None

    try:
        community_id, community_files = load_community_files(data_dir, args.file, args.community_id)
    except (ValueError, FileNotFoundError) as exc:
        print(str(exc), file=sys.stderr)
        return 2

    people = load_people(data_dir)

    ignore_re = re.compile(args.ignore_author_regex) if args.ignore_author_regex else None

    commits_path = data_dir / "commits.jsonl"
    if commits_path.exists():
        commit_iter = iter_commits_from_json(commits_path, since, until, args.date_field)
    else:
        if not args.repo:
            print("--repo is required when commits.jsonl is missing", file=sys.stderr)
            return 2
        commit_iter = iter_commits_from_git(args.repo, args.since, args.until, args.include_merges)

    commit_rows: list[tuple[dt.datetime, str, int, str, str]] = []
    for commit in commit_iter:
        if commit.get("is_merge") and not args.include_merges:
            continue
        files = commit.get("files", [])
        in_community = sum(1 for path in files if path in community_files)
        if in_community == 0:
            continue
        identity_name = commit.get(f"{args.identity}_name", "")
        identity_email = commit.get(f"{args.identity}_email", "")
        date_value = commit.get(f"{args.date_field}_date")
        if not date_value:
            print(
                "Missing committer fields in commits.jsonl. Re-run build or pass --repo.",
                file=sys.stderr,
            )
            return 2
        commit_date = parse_date(date_value)
        person_id = identity_email or identity_name
        if ignore_re and ignore_re.search(identity_name or ""):
            continue
        if ignore_re and ignore_re.search(identity_email or ""):
            continue
        touches = 1 if args.touch_mode == "commit" else in_community
        commit_rows.append((commit_date, person_id, touches, identity_name, identity_email))
        if person_id not in people:
            people[person_id] = {
                "name": identity_name,
                "email": identity_email,
                "primary_tz_offset": "",
            }

    if not commit_rows:
        print("No commits touching community files for the selected window.", file=sys.stderr)
        return 0

    commit_rows.sort(key=lambda row: row[0])
    period_counts: dict[str, Counter[str]] = defaultdict(Counter)
    period_totals: dict[str, float] = defaultdict(float)

    min_date = commit_rows[0][0]
    max_date = commit_rows[-1][0]
    if args.bucket == "quarter":
        period_cursor = quarter_start(min_date)
        period_end_anchor = quarter_start(max_date)
        step_months = 3
        key_func = quarter_key
        end_func = quarter_end
    else:
        period_cursor = dt.datetime(min_date.year, min_date.month, 1, tzinfo=dt.timezone.utc)
        period_end_anchor = dt.datetime(max_date.year, max_date.month, 1, tzinfo=dt.timezone.utc)
        step_months = 1
        key_func = month_key
        end_func = month_end

    while period_cursor <= period_end_anchor:
        bucket_end = end_func(period_cursor)
        bucket_key = key_func(bucket_end)
        if args.window_days > 0:
            window_start = bucket_end - dt.timedelta(days=args.window_days)

            def in_bucket(commit_date: dt.datetime) -> bool:
                return window_start <= commit_date <= bucket_end
        else:
            if args.bucket == "quarter":
                bucket_start = quarter_start(period_cursor)

                def in_bucket(commit_date: dt.datetime) -> bool:
                    return bucket_start <= commit_date <= bucket_end
            else:

                def in_bucket(commit_date: dt.datetime) -> bool:
                    return (
                        commit_date.year == bucket_end.year
                        and commit_date.month == bucket_end.month
                    )

        for commit_date, person_id, touches, _name, _email in commit_rows:
            if not in_bucket(commit_date):
                continue
            weight = 1.0
            if args.weight == "recency":
                age_days = (bucket_end - commit_date).total_seconds() / 86400.0
                weight = recency_weight(age_days, args.half_life_days)
            contribution = touches * weight
            period_counts[bucket_key][person_id] += contribution
            period_totals[bucket_key] += contribution

        period_cursor = add_months(period_cursor, step_months)

    writer = csv.writer(sys.stdout)
    writer.writerow(
        [
            "period",
            "rank",
            "name",
            "email",
            "primary_tz_offset",
            "community_touches",
            "touch_share",
        ]
    )

    for period in sorted(period_counts.keys()):
        total = period_totals[period]
        ranked = sorted(period_counts[period].items(), key=lambda item: item[1], reverse=True)
        rank = 0
        for person_id, touches in ranked:
            if touches < args.min_touches:
                continue
            share = touches / total if total else 0.0
            if share < args.min_share:
                continue
            rank += 1
            if rank > args.top:
                break
            person = people.get(person_id, {})
            if args.weight == "recency":
                touches_value = f"{touches:.4f}"
            else:
                touches_value = f"{touches:.0f}"
            writer.writerow(
                [
                    period,
                    rank,
                    person.get("name", ""),
                    person.get("email", person_id),
                    person.get("primary_tz_offset", ""),
                    touches_value,
                    f"{share:.4f}",
                ]
            )

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Query ownership-map outputs without loading everything into an LLM context."""

from __future__ import annotations

import argparse
import csv
import json
import sys
from collections import defaultdict
from pathlib import Path
from typing import Iterable


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Query ownership-map outputs with bounded JSON results."
    )
    parser.add_argument(
        "--data-dir",
        default="ownership-map-out",
        help="Directory containing people.csv, files.csv, edges.csv",
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    people = subparsers.add_parser("people", help="List people")
    people.add_argument("--limit", type=int, default=20)
    people.add_argument("--sort", default="touches")
    people.add_argument("--email-contains", default=None)
    people.add_argument("--min-touches", type=int, default=0)
    people.add_argument("--min-sensitive", type=float, default=0.0)

    files = subparsers.add_parser("files", help="List files")
    files.add_argument("--limit", type=int, default=20)
    files.add_argument("--sort", default="sensitivity_score")
    files.add_argument("--path-contains", default=None)
    files.add_argument("--tag", default=None)
    files.add_argument("--bus-factor-max", type=int, default=None)
    files.add_argument("--sensitivity-min", type=float, default=0.0)

    person = subparsers.add_parser("person", help="Show person details and top files")
    person.add_argument("--person", required=True, help="Exact email or substring")
    person.add_argument("--limit", type=int, default=20)
    person.add_argument("--sort", default="touches")

    file_cmd = subparsers.add_parser("file", help="Show file details and top people")
    file_cmd.add_argument("--file", required=True, help="Exact path or substring")
    file_cmd.add_argument("--limit", type=int, default=20)
    file_cmd.add_argument("--sort", default="touches")

    cochange = subparsers.add_parser("cochange", help="List co-change neighbors for a file")
    cochange.add_argument("--file", required=True, help="Exact path or substring")
    cochange.add_argument("--limit", type=int, default=20)
    cochange.add_argument("--sort", default="jaccard")
    cochange.add_argument("--min-jaccard", type=float, default=0.0)
    cochange.add_argument("--min-count", type=int, default=1)

    tag = subparsers.add_parser("tag", help="Show top people/files for a sensitive tag")
    tag.add_argument("--tag", required=True)
    tag.add_argument("--limit", type=int, default=20)

    summary = subparsers.add_parser("summary", help="Show summary.json sections")
    summary.add_argument("--section", default=None)

    communities = subparsers.add_parser("communities", help="List communities")
    communities.add_argument("--limit", type=int, default=10)
    communities.add_argument("--id", type=int, default=None)

    community = subparsers.add_parser("community", help="Show community maintainers")
    community.add_argument("--id", type=int, required=True)
    community.add_argument("--include-files", action="store_true")
    community.add_argument("--file-limit", type=int, default=50)

    return parser.parse_args()


def to_int(value: str) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def to_float(value: str) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def read_csv(path: Path) -> Iterable[dict[str, str]]:
    with path.open("r", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        yield from reader


def load_people(data_dir: Path) -> list[dict[str, object]]:
    people_path = data_dir / "people.csv"
    people = []
    for row in read_csv(people_path):
        person = dict(row)
        person["touches"] = to_int(row.get("touches", "0"))
        person["commit_count"] = to_int(row.get("commit_count", "0"))
        person["sensitive_touches"] = to_float(row.get("sensitive_touches", "0"))
        people.append(person)
    return people


def load_files(data_dir: Path) -> list[dict[str, object]]:
    files_path = data_dir / "files.csv"
    files = []
    for row in read_csv(files_path):
        file_entry = dict(row)
        file_entry["touches"] = to_int(row.get("touches", "0"))
        file_entry["commit_count"] = to_int(row.get("commit_count", "0"))
        file_entry["bus_factor"] = to_int(row.get("bus_factor", "0"))
        file_entry["sensitivity_score"] = to_float(row.get("sensitivity_score", "0"))
        tags = row.get("sensitivity_tags", "")
        file_entry["sensitivity_tags"] = [tag for tag in tags.split(";") if tag]
        files.append(file_entry)
    return files


def load_summary(data_dir: Path) -> dict[str, object]:
    summary_path = data_dir / "summary.json"
    with summary_path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def load_communities(data_dir: Path) -> list[dict[str, object]]:
    communities_path = data_dir / "communities.json"
    if not communities_path.exists():
        raise FileNotFoundError("communities.json not found; rerun build with --communities")
    with communities_path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def load_cochange_edges(data_dir: Path) -> Iterable[dict[str, object]]:
    edges_path = data_dir / "cochange_edges.csv"
    if not edges_path.exists():
        raise FileNotFoundError("cochange_edges.csv not found; rerun build without --no-cochange")
    for row in read_csv(edges_path):
        yield {
            "file_a": row.get("file_a"),
            "file_b": row.get("file_b"),
            "cochange_count": to_int(row.get("cochange_count", "0")),
            "jaccard": to_float(row.get("jaccard", "0")),
        }


def select_single(records: list[dict[str, object]], key: str, query: str) -> dict[str, object]:
    exact = [record for record in records if str(record.get(key, "")) == query]
    if exact:
        return exact[0]
    contains = [record for record in records if query in str(record.get(key, ""))]
    if len(contains) == 1:
        return contains[0]
    if not contains:
        raise ValueError(f"No match for {query}")
    candidates = [str(record.get(key, "")) for record in contains[:10]]
    raise ValueError(f"Multiple matches for {query}: {', '.join(candidates)}")


def top_edges_for_person(data_dir: Path, person_id: str) -> list[dict[str, object]]:
    edges_path = data_dir / "edges.csv"
    results = []
    for row in read_csv(edges_path):
        if row.get("person_id") != person_id:
            continue
        results.append(
            {
                "file_id": row.get("file_id"),
                "touches": to_int(row.get("touches", "0")),
                "recency_weight": to_float(row.get("recency_weight", "0")),
                "sensitive_weight": to_float(row.get("sensitive_weight", "0")),
                "last_seen": row.get("last_seen"),
            }
        )
    return results


def top_edges_for_file(data_dir: Path, file_id: str) -> list[dict[str, object]]:
    edges_path = data_dir / "edges.csv"
    results = []
    for row in read_csv(edges_path):
        if row.get("file_id") != file_id:
            continue
        results.append(
            {
                "person_id": row.get("person_id"),
                "touches": to_int(row.get("touches", "0")),
                "recency_weight": to_float(row.get("recency_weight", "0")),
                "sensitive_weight": to_float(row.get("sensitive_weight", "0")),
                "last_seen": row.get("last_seen"),
            }
        )
    return results


def sort_records(records: list[dict[str, object]], key: str) -> list[dict[str, object]]:
    return sorted(records, key=lambda item: item.get(key, 0), reverse=True)


def handle_people(args: argparse.Namespace, data_dir: Path) -> None:
    people = load_people(data_dir)
    if args.email_contains:
        people = [p for p in people if args.email_contains in p.get("email", "")]
    people = [p for p in people if p["touches"] >= args.min_touches]
    people = [p for p in people if p["sensitive_touches"] >= args.min_sensitive]
    people = sort_records(people, args.sort)[: args.limit]
    payload = [
        {
            "person_id": p.get("person_id"),
            "name": p.get("name"),
            "email": p.get("email"),
            "touches": p.get("touches"),
            "commit_count": p.get("commit_count"),
            "sensitive_touches": p.get("sensitive_touches"),
            "primary_tz_offset": p.get("primary_tz_offset"),
        }
        for p in people
    ]
    print(json.dumps(payload, indent=2))


def handle_files(args: argparse.Namespace, data_dir: Path) -> None:
    files = load_files(data_dir)
    if args.path_contains:
        files = [f for f in files if args.path_contains in f.get("path", "")]
    if args.tag:
        files = [f for f in files if args.tag in f.get("sensitivity_tags", [])]
    if args.bus_factor_max is not None:
        files = [f for f in files if f["bus_factor"] <= args.bus_factor_max]
    files = [f for f in files if f["sensitivity_score"] >= args.sensitivity_min]
    files = sort_records(files, args.sort)[: args.limit]
    payload = [
        {
            "file_id": f.get("file_id"),
            "path": f.get("path"),
            "touches": f.get("touches"),
            "bus_factor": f.get("bus_factor"),
            "sensitivity_score": f.get("sensitivity_score"),
            "sensitivity_tags": f.get("sensitivity_tags"),
            "last_seen": f.get("last_seen"),
        }
        for f in files
    ]
    print(json.dumps(payload, indent=2))


def handle_person(args: argparse.Namespace, data_dir: Path) -> None:
    people = load_people(data_dir)
    person = select_single(people, "person_id", args.person)
    files = load_files(data_dir)
    file_map = {f["file_id"]: f for f in files}
    edges = top_edges_for_person(data_dir, person["person_id"])
    edges = sort_records(edges, args.sort)[: args.limit]
    payload = {
        "person": {
            "person_id": person.get("person_id"),
            "name": person.get("name"),
            "email": person.get("email"),
            "touches": person.get("touches"),
            "commit_count": person.get("commit_count"),
            "sensitive_touches": person.get("sensitive_touches"),
            "primary_tz_offset": person.get("primary_tz_offset"),
            "timezone_offsets": person.get("timezone_offsets"),
        },
        "top_files": [
            {
                "file_id": edge.get("file_id"),
                "path": file_map.get(edge.get("file_id"), {}).get("path"),
                "touches": edge.get("touches"),
                "recency_weight": edge.get("recency_weight"),
                "sensitive_weight": edge.get("sensitive_weight"),
                "last_seen": edge.get("last_seen"),
                "sensitivity_tags": file_map.get(edge.get("file_id"), {}).get("sensitivity_tags"),
            }
            for edge in edges
        ],
    }
    print(json.dumps(payload, indent=2))


def handle_file(args: argparse.Namespace, data_dir: Path) -> None:
    files = load_files(data_dir)
    file_entry = select_single(files, "file_id", args.file)
    people = load_people(data_dir)
    people_map = {p["person_id"]: p for p in people}
    edges = top_edges_for_file(data_dir, file_entry["file_id"])
    edges = sort_records(edges, args.sort)[: args.limit]
    payload = {
        "file": {
            "file_id": file_entry.get("file_id"),
            "path": file_entry.get("path"),
            "touches": file_entry.get("touches"),
            "bus_factor": file_entry.get("bus_factor"),
            "sensitivity_score": file_entry.get("sensitivity_score"),
            "sensitivity_tags": file_entry.get("sensitivity_tags"),
            "last_seen": file_entry.get("last_seen"),
        },
        "top_people": [
            {
                "person_id": edge.get("person_id"),
                "name": people_map.get(edge.get("person_id"), {}).get("name"),
                "email": people_map.get(edge.get("person_id"), {}).get("email"),
                "touches": edge.get("touches"),
                "recency_weight": edge.get("recency_weight"),
                "sensitive_weight": edge.get("sensitive_weight"),
                "primary_tz_offset": people_map.get(edge.get("person_id"), {}).get(
                    "primary_tz_offset"
                ),
            }
            for edge in edges
        ],
    }
    print(json.dumps(payload, indent=2))


def handle_cochange(args: argparse.Namespace, data_dir: Path) -> None:
    files = load_files(data_dir)
    file_entry = select_single(files, "file_id", args.file)

    neighbors = []
    for row in load_cochange_edges(data_dir):
        file_a = row.get("file_a")
        file_b = row.get("file_b")
        if file_a == file_entry["file_id"]:
            other = file_b
        elif file_b == file_entry["file_id"]:
            other = file_a
        else:
            continue

        if row["cochange_count"] < args.min_count:
            continue
        if row["jaccard"] < args.min_jaccard:
            continue

        neighbors.append(
            {
                "file_id": other,
                "path": other,
                "cochange_count": row["cochange_count"],
                "jaccard": row["jaccard"],
            }
        )

    neighbors = sort_records(neighbors, args.sort)[: args.limit]
    payload = {
        "file": {
            "file_id": file_entry.get("file_id"),
            "path": file_entry.get("path"),
        },
        "neighbors": neighbors,
    }
    print(json.dumps(payload, indent=2))


def handle_tag(args: argparse.Namespace, data_dir: Path) -> None:
    files = load_files(data_dir)
    tagged_files = [f for f in files if args.tag in f.get("sensitivity_tags", [])]
    tagged_ids = {f["file_id"] for f in tagged_files}

    person_touch = defaultdict(int)
    edges_path = data_dir / "edges.csv"
    for row in read_csv(edges_path):
        if row.get("file_id") not in tagged_ids:
            continue
        person_touch[row.get("person_id")] += to_int(row.get("touches", "0"))

    people = load_people(data_dir)
    people_map = {p["person_id"]: p for p in people}
    top_people = [
        {
            "person_id": person_id,
            "name": people_map.get(person_id, {}).get("name"),
            "email": people_map.get(person_id, {}).get("email"),
            "touches": touches,
        }
        for person_id, touches in person_touch.items()
    ]
    top_people = sorted(top_people, key=lambda item: item.get("touches", 0), reverse=True)[
        : args.limit
    ]

    top_files = sorted(tagged_files, key=lambda item: item.get("touches", 0), reverse=True)[
        : args.limit
    ]

    payload = {
        "tag": args.tag,
        "top_people": top_people,
        "top_files": [
            {
                "file_id": entry.get("file_id"),
                "path": entry.get("path"),
                "touches": entry.get("touches"),
                "bus_factor": entry.get("bus_factor"),
            }
            for entry in top_files
        ],
    }
    print(json.dumps(payload, indent=2))


def handle_summary(args: argparse.Namespace, data_dir: Path) -> None:
    summary = load_summary(data_dir)
    if args.section:
        if args.section not in summary:
            raise ValueError(f"Section not found: {args.section}")
        payload = summary[args.section]
    else:
        payload = summary
    print(json.dumps(payload, indent=2))


def handle_communities(args: argparse.Namespace, data_dir: Path) -> None:
    communities = load_communities(data_dir)
    if args.id is not None:
        matches = [entry for entry in communities if entry.get("id") == args.id]
        if not matches:
            raise ValueError(f"Community id not found: {args.id}")
        payload = matches[0]
    else:
        payload = sorted(communities, key=lambda item: item.get("size", 0), reverse=True)[
            : args.limit
        ]
    print(json.dumps(payload, indent=2))


def handle_community(args: argparse.Namespace, data_dir: Path) -> None:
    communities = load_communities(data_dir)
    matches = [entry for entry in communities if entry.get("id") == args.id]
    if not matches:
        raise ValueError(f"Community id not found: {args.id}")
    entry = dict(matches[0])
    files = entry.pop("files", [])
    payload = entry
    if args.include_files:
        payload["files"] = files[: args.file_limit]
        payload["files_truncated"] = len(files) > args.file_limit
    print(json.dumps(payload, indent=2))


def main() -> int:
    args = parse_args()
    data_dir = Path(args.data_dir)
    if not data_dir.exists():
        print(f"Data directory not found: {data_dir}", file=sys.stderr)
        return 1

    try:
        if args.command == "people":
            handle_people(args, data_dir)
        elif args.command == "files":
            handle_files(args, data_dir)
        elif args.command == "person":
            handle_person(args, data_dir)
        elif args.command == "file":
            handle_file(args, data_dir)
        elif args.command == "cochange":
            handle_cochange(args, data_dir)
        elif args.command == "tag":
            handle_tag(args, data_dir)
        elif args.command == "summary":
            handle_summary(args, data_dir)
        elif args.command == "communities":
            handle_communities(args, data_dir)
        elif args.command == "community":
            handle_community(args, data_dir)
        else:
            raise ValueError(f"Unknown command: {args.command}")
    except (FileNotFoundError, ValueError) as exc:
        print(str(exc), file=sys.stderr)
        return 2

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""One-shot runner for building the security ownership map."""

from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import Path


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run build_ownership_map.py with sensible defaults."
    )
    parser.add_argument("--repo", default=".", help="Path to the git repo (default: .)")
    parser.add_argument(
        "--out",
        default="ownership-map-out",
        help="Output directory for graph artifacts",
    )
    parser.add_argument("--since", default=None, help="Limit git log to commits since date")
    parser.add_argument("--until", default=None, help="Limit git log to commits until date")
    parser.add_argument(
        "--identity",
        choices=("author", "committer"),
        default="author",
        help="Identity to attribute touches to",
    )
    parser.add_argument(
        "--date-field",
        choices=("author", "committer"),
        default="author",
        help="Date field to use for recency and bucketing",
    )
    parser.add_argument(
        "--include-merges",
        action="store_true",
        help="Include merge commits (excluded by default)",
    )
    parser.add_argument(
        "--emit-commits",
        action="store_true",
        help="Write commit list to commits.jsonl",
    )
    parser.add_argument(
        "--author-exclude-regex",
        action="append",
        default=[],
        help="Regex for author name/email to exclude (repeatable)",
    )
    parser.add_argument(
        "--no-default-author-excludes",
        action="store_true",
        help="Disable default author excludes (dependabot)",
    )
    parser.add_argument(
        "--graphml",
        action="store_true",
        help="Emit GraphML outputs",
    )
    parser.add_argument(
        "--sensitive-config",
        default=None,
        help="CSV file with pattern,tag,weight for sensitive paths",
    )
    parser.add_argument(
        "--cochange-max-files",
        type=int,
        default=50,
        help="Ignore commits touching more than this many files for co-change graph",
    )
    parser.add_argument(
        "--cochange-min-count",
        type=int,
        default=2,
        help="Minimum co-change count to keep file-file edge",
    )
    parser.add_argument(
        "--cochange-min-jaccard",
        type=float,
        default=0.05,
        help="Minimum Jaccard similarity to keep file-file edge",
    )
    parser.add_argument(
        "--cochange-exclude",
        action="append",
        default=[],
        help="Glob to exclude from co-change graph (repeatable)",
    )
    parser.add_argument(
        "--no-default-cochange-excludes",
        action="store_true",
        help="Disable default co-change excludes (lockfiles, .github, editor config)",
    )
    parser.add_argument(
        "--community-top-owners",
        type=int,
        default=5,
        help="Top maintainers saved per community",
    )
    parser.add_argument(
        "--bus-factor-threshold",
        type=int,
        default=1,
        help="Bus factor threshold for hotspots",
    )
    parser.add_argument(
        "--stale-days",
        type=int,
        default=365,
        help="Days since last touch to consider stale",
    )
    parser.add_argument(
        "--owner-threshold",
        type=float,
        default=0.5,
        help="Share threshold for hidden owner detection",
    )
    parser.add_argument(
        "--no-cochange",
        action="store_true",
        help="Disable co-change graph output",
    )
    parser.add_argument(
        "--no-communities",
        action="store_true",
        help="Disable community detection (not recommended)",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    try:
        import networkx  # noqa: F401
    except ImportError:
        print("networkx is required. Install with: pip install networkx", file=sys.stderr)
        return 2

    script_path = Path(__file__).resolve().parent / "build_ownership_map.py"
    cmd = [
        sys.executable,
        str(script_path),
        "--repo",
        args.repo,
        "--out",
        args.out,
        "--identity",
        args.identity,
        "--date-field",
        args.date_field,
        "--cochange-max-files",
        str(args.cochange_max_files),
        "--cochange-min-count",
        str(args.cochange_min_count),
        "--cochange-min-jaccard",
        str(args.cochange_min_jaccard),
        "--community-top-owners",
        str(args.community_top_owners),
        "--bus-factor-threshold",
        str(args.bus_factor_threshold),
        "--stale-days",
        str(args.stale_days),
        "--owner-threshold",
        str(args.owner_threshold),
    ]

    if args.since:
        cmd.extend(["--since", args.since])
    if args.until:
        cmd.extend(["--until", args.until])
    if args.include_merges:
        cmd.append("--include-merges")
    if args.emit_commits:
        cmd.append("--emit-commits")
    if args.graphml:
        cmd.append("--graphml")
    if args.sensitive_config:
        cmd.extend(["--sensitive-config", args.sensitive_config])
    if args.no_cochange:
        cmd.append("--no-cochange")
    if args.no_communities:
        cmd.append("--no-communities")
    if args.no_default_cochange_excludes:
        cmd.append("--no-default-cochange-excludes")
    for pattern in args.cochange_exclude:
        cmd.extend(["--cochange-exclude", pattern])
    if args.no_default_author_excludes:
        cmd.append("--no-default-author-excludes")
    for pattern in args.author_exclude_regex:
        cmd.extend(["--author-exclude-regex", pattern])

    result = subprocess.run(cmd, check=False)
    return result.returncode


if __name__ == "__main__":
    raise SystemExit(main())