digiserver-v2/docs/tools/sanitize_report.py

"""Accurate sanitization report for digiserver-v2.

Corrects two things the first pass got wrong:
  * blueprint -> file mapping now follows what app.py actually imports
  * route usage counts url_for() AND hardcoded URL paths in templates/JS

Also classifies routes that are intentionally consumed by the external Kivy
player rather than by any template.

Run: python tools/sanitize_report.py [--markdown out.md]
"""
from __future__ import annotations

import argparse
import ast
import os
import re
from collections import defaultdict

# Walk up until we find the project root (robust to the tool living in a
# nested folder such as docs/tools/).
_here = os.path.dirname(os.path.abspath(__file__))
while _here != os.path.dirname(_here) and not os.path.isdir(os.path.join(_here, 'app')):
    _here = os.path.dirname(_here)
REPO = _here
APP = os.path.join(REPO, 'app')


def read(p):
    return open(p, encoding='utf-8', errors='replace').read()


def rel(p):
    return os.path.relpath(p, REPO)


def walk_py(roots, skip=('__pycache__', 'legacy code')):
    for root_dir in roots:
        if not os.path.isdir(root_dir):
            continue
        for root, dirs, files in os.walk(root_dir):
            dirs[:] = [d for d in dirs if d not in skip]
            for f in sorted(files):
                if f.endswith('.py'):
                    yield os.path.join(root, f)


py_files = list(walk_py([APP, os.path.join(REPO, 'migrations')]))
src = {rel(p): read(p) for p in py_files}
app_src = src.get('app/app.py', '')

# ── 1. Which blueprint modules does app.py actually import? ─────────────────
imported_bp_modules = set(re.findall(
    r'from\s+(app\.blueprints\.\w+)\s+import', app_src))
registered_vars = set(re.findall(r'register_blueprint\((\w+)\)', app_src))

# ── 2. Every blueprint definition + the file that defines it ────────────────
bp_defs = []   # (var, blueprint_name, url_prefix, file, is_imported_by_app)
for path in py_files:
    r = rel(path)
    if not r.startswith('app/blueprints/') or os.path.basename(path).startswith('__'):
        continue
    s = src[r]
    m = re.search(r'^(\w+_bp)\s*=\s*Blueprint\(\s*[\'"]([\w-]+)[\'"]'
                  r'(?:\s*,\s*[^)]*?url_prefix\s*=\s*[\'"]([^\'"]+)[\'"])?',
                  s, re.M | re.S)
    if m:
        var, bpname, prefix = m.group(1), m.group(2), m.group(3) or ''
        dotted = 'app.blueprints.' + os.path.basename(path)[:-3]
        bp_defs.append({
            'var': var, 'name': bpname, 'prefix': prefix or '/',
            'file': r, 'imported': dotted in imported_bp_modules,
            'registered_var': var in registered_vars,
        })

# ── 3. Collect route usage signals ─────────────────────────────────────────
url_for_refs = set()
# Literal URL strings, but only ones that look like real request paths
# (avoids matching CSS selectors, regexes, mime types, etc.).
PATH_RE = re.compile(r"['\"](/[a-z][a-zA-Z0-9_\-/<>{}.:]*)['\"]")
hardcoded_paths = set()
consider_files = []
for root, dirs, files in os.walk(os.path.join(APP, 'templates')):
    for f in files:
        if f.endswith(('.html', '.js')):
            consider_files.append(os.path.join(root, f))
consider_files += [os.path.join(REPO, r) for r in src
                   if r.endswith(('.html', '.js'))]
for fp in consider_files:
    if not os.path.isfile(fp):
        continue
    s = read(fp)
    url_for_refs |= set(re.findall(r"url_for\(\s*['\"]([\w\.]+)['\"]", s))
    hardcoded_paths |= set(PATH_RE.findall(s))
for r, s in src.items():
    if r.endswith('.py'):
        url_for_refs |= set(re.findall(r"url_for\(\s*['\"]([\w\.]+)['\"]", s))
        hardcoded_paths |= set(PATH_RE.findall(s))

