path: root/scripts/newos4-maintain.py

#!/usr/bin/env python3
"""newOS4 maintenance: drift, bundles, syntax, perms, data hygiene, live health.
Safe fixes only with --fix. Never touches nginx, services, sudo, or user content."""
import glob, json, os, re, subprocess, sys
from datetime import datetime, timezone

ROOT = '/var/www/webx11.duckdns.org/spaces/newos4'
PAGE = '/var/www/webx11.duckdns.org/spaces/agando/downloads/changelog.html'
WEB = 'https://webx11.duckdns.org/spaces/newos4'
FIX = '--fix' in sys.argv
DEEP = '--deep' in sys.argv
issues = []
def add(l, m): issues.append((l, m))
def run(cmd, **kw): return subprocess.run(cmd, capture_output=True, text=True, **kw)
def sh(cmd): return run(cmd, shell=True).stdout.strip()

def check_versions():
    html = open(os.path.join(ROOT, 'main.html'), encoding='utf-8').read()
    loader = open(os.path.join(ROOT, 'modules/core/module-loader.js'), encoding='utf-8').read()
    m = re.search(r"const LATEST_VER\s*=\s*'([^']+)'", loader)
    ver = m.group(1) if m else '?'
    n = html.count(ver)
    add('PASS' if n == 3 else 'FAIL', f'version {ver} x{n} in main.html (expect 3)')

def syntax_failures():
    bad = []
    for p in glob.glob(os.path.join(ROOT, 'modules/core/*.js')) + glob.glob(os.path.join(ROOT, 'modules/ext/*.js')):
        if os.path.basename(p) in ('core-bundle.js', 'critical-ext-bundle.js'): continue
        if run(['node', '--check', p]).returncode: bad.append(os.path.basename(p))
    return bad

def check_bundles(bad_syntax=None):
    """Rebuild only when every module parses.

    rebuild-bundles.py concatenates blindly (no validation), so a module with a
    syntax error would be baked straight into the served core-bundle.js. Since
    --fix now runs daily, the rebuild is gated on check_syntax and the produced
    bundles are re-checked before the run is called clean.
    """
    paths = ('modules/core/core-bundle.js', 'modules/ext/critical-ext-bundle.js')
    before = {p: sh(f'md5sum {p}') for p in paths}
    if not FIX:
        add('PASS', 'bundle check skipped (use --fix)'); return
    if bad_syntax is None: bad_syntax = syntax_failures()
    if bad_syntax:
        add('FAIL', 'bundles rebuild BLOCKED: syntax fail in ' + ', '.join(bad_syntax[:5])); return
    r = run([sys.executable, 'scripts/rebuild-bundles.py'], cwd=ROOT)
    if r.returncode:
        add('FAIL', 'bundles FAILED'); return
    invalid = [p for p in paths if run(['node', '--check', os.path.join(ROOT, p)]).returncode]
    if invalid:
        add('FAIL', 'bundles rebuilt but INVALID: ' + ', '.join(invalid)); return
    after = {p: sh(f'md5sum {p}') for p in paths}
    changed = [p for p in paths if before[p] != after[p]]
    add('WARN' if changed else 'PASS',
        'bundles ' + ('rebuilt: ' + ', '.join(changed) if changed else 'clean'))

def check_syntax(bad=None):
    bad = syntax_failures() if bad is None else bad
    add('PASS' if not bad else 'FAIL', 'syntax ' + ('OK' if not bad else 'fail: ' + ', '.join(bad)))

def check_perms():
    targets = glob.glob(os.path.join(ROOT, 'course/*.json')) + glob.glob(os.path.join(ROOT, 'course/chapters/*.json')) \
        + [os.path.join(ROOT, 'modules/ext/course.js'), os.path.join(ROOT, 'course.html'), os.path.join(ROOT, 'api/course/event.php')]
    bad = []
    for p in targets:
        if not os.path.exists(p): continue
        if (os.stat(p).st_mode & 0o777) != 0o644 or sh(f'stat -c %G {p}') != 'www-data':
            if FIX:
                os.chmod(p, 0o644); sh(f'chgrp www-data {p}')
            else: bad.append(os.path.basename(p))
    add('PASS' if not bad else ('WARN' if FIX else 'FAIL'), 'perms ' + ('clean' if not bad else ('fixed' if FIX else 'bad: ' + ', '.join(bad[:5]))))

