`scripts/cactus-seam-purge.py`

194 lines
#!/usr/bin/env python3
"""cactus-seam-purge.py — rigorous seam-defect purge for the Cactus Curator.

Steve's directive: delete any cactus SEAMLESS TILE with fuzzy edges OR a fuzzy
internal center seam (the "2x2 internal-grid" defect, both horiz + vert
centerlines). Aggressive threshold: any of the 6 lenses (top/bottom/left/right/
h_mid/v_mid) over ΔE 5 (i.e. anything that isn't a clean PASS) is purged.

Scope guard: ONLY kind='seamless_tile'. Murals (kind in mural / mural_panel)
are single scenes / butt-join panels with no meaningful center seam — they are
NEVER touched.

Steps: (1) ensure full scan coverage by re-scanning every unscanned tile via
edges-scan.py's scan_path (free, local, per-lens). (2) report the exact doomed
count. (3) with --execute, soft-delete the doomed set: is_published=FALSE,
user_removed=TRUE, web_viewer=FALSE, quarantine the PNG (recoverable), tag
'seam-purge', append to data/cactus-decisions.jsonl. Files we could not scan
(missing PNG) are reported and KEPT, never blind-deleted.

  python3 scripts/cactus-seam-purge.py            # rescan + dry-run report
  python3 scripts/cactus-seam-purge.py --execute  # also perform the purge
"""
import importlib.util
import json
import os
import shutil
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
EXECUTE = '--execute' in sys.argv
PASS_MAX = 5.0          # any lens > this == "fuzzy" (not a clean PASS)
QUAR = ROOT / 'data' / 'generated_cactus_quarantine'
AUDIT = ROOT / 'data' / 'cactus-decisions.jsonl'

# import scan_path from the hyphenated edges-scan.py
_spec = importlib.util.spec_from_file_location('edges_scan', str(ROOT / 'scripts' / 'edges-scan.py'))
_es = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(_es)
scan_path = _es.scan_path