# ── 4. Enumerate routes ────────────────────────────────────────────────────
API_PREFIX = '/api'
routes = []
for path in py_files:
    r = rel(path)
    if not r.startswith('app/blueprints/'):
        continue
    s = src[r]
    bp_name = None
    file_prefix = ''
    for d in bp_defs:
        if d['file'] == r:
            bp_name = d['name']
            file_prefix = d['prefix']
    try:
        tree = ast.parse(s)
    except SyntaxError:
        continue
    for node in tree.body:
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        for dec in node.decorator_list:
            if not (isinstance(dec, ast.Call) and getattr(dec.func, 'attr', '') == 'route'):
                continue
            rule = dec.args[0].value if dec.args and isinstance(dec.args[0], ast.Constant) else '?'
            methods = 'GET'
            for kw in dec.keywords:
                if kw.arg == 'methods' and isinstance(kw.value, ast.List):
                    methods = ','.join(e.value for e in kw.value.elts
                                       if isinstance(e, ast.Constant))
            full = file_prefix.rstrip('/') + rule if rule != '/' else file_prefix.rstrip('/') + '/'
            endpoint = f'{bp_name}.{node.name}'
            routes.append({
                'endpoint': endpoint, 'func': node.name, 'rule': rule,
                'full': full, 'methods': methods, 'file': r,
                'lineno': node.lineno,
                'url_for_ref': endpoint in url_for_refs,
                'hardcoded_ref': full in hardcoded_paths or rule in hardcoded_paths,
                'is_api': (file_prefix or '').startswith(API_PREFIX),
            })

# ── 5. Broken model attribute references ──────────────────────────────────
model_attrs = defaultdict(set)
for path in py_files:
    r = rel(path)
    if not r.startswith('app/models/') or os.path.basename(path).startswith('__'):
        continue
    try:
        tree = ast.parse(src[r])
    except SyntaxError:
        continue
    for node in tree.body:
        if isinstance(node, ast.ClassDef):
            for sub in ast.walk(node):
                if isinstance(sub, ast.Assign):
                    for t in sub.targets:
                        if isinstance(t, ast.Name):
                            model_attrs[node.name].add(t.id)
                elif isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    model_attrs[node.name].add(sub.name)

VIA = {'via', 'e', 'exc', 'error', 'err', 'a', 'b', 'flask_', 'db_', 'fh', 'fd'}

# ── report ────────────────────────────────────────────────────────────────
lines = []
def out(s=''):
    lines.append(s)

out('# DigiServer v2 — Code Sanitization Report')
out()
out(f'Analysed **{len(py_files)} Python files** under `app/` and `migrations/`.')
out()

out('## 1. Dead modules (never imported)')
out()
out('| File | Status |')
out('|---|---|')
dead_files = []
for d in bp_defs:
    if not d['imported']:
        dead_files.append(d['file'])
        out(f"| `{d['file']}` | **DEAD** — defines `{d['var']}` but `app.py` imports another module |")
for path in py_files:
    r = rel(path)
    base = os.path.basename(path)[:-3]
    if not r.startswith('app/') or r.startswith('app/blueprints/'):
        continue
    if base == '__init__':
        continue
    dotted = r[:-3].replace('/', '.')
    referenced = any(dotted in s or f'.{base} import' in s
                     for k, s in src.items() if k != r)
    if not referenced:
        dead_files.append(r)
        out(f'| `{r}` | **DEAD** — never imported by any module |')
if not dead_files:
    out('| (none) | |')
out()

out('## 2. Blueprint registration')
out()
out('| Blueprint var | Module | Prefix | Registered |')
out('|---|---|---|---|')
for d in sorted(bp_defs, key=lambda x: x['name']):
    mark = 'yes' if (d['imported'] and d['registered_var']) else '**NO — DEAD**'
    out(f"| `{d['var']}` | `{d['file']}` | `{d['prefix']}` | {mark} |")
out()

used = [r for r in routes if r['url_for_ref'] or r['hardcoded_ref']]
api_routes = [r for r in routes if r['is_api']]
unused_non_api = [r for r in routes if not r['url_for_ref'] and not r['hardcoded_ref']
                  and not r['is_api']]

out('## 3. Routes')
out()
out(f'- Total routes: **{len(routes)}**')
out(f'- Referenced by a template/JS (`url_for` or hardcoded path): **{len(used)}**')
out(f'- API routes (`/api/*`, consumed by the external Kivy player, not templates): **{len(api_routes)}**')
out(f'- Non-API routes with **no template reference**: **{len(unused_non_api)}**')
out()
out('### 3a. Non-API routes with no template reference (candidates)')
out()
out('| Endpoint | Methods | Path | File:line |')
out('|---|---|---|---|')
for r in sorted(unused_non_api, key=lambda x: (x['file'], x['lineno'])):
    out(f"| `{r['endpoint']}` | {r['methods']} | `{r['full']}` | `{r['file']}:{r['lineno']}` |")
