scripts/docs_claims.py — Docs vs Code Checker

#!/usr/bin/env python3
"""Offline documentation claim discovery and evidence report helper.

By Zero Width Lab (zerowidthlab.com)
"""
from __future__ import annotations

import argparse
import errno
import hashlib
import json
import os
import re
import sys
import stat
from urllib.parse import unquote
import unicodedata
from pathlib import Path

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
SKIP = {".git", "node_modules", ".venv", "venv", "dist", "build", "__pycache__"}
CLAIM_PATTERN = r"(?i)(--[a-z][\w-]*|\b[A-Z][A-Z0-9_]{2,}\b|`[^`]+`|\b\d+(?:\.\d+)*\s*(?:tests?|files?|seconds?|ms|MB|GB|bytes?|%)\b|\b(?:version|default|exit code|supports?|windows|linux|macos|python)\b)"
FLAG_PATTERN = r"--[a-z][a-z0-9-]*"
ENV_PATTERN = r"(?:os\.environ(?:\.get)?\s*\[?\s*\(?\s*|(?:os\.)?getenv\s*\()\s*['\"]([A-Za-z_][A-Za-z0-9_]*)['\"]"
VERSION_PATTERN = r"\b(?:__version__|VERSION|version)\s*=\s*['\"]([^'\"]+)['\"]"


def display_text(value: object) -> str:
    """Keep untrusted content on one visible output line."""
    result = []
    for char in str(value):
        number = ord(char)
        if char == "\\":
            result.append("\\\\")
        elif char == "|":
            result.append("\\|")
        elif unicodedata.category(char).startswith("C") or unicodedata.category(char) in {"Zl", "Zp"}:
            result.append(f"\\u{number:04x}" if number <= 0xffff else f"\\U{number:08x}")
        else:
            result.append(char)
    return "".join(result)


def regular_file(path: Path) -> bool:
    return not path.is_symlink() and path.is_file()


MAX_BYTES = 1024 * 1024


def require_platform() -> None:
    if sys.platform not in {"darwin", "linux"} or os.name != "posix" or not all(hasattr(os, name) for name in ("O_DIRECTORY", "O_NOFOLLOW")) or os.open not in os.supports_dir_fd:
        raise ValueError("supported platforms: macOS and Linux with POSIX directory descriptors and no-follow APIs")


def validate_path(path: Path) -> None:
    raw = str(path)
    decoded = unquote(raw)
    if "\0" in raw or "\\" in raw or decoded != raw or ".." in path.parts:
        raise ValueError("path traversal, encoded paths and backslashes are refused")


def open_child_directory(name: str, parent: int) -> int:
    try:
        return os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
    except OSError as error:
        if error.errno in {errno.ENOTDIR, errno.ELOOP}:
            raise ValueError("a path component is a symlink; use the physical path (for example /private/tmp)") from error
        raise


def validate_entry_name(name: str) -> None:
    if name in {"", ".", ".."} or "\0" in name or "/" in name:
        raise ValueError("unsafe repository entry name")


