Blame for extractor/behaviour/inputs.py

"""Turn the three input shapes (package directory, unified diff, snippet) into analysis units.

A unit is one file's text plus an optional focus set of 1-based line numbers. In diff mode the text is the new side
of the file as far as the diff shows it (context and added lines at their real line numbers, unknown regions blank)
and the focus is the added lines; behaviours are reported only where they touch the focus.
"""
from __future__ import annotations

import os
import re
from dataclasses import dataclass, field
from pathlib import Path

MAX_PARSE_BYTES = 1_000_000

EXT_LANG = {
    ".py": "python", ".pyw": "python", ".pth": "pth",
    ".js": "javascript", ".mjs": "javascript", ".cjs": "javascript", ".jsx": "javascript",
    ".ts": "typescript", ".mts": "typescript", ".cts": "typescript", ".tsx": "tsx",
    ".vue": "sfc", ".svelte": "sfc",
    ".sh": "shell", ".bash": "shell", ".zsh": "shell", ".ksh": "shell", ".command": "shell",
    ".yml": "yaml", ".yaml": "yaml", ".json": "json", ".toml": "toml", ".cfg": "ini", ".ini": "ini",
}
DOC_EXT = {".md", ".rst", ".txt", ".adoc", ".markdown", ".svg", ".html", ".htm", ".css", ".scss", ".lock", ".csv",
           ".ipynb", ".sql", ".xml", ".po", ".pot"}
SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", ".tox", ".mypy_cache", ".pytest_cache"}
BINARY_EXT = {".node", ".so", ".dll", ".exe", ".dylib", ".pyd", ".pyc", ".pyo", ".wasm", ".bin", ".o", ".a",
              ".class", ".jar", ".dat"}
ARCHIVE_EXT = (".zip", ".tar", ".tgz", ".tar.gz", ".gz", ".bz2", ".xz", ".7z", ".rar", ".whl", ".egg", ".jar",
               ".nupkg", ".crx", ".vsix", ".tar.bz2", ".tar.xz", ".zst")


@dataclass
class Unit:
    path: str                       # relative, forward slashes
    lang: str                       # python|javascript|typescript|tsx|shell|yaml|json|... |other|doc|binary
    text: str = ""
    focus: set | None = None        # None = every line is in scope
    status: str = "file"            # file | added | modified | deleted | snippet
    size: int = 0
    raw_head: bytes = b""           # first bytes for binary/archive typing (package mode)
    binary_note: str = ""           # diff mode: "binary patch" etc.

    def in_focus(self, a: int, b: int | None = None) -> bool:
        if self.focus is None:
            return True
        b = a if b is None else b
        return any(x in self.focus for x in range(a, b + 1)) if b - a < 400 else any(a <= x <= b for x in self.focus)

    def first_focus(self) -> int:
        return min(self.focus) if self.focus else 1


def lang_of(path: str, text: str = "") -> str:
    p = path.lower()
    name = p.rsplit("/", 1)[-1]
    if name in ("dockerfile", "containerfile") or name.endswith(".dockerfile") or name.startswith("dockerfile."):
        return "dockerfile"
    if name in ("makefile", "gnumakefile") or name.endswith(".mk"):
        return "makefile"
    if name in (".npmrc", ".yarnrc", "pip.conf", "pip.ini", ".pypirc", ".netrc", ".env") or name.startswith(".env."):
        return "config"
    if name.startswith(".") and name.endswith("ignore") or name in ("license", "licence", "copying", "notice", "authors",
                                                                    "changelog", "changes", "codeowners", ".gitattributes",
                                                                    ".editorconfig", ".mailmap"):
        return "doc"
    if re.fullmatch(r"(requirements|constraints)[\w.-]*\.(txt|in)", name):
        return "requirements"
    for ext in sorted(EXT_LANG, key=len, reverse=True):
        if p.endswith(ext):
            return EXT_LANG[ext]
    if any(p.endswith(e) for e in BINARY_EXT):
        return "binary"
    if any(p.endswith(e) for e in DOC_EXT):
        return "doc"
    first = text.split("\n", 1)[0] if text else ""
    if first.startswith("#!"):
        if "python" in first:
            return "python"
        if "node" in first or "deno" in first or "bun" in first:
            return "javascript"
        if re.search(r"\b(ba|z|k|da)?sh\b", first):
            return "shell"
    if "/.husky/" in "/" + p or "/.git/hooks/" in "/" + p:
        return "shell"
    return "other"


def guess_snippet_lang(text: str) -> str:
    first = text.lstrip().split("\n", 1)[0]
    if first.startswith("#!"):
        return lang_of("snippet", first)
    py = len(re.findall(r"^\s*(def |import \w|from [\w.]+ import|print\(|with .+:\s*$|if __name__)", text, re.M))
    js = len(re.findall(r"\b(const|let|var|function|=>|require\(|console\.|await |import .+ from )", text))
    ts = len(re.findall(r":\s*(string|number|boolean|any|Promise<)|\binterface\s+\w+|\btype\s+\w+\s*=", text))
    sh = len(re.findall(r"^\s*(\$ )?(curl|wget|cd|ls|echo|export|sudo|apt|npm|pip|git|chmod|rm|mkdir|cat|grep)\b",
                        text, re.M)) + text.count("&&") + text.count(" | ")
    best = max((py, "python"), (js + ts, "typescript" if ts else "javascript"), (sh, "shell"))
    return best[1] if best[0] else "shell"


