Set up the GitHub check

name: PinDiff model dependencies

on:
  pull_request:

permissions:
  contents: read

jobs:
  pindiff:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - uses: ./.github/actions/audit
        with:
          token: ${{ secrets.PINDIFF_TOKEN }}
          api-url: https://pindiff.com

name: PinDiff model dependency check
description: >
  Scan a checkout for Hugging Face model dependencies and report pinned, floating
  and unknown revisions, plus relevant upstream changes. The scan is static: model
  code is not executed. A completed check is informational and does not fail the
  job because a dependency is floating or a model changed upstream. The job fails
  only when the scanner cannot run or the PinDiff token is rejected.
inputs:
  token:
    description: PinDiff API token with the checks scope. Store it as a secret named PINDIFF_TOKEN.
    required: true
  api-url:
    description: PinDiff base URL. Defaults to https://pindiff.com.
    required: false
    default: https://pindiff.com
  path:
    description: Directory or Python file to scan.
    required: false
    default: .
  compare-base:
    description: Compare with the pull request base ref when it is available in the checkout.
    required: false
    default: "true"
runs:
  using: composite
  steps:
    - name: Report model dependencies
      shell: bash
      env:
        PINDIFF_TOKEN: ${{ inputs.token }}
        PINDIFF_API_URL: ${{ inputs.api-url }}
        PINDIFF_PATH: ${{ inputs.path }}
        PINDIFF_COMPARE_BASE: ${{ inputs.compare-base }}
        PINDIFF_REPOSITORY: ${{ github.repository }}
        PINDIFF_PULL_REQUEST: ${{ github.event.pull_request.number }}
        PINDIFF_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
        PINDIFF_BASE_REF: ${{ github.base_ref }}
      run: python3 "$GITHUB_ACTION_PATH/check.py"

"""Report a PinDiff model-dependency check from GitHub Actions.

Scans with the PinDiff CLI (static analysis only; customer code is not executed)
and posts dependency metadata to the PinDiff checks API. The token is never printed.

Exit 1 only when the scanner cannot run or the token is missing or rejected.
Floating revisions, unknown revisions and upstream changes do not fail the job.
If the API cannot be reached, the local scan is still printed and the job succeeds.
"""

import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request

REPO = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$")
REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,250}$")
SHA = re.compile(r"^[0-9a-fA-F]{7,64}$")
CONFIG_STATUSES = {401, 403, 422}


def cli_root():
    override = os.environ.get("PINDIFF_CLI")
    candidates = []
    if override:
        candidates.append(override)
    here = os.path.dirname(os.path.abspath(__file__))
    candidates.append(here)
    candidates.append(os.path.abspath(os.path.join(here, "..", "..", "..", "cli")))
    for path in candidates:
        if os.path.isdir(os.path.join(path, "pindiff_cli")):
            return path
    return None


def classify_http(status):
    """ok, config (fail the job) or unavailable (keep the local scan, do not fail)."""
    if status == 200:
        return "ok"
    if status in CONFIG_STATUSES:
        return "config"
    return "unavailable"


def redact(text, secret):
    if secret and text:
        text = text.replace(secret, "[redacted]")
    return text


def error_message(raw):
    try:
        data = json.loads(raw)
    except (json.JSONDecodeError, TypeError):
        return (raw or "")[:500]
    if isinstance(data, dict):
        message = data.get("message")
        if isinstance(message, str) and message:
            return message
        errors = data.get("errors")
        if isinstance(errors, dict):
            parts = []
            for msgs in errors.values():
                if isinstance(msgs, list):
                    parts.extend(str(item) for item in msgs)
                elif isinstance(msgs, str):
                    parts.append(msgs)
            if parts:
                return "; ".join(parts)
    return (raw or "")[:500]


def git(cwd, args):
    try:
        proc = subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.TimeoutExpired):
        return None
    if proc.returncode != 0:
        return None
    return proc.stdout.strip()


