backend/modules/ingestion/parser.py

"""Parse dependency files from common package managers."""

from __future__ import annotations

import json
import os
import re
import xml.etree.ElementTree as ET

import toml
import yaml


def parse_dependencies(repo_path: str) -> dict:
    deps = {"npm": [], "pip": [], "maven": [], "gem": [], "go": [], "cargo": [], "composer": []}
    deps["npm"] = _parse_package_json(os.path.join(repo_path, "package.json"))
    deps["pip"] = _parse_requirements(os.path.join(repo_path, "requirements.txt"))
    deps["pip"].extend(_parse_pyproject(os.path.join(repo_path, "pyproject.toml")))
    deps["maven"] = _parse_pom(os.path.join(repo_path, "pom.xml"))
    deps["gem"] = _parse_gemfile(os.path.join(repo_path, "Gemfile"))
    deps["go"] = _parse_go_mod(os.path.join(repo_path, "go.mod"))
    deps["cargo"] = _parse_cargo(os.path.join(repo_path, "Cargo.toml"))
    deps["composer"] = _parse_composer(os.path.join(repo_path, "composer.json"))
    return {key: sorted(set(value)) for key, value in deps.items()}


def _parse_package_json(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    try:
        with open(path, "r", encoding="utf-8") as file:
            pkg = json.load(file)
        merged = {}
        for key in ("dependencies", "devDependencies", "peerDependencies", "optionalDependencies"):
            merged.update(pkg.get(key, {}))
        return list(merged.keys())
    except (OSError, json.JSONDecodeError):
        return []


def _parse_requirements(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    deps = []
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as file:
            for raw_line in file:
                line = raw_line.strip()
                if not line or line.startswith("#") or line.startswith(("-r", "--")):
                    continue
                package = re.split(r"[<>=~!\[]", line, maxsplit=1)[0].strip()
                if package:
                    deps.append(package)
    except OSError:
        return []
    return deps


def _parse_pyproject(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    try:
        data = toml.load(path)
    except Exception:
        return []
    deps = []
    project = data.get("project", {})
    for dep in project.get("dependencies", []):
        package = re.split(r"[<>=~!\[]", dep, maxsplit=1)[0].strip()
        if package:
            deps.append(package)
    poetry_deps = data.get("tool", {}).get("poetry", {}).get("dependencies", {})
    deps.extend(key for key in poetry_deps.keys() if key.lower() != "python")
    return deps


def _parse_pom(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    try:
        tree = ET.parse(path)
        root = tree.getroot()
        deps = []
        for dependency in root.findall(".//{*}dependency"):
            group = dependency.find("{*}groupId")
            artifact = dependency.find("{*}artifactId")
            if artifact is not None:
                deps.append(f"{group.text if group is not None else ''}:{artifact.text}".strip(":"))
        return deps
    except Exception:
        return []


def _parse_gemfile(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    deps = []
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as file:
            for line in file:
                match = re.match(r"\s*gem\s+['\"]([^'\"]+)['\"]", line)
                if match:
                    deps.append(match.group(1))
    except OSError:
        return []
    return deps


def _parse_go_mod(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    deps = []
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as file:
            for line in file:
                stripped = line.strip()
                if not stripped or stripped.startswith(("module ", "go ", "//", "require (", ")")):
                    continue
                if stripped.startswith("require "):
                    parts = stripped.split()
                    if len(parts) >= 2:
                        deps.append(parts[1])
                elif "/" in stripped:
                    deps.append(stripped.split()[0])
    except OSError:
        return []
    return deps


def _parse_cargo(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    try:
        data = toml.load(path)
        return list(data.get("dependencies", {}).keys()) + list(data.get("dev-dependencies", {}).keys())
    except Exception:
        return []


def _parse_composer(path: str) -> list[str]:
    if not os.path.exists(path):
        return []
    try:
        with open(path, "r", encoding="utf-8") as file:
            data = json.load(file)
        merged = {}
        merged.update(data.get("require", {}))
        merged.update(data.get("require-dev", {}))
        return [key for key in merged.keys() if key != "php"]
    except (OSError, json.JSONDecodeError):
        return []


def parse_yaml_file(path: str) -> dict:
    if not os.path.exists(path):
        return {}
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as file:
            return yaml.safe_load(file) or {}
    except Exception:
        return {}