def open_relative_directory(root_fd: int, parts: tuple[str, ...]) -> int:
    descriptor = os.dup(root_fd)
    try:
        for part in parts:
            validate_entry_name(part)
            new = open_child_directory(part, descriptor)
            os.close(descriptor)
            descriptor = new
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def open_directory(path: Path) -> int:
    validate_path(path)
    absolute = path.absolute()
    descriptor = os.open(absolute.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        for part in absolute.parts[1:]:
            new = open_child_directory(part, descriptor)
            os.close(descriptor)
            descriptor = new
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def identity(descriptor: int) -> list[int]:
    value = os.fstat(descriptor)
    return [value.st_dev, value.st_ino]


def read_bytes(path: Path, limit: int = MAX_BYTES, root_fd: int | None = None, root: Path | None = None) -> bytes:
    if root_fd is None:
        validate_path(path)
        parent = open_directory(path.parent)
    else:
        relative = path.relative_to(root)
        validate_entry_name(relative.name)
        parent = open_relative_directory(root_fd, relative.parts[:-1])
    try:
        descriptor = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
        with os.fdopen(descriptor, "rb") as stream:
            if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
                raise ValueError("input must be a regular file")
            content = stream.read(limit + 1)
            if len(content) > limit:
                raise ValueError(f"file exceeds 1 MiB: {display_text(path)}")
            return content
    finally:
        os.close(parent)


def read_lines(path: Path) -> list[str]:
    return read_bytes(path).decode("utf-8").splitlines()


def repository_files(root: Path, root_fd: int):
    def entries(directory_fd: int):
        with os.scandir(directory_fd) as listing:
            return iter(sorted(entry.name for entry in listing))

    descriptor = os.dup(root_fd)
    try:
        initial_entries = entries(descriptor)
    except BaseException:
        os.close(descriptor)
        raise
    stack = [(descriptor, Path(), initial_entries)]
    try:
        while stack:
            directory_fd, relative, names = stack[-1]
            name = next(names, None)
            if name is None:
                os.close(stack.pop()[0])
                continue
            validate_entry_name(name)
            metadata = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
            if stat.S_ISDIR(metadata.st_mode) and name not in SKIP:
                child = open_child_directory(name, directory_fd)
                try:
                    children = entries(child)
                except BaseException:
                    os.close(child)
                    raise
                stack.append((child, relative / name, children))
            elif stat.S_ISREG(metadata.st_mode):
                yield root / relative / name
    finally:
        for directory_fd, _, _ in stack:
            os.close(directory_fd)


def read_repository_lines(path: Path, root_fd: int, root: Path) -> list[str] | None:
    content = read_bytes(path, root_fd=root_fd, root=root)
    if b"\0" in content[:8192]:
        return None
    return content.decode("utf-8").splitlines()


def extract(paths: list[str]) -> tuple[list[dict], list[str]]:
    claims, skipped = [], []
    for name in paths:
        path = Path(name)
        try:
            lines = read_lines(path)
        except (OSError, UnicodeError, ValueError) as error:
            skipped.append(f"{name}: {error}")
            continue
        for number, line in enumerate(lines, 1):
            if line.lstrip().startswith(("#", "```", "~~~")) or line.strip() == SIGNATURE:
                continue
            for sentence in re.split(r"(?<=[.!?])\s+(?=[A-Z`])", line.strip()):
                if re.search(CLAIM_PATTERN, sentence):
                    claims.append({"id": f"C{len(claims)+1:04d}", "file": name, "line": number, "text": sentence})
    return claims, skipped


def scan(root: Path, token: str | None, maximum: int) -> tuple[list[dict], list[str], int]:
    matches, skipped, excluded_binary = [], [], 0
    root_fd = open_directory(root)
    try:
        for path in repository_files(root, root_fd):
            try:
                lines = read_repository_lines(path, root_fd, root)
            except (OSError, UnicodeError, ValueError) as error:
                skipped.append(f"{path.relative_to(root)}: {error}")
                continue
            if lines is None:
                excluded_binary += 1
                continue
            for number, line in enumerate(lines, 1):
                if len(matches) < maximum and (token is None or token in line):
                    matches.append({"file": str(path.relative_to(root)), "line": number, "text": line})
    finally:
        os.close(root_fd)
    return matches, skipped, excluded_binary


def inventory(root: Path) -> tuple[dict, list[str], int]:
    flags, env, versions, skipped, excluded_binary = [], [], [], [], 0
    root_fd = open_directory(root)
    try:
        for path in repository_files(root, root_fd):
            try:
                lines = read_repository_lines(path, root_fd, root)
            except (OSError, UnicodeError, ValueError) as error:
                skipped.append(f"{path.relative_to(root)}: {error}")
                continue
            if lines is None:
                excluded_binary += 1
                continue
            for number, line in enumerate(lines, 1):
                source = {"file": str(path.relative_to(root)), "line": number}
                for flag in re.findall(FLAG_PATTERN, line):
                    flags.append({**source, "value": flag})
                for variable in re.findall(ENV_PATTERN, line):
                    env.append({**source, "value": variable})
                for version in re.findall(VERSION_PATTERN, line):
                    versions.append({**source, "value": version})
    finally:
        os.close(root_fd)
    return {"flags": flags, "environment": env, "versions": versions}, skipped, excluded_binary


def load_array(path: Path) -> list[dict]:
    try:
        value = json.loads("\n".join(read_lines(path)))
    except (json.JSONDecodeError, RecursionError) as error:
        raise ValueError("invalid or excessively nested JSON input") from error
    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
        raise ValueError(f"expected JSON array of objects: {display_text(path)}")
    return value


def make_report(claims: list[dict], verdicts: list[dict]) -> str:
    by_id = {}
    for item in verdicts:
        key = item.get("id")
        if not isinstance(key, str) or key in by_id:
            raise ValueError("verdict IDs must be unique strings")
        by_id[key] = item
    rows, counts = [], {word: 0 for word in ("TRUE", "FALSE", "STALE", "UNVERIFIABLE")}
    seen = set()
    for claim in claims:
        key = claim.get("id")
        if not isinstance(key, str) or key in seen or key not in by_id:
            raise ValueError("each claim needs one matching verdict")
        seen.add(key)
        item = by_id[key]
        verdict = item.get("verdict")
        if not isinstance(verdict, str) or verdict not in counts:
            raise ValueError(f"invalid verdict for {display_text(key)}")
        evidence = item.get("evidence")
        if not isinstance(evidence, list) or (verdict != "UNVERIFIABLE" and not evidence):
            raise ValueError(f"evidence required for {display_text(key)}")
        pieces = []
        for entry in evidence:
            if not isinstance(entry, dict) or not isinstance(entry.get("file"), str) or type(entry.get("line")) is not int or entry["line"] < 1:
                raise ValueError(f"invalid evidence for {display_text(key)}")
            pieces.append(f"{display_text(entry['file'])}:{entry['line']}")
        if not isinstance(claim.get("file"), str) or type(claim.get("line")) is not int or claim["line"] < 1 or not isinstance(claim.get("text"), str):
            raise ValueError("invalid claim")
        if "reason" in item and not isinstance(item["reason"], str):
            raise ValueError("verdict reason must be text")
        counts[verdict] += 1
        rows.append(f"| {display_text(key)} | {display_text(claim['file'])}:{claim['line']} | {display_text(claim['text'])} | {verdict} | {', '.join(pieces) or 'None'} | {display_text(item.get('reason', ''))} |")
    if set(by_id) != seen:
        raise ValueError("verdict file has unknown IDs")
    summary = ", ".join(f"{key} {value}" for key, value in counts.items())
    return "# Docs vs Code Checker report\n\n" + summary + "\n\n| ID | Document | Claim | Verdict | Evidence | Reason |\n| --- | --- | --- | --- | --- | --- |\n" + "\n".join(rows) + "\n\n" + SIGNATURE + "\n"


def destination_snapshot(path: Path) -> dict:
    validate_path(path)
    directory = open_directory(path.parent)
    try:
        parent = open_directory(path.parent.absolute().parent)
    except BaseException:
        os.close(directory)
        raise
    try:
        try:
            os.stat(path.name, dir_fd=directory, follow_symlinks=False)
        except FileNotFoundError:
            pass
        else:
            raise ValueError("report destination already exists")
        return {"path": str(path.absolute()), "resolved": str(path.resolve()),
                "directory": identity(directory), "parent": identity(parent)}
    finally:
        os.close(directory)
        os.close(parent)


def approval_digest(path: Path, content: str, snapshot: dict | None = None) -> str:
    payload = json.dumps({"destination": snapshot if snapshot is not None else destination_snapshot(path), "content": content}, sort_keys=True, ensure_ascii=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def create_only(path: Path, content: str, expected: dict) -> None:
    directory = open_directory(path.parent)
    parent = open_directory(path.parent.absolute().parent)
    try:
        if identity(directory) != expected["directory"] or identity(parent) != expected["parent"] or str(path.resolve()) != expected["resolved"]:
            raise ValueError("destination directory changed after approval")
        descriptor = os.open(path.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            stream.write(content)
    finally:
        os.close(directory)
        os.close(parent)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Offline documentation claim helper", epilog=SIGNATURE)
    sub = parser.add_subparsers(dest="command", required=True)
    extract_parser = sub.add_parser("extract", epilog=SIGNATURE)
    extract_parser.add_argument("docs", nargs="+")
    extract_parser.add_argument("--json", action="store_true")
    extract_parser.add_argument("--allow-partial", action="store_true")
    locate_parser = sub.add_parser("locate", epilog=SIGNATURE)
    locate_parser.add_argument("token")
    locate_parser.add_argument("--root", default=".")
    locate_parser.add_argument("--max", type=int, default=100)
    locate_parser.add_argument("--allow-partial", action="store_true")
    inventory_parser = sub.add_parser("inventory", epilog=SIGNATURE)
    inventory_parser.add_argument("--root", default=".")
    inventory_parser.add_argument("--allow-partial", action="store_true")
    report_parser = sub.add_parser("report", epilog=SIGNATURE)
    report_parser.add_argument("--claims", required=True)
    report_parser.add_argument("--verdicts", required=True)
    report_parser.add_argument("--out", required=True)
    report_parser.add_argument("--dry-run", action="store_true")
    report_parser.add_argument("--approve")
    args = parser.parse_args(argv)
    try:
        require_platform()
        if args.command == "report":
            content = make_report(load_array(Path(args.claims)), load_array(Path(args.verdicts)))
            path = Path(args.out)
            snapshot = destination_snapshot(path)
            digest = approval_digest(path, content, snapshot)
            if args.dry_run:
                print(f"dry-run: create {display_text(path)}\napproval digest: {digest}")
            else:
                if args.approve != digest:
                    raise ValueError("missing or mismatched --approve digest; run report --dry-run first")
                create_only(path, content, snapshot)
                print(f"created {display_text(path)}")
            return 0
        if args.command == "extract":
            result, skipped = extract(args.docs)
            if args.json:
                print(json.dumps({"status": "incomplete" if skipped else "complete", "claims": result, "skipped": skipped}, ensure_ascii=True))
            else:
                for item in result:
                    print(f"{item['id']} {display_text(item['file'])}:{item['line']} {display_text(item['text'])}")
        elif args.command == "locate":
            if args.max < 1:
                raise ValueError("--max must be positive")
            result, skipped, excluded_binary = scan(Path(args.root), args.token, args.max)
            for item in result:
                print(f"{display_text(item['file'])}:{item['line']} {display_text(item['text'])}")
        else:
            result, skipped, excluded_binary = inventory(Path(args.root))
            for kind, entries in result.items():
                for item in entries:
                    print(f"{kind} {display_text(item['value'])} {display_text(item['file'])}:{item['line']}")
        if args.command in {"locate", "inventory"}:
            print(f"excluded_binary: {excluded_binary}")
        if skipped:
            print("incomplete: skipped " + "; ".join(display_text(item) for item in skipped), file=sys.stderr)
            return 0 if args.allow_partial else 2
        return 0
    except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as error:
        print(f"error: {display_text(error)}", file=sys.stderr)
        return 1


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

License: MIT (read it).

Back to the listing