def scan_base(scan_path, head_path, base_ref):
    """Return (result, reason). reason is set when the base tree could not be scanned."""
    if not base_ref or not REF.match(base_ref) or ".." in base_ref:
        return None, "no base ref"
    toplevel = git(head_path, ["rev-parse", "--show-toplevel"])
    if not toplevel:
        return None, "this checkout is not a git repository"
    rel = os.path.relpath(os.path.abspath(head_path), toplevel)
    ref = None
    for candidate in ("origin/" + base_ref, base_ref):
        if git(toplevel, ["rev-parse", "--verify", "--quiet", candidate]) is not None:
            ref = candidate
            break
    if ref is None:
        return None, "base ref {} is not in this checkout".format(base_ref)

    tmp = tempfile.mkdtemp(prefix="pindiff-base-")
    added = False
    try:
        if git(toplevel, ["worktree", "add", "--detach", tmp, ref]) is None:
            return None, "could not check out {}".format(base_ref)
        added = True
        target = tmp if rel in (".", "") else os.path.join(tmp, rel)
        if not os.path.exists(target):
            return None, "the scanned path does not exist on the base branch"
        return scan_path(target), None
    finally:
        if added:
            git(toplevel, ["worktree", "remove", "--force", tmp])
        shutil.rmtree(tmp, ignore_errors=True)


def post(url, token, body):
    data = json.dumps(body).encode("utf-8")
    request = urllib.request.Request(url, data=data, method="POST", headers={
        "Authorization": "Bearer " + token,
        "Content-Type": "application/json",
        "Accept": "application/json",
        "User-Agent": "pindiff-github-action",
    })
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.status, response.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode("utf-8", "replace")
    except urllib.error.URLError as exc:
        return 0, str(exc.reason)


def write_summary(text):
    print(text)
    path = os.environ.get("GITHUB_STEP_SUMMARY")
    if not path:
        return
    with open(path, "a", encoding="utf-8") as handle:
        handle.write(text)
        if not text.endswith("\n"):
            handle.write("\n")


def env_int(name):
    value = os.environ.get(name) or ""
    return int(value) if value.isdigit() else None


def main():
    token = os.environ.get("PINDIFF_TOKEN") or ""
    if not token:
        print("PinDiff token is missing. Create one on the GitHub integration page and store it as PINDIFF_TOKEN.", file=sys.stderr)
        return 1

    repository = os.environ.get("PINDIFF_REPOSITORY") or os.environ.get("GITHUB_REPOSITORY") or ""
    if not REPO.match(repository):
        print("PinDiff needs the GitHub repository as owner/name.", file=sys.stderr)
        return 1

    root = cli_root()
    if root is None:
        print("PinDiff CLI was not found next to this action. Download the PinDiff GitHub Action bundle, which includes .github/actions/audit/pindiff_cli/.", file=sys.stderr)
        return 1
    if root not in sys.path:
        sys.path.insert(0, root)

    from pindiff_cli.__main__ import payload, render
    from pindiff_cli.scanner import scan_path

    path = os.environ.get("PINDIFF_PATH") or "."
    if not os.path.exists(path):
        print("pindiff: no such file or directory: {}".format(path), file=sys.stderr)
        return 1

    try:
        head = scan_path(path)
    except OSError as exc:
        print("pindiff: could not scan {}: {}".format(path, exc), file=sys.stderr)
        return 1

    compare = (os.environ.get("PINDIFF_COMPARE_BASE") or "true").lower() != "false"
    base_ref = os.environ.get("PINDIFF_BASE_REF") or os.environ.get("GITHUB_BASE_REF") or ""
    base_doc = None
    base_unavailable = False
    if compare and base_ref:
        base, reason = scan_base(scan_path, path, base_ref)
        if base is None:
            base_unavailable = True
            print("PinDiff could not compare with {}: {}.".format(base_ref, reason), file=sys.stderr)
        else:
            base_doc = payload(base)

    sha = (os.environ.get("PINDIFF_HEAD_SHA") or os.environ.get("GITHUB_SHA") or "").strip()
    body = {
        "repository": repository,
        "pull_request": env_int("PINDIFF_PULL_REQUEST"),
        "head_sha": sha if SHA.match(sha) else None,
        "base_ref": base_ref if REF.match(base_ref) and ".." not in base_ref else None,
        "base_unavailable": base_unavailable,
        "head": payload(head),
        "base": base_doc,
    }

    api = (os.environ.get("PINDIFF_API_URL") or "https://pindiff.com").rstrip("/")
    status, raw = post(api + "/api/checks", token, body)
    kind = classify_http(status)

    if kind == "ok":
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            parsed = {}
        summary = parsed.get("summary") if isinstance(parsed, dict) else None
        if not isinstance(summary, str) or not summary:
            summary = "PinDiff completed the dependency check."
        write_summary(redact(summary, token))
        return 0

    local = io.StringIO()
    render(head, local, color=False)
    note = local.getvalue().rstrip()
    if kind == "config":
        print(redact(error_message(raw), token), file=sys.stderr)
        if note:
            print(note, file=sys.stderr)
        return 1

    warning = "PinDiff could not add upstream intelligence because the API did not respond. The dependency scan itself completed and this job is not failed."
    text = warning + ("\n\n" + note if note else "")
    write_summary(redact(text, token))
    return 0


