Security Ownership Map

Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

1. Definitions.

   "License" shall mean the terms and conditions for use, reproduction,
   and distribution as defined by Sections 1 through 9 of this document.

   "Licensor" shall mean the copyright owner or entity authorized by
   the copyright owner that is granting the License.

   "Legal Entity" shall mean the union of the acting entity and all
   other entities that control, are controlled by, or are under common
   control with that entity. For the purposes of this definition,
   "control" means (i) the power, direct or indirect, to cause the
   direction or management of such entity, whether by contract or
   otherwise, or (ii) ownership of fifty percent (50%) or more of the
   outstanding shares, or (iii) beneficial ownership of such entity.

   "You" (or "Your") shall mean an individual or Legal Entity
   exercising permissions granted by this License.

   "Source" form shall mean the preferred form for making modifications,
   including but not limited to software source code, documentation
   source, and configuration files.

   "Object" form shall mean any form resulting from mechanical
   transformation or translation of a Source form, including but
   not limited to compiled object code, generated documentation,
   and conversions to other media types.

   "Work" shall mean the work of authorship, whether in Source or
   Object form, made available under the License, as indicated by a
   copyright notice that is included in or attached to the work
   (an example is provided in the Appendix below).

   "Derivative Works" shall mean any work, whether in Source or Object
   form, that is based on (or derived from) the Work and for which the
   editorial revisions, annotations, elaborations, or other modifications
   represent, as a whole, an original work of authorship. For the purposes
   of this License, Derivative Works shall not include works that remain
   separable from, or merely link (or bind by name) to the interfaces of,
   the Work and Derivative Works thereof.

   "Contribution" shall mean any work of authorship, including
   the original version of the Work and any modifications or additions
   to that Work or Derivative Works thereof, that is intentionally
   submitted to Licensor for inclusion in the Work by the copyright owner
   or by an individual or Legal Entity authorized to submit on behalf of
   the copyright owner. For the purposes of this definition, "submitted"
   means any form of electronic, verbal, or written communication sent
   to the Licensor or its representatives, including but not limited to
   communication on electronic mailing lists, source code control systems,
   and issue tracking systems that are managed by, or on behalf of, the
   Licensor for the purpose of discussing and improving the Work, but
   excluding communication that is conspicuously marked or otherwise
   designated in writing by the copyright owner as "Not a Contribution."

   "Contributor" shall mean Licensor and any individual or Legal Entity
   on behalf of whom a Contribution has been received by Licensor and
   subsequently incorporated within the Work.

2. Grant of Copyright License. Subject to the terms and conditions of
   this License, each Contributor hereby grants to You a perpetual,
   worldwide, non-exclusive, no-charge, royalty-free, irrevocable
   copyright license to reproduce, prepare Derivative Works of,
   publicly display, publicly perform, sublicense, and distribute the
   Work and such Derivative Works in Source or Object form.

3. Grant of Patent License. Subject to the terms and conditions of
   this License, each Contributor hereby grants to You a perpetual,
   worldwide, non-exclusive, no-charge, royalty-free, irrevocable
   (except as stated in this section) patent license to make, have made,
   use, offer to sell, sell, import, and otherwise transfer the Work,
   where such license applies only to those patent claims licensable
   by such Contributor that are necessarily infringed by their
   Contribution(s) alone or by combination of their Contribution(s)
   with the Work to which such Contribution(s) was submitted. If You
   institute patent litigation against any entity (including a
   cross-claim or counterclaim in a lawsuit) alleging that the Work
   or a Contribution incorporated within the Work constitutes direct
   or contributory patent infringement, then any patent licenses
   granted to You under this License for that Work shall terminate
   as of the date such litigation is filed.

