planet/scripts/compute_aiprovider_dependency_fingerprint.py

#!/usr/bin/env python3

from __future__ import annotations

import ast
import json
import re
import sys
from pathlib import Path
from typing import Any


IMPORT_TO_DISTRIBUTION = {
    "pydantic_settings": "pydantic-settings",
}

# Some runtime dependencies are referenced indirectly:
# - uvicorn is launched by the container command rather than imported.
# - python-dotenv is used by pydantic-settings when loading the local .env file.
EXTRA_RUNTIME_DISTRIBUTIONS = {
    "python-dotenv",
    "uvicorn",
}


def normalize_name(value: str) -> str:
    return value.strip().lower().replace("_", "-").replace(".", "-")


def extract_table(text: str, table_name: str) -> str:
    pattern = re.compile(
        rf"(?ms)^\[{re.escape(table_name)}\]\s*$\n(.*?)(?=^\[[^\]]+\]\s*$|\Z)"
    )
    match = pattern.search(text)
    return match.group(1) if match else ""


def extract_string_value(text: str, key: str) -> str | None:
    match = re.search(rf'(?m)^{re.escape(key)}\s*=\s*"([^"]+)"\s*$', text)
    return match.group(1) if match else None


def extract_string_list_value(text: str, key: str) -> list[str]:
    match = re.search(rf"(?ms)^{re.escape(key)}\s*=\s*\[(.*?)\]\s*$", text)
    if not match:
        return []
    return re.findall(r'"([^"]+)"', match.group(1))


def project_dependency_map(pyproject_text: str) -> tuple[str | None, dict[str, str]]:
    project_block = extract_table(pyproject_text, "project")
    requires_python = extract_string_value(project_block, "requires-python")
    dependencies = extract_string_list_value(project_block, "dependencies")
    mapping: dict[str, str] = {}
    for spec in dependencies:
        name = spec.split(";", 1)[0].strip()
        for marker in ("<", ">", "=", "!", "~"):
            if marker in name:
                name = name.split(marker, 1)[0].strip()
        if "[" in name:
            name = name.split("[", 1)[0].strip()
        mapping[normalize_name(name)] = spec.strip()
    return requires_python, mapping


def collect_runtime_imports(aiprovider_dir: Path) -> set[str]:
    imports: set[str] = set()
    stdlib = set(sys.stdlib_module_names)

    for path in sorted(aiprovider_dir.rglob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    imports.add(alias.name.split(".", 1)[0])
            elif isinstance(node, ast.ImportFrom) and node.module:
                imports.add(node.module.split(".", 1)[0])

    return {
        name
        for name in imports
        if name not in stdlib and name != "aiprovider"
    }


def resolve_relevant_root_dependencies(
    pyproject_deps: dict[str, str],
    imported_modules: set[str],
) -> dict[str, str]:
    relevant_names: set[str] = set(EXTRA_RUNTIME_DISTRIBUTIONS)

    for module_name in imported_modules:
        mapped_name = IMPORT_TO_DISTRIBUTION.get(module_name, module_name)
        normalized_name = normalize_name(mapped_name)
        if normalized_name in pyproject_deps:
            relevant_names.add(normalized_name)

    return {
        name: pyproject_deps[name]
        for name in sorted(relevant_names)
        if name in pyproject_deps
    }


def lock_metadata(lock_text: str) -> dict[str, str | None]:
    return {
        "version": extract_string_value(lock_text, "version") or re.search(r"(?m)^version\s*=\s*(\d+)\s*$", lock_text).group(1),
        "revision": extract_string_value(lock_text, "revision") or re.search(r"(?m)^revision\s*=\s*(\d+)\s*$", lock_text).group(1),
        "requires_python": extract_string_value(lock_text, "requires-python"),
    }


def lock_package_map(lock_text: str) -> dict[str, dict[str, Any]]:
    mapping: dict[str, dict[str, Any]] = {}
    sections = re.split(r"(?m)^\[\[package\]\]\s*$\n?", lock_text)
    for section in sections[1:]:
        raw_section = section.strip()
        name = extract_string_value(raw_section, "name")
        if not name:
            continue
        version = extract_string_value(raw_section, "version")
        dependency_names = [
            normalize_name(dep_name)
            for dep_name in re.findall(r'\{\s*name\s*=\s*"([^"]+)"', raw_section)
        ]
        mapping[normalize_name(name)] = {
            "name": name,
            "version": version,
            "dependencies": dependency_names,
            "raw": raw_section,
        }
    return mapping


def dependency_closure(
    lock_packages: dict[str, dict[str, Any]],
    root_dependencies: dict[str, str],
) -> list[dict[str, Any]]:
    pending = list(root_dependencies.keys())
    visited: set[str] = set()
    resolved: list[dict[str, Any]] = []

    while pending:
        current = pending.pop()
        if current in visited:
            continue
        visited.add(current)

        package = lock_packages.get(current)
        if package is None:
            continue

        resolved.append(package)
        for dep_name in package.get("dependencies", []):
            pending.append(dep_name)

    resolved.sort(key=lambda pkg: (normalize_name(pkg["name"]), pkg.get("version", "")))
    return resolved


def main() -> int:
    repo_root = Path(__file__).resolve().parents[1]
    pyproject_path = repo_root / "pyproject.toml"
    uv_lock_path = repo_root / "uv.lock"
    aiprovider_dir = repo_root / "aiprovider"

    pyproject_text = pyproject_path.read_text(encoding="utf-8")
    uv_lock_text = uv_lock_path.read_text(encoding="utf-8")

    requires_python, pyproject_deps = project_dependency_map(pyproject_text)
    imported_modules = collect_runtime_imports(aiprovider_dir)
    relevant_roots = resolve_relevant_root_dependencies(pyproject_deps, imported_modules)
    relevant_packages = dependency_closure(lock_package_map(uv_lock_text), relevant_roots)

    fingerprint_payload = {
        "requires_python": requires_python,
        "relevant_root_dependencies": relevant_roots,
        "lock": {**lock_metadata(uv_lock_text), "packages": relevant_packages},
    }

    print(json.dumps(fingerprint_payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")))
    return 0


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