if __name__ == "__main__":
    sys.exit(main())

__version__ = "0.1.0"

"""pindiff audit: find Hugging Face model dependencies and whether they are pinned, floating or unknown.

Scanning is local and static. Nothing is sent anywhere; --json prints only dependency
metadata (repository ids, revisions, file paths, line numbers) for upload to PinDiff.
"""

import argparse
import json
import os
import sys
from collections import OrderedDict
from datetime import datetime, timezone
from typing import List

from . import __version__
from . import revision as rev
from .scanner import FORMAT, Finding, ScanResult, scan_path

STATUS_ORDER = [rev.PINNED, rev.FLOATING, rev.UNKNOWN]
STATUS_MARK = {rev.PINNED: "✓", rev.FLOATING: "⚠", rev.UNKNOWN: "?"}
STATUS_MARK_ASCII = {rev.PINNED: "+", rev.FLOATING: "!", rev.UNKNOWN: "?"}
SUMMARY = {
    rev.PINNED: "pinned to a specific revision",
    rev.FLOATING: "using floating revisions (main, tags, branches, no revision)",
    rev.UNKNOWN: "with unknown revision or repository",
}

# ANSI SGR codes
GREEN, YELLOW, RED, BOLD, DIM = "32", "33", "31", "1", "2"
STATUS_COLOR = {rev.PINNED: GREEN, rev.FLOATING: YELLOW, rev.UNKNOWN: RED}

MAX_REPO = 48
MAX_REVISION = 24


def group(findings: List[Finding]) -> List[List[Finding]]:
    """One dependency per repository and revision; each unresolved reference stands alone."""
    groups: "OrderedDict[tuple, List[Finding]]" = OrderedDict()
    for index, finding in enumerate(findings):
        if finding.repo_id is None:
            key = ("unresolved", index)
        else:
            key = (finding.repo_id, finding.revision_source, finding.revision)
        groups.setdefault(key, []).append(finding)
    return sorted(groups.values(), key=lambda g: (STATUS_ORDER.index(g[0].revision_type), (g[0].repo_id or "~").lower()))


def revision_label(finding: Finding) -> str:
    if finding.revision_source == rev.DEFAULT:
        return "main (default)"
    if finding.revision_source == rev.EXPLICIT and finding.revision:
        return finding.revision[:7] if rev.is_commit(finding.revision) else finding.revision
    if finding.revision_source == rev.DYNAMIC:
        return "{} (dynamic)".format(finding.revision_expression)
    return "(unknown)"


def repository_label(finding: Finding) -> str:
    if finding.repo_id is not None:
        return finding.repo_id
    expression = finding.repo_expression or ""
    literal = expression[:1] in ("'", '"')
    return "{} ({})".format(expression, "not org/name" if literal else "dynamic")


def use_color(mode: str, stream) -> bool:
    if mode != "auto":
        return mode == "always"
    if os.environ.get("NO_COLOR"):
        return False
    if os.environ.get("FORCE_COLOR"):
        return True
    return os.environ.get("TERM") != "dumb" and hasattr(stream, "isatty") and stream.isatty()


def _fit(text: str, width: int) -> str:
    return text if len(text) <= width else text[: width - 1] + "…"