4. Redistribution. You may reproduce and distribute copies of the
   Work or Derivative Works thereof in any medium, with or without
   modifications, and in Source or Object form, provided that You
   meet the following conditions:

   (a) You must give any other recipients of the Work or
       Derivative Works a copy of this License; and

   (b) You must cause any modified files to carry prominent notices
       stating that You changed the files; and

   (c) You must retain, in the Source form of any Derivative Works
       that You distribute, all copyright, patent, trademark, and
       attribution notices from the Source form of the Work,
       excluding those notices that do not pertain to any part of
       the Derivative Works; and

   (d) If the Work includes a "NOTICE" text file as part of its
       distribution, then any Derivative Works that You distribute must
       include a readable copy of the attribution notices contained
       within such NOTICE file, excluding those notices that do not
       pertain to any part of the Derivative Works, in at least one
       of the following places: within a NOTICE text file distributed
       as part of the Derivative Works; within the Source form or
       documentation, if provided along with the Derivative Works; or,
       within a display generated by the Derivative Works, if and
       wherever such third-party notices normally appear. The contents
       of the NOTICE file are for informational purposes only and
       do not modify the License. You may add Your own attribution
       notices within Derivative Works that You distribute, alongside
       or as an addendum to the NOTICE text from the Work, provided
       that such additional attribution notices cannot be construed
       as modifying the License.

   You may add Your own copyright statement to Your modifications and
   may provide additional or different license terms and conditions
   for use, reproduction, or distribution of Your modifications, or
   for any such Derivative Works as a whole, provided Your use,
   reproduction, and distribution of the Work otherwise complies with
   the conditions stated in this License.

5. Submission of Contributions. Unless You explicitly state otherwise,
   any Contribution intentionally submitted for inclusion in the Work
   by You to the Licensor shall be under the terms and conditions of
   this License, without any additional terms or conditions.
   Notwithstanding the above, nothing herein shall supersede or modify
   the terms of any separate license agreement you may have executed
   with Licensor regarding such Contributions.

6. Trademarks. This License does not grant permission to use the trade
   names, trademarks, service marks, or product names of the Licensor,
   except as required for reasonable and customary use in describing the
   origin of the Work and reproducing the content of the NOTICE file.

7. Disclaimer of Warranty. Unless required by applicable law or
   agreed to in writing, Licensor provides the Work (and each
   Contributor provides its Contributions) on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   implied, including, without limitation, any warranties or conditions
   of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
   PARTICULAR PURPOSE. You are solely responsible for determining the
   appropriateness of using or redistributing the Work and assume any
   risks associated with Your exercise of permissions under this License.

8. Limitation of Liability. In no event and under no legal theory,
   whether in tort (including negligence), contract, or otherwise,
   unless required by applicable law (such as deliberate and grossly
   negligent acts) or agreed to in writing, shall any Contributor be
   liable to You for damages, including any direct, indirect, special,
   incidental, or consequential damages of any character arising as a
   result of this License or out of the use or inability to use the
   Work (including but not limited to damages for loss of goodwill,
   work stoppage, computer failure or malfunction, or any and all
   other commercial damages or losses), even if such Contributor
   has been advised of the possibility of such damages.

9. Accepting Warranty or Additional Liability. While redistributing
   the Work or Derivative Works thereof, You may choose to offer,
   and charge a fee for, acceptance of support, warranty, indemnity,
   or other liability obligations and/or rights consistent with this
   License. However, in accepting such obligations, You may act only
   on Your own behalf and on Your sole responsibility, not on behalf of
   any other Contributor, and only if You agree to indemnify,
   defend, and hold each Contributor harmless for any liability
   incurred by, or claims asserted against, such Contributor by reason
   of your accepting any such warranty or additional liability.

END OF TERMS AND CONDITIONS

APPENDIX: How to apply the Apache License to your work.

   To apply the Apache License to your work, attach the following
   boilerplate notice, with the fields enclosed by brackets "[]"
   replaced with your own identifying information. (Don\'t include
   the brackets!)  The text should be enclosed in the appropriate
   comment syntax for the file format. We also recommend that a
   file or class name and description of purpose be included on the
   same "printed page" as the copyright notice for easier
   identification within third-party archives.

Copyright [yyyy] [name of copyright owner]

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

#!/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())