#!/usr/bin/env python3
"""Compare password-manager CSV exports without printing secret-bearing fields.

The tool accepts browser, Bitwarden, and KeePassXC-style CSV headers. It emits
aggregate counts only. Credentials are compared as in-memory SHA-256 digests;
digests, usernames, passwords, and URLs are never written to the report.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
from collections import Counter
from pathlib import Path
from urllib.parse import urlsplit


SCHEMAS = {
    "bitwarden": ("login_uri", "login_username", "login_password"),
    "keepassxc": ("url", "username", "password"),
    "browser": ("url", "username", "password"),
}


def normalized_headers(fieldnames: list[str] | None) -> list[str]:
    return [(name or "").lstrip("\ufeff").strip().lower() for name in (fieldnames or [])]


def detect_schema(headers: list[str]) -> tuple[str, tuple[str, str, str]]:
    header_set = set(headers)
    if set(SCHEMAS["bitwarden"]).issubset(header_set):
        return "bitwarden", SCHEMAS["bitwarden"]
    if set(SCHEMAS["browser"]).issubset(header_set):
        if {"group", "title"}.issubset(header_set):
            return "keepassxc", SCHEMAS["keepassxc"]
        return "browser", SCHEMAS["browser"]
    raise ValueError("Unsupported CSV header: expected URL, username, and password fields")


def credential_digest(url: str, username: str, password: str) -> str:
    material = "\0".join((url.strip(), username.strip(), password))
    return hashlib.sha256(material.encode("utf-8")).hexdigest()


def valid_url(value: str) -> bool:
    try:
        parsed = urlsplit(value)
    except ValueError:
        return False
    return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc)


def load_csv(path: Path) -> dict:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        raw_headers = reader.fieldnames or []
        headers = normalized_headers(raw_headers)
        schema_name, (url_field, user_field, password_field) = detect_schema(headers)
        header_map = dict(zip(raw_headers, headers))
        rows = []
        for raw_row in reader:
            row = {header_map[key]: (value or "") for key, value in raw_row.items() if key is not None}
            if schema_name == "bitwarden" and row.get("type", "login").strip().lower() not in {"", "login"}:
                continue
            rows.append(row)

    digests = Counter(
        credential_digest(row.get(url_field, ""), row.get(user_field, ""), row.get(password_field, ""))
        for row in rows
    )
    urls = [row.get(url_field, "").strip() for row in rows]
    return {
        "format": schema_name,
        "rows": len(rows),
        "digests": digests,
        "exact_duplicates": sum(count - 1 for count in digests.values() if count > 1),
        "blank_usernames": sum(not row.get(user_field, "").strip() for row in rows),
        "blank_passwords": sum(not row.get(password_field, "") for row in rows),
        "invalid_urls": sum(not valid_url(url) for url in urls),
        "http_urls": sum(urlsplit(url).scheme.lower() == "http" for url in urls if valid_url(url)),
    }


def compare(before: dict, after: dict) -> dict:
    matched = sum((before["digests"] & after["digests"]).values())
    missing = sum((before["digests"] - after["digests"]).values())
    unexpected = sum((after["digests"] - before["digests"]).values())
    coverage = 100.0 if before["rows"] == 0 else round(100 * matched / before["rows"], 1)
    return {
        "source_format": before["format"],
        "destination_format": after["format"],
        "source_rows": before["rows"],
        "destination_rows": after["rows"],
        "row_count_match": before["rows"] == after["rows"],
        "exact_credentials_matched": matched,
        "exact_coverage_percent": coverage,
        "missing_source_credentials": missing,
        "unexpected_destination_credentials": unexpected,
        "source_exact_duplicates": before["exact_duplicates"],
        "destination_exact_duplicates": after["exact_duplicates"],
        "source_blank_usernames": before["blank_usernames"],
        "source_blank_passwords": before["blank_passwords"],
        "source_invalid_urls": before["invalid_urls"],
        "source_http_urls": before["http_urls"],
        "result": "PASS" if missing == 0 and unexpected == 0 else "HOLD",
    }


def write_csv(path: Path, report: dict) -> None:
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(("check", "value"))
        writer.writerows(report.items())


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("before", type=Path, help="source CSV export")
    parser.add_argument("after", type=Path, help="destination CSV export after migration")
    parser.add_argument("--csv-out", type=Path, help="write a secret-free two-column report")
    args = parser.parse_args()

    report = compare(load_csv(args.before), load_csv(args.after))
    if args.csv_out:
        write_csv(args.csv_out, report)
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0 if report["result"] == "PASS" else 2


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