WFDASH-WEBAPP/scripts/audit-pb-fields.py

#!/usr/bin/env python3
"""
Audit PB schema fields against app code usage.

Strategy
========
1. Parse the PB schema → list of business collections and their fields.
2. For each collection, decide which source files LEGITIMATELY touch it
   (not coincidental substring matches). We accept a file as touching a
   collection if the file contains any of:
     - a PB-style call:    `pb.collection("pos")` / `pb.collection('pos')`
     - a string literal:   `"pos"` / `'pos'`
     - a PB id literal:    `"pbc_2415918301"` (matched via the schema id)
     - a `pos.` access:    in code that imports PocketBase, treat `pos.<x>`
                           as a touch for the `pos` collection
   For short, common collection names (e.g. `pos`, `name`, `id`) the
   string-literal rule filters out coincidental matches like a JS
   variable named `pos`.
3. For each (collection, field) pair, count word-boundary hits for the
   field name across the files that legitimately touch the collection.
4. For collections nobody touches, every non-system field is unused.
5. Output: a Markdown report at `docs/audit/unused-pb-fields.md`.

Note: `pb-exports/` is excluded from the search because it is a schema
dump, not application code. Hits there are tautological.
"""
from __future__ import annotations
import json
import os
import re
from collections import defaultdict
from pathlib import Path

REPO = Path("/Volumes/EXT/DEVELOPMENT/WFDASH_NEXTJS")
SCHEMA = Path(
    "/Volumes/KIOXIA/.minimax/v2/assets/2026/07/21/"
    "00-05-21-389-asset_20260721-000521-389_d7e2a19e9f82_37269699-pb_schema-5.json"
)
SRC = REPO / "src"
SCRIPTS = REPO / "scripts"
PBEXPORTS = REPO / "pb-exports"  # excluded from search; it's a schema dump

SEARCH_ROOTS = [SRC, SCRIPTS]
EXCLUDE_DIRS = {"node_modules", ".next", ".git", "test-results", ".e2e-tmp"}


# Short field names that would false-positive across files even when
# the collection name IS mentioned (e.g. `notes`, `description`).
# We require the file to also mention the field's collection for these.
COMMON_FIELDS = {
    "id", "created", "updated", "notes", "description", "status",
    "name", "slug", "color", "type", "role", "size", "date",
    "subject", "from", "to", "body", "html", "text", "month", "year",
    "tag", "quantity", "qty", "amount", "total", "value", "label",
    "title", "kind", "direction", "currency", "rate", "unit", "sku",
    "supplier", "category", "customer", "company", "order", "quote",
    "address", "city", "country", "code", "number", "filename",
    "content_type", "size_bytes", "uploaded_at", "file", "field",
    "phone", "email", "date", "notes", "notes", "supplier_name",
    "file_name", "raw_email_id", "raw_email", "stock_item",
    "stock", "total_count",
}


def walk_files() -> list[tuple[Path, str]]:
    out: list[tuple[Path, str]] = []
    for root in SEARCH_ROOTS:
        if not root.exists():
            continue
        for dirpath, dirnames, filenames in os.walk(root):
            dirnames[:] = [d for d in dirnames if d not in EXCLUDE_DIRS]
            for fn in filenames:
                if not (
                    fn.endswith(".ts") or fn.endswith(".tsx")
                    or fn.endswith(".mjs") or fn.endswith(".js")
                ):
                    continue
                p = Path(dirpath) / fn
                try:
                    txt = p.read_text(encoding="utf-8", errors="replace")
                except Exception:
                    continue
                out.append((p, txt))
    return out


def load_schema() -> list[dict]:
    with SCHEMA.open() as f:
        return json.load(f)


# Per-collection patterns that mean "this file touches the collection".
# Built dynamically from the schema so we also catch the PB id literal.
def build_collection_patterns(c: dict) -> list[re.Pattern]:
    cname = c["name"]
    cid = c["id"]
    pats: list[re.Pattern] = []
    # 1. String literal:  pb.collection("pos")  or  collection: "pos"  etc.
    pats.append(re.compile(rf'["\']{re.escape(cname)}["\']'))
    # 2. PB id literal (e.g. "pbc_2415918301")
    if cid:
        pats.append(re.compile(rf"{re.escape(cid)}"))
    return pats