def render(result: ScanResult, out, color: bool = False) -> None:
    def paint(text: str, *codes: str) -> str:
        return "\033[{}m{}\033[0m".format(";".join(codes), text) if color and codes and text else text

    def say(text: str = "") -> None:
        print(text, file=out)

    marks, rule = STATUS_MARK, "─"
    try:
        ("".join(marks.values()) + rule + "…").encode(out.encoding or "utf-8")
    except (UnicodeEncodeError, LookupError):
        marks, rule = STATUS_MARK_ASCII, "-"

    groups = group(result.findings)
    counts = {status: sum(1 for g in groups if g[0].revision_type == status) for status in STATUS_ORDER}

    say(paint("Scanning project for Hugging Face model dependencies...", DIM))
    say()
    noun = "dependency" if len(groups) == 1 else "dependencies"
    say(paint("Found {} Hugging Face {} in {} Python files{}".format(
        len(groups), noun, result.files_scanned, ":" if groups else "."), BOLD))

    if groups:
        for status in STATUS_ORDER:
            say(paint("{} {} {}".format(marks[status], counts[status], SUMMARY[status]), STATUS_COLOR[status]))

        rows = []
        for findings in groups:
            first = findings[0]
            locations = ["{}:{}".format(f.source_file, f.source_line) for f in findings]
            rows.append((first.revision_type, _fit(repository_label(first), MAX_REPO), _fit(revision_label(first), MAX_REVISION), locations))

        headers = ("STATUS", "REPOSITORY", "REVISION", "FILE")
        widths = [
            max(len(headers[0]), max(len(marks[s]) + 1 + len(s) for s, _, _, _ in rows)),
            max(len(headers[1]), max(len(r) for _, r, _, _ in rows)),
            max(len(headers[2]), max(len(v) for _, _, v, _ in rows)),
            max(len(headers[3]), max(len(loc) for _, _, _, locs in rows for loc in locs)),
        ]
        gap = "   "

        say()
        say(paint(gap.join(h.ljust(w) for h, w in zip(headers, widths)).rstrip(), DIM))
        say(paint(rule * (sum(widths) + len(gap) * 3), DIM))
        for status, repo, revision, locations in rows:
            tone = STATUS_COLOR[status]
            cells = [
                paint((marks[status] + " " + status.upper()).ljust(widths[0]), tone, BOLD),
                paint(repo.ljust(widths[1]), DIM) if status == rev.UNKNOWN and repo.endswith(")") else repo.ljust(widths[1]),
                revision.ljust(widths[2]) if status == rev.PINNED else paint(revision.ljust(widths[2]), tone),
                paint(locations[0], DIM),
            ]
            say(gap.join(cells))
            for location in locations[1:]:
                say(" " * (sum(widths[:3]) + len(gap) * 3) + paint(location, DIM))

        if counts[rev.UNKNOWN]:
            say()
            say(paint("PinDiff does not evaluate expressions. After uploading, record the model each unknown reference loads.", DIM))

    if result.skipped:
        say()
        say(paint("Skipped {} file(s):".format(len(result.skipped)), YELLOW))
        for path, reason in result.skipped:
            say(paint("  {}  ({})".format(path, reason), DIM))

    if groups:
        url = os.environ.get("PINDIFF_URL", "https://pindiff.com").rstrip("/")
        say()
        say("To monitor these, run " + paint("pindiff audit . --json > pindiff-audit.json", BOLD))
        say("and upload the file at " + paint("{}/app/audit".format(url), BOLD))


def payload(result: ScanResult) -> dict:
    return {
        "format": FORMAT,
        "tool": {"name": "pindiff", "version": __version__},
        "project": os.path.basename(result.root.rstrip(os.sep)) or "project",
        "generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
        "files_scanned": result.files_scanned,
        "files_skipped": len(result.skipped),
        "findings": [f.to_dict() for f in result.findings],
    }


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(prog="pindiff", description="PinDiff model dependency tools.")
    parser.add_argument("--version", action="version", version="pindiff " + __version__)
    commands = parser.add_subparsers(dest="command")

    audit = commands.add_parser("audit", help="Find Hugging Face model dependencies in a project.")
    audit.add_argument("path", nargs="?", default=".", help="Project directory or Python file (default: .)")
    audit.add_argument("--json", action="store_true", help="Print dependency metadata as JSON for upload to PinDiff.")
    audit.add_argument("-o", "--output", help="Write the JSON to this file instead of stdout (implies --json).")
    audit.add_argument("--color", choices=["auto", "always", "never"], default="auto",
                       help="Colour the summary (default: auto, when writing to a terminal; NO_COLOR disables).")

    args = parser.parse_args(argv)
    if args.command != "audit":
        parser.print_help()
        return 2

    if not os.path.exists(args.path):
        print("pindiff: no such file or directory: {}".format(args.path), file=sys.stderr)
        return 2

    result = scan_path(args.path)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as handle:
            json.dump(payload(result), handle, indent=2)
            handle.write("\n")
        print("Wrote {} references to {}".format(len(result.findings), args.output), file=sys.stderr)
    elif args.json:
        json.dump(payload(result), sys.stdout, indent=2)
        sys.stdout.write("\n")
    else:
        color = use_color(args.color, sys.stdout)
        if color and os.name == "nt":
            os.system("")  # enables ANSI escape processing in the Windows console
        render(result, sys.stdout, color)
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Static scanner for Hugging Face model references in Python source.