def read_package(root: Path, include_node_modules: bool = False):
    units = []
    root = root.resolve()
    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
        dirnames[:] = sorted(d for d in dirnames if include_node_modules or d not in SKIP_DIRS)
        for fn in sorted(filenames):
            full = Path(dirpath) / fn
            rel = full.relative_to(root).as_posix()
            if full.is_symlink():
                units.append(Unit(rel, "other", text="", size=0, binary_note="symlink -> " + os.readlink(full)))
                continue
            try:
                size = full.stat().st_size
                with open(full, "rb") as fh:
                    head = fh.read(MAX_PARSE_BYTES + 1)
            except OSError:
                continue
            u = Unit(rel, "other", size=size, raw_head=head[:4096])
            if b"\x00" in head[:8192]:
                u.lang = "binary"
            else:
                u.text = head[:MAX_PARSE_BYTES].decode("utf-8", errors="replace")
                u.lang = lang_of(rel, u.text)
                if u.lang == "binary":
                    u.lang = "other"
            units.append(u)
    return units


_HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")


def _strip_prefix(p: str) -> str:
    p = p.strip().split("\t", 1)[0]
    if p.startswith('"') and p.endswith('"'):
        p = p[1:-1]
    if p in ("/dev/null",):
        return p
    return p[2:] if p[:2] in ("a/", "b/") else p


def parse_diff(diff: str):
    """Return units for every file in a unified (git) diff. Only the new side is kept."""
    files = []
    cur = None

    def start(path):
        nonlocal cur
        cur = {"path": path, "lines": {}, "added": set(), "status": "modified", "binary": "", "old": path}
        files.append(cur)

    new_ln = 0
    old_left = new_left = 0          # remaining hunk body lines, so "--- x" inside a hunk is a removed line
    git_header = False
    for raw in diff.splitlines():
        if cur is not None and (old_left > 0 or new_left > 0):
            tag, body = raw[:1], raw[1:]
            if tag == "\\":
                continue
            if tag == "+":
                cur["lines"][new_ln] = body
                cur["added"].add(new_ln)
                new_ln += 1
                new_left -= 1
                continue
            if tag == "-":
                old_left -= 1
                continue
            if tag == " " or raw == "":
                cur["lines"][new_ln] = body
                new_ln += 1
                new_left -= 1
                old_left -= 1
                continue
            old_left = new_left = 0      # malformed hunk: fall through to header handling
        if raw.startswith("diff --git "):
            m = re.match(r'diff --git "?a/(.+?)"? "?b/(.+?)"?$', raw)
            start(m.group(2) if m else raw.split()[-1].removeprefix("b/"))
            git_header = True
            continue
        if raw.startswith("--- "):
            old = _strip_prefix(raw[4:])
            if not git_header:
                start(old)
            git_header = False
            cur["old"] = old
            if old == "/dev/null":
                cur["status"] = "added"
            continue
        if raw.startswith("+++ "):
            new = _strip_prefix(raw[4:])
            if cur is None:
                start(new)
            if new == "/dev/null":
                cur["status"] = "deleted"
            else:
                cur["path"] = new
            continue
        m = _HUNK.match(raw)
        if m and cur is not None:
            new_ln = int(m.group(3))
            old_left = int(m.group(2)) if m.group(2) is not None else 1
            new_left = int(m.group(4)) if m.group(4) is not None else 1
            git_header = False
            continue
        if cur is None:
            continue
        if raw.startswith("new file mode"):
            cur["status"] = "added"
        elif raw.startswith("deleted file mode"):
            cur["status"] = "deleted"
        elif raw.startswith("rename to "):
            cur["path"] = raw[len("rename to "):]
        elif raw.startswith("Binary files ") or raw.startswith("GIT binary patch"):
            cur["binary"] = raw.strip()
        elif raw.startswith("literal ") and cur.get("binary") and "bytes" not in cur["binary"]:
            cur["binary"] += f" ({raw.split()[1]} bytes)"
    units = []
    for f in files:
        path = f["path"]
        if f["binary"]:
            if f["status"] != "deleted":
                units.append(Unit(path, "binary", status=f["status"], binary_note=f["binary"]))
            continue
        if f["status"] == "deleted":
            continue
        n = max(f["lines"], default=0)
        text = "\n".join(f["lines"].get(i, "") for i in range(1, n + 1))
        lang = lang_of(path, f["lines"].get(1, ""))
        units.append(Unit(path, lang, text=text, focus=set(f["added"]), status=f["status"],
                          size=sum(len(f["lines"][i]) + 1 for i in f["added"])))
    return units


def snippet_unit(text: str, lang: str | None = None, name: str = "snippet") -> Unit:
    lang = {"js": "javascript", "ts": "typescript", "py": "python", "sh": "shell", "bash": "shell"}.get(lang or "", lang)
    return Unit(name, lang or guess_snippet_lang(text), text=text, status="snippet", size=len(text))