""" doc-hygiene scanner — deterministic, no model, no network. Produces an intermediate artifact: { "project_root": str, "scope_globs": [str, ...], "excluded_dirs": [str, ...], "files_scanned": int, "shortlist": [str, ...], # project-root-relative paths "signals": { path: [{"name": str, "detail": str}, ...] } } Signal shape matches entries[].signals in the frozen report schema. """ from __future__ import annotations import fnmatch import json import os import re import subprocess import sys from pathlib import Path from typing import Callable, Iterator, List, Optional from token_estimator import TokenEstimator, default_estimator # --------------------------------------------------------------------------- # Glob matching (Python 3.14-compatible) # --------------------------------------------------------------------------- def _file_matches_glob(rel_path_str: str, glob_pattern: str) -> bool: """Return True if *rel_path_str* matches *glob_pattern*. Works around a Python 3.14 behavior where ``Path('a.md').match('**/*.md')`` returns False for root-level files (no parent component). For ``**/X`` patterns we fall back to matching just ``X`` against the filename so that root-level files are included. """ p = Path(rel_path_str) # Try direct PurePath.match() first — handles nested files correctly if p.match(glob_pattern): return True # Fallback for '**/' patterns: match just the trailing pattern against # the filename, so root-level 'a.md' is caught by '**/*.md'. if glob_pattern.startswith("**/"): suffix_pat = glob_pattern[3:] if fnmatch.fnmatch(p.name, suffix_pat): return True return False # --------------------------------------------------------------------------- # Defaults # --------------------------------------------------------------------------- DEFAULT_SCOPE_GLOBS: List[str] = ["**/*.md"] DEFAULT_MAX_LINES: int = 400 DEFAULT_MAX_TOKENS: int = 4000 DEFAULT_EXCLUDED_DIRS: List[str] = [ "build", "vendor", "archive", "graphify-out", ".cc-os", ".dochygiene", # legacy state dir (pre-ADR-027); still excluded for old trees # Test-fixture dirs intentionally contain stale/bloated docs as scanner # inputs. Bare directory-*name* match (consistent with the entries above): # any child dir named "fixtures" at any depth is pruned — same breadth as # "archive"/"vendor". "fixtures", # Golden classifier inputs also contain deliberately-stale docs, but a bare # "golden" name-match would silently skip legitimate `golden/` dirs in # unrelated projects (doc-hygiene installs globally). So this entry is # PATH-AWARE: an entry containing "/" is matched as a parent/child name pair # anywhere in the tree — here, a dir named "golden" whose immediate parent is # named "examples". A root-level `golden/` with a different parent is still # scanned. See Scanner._is_excluded_child for the matcher. "examples/golden", ] # --------------------------------------------------------------------------- # Helpers — frontmatter # --------------------------------------------------------------------------- def _parse_frontmatter_value(path: Path, key: str) -> Optional[str]: """Return the string value of *key* in YAML frontmatter, or None. Reads only the frontmatter block (between the first pair of ``---`` lines). Uses stdlib only (no PyYAML). """ try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return None lines = text.splitlines() if not lines or lines[0].strip() != "---": return None in_block = True for line in lines[1:]: stripped = line.strip() if stripped == "---" or stripped == "...": break # Match: key: value (simple scalar, no quoting) m = re.match(r"^(\w[\w\-]*):\s*(.+)$", stripped) if m and m.group(1) == key: return m.group(2).strip() return None def _is_frozen(path: Path) -> bool: """Return True if the file has ``hygiene: frozen`` in its frontmatter.""" return _parse_frontmatter_value(path, "hygiene") == "frozen" # --------------------------------------------------------------------------- # Helpers — append-only detection (content-based, no git) # --------------------------------------------------------------------------- _APPEND_ONLY_MARKER_RE = re.compile( r"|", re.IGNORECASE, ) # A dated section header: starts with a 4-digit year, or ISO date, or # common changelog-style prefix e.g. "## 2026-06-18" or "# v1.2 (2026-06-18)" _DATED_HEADER_RE = re.compile( r"^#{1,3}\s+(?:v?\d+\.\d|\d{4}[-/]\d{2}[-/]\d{2}|\d{4}-\d{2})" ) def _is_append_only(path: Path) -> bool: """Heuristic: return True if the file looks like an append-only log. Two independent paths (either is sufficient): 1. File contains an explicit ```` marker. 2. Every non-blank, non-header line cluster lives under a dated section header, and the section count is ≥ 2 (a multi-entry changelog pattern). Deterministic: reads content only, no git, no network. """ try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return False # Path 1: explicit marker anywhere in the file if _APPEND_ONLY_MARKER_RE.search(text): return True # Path 2: structural — look for ≥2 dated-section headers and NO prose # outside of dated sections (aside from a preamble). lines = text.splitlines() dated_header_count = 0 in_dated_section = False non_section_content_found = False preamble_done = False # first dated header ends the preamble for line in lines: stripped = line.strip() if not stripped: continue if _DATED_HEADER_RE.match(stripped): dated_header_count += 1 in_dated_section = True preamble_done = True elif not preamble_done: # still in preamble (before first dated header) — allowed pass elif not in_dated_section: # content outside any dated section after preamble non_section_content_found = True break if dated_header_count >= 2 and not non_section_content_found: return True return False # --------------------------------------------------------------------------- # Helpers — .dochygiene-ignore # --------------------------------------------------------------------------- def _load_ignore_patterns(root: Path) -> List[str]: """Load patterns from ``/.dochygiene-ignore``. One pattern per line.""" ignore_file = root / ".dochygiene-ignore" try: text = ignore_file.read_text(encoding="utf-8", errors="replace") return [ line.strip() for line in text.splitlines() if line.strip() and not line.strip().startswith("#") ] except OSError: return [] def _matches_ignore(rel_path: str, patterns: List[str]) -> bool: """Return True if *rel_path* matches any ignore pattern (fnmatch).""" for pat in patterns: if fnmatch.fnmatch(rel_path, pat) or fnmatch.fnmatch( os.path.basename(rel_path), pat ): return True return False # --------------------------------------------------------------------------- # Signal computation # --------------------------------------------------------------------------- class SignalComputer: """Compute objective signals for a single file. Injected dependencies ---------------------- git_log_fn : callable(path: Path) -> list[str] Returns git log lines for the file. Production uses ``_git_log_real``; tests pass a fake. Pass ``None`` to disable git-based signals. now_fn : callable() -> float Returns current time as a Unix timestamp. max_lines : int Line-count threshold for the ``file_length`` signal. max_tokens : int Estimated-token threshold for the ``file_length`` signal. token_estimator : TokenEstimator, optional Injectable estimator used for the ``file_length`` signal's token count. Defaults to ``token_estimator.default_estimator()``. Tests can inject a deterministic stub. """ def __init__( self, root: Path, git_log_fn: Optional[Callable[[Path], List[str]]] = None, now_fn: Optional[Callable[[], float]] = None, max_lines: int = DEFAULT_MAX_LINES, max_tokens: int = DEFAULT_MAX_TOKENS, token_estimator: Optional[TokenEstimator] = None, ) -> None: self._root = root self._git_log_fn = git_log_fn self._now_fn = now_fn or (lambda: __import__("time").time()) self._max_lines = max_lines self._max_tokens = max_tokens self._token_estimator = token_estimator or default_estimator() # ------------------------------------------------------------------ # Public # ------------------------------------------------------------------ def compute(self, path: Path) -> List[dict]: """Return a list of signal dicts for *path*.""" signals: List[dict] = [] rel = str(path.relative_to(self._root)) signals.extend(self._broken_references(path, rel)) signals.extend(self._version_skew(path)) signals.extend(self._edit_recency_vs_churn(path)) signals.extend(self._location_signals(path, rel)) signals.extend(self._archive_to_live_ratio(path)) signals.extend(self._frontmatter_markers(path)) signals.extend(self._file_length(path)) return signals # ------------------------------------------------------------------ # Individual signal detectors # ------------------------------------------------------------------ def _broken_references(self, path: Path, rel: str) -> List[dict]: """Detect Markdown links/images whose targets do not exist on disk.""" signals = [] try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return signals # Match [text](target) — skip anchors, URLs, and mailto link_re = re.compile(r"\[(?:[^\[\]]*)\]\(([^)]+)\)") for m in link_re.finditer(text): target = m.group(1).split("#")[0].strip() # strip fragment if not target: continue if re.match(r"https?://|mailto:", target): continue # Resolve relative to the file's directory if target.startswith("/"): resolved = self._root / target.lstrip("/") else: resolved = path.parent / target if not resolved.exists(): signals.append( { "name": "broken_reference", "detail": f"links to '{target}' which does not exist", } ) return signals def _version_skew(self, path: Path) -> List[dict]: """Detect explicit version declarations that differ from the repo version. Looks for patterns like ``version: X.Y.Z`` or ``# vX.Y.Z`` in the file and compares against a ``pyproject.toml`` or ``package.json`` at the project root, if present. Objective fact only — no judgment. """ signals = [] try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return signals # Find declared versions in the doc (e.g. ``version: 1.2.3``) doc_versions = re.findall(r"\bversion[:\s]+['\"]?(\d+\.\d+[\.\d]*)['\"]?", text, re.IGNORECASE) if not doc_versions: return signals # Try to read authoritative version from pyproject.toml or package.json repo_version: Optional[str] = None pyproject = self._root / "pyproject.toml" if pyproject.exists(): try: pp_text = pyproject.read_text(encoding="utf-8", errors="replace") m = re.search(r'^version\s*=\s*["\'](\d+\.\d+[\.\d]*)["\']', pp_text, re.MULTILINE) if m: repo_version = m.group(1) except OSError: pass if repo_version is None: pkg_json = self._root / "package.json" if pkg_json.exists(): try: pkg = json.loads(pkg_json.read_text(encoding="utf-8", errors="replace")) repo_version = pkg.get("version") except (OSError, json.JSONDecodeError): pass if repo_version is None: return signals for doc_ver in doc_versions: if doc_ver != repo_version: signals.append( { "name": "version_skew", "detail": ( f"doc declares version {doc_ver!r} but repo version is {repo_version!r}" ), } ) break # one signal per file is sufficient return signals def _edit_recency_vs_churn(self, path: Path) -> List[dict]: """Flag files edited very recently relative to their historical git churn. Signal: file was modified within the last 7 days (mtime) but has a high commit frequency (≥ 5 commits in git history), suggesting it is actively evolving and may have outpaced its documentation. If no git_log_fn is provided this signal is skipped. """ if self._git_log_fn is None: return [] try: stat = path.stat() except OSError: return [] age_days = (self._now_fn() - stat.st_mtime) / 86400.0 if age_days > 7: return [] # not recently edited try: log_lines = self._git_log_fn(path) except Exception: return [] commit_count = len([l for l in log_lines if l.strip()]) if commit_count >= 5: return [ { "name": "edit_recency_vs_churn", "detail": ( f"file modified {age_days:.1f} days ago and has {commit_count} commits" " — actively changing, may need review" ), } ] return [] def _location_signals(self, path: Path, rel: str) -> List[dict]: """Flag files whose location suggests staleness risk. Detects: - Files named ``*old*``, ``*deprecated*``, ``*legacy*``, ``*obsolete*`` that are NOT inside an archive/ dir (which is excluded). - Files in a ``docs/`` subdirectory that have no counterpart in the current codebase module they claim to document (heuristic: if the filename mentions a module that no longer exists). """ signals = [] basename = path.stem.lower() stale_names = {"old", "deprecated", "legacy", "obsolete", "archive"} for keyword in stale_names: if keyword in basename: signals.append( { "name": "stale_name_location", "detail": ( f"filename '{path.name}' contains '{keyword}' — " "may indicate superseded content" ), } ) break return signals def _archive_to_live_ratio(self, path: Path) -> List[dict]: """Detect high proportion of 'archived' / resolved-problem content. Intra-doc heuristic: count lines under headings that contain 'archive', 'resolved', 'completed', 'done', 'old', 'deprecated', 'legacy' vs total non-blank lines. If > 60% of content is under such headings, emit a signal. """ signals = [] try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return signals lines = text.splitlines() archive_heading_re = re.compile( r"^#{1,4}\s+.*\b(archive|resolved|completed|done|old|deprecated|legacy)\b", re.IGNORECASE, ) in_archive_section = False archive_lines = 0 total_lines = 0 for line in lines: stripped = line.strip() if not stripped: continue total_lines += 1 if re.match(r"^#{1,4}\s+", stripped): in_archive_section = bool(archive_heading_re.match(stripped)) if in_archive_section: archive_lines += 1 if total_lines > 0 and (archive_lines / total_lines) > 0.6: pct = int(100 * archive_lines / total_lines) signals.append( { "name": "archive_to_live_ratio", "detail": ( f"{pct}% of content is under archive/resolved/completed headings" ), } ) return signals def _frontmatter_markers(self, path: Path) -> List[dict]: """Emit signals for notable frontmatter keys that flag provisional content. Objective markers only: ``status: draft``, ``status: provisional``, ``status: wip``, ``draft: true``. No classification — just fact. """ signals = [] for key, expected_values in [ ("status", {"draft", "provisional", "wip", "todo"}), ("draft", {"true", "yes"}), ]: value = _parse_frontmatter_value(path, key) if value and value.lower() in expected_values: signals.append( { "name": "frontmatter_marker", "detail": f"frontmatter has {key}: {value!r}", } ) return signals def _file_length(self, path: Path) -> List[dict]: """Flag files whose line count or estimated token count exceeds threshold. Objective fact only: reports the measured line count and estimated token count against the configured thresholds. Fires if EITHER threshold is exceeded. No classification. """ signals = [] try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: return signals line_count = len(text.splitlines()) token_count = self._token_estimator.estimate(text) if line_count > self._max_lines or token_count > self._max_tokens: token_count_k = token_count / 1000.0 signals.append( { "name": "file_length", "detail": ( f"{line_count} lines (~{token_count_k:.1f}k tokens est), " f"threshold {self._max_lines} lines / {self._max_tokens} tokens" ), } ) return signals # --------------------------------------------------------------------------- # Production git helper # --------------------------------------------------------------------------- def _git_log_real(path: Path) -> List[str]: """Return one line per commit touching *path* (commit hash only).""" try: result = subprocess.run( ["git", "log", "--oneline", "--follow", "--", str(path)], capture_output=True, text=True, timeout=10, ) return result.stdout.splitlines() except Exception: return [] # --------------------------------------------------------------------------- # Scanner # --------------------------------------------------------------------------- class Scanner: """Deterministic doc-hygiene scanner. Parameters ---------- root : Path Resolved project root (absolute). All output paths are relative to this directory. scope_globs : list[str] Glob patterns to match candidate files (relative to *root*). excluded_dirs : list[str] Entries to prune during walk. A bare entry (no ``/``) is a directory *name* matched at any depth; an entry of the form ``parent/child`` is a path-aware parent/child name pair pruned only where the child's immediate parent matches ``parent`` (depth-independent). git_log_fn : callable, optional Injected git-log provider. Default: real git subprocess. now_fn : callable, optional Injected clock. Default: ``time.time``. max_lines : int Line-count threshold for the ``file_length`` signal. Default 400. max_tokens : int Estimated-token threshold for the ``file_length`` signal. Default 4000. token_estimator : TokenEstimator, optional Injectable estimator for the ``file_length`` signal. Default: ``token_estimator.default_estimator()``. """ def __init__( self, root: Path, scope_globs: Optional[List[str]] = None, excluded_dirs: Optional[List[str]] = None, git_log_fn: Optional[Callable[[Path], List[str]]] = None, now_fn: Optional[Callable[[], float]] = None, max_lines: int = DEFAULT_MAX_LINES, max_tokens: int = DEFAULT_MAX_TOKENS, token_estimator: Optional[TokenEstimator] = None, ) -> None: self._root = root.resolve() self._scope_globs = scope_globs if scope_globs is not None else DEFAULT_SCOPE_GLOBS # Always include .cc-os (canonical, ADR-027) and .dochygiene (legacy) in # excluded dirs (self-exclusion invariant #9). Preserve insertion order # so the artifact echoes the canonical default order (matches # valid_report.json golden fixture). base_excluded = excluded_dirs if excluded_dirs is not None else DEFAULT_EXCLUDED_DIRS seen: set = set() deduped: List[str] = [] for d in list(base_excluded) + [".cc-os", ".dochygiene"]: if d not in seen: seen.add(d) deduped.append(d) self._excluded_dirs: List[str] = deduped # Partition the exclude list into two matchers (both echoed verbatim in # the artifact's ``excluded_dirs``): # * bare names (no "/") → prune any child dir with that exact name. # * "parent/child" pairs → prune a child dir named ``child`` only # when its immediate parent is ``parent``, # anywhere in the tree (path-aware). # Only the trailing two segments of a "/"-entry are used as the pair; the # match is depth-independent (not anchored to the project root) so it # catches e.g. ``doc-hygiene/examples/golden`` in a monorepo. self._excluded_name_set: set = {d for d in deduped if "/" not in d} self._excluded_pairs: List[tuple] = [] for d in deduped: if "/" in d: parent, child = d.rsplit("/", 1) parent_name = parent.rsplit("/", 1)[-1] # trailing segment only self._excluded_pairs.append((parent_name, child)) self._git_log_fn = git_log_fn self._now_fn = now_fn self._max_lines = max_lines self._max_tokens = max_tokens self._token_estimator = token_estimator self._ignore_patterns: List[str] = _load_ignore_patterns(self._root) def _is_excluded_child(self, parent_dirpath: str, child_name: str) -> bool: """Return True if walking would prune *child_name* under *parent_dirpath*. Bare-name excludes match on ``child_name`` alone; path-aware excludes (``parent/child``) match only when the child's immediate parent name equals ``parent``. """ if child_name in self._excluded_name_set: return True parent_name = os.path.basename(parent_dirpath) for ex_parent, ex_child in self._excluded_pairs: if child_name == ex_child and parent_name == ex_parent: return True return False # ------------------------------------------------------------------ # Public # ------------------------------------------------------------------ def run(self) -> dict: """Execute the scan and return the intermediate artifact dict.""" signal_computer = SignalComputer( root=self._root, git_log_fn=self._git_log_fn, now_fn=self._now_fn, max_lines=self._max_lines, max_tokens=self._max_tokens, token_estimator=self._token_estimator, ) ignore_patterns = self._ignore_patterns files_scanned = 0 shortlist: List[str] = [] signals_map: dict = {} for path in self._walk_scoped(): files_scanned += 1 rel = str(path.relative_to(self._root)) # Exclusion pipeline (short-circuit, ordered per D7) # (1) dir-prune already applied by _walk_scoped at walk time # (2) ignore-match if _matches_ignore(rel, ignore_patterns): continue # (3) frozen frontmatter if _is_frozen(path): continue # (4) append-only if _is_append_only(path): continue # Survived exclusions — compute signals sigs = signal_computer.compute(path) shortlist.append(rel) if sigs: signals_map[rel] = sigs return { "project_root": str(self._root), "scope_globs": self._scope_globs, "excluded_dirs": self._excluded_dirs, "files_scanned": files_scanned, "shortlist": shortlist, "signals": signals_map, } # ------------------------------------------------------------------ # Walk # ------------------------------------------------------------------ def _walk_scoped(self) -> Iterator[Path]: """Yield files matching scope_globs, with excluded dirs pruned at walk time.""" for dirpath, dirnames, filenames in os.walk(str(self._root)): # Prune excluded dirs IN-PLACE so os.walk never descends them (D7). # Path-aware pairs (e.g. examples/golden) are matched against the # current dirpath's basename so only the right parent prunes them. dirnames[:] = [ d for d in dirnames if not self._is_excluded_child(dirpath, d) ] for filename in filenames: filepath = Path(dirpath) / filename rel = filepath.relative_to(self._root) rel_str = str(rel) for glob in self._scope_globs: if _file_matches_glob(rel_str, glob): yield filepath break # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- def _resolve_project_root(start: Path) -> Path: """Walk upward from *start* to find a .git directory; fallback to *start*.""" current = start.resolve() for parent in [current, *current.parents]: if (parent / ".git").exists(): return parent return current def main(argv: Optional[List[str]] = None) -> int: import argparse parser = argparse.ArgumentParser( description="doc-hygiene scanner — emits intermediate artifact JSON" ) parser.add_argument( "--root", default=None, help="Project root (default: auto-resolved from cwd)", ) parser.add_argument( "--globs", nargs="*", default=None, help="Scope globs (default: **/*.md)", ) parser.add_argument( "--excluded-dirs", nargs="*", default=None, help="Directory names to exclude", ) parser.add_argument( "--max-lines", type=int, default=DEFAULT_MAX_LINES, help=f"Line-count threshold for the file_length signal (default: {DEFAULT_MAX_LINES})", ) parser.add_argument( "--max-tokens", type=int, default=DEFAULT_MAX_TOKENS, help=f"Estimated-token threshold for the file_length signal (default: {DEFAULT_MAX_TOKENS})", ) args = parser.parse_args(argv) root = Path(args.root) if args.root else _resolve_project_root(Path.cwd()) scanner = Scanner( root=root, scope_globs=args.globs, excluded_dirs=args.excluded_dirs, git_log_fn=_git_log_real, max_lines=args.max_lines, max_tokens=args.max_tokens, ) artifact = scanner.run() print(json.dumps(artifact, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())