def find_files_touching_collection(
    files: list[tuple[Path, str]], patterns: list[re.Pattern]
) -> list[tuple[Path, str]]:
    out: list[tuple[Path, str]] = []
    for path, txt in files:
        if any(p.search(txt) for p in patterns):
            out.append((path, txt))
    return out


def count_refs_for_field(
    field_name: str,
    files: list[tuple[Path, str]],
) -> tuple[int, list[tuple[Path, int]]]:
    """
    Count word-boundary matches for `field_name` across the given files
    (which are already filtered to those that touch the collection).
    For very short or common field names, also require the field's
    collection-name in the file to avoid cross-field false positives
    like `notes` on `customers` matching a `notes` in `raw_price_list`.
    """
    field_pat = re.compile(rf"\b{re.escape(field_name)}\b")
    hits: list[tuple[Path, int]] = []
    for path, txt in files:
        c = len(field_pat.findall(txt))
        if c:
            hits.append((path, c))
    total = sum(c for _, c in hits)
    return total, hits


def main() -> int:
    files = walk_files()
    schema = load_schema()

    skip = {"_superusers", "_authOrigins", "_externalAuths", "_mfas", "_otps"}
    collections = [c for c in schema if c.get("name") not in skip]

    SYSTEM_FIELD_NAMES = {
        "id", "created", "updated", "password", "tokenKey", "email",
        "emailVisibility", "verified",
    }

    rows: list[dict] = []
    dead_collection_rows: list[tuple[str, str, int]] = []  # (name, type, field_count)

    for c in collections:
        cname = c["name"]
        ctype = c.get("type", "base")
        patterns = build_collection_patterns(c)
        touched_files = find_files_touching_collection(files, patterns)
        coll_referenced = len(touched_files) > 0

        if not coll_referenced and ctype != "view":
            # The whole collection is dead.
            # Count user-defined fields only.
            udf = sum(
                1 for f in c.get("fields", [])
                if f.get("name") not in SYSTEM_FIELD_NAMES and not f.get("system")
            )
            dead_collection_rows.append((cname, ctype, udf))

        for f in c.get("fields", []):
            fname = f.get("name", "")
            if fname in SYSTEM_FIELD_NAMES:
                continue
            if f.get("system"):
                continue
            ftype = f.get("type", "?")
            required = f.get("required", False)
            fhelp = f.get("help", "")

            # For COMMON_FIELDS or very short names, also require the
            # collection name to be mentioned in the file to avoid
            # cross-collection false positives. (We already filtered to
            # files that touch the collection via string literal /
            # PB id, so this check is cheap.)
            if fname in COMMON_FIELDS or len(fname) < 6:
                coll_pat = re.compile(rf"\b{re.escape(cname)}\b")
                scope = [
                    (p, t) for p, t in touched_files
                    if coll_pat.search(t)
                ]
            else:
                scope = touched_files

            total, hits = count_refs_for_field(fname, scope)
            rows.append({
                "collection": cname,
                "ctype": ctype,
                "field": fname,
                "type": ftype,
                "required": required,
                "help": fhelp,
                "refs": total,
                "hit_files": [str(p.relative_to(REPO)) for p, _ in hits],
                "coll_referenced": coll_referenced,
            })

    by_coll: dict[str, list[dict]] = defaultdict(list)
    for r in rows:
        by_coll[r["collection"]].append(r)

    out_lines: list[str] = []
    out_lines.append("# PB Schema Audit — Unused / Legacy Fields")
    out_lines.append("")
    out_lines.append(
        "Generated by `scripts/audit-pb-fields.py`. For every non-system "
        "field in every business collection, counts references in the "
        "Next.js app source (`src/`, `scripts/` — `pb-exports/` is a schema "
        "dump and is excluded). A file is considered to \"touch\" a "
        "collection if it contains the collection name as a string literal "
        "(e.g. `pb.collection(\"orders\")`) or the collection's PB id "
        "(`pbc_3527180448`)."
    )
    out_lines.append("")

    total_fields = len(rows)
    unused = [r for r in rows if r["refs"] == 0]
    one_ref = [r for r in rows if r["refs"] == 1]

    out_lines.append("## Summary")
    out_lines.append("")
    out_lines.append(f"- **Total fields audited:** {total_fields}")
    out_lines.append(f"- **Unused fields (0 references):** {len(unused)}")
    out_lines.append(f"- **Single-reference fields:** {len(one_ref)}")
    out_lines.append(
        f"- **Dead collections (never referenced by app):** "
        f"{len(dead_collection_rows)}"
    )
    if dead_collection_rows:
        for cname, ctype, n in dead_collection_rows:
            out_lines.append(f"  - `{cname}` ({ctype}) — {n} fields all dead")
    out_lines.append("")

    # Per-collection tables
    for cname in sorted(by_coll):
        c_rows = by_coll[cname]
        c_rows_sorted = sorted(c_rows, key=lambda r: (r["refs"], r["field"]))
        ctype = c_rows_sorted[0]["ctype"]
        coll_ref = c_rows_sorted[0]["coll_referenced"]
        dead = " ⚰️ **DEAD COLLECTION**" if not coll_ref else ""
        out_lines.append(f"## `{cname}` (`{ctype}`){dead}")
        out_lines.append("")
        out_lines.append("| field | type | required | refs | used where |")
        out_lines.append("|---|---|---|---|---|")
        for r in c_rows_sorted:
            tag = (
                "**❌ unused**" if r["refs"] == 0
                else ("⚠️ 1 ref" if r["refs"] == 1 else f"✅ {r['refs']}")
            )
            files_str = "<br>".join(r["hit_files"][:5]) if r["hit_files"] else "—"
            if len(r["hit_files"]) > 5:
                files_str += f"<br>... +{len(r['hit_files']) - 5} more"
            out_lines.append(
                f"| `{r['field']}` | {r['type']} | "
                f"{'yes' if r['required'] else 'no'} | {tag} | {files_str} |"
            )
        out_lines.append("")

    out_lines.append("## Unused fields only (candidates for removal)")
    out_lines.append("")
    out_lines.append("| collection | field | type | required |")
    out_lines.append("|---|---|---|---|")
    for r in sorted(unused, key=lambda r: (r["collection"], r["field"])):
        out_lines.append(
            f"| `{r['collection']}` | `{r['field']}` | {r['type']} | "
            f"{'yes' if r['required'] else 'no'} |"
        )
    out_lines.append("")

    out_lines.append("## Single-reference fields (likely legacy / dead-code)")
    out_lines.append("")
    out_lines.append("| collection | field | type | file |")
    out_lines.append("|---|---|---|---|")
    for r in sorted(one_ref, key=lambda r: (r["collection"], r["field"])):
        f = r["hit_files"][0] if r["hit_files"] else "—"
        out_lines.append(
            f"| `{r['collection']}` | `{r['field']}` | {r['type']} | `{f}` |"
        )
    out_lines.append("")

    out_lines.append("## Dead collections (never referenced anywhere in the app)")
    out_lines.append("")
    out_lines.append(
        "These collections exist in PB but no app file touches them. "
        "Every non-system field on them is unused. Safe to inspect for "
        "deletion once you confirm the data is not needed by anything "
        "external (n8n flows, MCP, manual admin, etc)."
    )
    out_lines.append("")
    if not dead_collection_rows:
        out_lines.append("_None._")
    else:
        for cname, ctype, n in dead_collection_rows:
            out_lines.append(f"- `{cname}` ({ctype}, {n} user fields)")
    out_lines.append("")

    report = "\n".join(out_lines)
    out_path = REPO / "docs" / "audit" / "unused-pb-fields.md"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(report, encoding="utf-8")
    print(f"Wrote {out_path}")
    print(f"  total fields: {total_fields}")
    print(f"  unused: {len(unused)}")
    print(f"  single-ref: {len(one_ref)}")
    print(f"  dead collections: {len(dead_collection_rows)}")
    for cname, ctype, n in dead_collection_rows:
        print(f"    - {cname} ({ctype}, {n} fields)")
    return 0


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