cc-os/plugins/os-doc-hygiene/scripts/scanner.py

784 lines
29 KiB
Python
Raw Permalink Normal View History

"""
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 '**/<pat>' 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"<!--\s*append[-_]only\s*-->|<!--\s*hygiene:\s*append[-_]only\s*-->",
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 ``<!-- append-only -->`` 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 ``<root>/.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())