#!/usr/bin/env python3
# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import argparse
import base64
import binascii
import hashlib
import hmac
import io
import json
import re
import subprocess
import sys
import tarfile
import tomllib
import urllib.request
import zipfile
from email.parser import Parser
from pathlib import Path
from typing import Any, TypedDict, cast

ATTRIBUTIONS_MD_LICENSE_PREFIX = """<!--
SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
SPDX-License-Identifier: Apache-2.0
-->
"""

ROOT = Path(__file__).resolve().parents[2]
NODE = ROOT
MARKDOWN_CODE_FENCE = "```\n"
MARKDOWN_CODE_BLOCK_END = "```\n\n"
_NODE_TARBALL_LICENSE_CACHE: dict[tuple[str, str], str | None] = {}


class RenderedPythonPackage(TypedDict):
    """Normalized Python package data ready to be rendered into markdown."""

    name: str
    version: str
    license_name: str
    license_texts: list[tuple[str, str]]


class LicenseInventoryEntry(TypedDict):
    """Minimal package license data used by machine-readable inventories."""

    package: str
    version: str
    license: str


class NodeAttributionPackage(TypedDict):
    """Normalized Node.js package data ready to be rendered into markdown."""

    name: str
    version: str
    license_name: str
    key: str
    resolved: str
    integrity: str
    platform_gated: bool


RUST_HEADER = (
    ATTRIBUTIONS_MD_LICENSE_PREFIX
    + "\n# Third-Party Software Attributions\n\n"
    + "This project uses the following third-party libraries. "
    + "Each library is open-source and licensed under the terms indicated below.\n\n"
    + "This file is automatically generated. Please do not edit it directly. "
    + "Regenerate with `./scripts/generate_attributions.sh rust`.\n\n"
)

PYTHON_HEADER = f"""{ATTRIBUTIONS_MD_LICENSE_PREFIX}

# Third-Party Software Attributions (Python)

This project uses the following third-party Python packages.
Each package is open-source and licensed under the terms indicated below.

This file is automatically generated from **uv.lock**
Please do not edit it directly. Regenerate with `./scripts/generate_attributions.sh python`
"""

NODE_HEADER = f"""{ATTRIBUTIONS_MD_LICENSE_PREFIX}

# Third-Party Software Attributions (Node.js)

This project uses the following third-party npm packages for the Node.js bindings.
Each package is open-source and licensed under the terms indicated below.

This file is automatically generated from **package-lock.json**
Please do not edit it directly. Regenerate with `./scripts/generate_attributions.sh node`
"""


def spdx_url(license_field: str, *, fallback: str) -> str:
    """Map a simple SPDX-looking license string to an SPDX page, otherwise use a generic fallback."""
    s = license_field.strip()
    if not s or s == "UNKNOWN":
        return fallback
    m = re.match(r"^([A-Za-z0-9.\-]+)$", s)
    if m:
        return f"https://spdx.org/licenses/{m.group(1)}.html"
    if " OR " in s or " AND " in s or "(" in s:
        return "https://spdx.org/licenses/"
    return fallback


def _pypi_license_path_display(license_file: str) -> str:
    """Match Dynamo ATTRIBUTIONS-Python.md: paths like ``licenses/LICENSE`` under .dist-info."""
    if not license_file or license_file == "UNKNOWN":
        return "LICENSE"
    p = license_file.replace("\\", "/")
    for marker in (".dist-info/", ".egg-info/"):
        if marker in p:
            return p.split(marker, 1)[1]
    name = Path(p).name
    return name if name else "LICENSE"


def _relative_archive_path_display(path: str) -> str:
    """Drop the sdist root directory so rendered paths match the installed-package style."""
    p = path.replace("\\", "/")
    if "/" in p:
        return p.split("/", 1)[1]
    return p


def _md_inline_safe(s: str) -> str:
    """Wrap s in markdown inline code, using double-backtick delimiters if s contains a backtick."""
    if "`" in s:
        return f"`` {s} ``"
    return f"`{s}`"


def _is_unknown_value(s: str) -> bool:
    """Return true for empty metadata values and common unknown placeholders."""
    return not s.strip() or s.strip().upper() == "UNKNOWN"


def _is_weak_license_name(s: str) -> bool:
    """Return true when package metadata does not provide a useful license label."""
    normalized = " ".join(s.strip().split()).lower()
    return _is_unknown_value(s) or normalized == "see package metadata"


def _is_useful_license_text(text: str) -> bool:
    """Return true when text is usable license content rather than a placeholder."""
    return not _is_unknown_value(text)


def _normalize_license_name(license_name: str) -> str:
    """Normalize common license labels that are equivalent but noisy across metadata sources."""
    normalized = " ".join(license_name.strip().split()).lower()
    label_map = {
        "mit license": "MIT",
        "the mit license": "MIT",
        "the mit license (mit)": "MIT",
    }
    return label_map.get(normalized, license_name)


def _looks_like_mit_license(text: str) -> bool:
    """Return true when license text matches the canonical MIT license shape."""
    normalized = " ".join(text.lower().split())
    return (
        "permission is hereby granted, free of charge" in normalized
        and "without restriction" in normalized
        and "use, copy, modify, merge, publish, distribute, sublicense" in normalized
        and "the above copyright notice and this permission notice" in normalized
        and "the software is provided" in normalized
        and "without warranty of any kind" in normalized
    )