Files are parsed with the standard-library ast module. Scanned code is never imported,
executed or evaluated; anything that is not a plain string literal is reported as dynamic.

Detects:
- <obj>.from_pretrained(...)    transformers, diffusers, peft and other from_pretrained loaders
- snapshot_download(...)        huggingface_hub
- hf_hub_download(...)          huggingface_hub
- huggingface.co model URLs in string literals
"""

import ast
import os
import re
from dataclasses import asdict, dataclass
from typing import Callable, Dict, List, Optional, Set, Tuple

from . import revision as rev

FORMAT = "pindiff.audit.v1"

SKIP_DIRS = {
    "node_modules", "__pycache__", "venv", "env", "site-packages",
    "build", "dist", "htmlcov",
}
MAX_FILE_BYTES = 2_000_000
MAX_EXPRESSION = 120

REPO_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*$")
HF_URL = re.compile(
    r"https?://(?:www\.)?(?:huggingface\.co|hf\.co)/"
    r"([A-Za-z0-9][A-Za-z0-9_.-]*)/([A-Za-z0-9][A-Za-z0-9_.-]*)"
    r"(?:/(?:resolve|tree|blob|commit)/([A-Za-z0-9][A-Za-z0-9._-]*))?"
)
# First URL path segments that are not model organisations.
URL_RESERVED = {
    "datasets", "spaces", "docs", "blog", "api", "models", "papers", "collections",
    "organizations", "settings", "join", "login", "pricing", "tasks", "learn", "posts",
    "chat", "new", "enterprise", "search", "transformers", "huggingface_hub",
}

FROM_PRETRAINED_LIBRARIES = {"transformers", "diffusers", "peft"}
HUB_FUNCTIONS = {"snapshot_download", "hf_hub_download"}

HIGH = "high"
MEDIUM = "medium"
LOW = "low"


@dataclass
class Finding:
    repo_id: Optional[str]
    repo_expression: Optional[str]
    revision: Optional[str]
    revision_expression: Optional[str]
    revision_source: str
    revision_type: str
    source_file: str
    source_line: int
    detector: str
    confidence: str

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass
class ScanResult:
    root: str
    files_scanned: int
    skipped: List[Tuple[str, str]]
    findings: List[Finding]


def is_local_path(value: str) -> bool:
    return (
        value == ""
        or value.startswith(("/", "./", "../", "~", "\\"))
        or "\\" in value
        or value.count("/") > 1
        or re.match(r"^[A-Za-z]:", value) is not None
    )


def _collapse(text: str) -> str:
    text = re.sub(r"\s+", " ", text).strip()
    return text if len(text) <= MAX_EXPRESSION else text[: MAX_EXPRESSION - 1] + "…"


def _callee(func: ast.expr) -> Tuple[Optional[str], Optional[str], bool]:
    """Call target => (final name, root name of an attribute chain, is attribute)."""
    if isinstance(func, ast.Name):
        return func.id, func.id, False
    if isinstance(func, ast.Attribute):
        node = func.value
        while isinstance(node, ast.Attribute):
            node = node.value
        return func.attr, node.id if isinstance(node, ast.Name) else None, True
    return None, None, False


class _FileScanner(ast.NodeVisitor):
    def __init__(self, source: str, rel_path: str, local_exists: Callable[[str], bool]):
        self.source = source
        self.rel_path = rel_path
        self.local_exists = local_exists
        self.imports: Dict[str, str] = {}
        self.consumed: Set[int] = set()
        self.findings: List[Finding] = []

    def scan(self, tree: ast.AST) -> List[Finding]:
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    top = alias.name.split(".")[0]
                    self.imports[alias.asname or top] = top
            elif isinstance(node, ast.ImportFrom) and node.module and not node.level:
                top = node.module.split(".")[0]
                for alias in node.names:
                    self.imports[alias.asname or alias.name] = top
            elif isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
                self.consumed.add(id(node.value))
        self.visit(tree)
        return self.findings

    def _text(self, node: ast.AST) -> str:
        segment = ast.get_source_segment(self.source, node)
        if segment is None:
            segment = ast.unparse(node) if hasattr(ast, "unparse") else "<expression>"
        return _collapse(segment)

    def visit_Call(self, node: ast.Call) -> None:
        name, root, is_attribute = _callee(node.func)

        if name == "from_pretrained" and is_attribute:
            library = self.imports.get(root) if root else None
            if library in FROM_PRETRAINED_LIBRARIES:
                detector = library + ".from_pretrained"
            else:
                detector = "from_pretrained"
            if library == "peft":
                self._loader(node, detector, 1, "model_id", known=True)
            else:
                self._loader(node, detector, 0, "pretrained_model_name_or_path", known=detector != "from_pretrained")
        elif name in HUB_FUNCTIONS:
            library = self.imports.get(root if is_attribute else name) if root else None
            if not self._non_model_repo_type(node):
                self._loader(node, "huggingface_hub." + name, 0, "repo_id", known=library == "huggingface_hub")

        self.generic_visit(node)

    def visit_Constant(self, node: ast.Constant) -> None:
        if isinstance(node.value, str) and id(node) not in self.consumed:
            for match in HF_URL.finditer(node.value):
                org, name, revision = match.group(1), match.group(2), match.group(3)
                if org.lower() in URL_RESERVED:
                    continue
                repo_id = org + "/" + name
                source = rev.EXPLICIT if revision else rev.UNSPECIFIED
                self.findings.append(Finding(
                    repo_id=repo_id,
                    repo_expression=_collapse(match.group(0)),
                    revision=revision,
                    revision_expression=None,
                    revision_source=source,
                    revision_type=rev.revision_type(repo_id, revision, source),
                    source_file=self.rel_path,
                    source_line=node.lineno,
                    detector="huggingface.url",
                    confidence=MEDIUM,
                ))

    def _non_model_repo_type(self, node: ast.Call) -> bool:
        for keyword in node.keywords:
            if keyword.arg == "repo_type" and isinstance(keyword.value, ast.Constant):
                return isinstance(keyword.value.value, str) and keyword.value.value != "model"
        return False

    def _loader(self, node: ast.Call, detector: str, position: int, keyword: str, known: bool) -> None:
        keywords = {k.arg: k.value for k in node.keywords if k.arg is not None}
        star_kwargs = next((k for k in node.keywords if k.arg is None), None)

        repo_node: Optional[ast.AST] = keywords.get(keyword)
        if repo_node is None:
            for index, arg in enumerate(node.args):
                if isinstance(arg, ast.Starred):
                    repo_node = arg
                    break
                if index == position:
                    repo_node = arg
                    break
        if repo_node is None and star_kwargs is not None:
            repo_node = star_kwargs
        if repo_node is None:
            return

        repo_id: Optional[str] = None
        if isinstance(repo_node, ast.Constant) and isinstance(repo_node.value, str):
            self.consumed.add(id(repo_node))
            value = repo_node.value.strip()
            if is_local_path(value) or self.local_exists(value):
                return
            if REPO_ID.match(value):
                repo_id = value
        repo_expression = self._text(repo_node)

        revision: Optional[str] = None
        revision_expression: Optional[str] = None
        revision_node = keywords.get("revision")
        if revision_node is None:
            if star_kwargs is not None:
                source, revision_expression = rev.DYNAMIC, self._text(star_kwargs)
            else:
                source = rev.DEFAULT
        elif isinstance(revision_node, ast.Constant) and revision_node.value is None:
            source, revision_expression = rev.DEFAULT, self._text(revision_node)
        elif isinstance(revision_node, ast.Constant) and isinstance(revision_node.value, str):
            source, revision, revision_expression = rev.EXPLICIT, revision_node.value, self._text(revision_node)
        else:
            source, revision_expression = rev.DYNAMIC, self._text(revision_node)

        self.findings.append(Finding(
            repo_id=repo_id,
            repo_expression=repo_expression,
            revision=revision,
            revision_expression=revision_expression,
            revision_source=source,
            revision_type=rev.revision_type(repo_id, revision, source),
            source_file=self.rel_path,
            source_line=node.lineno,
            detector=detector,
            confidence=LOW if repo_id is None else (HIGH if known else MEDIUM),
        ))


def scan_source(source: str, rel_path: str, local_exists: Callable[[str], bool] = lambda _: False) -> List[Finding]:
    """Findings in one file's source. Raises SyntaxError/ValueError when it cannot be parsed."""
    tree = ast.parse(source, filename=rel_path)
    findings = _FileScanner(source, rel_path, local_exists).scan(tree)
    return sorted(findings, key=lambda f: f.source_line)