def check_index():
    idxp = os.path.join(ROOT, 'modules-index.json')
    atlas = json.load(open(os.path.join(ROOT, 'course/atlas.json')))
    idx = json.load(open(idxp))
    index_cmds = set(c for m in idx.get('modules', {}).values() for c in m.get('commands', []))
    missing = sorted(c for c, e in atlas.get('commands', {}).items() if not e.get('builtin') and c not in index_cmds)
    if not missing:
        add('PASS', f'modules-index covers {len(index_cmds)} commands'); return
    add('WARN', f'modules-index missing {len(missing)}: ' + ', '.join(missing[:5]))
    if not FIX: return
    for name, m in atlas['modules'].items():
        e = idx['modules'].get(name, {'name': name, 'file': m['file'], 'type': 'core' if m['tier'] == 'core-bundle' else 'ext',
                                      'class': '', 'auto_load': m['tier'] != 'ext-manual', 'description': ''})
        e.update({'commands': m['commands'], 'file': m['file'], 'tier': m['tier']})
        idx['modules'][name] = e
    idx['version'] = datetime.now(timezone.utc).strftime('%Y-%m-%d') + '-atlas'
    tmp = idxp + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as f: json.dump(idx, f, ensure_ascii=False, indent=1)
    os.replace(tmp, idxp); os.chmod(idxp, 0o644); sh(f'chgrp www-data {idxp}')
    add('PASS', 'modules-index regenerated')

def check_data():
    err = os.path.join(ROOT, 'data/debug/errors.json')
    if os.path.exists(err) and os.path.getsize(err) > 5 * 1024 * 1024:
        if FIX:
            arc = os.path.join(ROOT, 'data/debug/archive'); os.makedirs(arc, exist_ok=True)
            os.replace(err, os.path.join(arc, 'errors-' + datetime.now().strftime('%Y%m') + '.json'))
            open(err, 'w').write('[]'); os.chmod(err, 0o664); sh(f'chgrp www-data {err}')
            add('PASS', 'errors.json rotated')
        else: add('WARN', 'errors.json >5MB')
    else: add('PASS', 'debug data OK')

def check_live():
    for url in (f'{WEB}/main.html', f'{WEB}/course.html', f'{WEB}/course/course.json', f'{WEB}/modules/core/core-bundle.js'):
        code = sh(f"curl -s -o /dev/null -w '%{{http_code}}' '{url}'")
        add('PASS' if code == '200' else 'FAIL', f'HTTP {code} {url.replace(WEB, "")}')

def check_atlas():
    p = os.path.join(ROOT, 'course/atlas.json')
    before = (json.load(open(p)).get('coverage', {}).get('commands') if os.path.exists(p) else None)
    r = run([sys.executable, 'scripts/gen-course-atlas.py'], cwd=ROOT)
    after = (json.load(open(p)).get('coverage', {}).get('commands') if os.path.exists(p) else None)
    add('PASS' if r.returncode == 0 else 'FAIL',
        f'atlas {after} commands' + (' (refreshed from ' + str(before) + ')' if before != after else ''))

def check_course():
    r = run([sys.executable, 'scripts/course-validate.py'], cwd=ROOT)
    add('PASS' if r.returncode == 0 else 'FAIL', 'course ' + ('valid' if r.returncode == 0 else 'INVALID: ' + r.stdout[-160:]))