def _license_name_from_file_heading(text: str) -> str | None:
    """Infer a license label from the first non-empty line of a license file."""
    heading_map = {
        "BSD 2-Clause License": "BSD-2-Clause",
        "BSD 3-Clause License": "BSD-3-Clause",
        "ISC License": "ISC",
        "MIT License": "MIT",
        "Mozilla Public License Version 2.0": "MPL-2.0",
        "NVIDIA LICENSE AGREEMENT": "NVIDIA License Agreement",
        "The MIT License": "MIT",
        "The MIT License (MIT)": "MIT",
    }
    for line in text.splitlines():
        heading = line.strip()
        if heading:
            mapped = heading_map.get(heading)
            if mapped:
                return mapped
            break

    normalized = text.lower()
    if "under the isc license" in normalized:
        return "ISC"
    if _looks_like_mit_license(text):
        return "MIT"
    return None


def _normalize_package_name(name: str) -> str:
    """Normalize names per Python package matching rules so lockfile/install rows can be joined reliably."""
    return re.sub(r"[-_.]+", "-", name).lower()


def _lockfile_registry_packages() -> dict[str, dict[str, Any]]:
    """Return {normalized_name: package_entry} for all third-party registry packages in uv.lock."""
    import tomllib

    with open(ROOT / "uv.lock", "rb") as f:
        lock: dict[str, Any] = tomllib.load(f)
    return {
        _normalize_package_name(str(pkg["name"])): pkg
        for pkg in cast(list[dict[str, Any]], lock.get("package", []))
        if pkg.get("name") and pkg.get("version") and cast(dict[str, Any], pkg.get("source", {})).get("registry")
    }


def _render_python_package(
    parts: list[str], *, name: str, version: str, license_name: str, license_texts: list[tuple[str, str]]
) -> None:
    """Append one Python package section to the markdown output."""
    parts.append(f"## {name} ({version})\n\n")
    parts.append("### Licenses\n")
    parts.append(f"License: {_md_inline_safe(license_name)}\n\n")
    for path_label, text in license_texts:
        parts.append(f"  - {_md_inline_safe(path_label)}:\n")
        parts.append(MARKDOWN_CODE_FENCE)
        parts.append(text if text.endswith("\n") else text + "\n")
        parts.append(MARKDOWN_CODE_BLOCK_END)