def psql(sql, read=True):
    r = subprocess.run(['psql', 'dw_unified', '-At', '-F', '|', '-c', sql],
                       capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError('psql: ' + r.stderr.strip())
    return r.stdout.strip()


def psql_stdin(sql):
    r = subprocess.run(['psql', 'dw_unified', '-At', '-q'], input=sql,
                       capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError('psql: ' + r.stderr.strip())
    return r.stdout.strip()


def seam_quality(edge, mid):
    worst = max(edge or 0, mid or 0)
    return max(0, min(100, round((1 - worst / 12.0) * 100)))


# ── per-lens columns (rigorous audit trail) ────────────────────────────────
psql_stdin("""
ALTER TABLE wallco_cactus_rank ADD COLUMN IF NOT EXISTS seam_hmid_raw numeric;
ALTER TABLE wallco_cactus_rank ADD COLUMN IF NOT EXISTS seam_vmid_raw numeric;
INSERT INTO wallco_cactus_rank (design_id)
  SELECT id FROM all_designs WHERE category ILIKE '%cactus%' AND kind='seamless_tile'
  ON CONFLICT (design_id) DO NOTHING;
""")

# ── 1. rescan unscanned seamless_tile cactus ───────────────────────────────
unscanned = psql("""
  SELECT d.id, d.local_path FROM all_designs d
  JOIN wallco_cactus_rank r ON r.design_id=d.id
  WHERE d.category ILIKE '%cactus%' AND d.kind='seamless_tile'
    AND NOT COALESCE(d.user_removed,false)
    AND r.seam_edge_raw IS NULL AND d.local_path IS NOT NULL;
""")
rows = [ln.split('|', 1) for ln in unscanned.split('\n') if ln.strip()]
print(f'[purge] rescanning {len(rows)} unscanned tiles…')


def _flush(batch):
    if not batch:
        return
    tup = ','.join(
        f"({i},{sq},{e},{m},{hm},{vm},'{v}',now())" for (i, sq, e, m, hm, vm, v) in batch)
    psql_stdin(f"""
      INSERT INTO wallco_cactus_rank
        (design_id, seam_score, seam_edge_raw, seam_mid_raw, seam_hmid_raw, seam_vmid_raw, seam_verdict, seam_scored_at)
      VALUES {tup}
      ON CONFLICT (design_id) DO UPDATE SET
        seam_score=EXCLUDED.seam_score, seam_edge_raw=EXCLUDED.seam_edge_raw,
        seam_mid_raw=EXCLUDED.seam_mid_raw, seam_hmid_raw=EXCLUDED.seam_hmid_raw,
        seam_vmid_raw=EXCLUDED.seam_vmid_raw, seam_verdict=EXCLUDED.seam_verdict,
        seam_scored_at=EXCLUDED.seam_scored_at;
    """)


missing, scanned, vals = 0, 0, []
for i, (rid, lp) in enumerate(rows):
    if not lp or not os.path.isfile(lp):
        missing += 1
        continue
    try:
        r = scan_path(Path(lp))
    except Exception as e:
        missing += 1
        continue
    if not r.get('ok'):
        missing += 1
        continue
    e_ = r['edges_max']; m_ = r['mids_max']
    hm = r['lenses']['h_mid']['score']; vm = r['lenses']['v_mid']['score']
    vals.append((int(rid), seam_quality(e_, m_), e_, m_, hm, vm, r['verdict']))
    scanned += 1
    if len(vals) >= 400:
        _flush(vals); vals = []
    if (i + 1) % 100 == 0:
        print(f'\r[purge]   scanned {i+1}/{len(rows)} (missing {missing})', end='')

_flush(vals)
print(f'\n[purge] rescan complete — scanned {scanned}, unscannable(missing PNG) {missing}')

# ── 2. doomed set: any lens > PASS_MAX (edges OR center fuzzy) ──────────────
doomed = psql(f"""
  SELECT d.id, d.local_path FROM all_designs d
  JOIN wallco_cactus_rank r ON r.design_id=d.id
  WHERE d.category ILIKE '%cactus%' AND d.kind='seamless_tile'
    AND NOT COALESCE(d.user_removed,false)
    AND r.seam_edge_raw IS NOT NULL
    AND (r.seam_edge_raw > {PASS_MAX} OR r.seam_mid_raw > {PASS_MAX});
""")
doomed_rows = [ln.split('|', 1) for ln in doomed.split('\n') if ln.strip()]

clean = psql(f"""
  SELECT count(*) FROM all_designs d JOIN wallco_cactus_rank r ON r.design_id=d.id
  WHERE d.category ILIKE '%cactus%' AND d.kind='seamless_tile' AND NOT COALESCE(d.user_removed,false)
    AND r.seam_edge_raw IS NOT NULL AND r.seam_edge_raw<={PASS_MAX} AND r.seam_mid_raw<={PASS_MAX};
""")
unscannable = psql(f"""
  SELECT count(*) FROM all_designs d JOIN wallco_cactus_rank r ON r.design_id=d.id
  WHERE d.category ILIKE '%cactus%' AND d.kind='seamless_tile' AND NOT COALESCE(d.user_removed,false)
    AND r.seam_edge_raw IS NULL;
""")
murals = psql("""SELECT count(*) FROM all_designs WHERE category ILIKE '%cactus%'
  AND kind IN ('mural','mural_panel') AND NOT COALESCE(user_removed,false);""")

print('\n===== SEAM PURGE REPORT (kind=seamless_tile only) =====')
print(f'  DOOMED (fuzzy edge or center, ΔE>{PASS_MAX:.0f}): {len(doomed_rows)}')
print(f'  KEEP — clean PASS all 6 lenses           : {clean}')
print(f'  KEEP — unscannable (missing PNG)          : {unscannable}')
print(f'  UNTOUCHED — murals / mural_panels         : {murals}')
print('=======================================================')

if not EXECUTE:
    print('\n[dry-run] no changes made. Re-run with --execute to purge.')
    sys.exit(0)

# ── 3. execute soft-delete: quarantine PNG + flag + audit ──────────────────
QUAR.mkdir(parents=True, exist_ok=True)
moved, ids = 0, []
with open(AUDIT, 'a') as af:
    for rid, lp in doomed_rows:
        ids.append(int(rid))
        if lp and os.path.isfile(lp):
            dest = QUAR / os.path.basename(lp)
            try:
                if not dest.exists():
                    shutil.move(lp, dest); moved += 1
            except Exception as e:
                print(f'  quarantine fail {rid}: {e}')
        af.write(json.dumps({'ts': datetime.now(timezone.utc).isoformat(),
                             'id': int(rid), 'action': 'bad', 'reason': 'seam-purge-aggressive'}) + '\n')

# bulk DB flip in chunks
for k in range(0, len(ids), 500):
    chunk = ids[k:k + 500]
    idlist = ','.join(str(x) for x in chunk)
    psql_stdin(f"""
      UPDATE all_designs SET is_published=FALSE, user_removed=TRUE, web_viewer=FALSE,
        tags = array_remove(COALESCE(tags,ARRAY[]::text[]),'seam-purge') || ARRAY['seam-purge']::text[]
      WHERE id IN ({idlist});
    """)

print(f'\n[purge] EXECUTED — soft-deleted {len(ids)} tiles, quarantined {moved} PNGs.')
print(f'[purge] recoverable from {QUAR}')
remaining = psql("""SELECT count(*) FROM all_designs WHERE category ILIKE '%cactus%'
  AND kind='seamless_tile' AND NOT COALESCE(user_removed,false);""")
print(f'[purge] live seamless_tile cactus remaining: {remaining}')