def check_changelog():
    """The changelog page must describe the build that is actually running.

    docs/changelog.html was hand-built from VFS notes and had stopped at v4.6.101 while
    the app was on v4.6.119 -- a docs page that lags the release is worse than no page,
    because it reads as current. build-changelog.py derives every entry from git, so the
    only thing left to check is whether it has been run since the last version bump.
    """
    script = os.path.join(ROOT, 'scripts/build-changelog.py')
    if not os.path.exists(script):
        add('FAIL', 'changelog: scripts/build-changelog.py missing')
        return
    r = run([sys.executable, script, '--check'], cwd=ROOT)
    if r.returncode == 0:
        add('PASS', 'changelog: ' + (r.stdout.strip() or 'current'))
        # A version match is not freshness: if the cron dies, the page and the build stay in
        # agreement forever while the page stops being regenerated.
        import datetime as _dt
        m = re.search(r'"generated"\s*:\s*"([^"]+)"', open(PAGE, encoding='utf-8').read())
        if not m:
            add('FAIL', 'changelog: page has no generated timestamp')
            return
        try:
            gen = _dt.datetime.strptime(m.group(1), '%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=_dt.timezone.utc)
        except ValueError:
            add('FAIL', 'changelog: unreadable generated timestamp ' + m.group(1))
            return
        age_h = (_dt.datetime.now(_dt.timezone.utc) - gen).total_seconds() / 3600.0
        if age_h > 48:
            add('FAIL', 'changelog: generated %.0fh ago (>48h) - the daily job is not running' % age_h)
        else:
            add('PASS', 'changelog age %.1fh' % age_h)
        return
    stale = (r.stdout or r.stderr).strip().splitlines()
    stale = stale[-1][:150] if stale else 'stale'
    if FIX:
        r2 = run([sys.executable, script], cwd=ROOT)
        if r2.returncode == 0:
            add('PASS', 'changelog regenerated: ' + (r2.stdout.strip()[:130] or 'ok'))
            return
        add('FAIL', 'changelog regen failed: ' + (r2.stderr.strip()[:150] or '?'))
        return
    add('FAIL', 'changelog ' + stale)

def check_help_coverage():
    """Every command the code registers must appear in help.html (2026-10-22).

    help.html had a 74-command backlog because nothing compared the two. This
    check makes the gap visible on every maintenance run.
    """
    cmds, term = set(), open(os.path.join(ROOT, 'modules/core/terminal.js'), encoding='utf-8').read()
    for m in re.finditer(r'this\._baseWhitelist = \[(.*?)\];', term, re.S):
        cmds.update(re.findall(r"'([^']+)'", m.group(1)))
    for m in re.finditer(r'const internals = \[(.*?)\];', term, re.S):
        cmds.update(re.findall(r"'([^']+)'", m.group(1)))
    cmds.update(re.findall(r"cmd === ['\"]([a-z0-9_-]+)['\"]", term))
    for p in glob.glob(os.path.join(ROOT, 'modules/**/*.js'), recursive=True):
        if 'bundle' in p or '/test/' in p or '.bkp' in p:
            continue
        src = open(p, encoding='utf-8', errors='ignore').read()
        cmds.update(re.findall(r"commands\[['\"]([a-z0-9_-]+)['\"]\]\s*=", src))
    doc = open(os.path.join(ROOT, 'help.html'), encoding='utf-8').read()
    documented = set()
    for tok in re.findall(r'<code>([^<]+)</code>', doc):
        first = tok.replace('&lt;', ' ').replace('&gt;', ' ').split()
        if first and re.fullmatch(r'[a-z][a-z0-9_-]{0,40}', first[0].lstrip('./')):
            documented.add(first[0].lstrip('./'))
    missing = sorted(cmds - documented)
    add('PASS' if not missing else 'WARN',
        f'help coverage {len(cmds) - len(missing)}/{len(cmds)} commands'
        + ('' if not missing else f' — undocumented: {", ".join(missing[:8])}'))

def check_regression_tests():
    """Every node-runnable test under modules/test/ must pass (2026-10-27).

    The v4.6.102 checkpoint regression was written but no job executed it, so drift could ship
    silently. Two header conventions exist: '// ... run: node' and '// node modules/test/x.js'.
    test-suite.js starts with a class (browser-only) and is correctly skipped.
    """
    tests, failed = [], []
    for p in sorted(glob.glob(os.path.join(ROOT, 'modules/test/test-*.js'))):
        head = ''.join(open(p, encoding='utf-8', errors='ignore').readlines()[:3])
        if 'run: node' not in head and not head.startswith('// node modules/test/'):
            continue
        tests.append(p)
        try:
            r = subprocess.run(['node', p], cwd=ROOT, capture_output=True, text=True, timeout=120)
        except subprocess.TimeoutExpired:
            failed.append(os.path.basename(p) + ': timeout')
            continue
        if r.returncode != 0:
            lines = (r.stdout or r.stderr).strip().splitlines() or ['no output']
            failed.append(os.path.basename(p) + ': ' + lines[-1][:120])
    add('PASS' if not failed else 'FAIL',
        'node tests %d/%d pass' % (len(tests) - len(failed), len(tests))
        + ('' if not failed else ' — ' + '; '.join(failed[:3])))

def check_links():
    if not DEEP: return
    broken = []
    for page in ('main.html', 'course.html', 'help.html', 'modules.html', 'guide.html'):
        p = os.path.join(ROOT, page)
        if not os.path.exists(p): continue
        for href in re.findall(r'(?:href|src)="([^"#]+)"', open(p, encoding='utf-8').read()):
            if any(x in href for x in ("'", '+', '{', '}')): continue
            if href.startswith(('http', '//', 'mailto:', 'javascript:', 'data:', '/api/')): continue
            target = os.path.normpath(os.path.join(ROOT, href.split('?')[0]))
            if not os.path.exists(target): broken.append(f'{page}:{href}')
    add('PASS' if not broken else 'WARN', 'links ' + ('OK' if not broken else f'{len(broken)} broken (first: {broken[0]})'))

def main():
    os.chdir(ROOT)
    bad = syntax_failures()          # computed once, gates the bundle rebuild
    check_versions(); check_syntax(bad); check_regression_tests(); check_bundles(bad); check_perms(); check_index(); check_data(); check_live(); check_atlas(); check_course(); check_help_coverage(); check_links(); check_changelog()
    fails = [m for l, m in issues if l == 'FAIL']
    print('newOS4 maintenance — ' + datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%MZ') + (' [fix]' if FIX else '') + (' [deep]' if DEEP else ''))
    for l, m in issues: print(f'{l}: {m}')
    print('maintenance result: ' + ('FAIL' if fails else 'OK'))
    print('needs OK: ' + ('none' if not fails else '; '.join(fails)[:400]))
    return 1 if fails else 0

if __name__ == '__main__':
    sys.exit(main())