def _python_files(root: str):
    if os.path.isfile(root):
        yield root
        return
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = sorted(d for d in dirnames if d not in SKIP_DIRS and not d.startswith("."))
        for filename in sorted(filenames):
            path = os.path.join(dirpath, filename)
            if filename.endswith(".py") and not os.path.islink(path):
                yield path


def scan_path(root: str) -> ScanResult:
    root = os.path.abspath(root)
    base = os.path.dirname(root) if os.path.isfile(root) else root
    findings: List[Finding] = []
    skipped: List[Tuple[str, str]] = []
    scanned = 0

    for path in _python_files(root):
        rel_path = os.path.relpath(path, base).replace(os.sep, "/")
        if os.path.getsize(path) > MAX_FILE_BYTES:
            skipped.append((rel_path, "larger than 2 MB"))
            continue
        try:
            with open(path, "rb") as handle:
                source = handle.read().decode("utf-8-sig")
        except (OSError, UnicodeDecodeError):
            skipped.append((rel_path, "could not be read as UTF-8"))
            continue

        file_dir = os.path.dirname(path)

        def local_exists(value: str, file_dir: str = file_dir) -> bool:
            return os.path.exists(os.path.join(base, value)) or os.path.exists(os.path.join(file_dir, value))

        try:
            findings.extend(scan_source(source, rel_path, local_exists))
        except (SyntaxError, ValueError, RecursionError, MemoryError):
            skipped.append((rel_path, "could not be parsed as Python"))
            continue
        scanned += 1

    return ScanResult(root=root, files_scanned=scanned, skipped=skipped, findings=findings)