out()

out('## 4. Broken attribute references (would raise at runtime)')
out()
out('| Location | Reference | Problem |')
out('|---|---|---|')
# Locals that are known to hold a specific model instance.
VAR_TO_MODEL = {
    'player': 'Player', 'content': 'Content', 'playlist': 'Playlist',
    'assigned_playlist': 'Playlist', 'feedback': 'PlayerFeedback',
    'latest_feedback': 'PlayerFeedback', 'edit_record': 'PlayerEdit',
    'admin': 'User', 'user': 'User', 'log': 'ServerLog',
    'group': 'Group', 'new_user': 'PlayerUser', 'existing_user': 'PlayerUser',
}
# Python/library attributes that would otherwise be false positives.
SAFE = {
    'query', 'get', 'filter_by', 'first', 'all', 'count', 'id', 'session', 'add',
    'commit', 'rollback', 'delete', 'get_or_404', 'order_by', 'isoformat', 'route',
    'methods', 'get_json', 'args', 'files', 'form', 'json', 'headers', 'values',
    'remote_addr', 'host_url', 'script_root', 'scheme', 'host', 'root_path',
    'config', 'utcnow', 'now', 'total_seconds', 'name', 'value', 'keys', 'items',
    'groups', 'contents', 'players', 'append', 'lower', 'upper', 'strip', 'split',
    'join', 'replace', 'encode', 'decode', 'check_password', 'check_quickconnect_code',
    'set_password', 'set_quickconnect_code', 'authenticate', 'update_status',
    'to_dict', 'is_online', 'is_admin', 'is_active', 'file_size_mb', 'group_count',
    'player_count', 'content_count', 'original_display_name', 'original_media_path',
    'current_media_path', 'get_content_ordered', 'version', 'filename', 'content_type',
    'duration', 'url', 'description', 'file_size', 'uploaded_at', 'hostname',
    'location', 'auth_code', 'orientation', 'status', 'playlist_id', 'last_seen',
    'created_at', 'updated_at', 'deployment_status', 'last_deployment_at',
    'last_deployment_status', 'last_deployment_message', 'original_filename',
    'time_of_modification', 'metadata_path', 'edited_file_path', 'new_name',
    'original_name', 'username', 'role', 'password', 'level', 'message', 'user_code',
    'user_name', 'email', 'func', 'rules', 'player_id', 'content_id', 'error',
    'show', 'seek', 'load', 'play', 'pause', 'stop', 'text', 'bind', 'add_widget',
    'clear_widgets', 'current', 'parent', 'children', 'ids', 'size', 'pos', 'opacity',
    'source', 'state', 'duration', 'position', 'muted', 'audio', 'describe',
}
broken = []
for r, s in src.items():
    if not (r.startswith('app/blueprints/') or r.startswith('app/utils/')):
        continue
    try:
        tree = ast.parse(s)
    except SyntaxError:
        continue
    for node in ast.walk(tree):
        if not (isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name)):
            continue
        var, attr = node.value.id, node.attr
        model = VAR_TO_MODEL.get(var)
        if not model or model not in model_attrs:
            continue
        if attr in model_attrs[model] or attr in SAFE:
            continue
        broken.append((r, node.lineno, f'{var}.{attr}', f'`{model}` has no `{attr}`'))
for r, ln, ref, prob in broken:
    out(f'| `{r}:{ln}` | `{ref}` | {prob} |')
if not broken:
    out('| (none) | | |')
out()

out('## 5. How to use this report')
out()
out('Each section lists deletion candidates. Reply with the section/symbol names')
out('you want removed and they will be deleted (the snapshot in `legacy code/`')
out('preserves the originals).')

text = '\n'.join(lines)

ap = argparse.ArgumentParser()
ap.add_argument('--markdown', default=None)
args = ap.parse_args()

if args.markdown:
    open(args.markdown, 'w').write(text)
    print(f'wrote {args.markdown}')
else:
    print(text)