| """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 |
| lang: str |
| text: str = "" |
| focus: set | None = None |
| status: str = "file" |
| size: int = 0 |
| raw_head: bytes = b"" |
| binary_note: str = "" |
|
|
| 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 |
| 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 |
| 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)) |
|
|