"""Revision classification, matching the PinDiff server (App\\Support\\Revision).

PINNED: an exact commit SHA (immutable).
FLOATING: a branch, main/latest, tag or the implicit default branch; may resolve differently later.
UNKNOWN: the revision cannot be determined statically.
"""

import re
from typing import Optional

PINNED = "pinned"
FLOATING = "floating"
UNKNOWN = "unknown"

# How the revision was found in code.
EXPLICIT = "explicit"  # a string literal, e.g. revision="82fa41f"
DEFAULT = "default"  # no revision argument: the loader uses the default branch
DYNAMIC = "dynamic"  # an expression PinDiff will not evaluate, e.g. revision=REV
UNSPECIFIED = "unspecified"  # a reference that names no revision, e.g. a model page URL

_COMMIT = re.compile(r"^[0-9a-f]{7,40}$", re.IGNORECASE)
_REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")


def is_commit(value: str) -> bool:
    return bool(_COMMIT.match(value))


def is_valid_ref(value: str) -> bool:
    return len(value) <= 200 and bool(_REF.match(value)) and ".." not in value and not value.endswith("/")


def revision_type(repo_id: Optional[str], revision: Optional[str], source: str) -> str:
    if repo_id is None:
        return UNKNOWN
    if source == DEFAULT:
        return FLOATING
    if source != EXPLICIT or revision is None:
        return UNKNOWN
    value = revision.strip()
    if value == "" or value.lower() == UNKNOWN or not is_valid_ref(value):
        return UNKNOWN
    return PINNED if is_commit(value) else FLOATING