def _cargo_workspace_members() -> set[str]:
    """Return Cargo package IDs for the current workspace members."""
    proc = subprocess.run(  # noqa: S607
        [
            "cargo",
            "metadata",
            "--format-version=1",
            "--no-deps",
            "--manifest-path",
            str(ROOT / "Cargo.toml"),
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=True,
    )
    metadata: dict[str, Any] = json.loads(proc.stdout)
    return {str(member) for member in metadata.get("workspace_members", [])}


def _cargo_fetch_locked() -> None:
    """Ensure locked registry crate sources are available for fallback license file reads."""
    subprocess.run(  # noqa: S607
        ["cargo", "fetch", "--locked", "--manifest-path", str(ROOT / "Cargo.toml")],
        cwd=ROOT,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=True,
    )


def _cargo_metadata() -> dict[str, Any]:
    """Return full Cargo metadata for the all-features workspace graph."""
    _cargo_fetch_locked()
    proc = subprocess.run(  # noqa: S607
        [
            "cargo",
            "metadata",
            "--format-version=1",
            "--all-features",
            "--manifest-path",
            str(ROOT / "Cargo.toml"),
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=True,
    )
    return cast(dict[str, Any], json.loads(proc.stdout))


def _cargo_about_json() -> dict[str, Any]:
    """Generate cargo-about JSON for the workspace dependency graph."""
    _cargo_fetch_locked()
    proc = subprocess.run(  # noqa: S607
        [
            "cargo",
            "about",
            "generate",
            "--format",
            "json",
            "-m",
            str(ROOT / "Cargo.toml"),
            "--all-features",
            "--workspace",
            "--fail",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=True,
    )
    return cast(dict[str, Any], json.loads(proc.stdout))


def _rust_package_key(name: str, version: str) -> tuple[str, str]:
    """Return the stable key used to reconcile Cargo.lock, metadata, and cargo-about rows."""
    return name, version


def _cargo_lock_registry_package_keys() -> set[tuple[str, str]]:
    """Return third-party package keys from Cargo.lock."""
    with open(ROOT / "Cargo.lock", "rb") as f:
        lock: dict[str, Any] = tomllib.load(f)

    keys: set[tuple[str, str]] = set()
    for pkg in cast(list[dict[str, Any]], lock.get("package", [])):
        name = str(pkg.get("name") or "")
        version = str(pkg.get("version") or "")
        source = str(pkg.get("source") or "")
        if name and version and source:
            keys.add(_rust_package_key(name, version))
    return keys


def _cargo_metadata_packages_by_key() -> dict[tuple[str, str], dict[str, Any]]:
    """Return Cargo metadata package records keyed by package name and version."""
    metadata = _cargo_metadata()
    packages: dict[tuple[str, str], dict[str, Any]] = {}
    for pkg in cast(list[dict[str, Any]], metadata.get("packages", [])):
        name = str(pkg.get("name") or "")
        version = str(pkg.get("version") or "")
        source = str(pkg.get("source") or "")
        if name and version and source:
            packages.setdefault(_rust_package_key(name, version), pkg)
    return packages


def _rust_license_files(crate: dict[str, Any]) -> list[tuple[str, str]]:
    """Return root license files from a downloaded Cargo package."""
    manifest_path = str(crate.get("manifest_path") or "")
    if not manifest_path:
        return []
    package_dir = Path(manifest_path).parent
    candidates: list[Path] = []

    license_file = str(crate.get("license_file") or "")
    if license_file:
        p = Path(license_file)
        candidates.append(p if p.is_absolute() else package_dir / p)

    if package_dir.is_dir():
        for child in package_dir.iterdir():
            name = child.name.lower()
            if child.is_file() and (name.startswith(("license", "licence")) or name.startswith("copying")):
                candidates.append(child)

    seen: set[Path] = set()
    texts: list[tuple[str, str]] = []
    for path in sorted(candidates, key=lambda item: item.name.lower()):
        resolved = path.resolve()
        if resolved in seen or not path.is_file():
            continue
        seen.add(resolved)
        try:
            text = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            text = path.read_text(encoding="utf-8", errors="replace")
        label = path.name if path.parent == package_dir else str(path)
        if _is_useful_license_text(text):
            texts.append((label, text))
    return texts


def _render_rust_metadata_fallback_attribution(crate: dict[str, Any]) -> tuple[str, str, str]:
    """Render one Rust Cargo.lock package omitted by cargo-about."""
    name = str(crate.get("name") or "unknown")
    version = str(crate.get("version") or "")
    repo = str(crate.get("repository") or "").strip()
    if not repo:
        repo = f"https://crates.io/crates/{name}"
    license_name = _normalize_license_name(str(crate.get("license") or "UNKNOWN"))

    parts = [
        f"## {name} - {version}\n",
        f"**Repository URL**: {repo}\n",
        f"**License Type(s)**: {license_name}\n",
        f"### License: {spdx_url(license_name, fallback='https://spdx.org/licenses/')}\n",
    ]
    license_files = _rust_license_files(crate)
    if license_files:
        for label, text in license_files:
            parts.append(f"### License File: {label}\n")
            parts.append(MARKDOWN_CODE_FENCE)
            parts.append(text if text.endswith("\n") else text + "\n")
            parts.append(MARKDOWN_CODE_BLOCK_END)
    else:
        parts.append(MARKDOWN_CODE_FENCE)
        parts.append(f"License expression from Cargo metadata: {license_name}\n")
        parts.append("No package license file was found in the downloaded crate archive.\n")
        parts.append(MARKDOWN_CODE_BLOCK_END)

    return name, version, "".join(parts)


def _rust_missing_cargo_about_packages(rendered_keys: set[tuple[str, str]]) -> list[dict[str, Any]]:
    """Return Cargo.lock packages that cargo-about did not render."""
    lock_keys = _cargo_lock_registry_package_keys()
    metadata_by_key = _cargo_metadata_packages_by_key()
    missing: list[dict[str, Any]] = []
    for key in sorted(lock_keys - rendered_keys, key=lambda item: (item[0].lower(), item[1])):
        pkg = metadata_by_key.get(key)
        if pkg is None:
            raise RuntimeError(f"Cargo.lock package {key[0]} {key[1]} is missing from cargo metadata")
        missing.append(pkg)
    return missing


def _render_rust_crate_attribution(
    crate: dict[str, Any], *, license_id: str, license_text: str, workspace_members: set[str]
) -> tuple[str, str, str] | None:
    """Render one Rust crate attribution, skipping local workspace crates."""
    if str(crate.get("id") or "") in workspace_members:
        return None

    name = str(crate.get("name") or "unknown")
    version = str(crate.get("version") or "")
    repo = str(crate.get("repository") or "").strip()
    if not repo:
        repo = f"https://crates.io/crates/{name}"

    rendered = "".join(
        [
            f"## {name} - {version}\n",
            f"**Repository URL**: {repo}\n",
            f"**License Type(s)**: {license_id}\n",
            f"### License: https://spdx.org/licenses/{license_id}.html\n",
            MARKDOWN_CODE_FENCE,
            license_text,
            "\n",
            MARKDOWN_CODE_BLOCK_END,
        ]
    )
    return name, version, rendered


def _render_rust_attributions(data: dict[str, Any], workspace_members: set[str]) -> str:
    """Render Rust attributions for every non-workspace package in Cargo.lock."""
    parts: list[str] = [RUST_HEADER]
    rendered_crates: list[tuple[str, str, str]] = []
    rendered_keys: set[tuple[str, str]] = set()
    for license_group in cast(list[dict[str, Any]], data.get("licenses", [])):
        license_id = str(license_group.get("id") or "UNKNOWN")
        license_text = str(license_group.get("text") or "")
        for used_by in cast(list[dict[str, Any]], license_group.get("used_by", [])):
            rendered = _render_rust_crate_attribution(
                cast(dict[str, Any], used_by.get("crate") or {}),
                license_id=license_id,
                license_text=license_text,
                workspace_members=workspace_members,
            )
            if rendered:
                rendered_keys.add(_rust_package_key(rendered[0], rendered[1]))
                rendered_crates.append(rendered)

    for crate in _rust_missing_cargo_about_packages(rendered_keys):
        rendered = _render_rust_metadata_fallback_attribution(crate)
        rendered_keys.add(_rust_package_key(rendered[0], rendered[1]))
        rendered_crates.append(rendered)

    for _, _, rendered in sorted(rendered_crates, key=lambda row: (row[0].lower(), row[1])):
        parts.append(rendered)

    return "".join(parts).rstrip("\n") + "\n"


def _license_inventory_entry(name: str, version: str, license_name: str) -> LicenseInventoryEntry:
    """Return one normalized package license inventory row."""
    return LicenseInventoryEntry(package=name, version=version, license=license_name or "UNKNOWN")


def _rendered_python_package_inventory(pkg: RenderedPythonPackage) -> LicenseInventoryEntry:
    """Return a minimal inventory row from the richer Python attribution record."""
    return _license_inventory_entry(pkg["name"], pkg["version"], _normalize_license_name(pkg["license_name"]))


def _rust_license_inventory(data: dict[str, Any], workspace_members: set[str]) -> list[LicenseInventoryEntry]:
    """Return Rust license rows for every non-workspace package in Cargo.lock."""
    rows: dict[tuple[str, str], set[str]] = {}
    for license_group in cast(list[dict[str, Any]], data.get("licenses", [])):
        license_id = str(license_group.get("id") or "UNKNOWN")
        for used_by in cast(list[dict[str, Any]], license_group.get("used_by", [])):
            crate = cast(dict[str, Any], used_by.get("crate") or {})
            if str(crate.get("id") or "") in workspace_members:
                continue
            name = str(crate.get("name") or "unknown")
            version = str(crate.get("version") or "")
            rows.setdefault(_rust_package_key(name, version), set()).add(license_id)

    for crate in _rust_missing_cargo_about_packages(set(rows)):
        name = str(crate.get("name") or "unknown")
        version = str(crate.get("version") or "")
        license_name = _normalize_license_name(str(crate.get("license") or "UNKNOWN"))
        rows.setdefault(_rust_package_key(name, version), set()).add(license_name)

    return [
        _license_inventory_entry(name, version, " OR ".join(sorted(licenses)))
        for (name, version), licenses in sorted(rows.items(), key=lambda item: (item[0][0].lower(), item[0][1]))
    ]


def _hash_matches(data: bytes, expected_hash: str) -> bool:
    """Validate an artifact against the hash recorded in uv.lock."""
    algo, _, value = expected_hash.partition(":")
    if algo != "sha256" or not value:
        raise ValueError(f"Unsupported hash format: {expected_hash}")
    return hashlib.sha256(data).hexdigest() == value


def _download_locked_artifact(url: str, expected_hash: str) -> bytes:
    """Download one locked artifact and fail closed if its hash does not match uv.lock."""
    with urllib.request.urlopen(url, timeout=10) as response:  # noqa: S310
        data = response.read()
    if not _hash_matches(data, expected_hash):
        raise ValueError(f"Hash mismatch for {url}")
    return data


def _metadata_license_classifiers(msg) -> list[str]:
    """Extract the license portion of Trove classifiers from parsed package metadata."""
    found: list[str] = []
    for classifier in msg.get_all("Classifier", []):
        if not classifier.startswith("License ::"):
            continue
        tail = classifier.split("::")[-1].strip()
        if tail and tail not in found:
            found.append(tail)
    return found


def _metadata_license_name(msg) -> str:
    """Choose the best human-readable license label from package metadata.

    Prefer modern SPDX-style fields, then legacy free-form license text, then
    Trove classifiers as the final fallback.
    """
    for field in ("License-Expression", "License"):
        value = (msg.get(field) or "").strip()
        value = " ".join(value.split())
        if value:
            if len(value) > 200:
                return "See package metadata"
            return value
    classifiers = _metadata_license_classifiers(msg)
    if classifiers:
        return " OR ".join(classifiers)
    return "UNKNOWN"


def _clean_license_text(text: str) -> str:
    """Normalize bundled license text without changing meaningful line structure."""
    return "\n".join(line.rstrip() for line in text.splitlines()).strip()


def _extract_zip_text(zf: zipfile.ZipFile, path: str) -> str | None:
    """Read a text file from a wheel, returning None when the path is absent."""
    try:
        return _clean_license_text(zf.read(path).decode("utf-8", errors="replace"))
    except KeyError:
        return None


def _extract_tar_text(tf: tarfile.TarFile, path: str) -> str | None:
    """Read a text file from an sdist, returning None when the path is absent."""
    try:
        member = tf.getmember(path)
    except KeyError:
        return None
    fileobj = tf.extractfile(member)
    if fileobj is None:
        return None
    return _clean_license_text(fileobj.read().decode("utf-8", errors="replace"))


def _common_license_candidates(paths: list[str]) -> list[str]:
    """Heuristic fallback for packages that ship license files without License-File metadata."""
    preferred = ("license", "licence", "copying", "notice", "authors")
    return [path for path in sorted(paths) if Path(path).name.lower().startswith(preferred)]


def _license_name_rank(path: str) -> int:
    """Rank more canonical license filenames ahead of generic notice-style files."""
    name = Path(path.replace("\\", "/")).name.lower()
    if name == "license":
        return 0
    if name.startswith("license."):
        return 1
    if "license" in name:
        return 2
    if name == "licence":
        return 3
    if name.startswith("licence."):
        return 4
    if "licence" in name:
        return 5
    if name == "copying":
        return 6
    if name.startswith("copying."):
        return 7
    if "copying" in name:
        return 8
    if name == "notice":
        return 9
    if name.startswith("notice."):
        return 10
    if "notice" in name:
        return 11
    if name == "authors":
        return 12
    if name.startswith("authors."):
        return 13
    if "authors" in name:
        return 14
    return 99


def _license_path_depth(path: str) -> int:
    """Count path nesting so package-level license files win over vendored subtree licenses."""
    normalized = path.replace("\\", "/").strip("/")
    return normalized.count("/") if normalized else 0


def _choose_declared_license(candidates: list[tuple[int, str, str, str]]) -> tuple[str, str] | None:
    """Choose one canonical declared License-File match.

    Candidates are ``(declared_index, relative_path, display_path, text)``.
    Only ``LICENSE``-style basenames participate so metadata does not pull in
    vendored subtree notices when a package-level license exists.
    """
    license_named = [
        candidate for candidate in candidates if "license" in Path(candidate[1].replace("\\", "/")).name.lower()
    ]
    if not license_named:
        return None
    _, _, display_path, text = min(
        license_named,
        key=lambda candidate: (_license_path_depth(candidate[1]), candidate[0], candidate[1].lower()),
    )
    return display_path, text


def _choose_heuristic_license(candidates: list[tuple[str, str, str]]) -> tuple[str, str] | None:
    """Choose one fallback license-like file when metadata is missing or unhelpful.

    Candidates are ``(relative_path, display_path, text)``.
    """
    if not candidates:
        return None
    _, display_path, text = min(
        candidates,
        key=lambda candidate: (
            _license_path_depth(candidate[0]),
            _license_name_rank(candidate[0]),
            candidate[0].lower(),
        ),
    )
    return display_path, text


def _declared_wheel_license_texts(
    zf: zipfile.ZipFile, dist_info_dir: str, declared_files: list[str]
) -> list[tuple[str, str]]:
    """Return the best declared wheel license file, if the wheel bundles one."""
    declared_candidates: list[tuple[int, str, str, str]] = []
    for declared_index, rel in enumerate(declared_files):
        candidates = [f"{dist_info_dir}/licenses/{rel}", f"{dist_info_dir}/{rel}"]
        for candidate in candidates:
            text = _extract_zip_text(zf, candidate)
            if text and _is_useful_license_text(text):
                declared_candidates.append(
                    (declared_index, rel, _pypi_license_path_display(candidate), text),
                )
                break
    selected_declared = _choose_declared_license(declared_candidates)
    return [selected_declared] if selected_declared else []


def _heuristic_wheel_license_texts(zf: zipfile.ZipFile, dist_info_dir: str) -> list[tuple[str, str]]:
    """Return a fallback wheel license file when metadata does not declare one."""
    heuristic_paths = _common_license_candidates(
        [
            name
            for name in zf.namelist()
            if name.startswith(f"{dist_info_dir}/licenses/") or name.startswith(f"{dist_info_dir}/")
        ]
    )
    heuristic_candidates: list[tuple[str, str, str]] = []
    for candidate in heuristic_paths:
        text = _extract_zip_text(zf, candidate)
        if text and _is_useful_license_text(text):
            display_path = _pypi_license_path_display(candidate)
            heuristic_candidates.append((display_path, display_path, text))
    selected_heuristic = _choose_heuristic_license(heuristic_candidates)
    return [selected_heuristic] if selected_heuristic else []


def _wheel_metadata_from_bytes(data: bytes, *, package_name: str) -> tuple[str, list[tuple[str, str]]]:
    """Extract a license label and bundled license texts from one wheel artifact."""
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        metadata_paths = [name for name in zf.namelist() if name.endswith(".dist-info/METADATA")]
        if not metadata_paths:
            raise ValueError(f"No METADATA found in wheel for {package_name}")
        metadata_path = metadata_paths[0]
        dist_info_dir = metadata_path.rsplit("/", 1)[0]
        metadata = Parser().parsestr(zf.read(metadata_path).decode("utf-8", errors="replace"))
        license_name = _metadata_license_name(metadata)
        license_texts = _declared_wheel_license_texts(zf, dist_info_dir, metadata.get_all("License-File", []))
        if not license_texts:
            # Older wheels often omit License-File entries but still bundle a LICENSE-like file.
            license_texts = _heuristic_wheel_license_texts(zf, dist_info_dir)
        return license_name, license_texts


def _declared_sdist_license_texts(
    tf: tarfile.TarFile, root_dir: str, declared_files: list[str]
) -> list[tuple[str, str]]:
    """Return the best declared sdist license file, if the sdist bundles one."""
    declared_candidates: list[tuple[int, str, str, str]] = []
    for declared_index, rel in enumerate(declared_files):
        candidate = f"{root_dir}/{rel}"
        text = _extract_tar_text(tf, candidate)
        if text and _is_useful_license_text(text):
            declared_candidates.append(
                (declared_index, rel, _relative_archive_path_display(candidate), text),
            )
    selected_declared = _choose_declared_license(declared_candidates)
    return [selected_declared] if selected_declared else []


def _heuristic_sdist_license_texts(tf: tarfile.TarFile, root_dir: str) -> list[tuple[str, str]]:
    """Return a fallback sdist license file when metadata does not declare one."""
    heuristic_paths = _common_license_candidates([name for name in tf.getnames() if name.startswith(f"{root_dir}/")])
    heuristic_candidates: list[tuple[str, str, str]] = []
    for candidate in heuristic_paths:
        text = _extract_tar_text(tf, candidate)
        if text and _is_useful_license_text(text):
            display_path = _relative_archive_path_display(candidate)
            heuristic_candidates.append((display_path, display_path, text))
    selected_heuristic = _choose_heuristic_license(heuristic_candidates)
    return [selected_heuristic] if selected_heuristic else []


def _sdist_metadata_from_bytes(data: bytes, *, package_name: str) -> tuple[str, list[tuple[str, str]]]:
    """Extract a license label and bundled license texts from one sdist artifact."""
    with tarfile.open(fileobj=io.BytesIO(data), mode="r:*") as tf:
        members: list[str] = [name for name in tf.getnames() if name.endswith("/PKG-INFO")]
        if not members:
            raise ValueError(f"No PKG-INFO found in sdist for {package_name}")
        metadata_path = cast(str, min(members, key=len))
        root_dir = metadata_path.split("/", 1)[0]
        raw = _extract_tar_text(tf, metadata_path)
        if raw is None:
            raise ValueError(f"Failed to read PKG-INFO from sdist for {package_name}")
        metadata = Parser().parsestr(raw)
        license_name = _metadata_license_name(metadata)
        license_texts = _declared_sdist_license_texts(tf, root_dir, metadata.get_all("License-File", []))
        if not license_texts:
            # Sdists are our best fallback for unresolved packages because they are typically
            # platform-agnostic and more likely to include complete source-level license files.
            license_texts = _heuristic_sdist_license_texts(tf, root_dir)
        return license_name, license_texts


def _artifact_metadata_from_lockfile(pkg: dict[str, Any]) -> tuple[str, list[tuple[str, str]]]:
    """Read license metadata from the locked artifacts for one package.

    Prefer the sdist when available because it is less platform-specific than a
    wheel. If sdist metadata does not provide a useful license label, inspect
    locked wheels before falling back to the sdist result.
    """
    package_name = pkg["name"]
    sdist = pkg.get("sdist")
    sdist_metadata: tuple[str, list[tuple[str, str]]] | None = None
    if isinstance(sdist, dict) and sdist.get("url") and sdist.get("hash"):
        data = _download_locked_artifact(sdist["url"], sdist["hash"])
        sdist_metadata = _sdist_metadata_from_bytes(data, package_name=package_name)
        if not _is_weak_license_name(sdist_metadata[0]):
            return sdist_metadata

    wheels = cast(list[dict[str, Any]], pkg.get("wheels") or [])
    first_wheel_metadata: tuple[str, list[tuple[str, str]]] | None = None
    for wheel in wheels:
        if wheel.get("url") and wheel.get("hash"):
            data = _download_locked_artifact(wheel["url"], wheel["hash"])
            wheel_metadata = _wheel_metadata_from_bytes(data, package_name=package_name)
            if first_wheel_metadata is None:
                first_wheel_metadata = wheel_metadata
            if not _is_weak_license_name(wheel_metadata[0]):
                if sdist_metadata and sdist_metadata[1]:
                    return wheel_metadata[0], sdist_metadata[1]
                return wheel_metadata

    if sdist_metadata is not None:
        return sdist_metadata
    if first_wheel_metadata is not None:
        return first_wheel_metadata

    raise ValueError(f"No downloadable artifact found in uv.lock for {package_name}")


def _infer_license_name(license_name: str, license_texts: list[tuple[str, str]]) -> str:
    """Infer a package license from bundled license text when metadata reports UNKNOWN."""
    if not _is_weak_license_name(license_name):
        return license_name
    for _, license_text in license_texts:
        inferred_license = _license_name_from_file_heading(license_text)
        if inferred_license:
            return inferred_license
    return license_name


def _lockfile_fallback_license_texts(package_name: str) -> list[tuple[str, str]]:
    """Return a stable placeholder when no bundled license file can be resolved from the artifact."""
    return [
        ("LICENSE", f"(No license file found in locked artifact for {package_name}; see package metadata or PyPI.)\n")
    ]


def _lockfile_only_python_package(pkg: dict[str, Any]) -> RenderedPythonPackage:
    """Render one package from the locked artifact metadata in uv.lock."""
    license_name, license_texts = _artifact_metadata_from_lockfile(pkg)
    if not license_texts:
        license_texts = _lockfile_fallback_license_texts(str(pkg["name"]))
    license_name = _infer_license_name(license_name, license_texts)
    license_name = _normalize_license_name(license_name)
    return RenderedPythonPackage(
        name=str(pkg["name"]),
        version=str(pkg["version"]),
        license_name=license_name,
        license_texts=license_texts,
    )


def _lockfile_python_packages(
    lockfile_pkgs: dict[str, dict[str, Any]], *, own_name: str, seen: set[str] | None = None
) -> list[RenderedPythonPackage]:
    """Render third-party Python packages directly from uv.lock."""
    seen = seen or set()
    packages: list[RenderedPythonPackage] = []
    for normalized, pkg in lockfile_pkgs.items():
        if normalized == own_name or normalized in seen:
            continue
        packages.append(_lockfile_only_python_package(pkg))
    return packages


def _python_attribution_packages() -> list[RenderedPythonPackage]:
    """Return Python attribution packages directly from uv.lock artifacts."""
    lockfile_pkgs = _lockfile_registry_packages()
    own_name = "nemo-flow"
    packages = _lockfile_python_packages(lockfile_pkgs, own_name=own_name)
    packages.sort(key=lambda r: (str(r["name"]).lower(), str(r["version"])))
    return packages


def _python_license_inventory() -> list[LicenseInventoryEntry]:
    """Return minimal Python dependency license rows directly from uv.lock artifacts."""
    lockfile_pkgs = _lockfile_registry_packages()
    own_name = "nemo-flow"
    return [
        _rendered_python_package_inventory(pkg) for pkg in _lockfile_python_packages(lockfile_pkgs, own_name=own_name)
    ]


def cmd_rust() -> int:
    """Generate ATTRIBUTIONS-Rust.md from cargo-about, excluding local workspace crates."""
    out = ROOT / "ATTRIBUTIONS-Rust.md"
    workspace_members = _cargo_workspace_members()
    rendered = _render_rust_attributions(_cargo_about_json(), workspace_members)
    out.write_text(rendered, encoding="utf-8")
    print(f"Wrote {out.relative_to(ROOT)}", file=sys.stderr)
    return 0


def cmd_python() -> int:
    """Generate ATTRIBUTIONS-Python.md from the current locked install plus archive fallback."""
    out = ROOT / "ATTRIBUTIONS-Python.md"
    packages = _python_attribution_packages()

    parts: list[str] = [PYTHON_HEADER]
    for pkg in packages:
        _render_python_package(
            parts,
            name=pkg["name"],
            version=pkg["version"],
            license_name=pkg["license_name"],
            license_texts=pkg["license_texts"],
        )

    out.write_text("".join(parts).rstrip("\n") + "\n", encoding="utf-8")
    print(f"Wrote {out.relative_to(ROOT)}", file=sys.stderr)
    return 0


def _node_package_key(meta: dict[str, Any]) -> str | None:
    """Return the npm package key for a package-lock entry that has a name and version."""
    name = str(meta.get("name") or "").strip()
    version = str(meta.get("version") or "").strip()
    if name and version:
        return f"{name}@{version}"
    return None


def _node_workspace_package_metas(lock: dict[str, Any]) -> list[dict[str, Any]]:
    """Return local package metadata entries from a package-lock file."""
    package_metas = [lock]
    packages = lock.get("packages") or {}
    if not isinstance(packages, dict):
        return package_metas
    for path, meta in packages.items():
        normalized_path = str(path).replace("\\", "/")
        if normalized_path and "node_modules" in normalized_path.split("/"):
            continue
        if isinstance(meta, dict):
            package_metas.append(meta)
    return package_metas


def _node_workspace_package_keys(lock: dict[str, Any]) -> set[str]:
    """Return npm package keys for local packages described by package-lock.json."""
    keys: set[str] = set()
    for meta in _node_workspace_package_metas(lock):
        key = _node_package_key(meta)
        if key:
            keys.add(key)

    return keys


def _node_lockfile_path() -> Path:
    """Return the Node.js package lockfile, preferring the workspace-root lockfile."""
    root_lockfile = NODE / "package-lock.json"
    if root_lockfile.exists():
        return root_lockfile
    return NODE / "crates" / "node" / "package-lock.json"


def _node_license_checker_data() -> tuple[dict[str, dict[str, str]], set[str]]:
    """Return license-checker package data and local package keys to omit."""
    lockfile = _node_lockfile_path()
    lock = cast(dict[str, Any], json.loads(lockfile.read_text(encoding="utf-8")))
    workspace_package_keys = _node_workspace_package_keys(lock)

    subprocess.run(["npm", "ci", "--ignore-scripts"], cwd=lockfile.parent, check=True)  # noqa: S607
    proc = subprocess.run(
        ["npx", "--yes", "license-checker@25.0.1", "--json"],  # noqa: S607
        cwd=lockfile.parent,
        capture_output=True,
        text=True,
        check=True,
    )
    data: dict[str, dict[str, str]] = json.loads(proc.stdout)
    return data, workspace_package_keys


def _node_attribution_packages(lock: dict[str, Any], workspace_package_keys: set[str]) -> list[NodeAttributionPackage]:
    """Return all third-party Node.js packages from package-lock.json.

    `license-checker` only sees packages installed for the current platform. The
    lockfile is the source of truth for attribution so platform-gated optional
    packages stay present on every machine that regenerates this file.
    """
    packages = lock.get("packages") or {}
    if not isinstance(packages, dict):
        return []

    rows: dict[str, NodeAttributionPackage] = {}
    for path, meta in packages.items():
        normalized_path = str(path).replace("\\", "/")
        if "node_modules" not in normalized_path.split("/"):
            continue
        if not isinstance(meta, dict):
            continue
        if meta.get("extraneous"):
            continue
        version = str(meta.get("version") or "").strip()
        if not version:
            continue
        name = str(meta.get("name") or _node_package_name_from_lock_path(normalized_path)).strip()
        if not name:
            continue
        key = f"{name}@{version}"
        if key in workspace_package_keys:
            continue
        license_name = str(meta.get("license") or "UNKNOWN")
        rows[key] = {
            "name": name,
            "version": version,
            "license_name": license_name,
            "key": key,
            "resolved": str(meta.get("resolved") or ""),
            "integrity": str(meta.get("integrity") or ""),
            "platform_gated": bool(meta.get("optional") or meta.get("os") or meta.get("cpu")),
        }

    return sorted(rows.values(), key=lambda row: (row["name"].lower(), row["version"], row["license_name"]))


def _node_integrity_matches(data: bytes, integrity: str) -> bool:
    """Validate downloaded npm artifact bytes against package-lock integrity."""
    if not integrity:
        return True
    algorithms = {
        "sha1": hashlib.sha1,
        "sha256": hashlib.sha256,
        "sha384": hashlib.sha384,
        "sha512": hashlib.sha512,
    }
    saw_supported = False
    for token in integrity.split():
        algorithm, sep, encoded = token.partition("-")
        if not sep or algorithm not in algorithms:
            continue
        saw_supported = True
        try:
            expected = base64.b64decode(encoded, validate=True)
        except (ValueError, binascii.Error):
            return False
        if hmac.compare_digest(algorithms[algorithm](data).digest(), expected):
            return True
    return not saw_supported


def _node_license_text_from_tarball(resolved: str, integrity: str) -> str:
    """Return license text from a locked npm tarball URL, if available."""
    if not resolved.startswith(("http://", "https://")):
        return ""
    cache_key = (resolved, integrity)
    cached = _NODE_TARBALL_LICENSE_CACHE.get(cache_key)
    if cached is not None:
        return cached
    if cache_key in _NODE_TARBALL_LICENSE_CACHE:
        return ""

    try:
        with urllib.request.urlopen(resolved, timeout=60) as response:  # noqa: S310
            raw = response.read()
    except OSError:
        _NODE_TARBALL_LICENSE_CACHE[cache_key] = None
        return ""
    if not _node_integrity_matches(raw, integrity):
        _NODE_TARBALL_LICENSE_CACHE[cache_key] = None
        return ""

    try:
        with tarfile.open(fileobj=io.BytesIO(raw), mode="r:*") as tf:
            heuristic_candidates: list[tuple[str, str, str]] = []
            for member_name in _common_license_candidates(tf.getnames()):
                text = _extract_tar_text(tf, member_name)
                if text and _is_useful_license_text(text):
                    display_path = member_name.split("/", 1)[1] if "/" in member_name else member_name
                    heuristic_candidates.append((display_path, display_path, text))
            selected = _choose_heuristic_license(heuristic_candidates)
    except (OSError, tarfile.TarError):
        selected = None

    text = selected[1] if selected else ""
    _NODE_TARBALL_LICENSE_CACHE[cache_key] = text or None
    return text


def _node_package_name_from_lock_path(path: str) -> str:
    """Infer an npm package name from a package-lock packages path."""
    parts = path.replace("\\", "/").split("node_modules/")
    return parts[-1].strip("/")


def _split_node_package_key(key: str) -> tuple[str, str]:
    """Split a license-checker package key into package name and version."""
    idx = key.rfind("@")
    if idx <= 0:
        return key, ""
    return key[:idx], key[idx + 1 :]


def _node_license_inventory() -> list[LicenseInventoryEntry]:
    """Return minimal Node.js dependency license rows directly from package-lock.json."""
    lock = cast(dict[str, Any], json.loads(_node_lockfile_path().read_text(encoding="utf-8")))
    packages = lock.get("packages") or {}
    if not isinstance(packages, dict):
        return []

    rows: dict[tuple[str, str, str], LicenseInventoryEntry] = {}
    for path, meta in packages.items():
        normalized_path = str(path).replace("\\", "/")
        if "node_modules" not in normalized_path.split("/"):
            continue
        if not isinstance(meta, dict):
            continue
        if meta.get("extraneous"):
            continue
        version = str(meta.get("version") or "")
        if not version:
            continue
        name = str(meta.get("name") or _node_package_name_from_lock_path(normalized_path))
        license_name = str(meta.get("license") or "UNKNOWN")
        rows[(name, version, license_name)] = _license_inventory_entry(name, version, license_name)

    return sorted(rows.values(), key=lambda row: (row["package"].lower(), row["version"], row["license"]))


def cmd_node() -> int:
    """Generate ATTRIBUTIONS-Node.md from package-lock.json."""
    out = ROOT / "ATTRIBUTIONS-Node.md"
    lockfile = _node_lockfile_path()
    lock = cast(dict[str, Any], json.loads(lockfile.read_text(encoding="utf-8")))
    data, workspace_package_keys = _node_license_checker_data()
    package_rows = _node_attribution_packages(lock, workspace_package_keys)

    parts: list[str] = [NODE_HEADER]
    for pkg in package_rows:
        name = pkg["name"]
        ver = pkg["version"]
        meta = {} if pkg["platform_gated"] else data.get(pkg["key"], {})

        lic = meta.get("licenses") or pkg["license_name"] or "UNKNOWN"
        repo = (meta.get("repository") or "").strip()
        if not repo:
            repo = f"https://www.npmjs.com/package/{name}"
        lic_url = spdx_url(str(lic), fallback="https://opensource.org/licenses/")
        text = _node_license_text_from_tarball(pkg["resolved"], pkg["integrity"]) if pkg["platform_gated"] else ""
        lf = meta.get("licenseFile") if not text else None
        if lf:
            try:
                raw = Path(lf).read_text(encoding="utf-8", errors="replace").strip()
                text = "\n".join(line.rstrip() for line in raw.splitlines())
            except OSError:
                text = ""
        if not text:
            text = _node_license_text_from_tarball(pkg["resolved"], pkg["integrity"])

        parts.append(f"## {name} - {ver}\n")
        parts.append(f"**Repository URL**: {repo}\n")
        parts.append(f"**License Type(s)**: {lic}\n")
        parts.append(f"### License: {lic_url}\n")
        parts.append(MARKDOWN_CODE_FENCE)
        fallback = f"(No license file read from locked npm artifact for {name}; see npm metadata.)\n"
        parts.append(text if text else fallback)
        parts.append(MARKDOWN_CODE_BLOCK_END)

    out.write_text("".join(parts).rstrip("\n") + "\n", encoding="utf-8")
    print(f"Wrote {out.relative_to(ROOT)}", file=sys.stderr)
    return 0


def main() -> int:
    """Dispatch the lockfile-to-markdown generator for the requested ecosystem."""
    p = argparse.ArgumentParser(description=__doc__)
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("rust", help="ATTRIBUTIONS-Rust.md from Cargo.lock")
    sub.add_parser("python", help="ATTRIBUTIONS-Python.md from uv.lock")
    sub.add_parser("node", help="ATTRIBUTIONS-Node.md from package-lock.json")
    args = p.parse_args()
    if args.cmd == "rust":
        return cmd_rust()
    if args.cmd == "python":
        return cmd_python()
    if args.cmd == "node":
        return cmd_node()
    return 2


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