405 lines
17 KiB
Python
405 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
scripts/audit_photos.py — Pre-cutover photo audit, PER TENANT (READ-ONLY)
|
|
===========================================================================
|
|
|
|
Multi-tenant port of the single-tenant script of the same name (ST §22 Phase 0).
|
|
Writes NOTHING to any database and NOTHING to the uploads tree. It only reads,
|
|
and emits a report.
|
|
|
|
Why this differs from the single-tenant version
|
|
-----------------------------------------------
|
|
The local backend does NOT prefix keys (see app/utils/storage.py, "Tenant
|
|
isolation"), so every tenant's photos sit intermixed in one shared directory:
|
|
|
|
app/static/uploads/issue_photos/<uuid>.jpg
|
|
|
|
A file on disk therefore carries no evidence of which tenant owns it. Ownership
|
|
is established the only way it can be — by walking each tenant's own database
|
|
and collecting the keys it references. That mapping is exactly what the R2 sync
|
|
needs, because on R2 the object lives at ``t<tenant_id>/uploads/...``.
|
|
|
|
What it does
|
|
------------
|
|
1. For every tenant (control DB, status in provisioning/active/suspended),
|
|
collects every photo key REFERENCED by that tenant's DB, across all 5 sources:
|
|
- issues.photo_path (string)
|
|
- issues.mobile_photo_paths[] (JSON list — iPad evidence)
|
|
- issues.result_photos[] (JSON list — web resolution photos)
|
|
- inspections.form_data{} (any 'uploads/...' string, incl. nested)
|
|
- inspection_results.photo_path (string)
|
|
(Signatures are stored inline as 'data:' base64, NOT files — correctly ignored.)
|
|
2. Checks each referenced key against the file actually on disk.
|
|
3. Walks the uploads tree and finds files referenced by NO tenant (orphans).
|
|
4. Flags any key referenced by MORE THAN ONE tenant (should be impossible —
|
|
keys are uuid4 — but a restore/copy could produce one, and it would mean the
|
|
same bytes get written under two tenant prefixes).
|
|
5. Prints a summary and writes a detailed JSON report kept as the baseline.
|
|
|
|
The number that matters, per tenant: **distinct referenced photos that resolve
|
|
on disk**. That count must be identical after cutover. Any "missing on disk"
|
|
here is a PRE-EXISTING broken reference — surfaced now so it can never be
|
|
blamed on the move.
|
|
|
|
Usage
|
|
-----
|
|
cd /home/jqc/lt_janitorial_quality_control
|
|
source venv/bin/activate
|
|
python scripts/audit_photos.py # all tenants -> /tmp report
|
|
python scripts/audit_photos.py --tenant lts # one tenant (id or slug)
|
|
python scripts/audit_photos.py --report /home/jqc/photo_audit.json
|
|
python scripts/audit_photos.py --list-orphans # also print orphan paths
|
|
|
|
Requires CONTROL_DATABASE_URL and CONTROL_FERNET_KEY in the environment (.env),
|
|
the same as any other control-plane command. Exit code is always 0 (audit is
|
|
informational). Nothing is modified.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import argparse
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
# Make the repo root importable when run from anywhere.
|
|
_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
|
sys.path.insert(0, _REPO_ROOT)
|
|
|
|
from dotenv import load_dotenv # noqa: E402
|
|
load_dotenv(os.path.join(_REPO_ROOT, '.env'))
|
|
|
|
from sqlalchemy import create_engine, text # noqa: E402
|
|
|
|
from control.base import control_session # noqa: E402
|
|
from control.models import Tenant # noqa: E402
|
|
|
|
# A deleted tenant's DB may be gone; everything else is audited.
|
|
AUDITABLE_STATUSES = ('provisioning', 'active', 'suspended')
|
|
|
|
STATIC_FOLDER = os.path.join(_REPO_ROOT, 'app', 'static')
|
|
UPLOADS_ROOT = os.path.join(STATIC_FOLDER, 'uploads')
|
|
|
|
|
|
# ── Path helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def normalize(path):
|
|
"""
|
|
Return a canonical 'uploads/...' key, or None if this value is not a stored
|
|
photo file (base64 signatures, unsynced iPad sentinels, remote URLs, blanks).
|
|
"""
|
|
if not isinstance(path, str):
|
|
return None
|
|
p = path.strip()
|
|
if not p:
|
|
return None
|
|
# Not on-disk files:
|
|
if p.startswith('data:') or p.startswith('local://'):
|
|
return None
|
|
if p.startswith('http://') or p.startswith('https://'):
|
|
return None
|
|
# Tolerate a few stored variants and reduce to 'uploads/...'
|
|
for prefix in ('/static/', 'static/', '/'):
|
|
if p.startswith(prefix):
|
|
p = p[len(prefix):]
|
|
if not p.startswith('uploads/'):
|
|
return None
|
|
return p
|
|
|
|
|
|
def walk_strings(value):
|
|
"""Yield every string found in an arbitrarily nested JSON value (dict/list/str)."""
|
|
if isinstance(value, str):
|
|
yield value
|
|
elif isinstance(value, dict):
|
|
for v in value.values():
|
|
yield from walk_strings(v)
|
|
elif isinstance(value, (list, tuple)):
|
|
for v in value:
|
|
yield from walk_strings(v)
|
|
|
|
|
|
def as_json(value):
|
|
"""Decode a JSON column that may arrive as str (PyMySQL) or already-parsed."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (dict, list)):
|
|
return value
|
|
if isinstance(value, (bytes, bytearray)):
|
|
value = value.decode('utf-8', 'replace')
|
|
if isinstance(value, str):
|
|
try:
|
|
return json.loads(value)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
return None
|
|
|
|
|
|
# ── Reference collection (one tenant DB) ──────────────────────────────────────
|
|
|
|
def collect_referenced(db_uri):
|
|
"""
|
|
Return (referenced, by_source) for a single tenant database.
|
|
|
|
referenced : { normalized_key -> [ "issue:123:photo_path", ... ] }
|
|
by_source : { source_name -> count_of_refs }
|
|
|
|
Raw SQL on purpose: this runs outside a Flask app context, so the ORM models
|
|
(bound to app.db) are not usable without dragging in create_app and its
|
|
default bind — which is the wrong database for every tenant.
|
|
"""
|
|
referenced = defaultdict(list)
|
|
by_source = defaultdict(int)
|
|
|
|
def add(path, ref_label, source_name):
|
|
norm = normalize(path)
|
|
if norm is None:
|
|
return
|
|
referenced[norm].append(ref_label)
|
|
by_source[source_name] += 1
|
|
|
|
engine = create_engine(db_uri, pool_pre_ping=True, future=True)
|
|
try:
|
|
with engine.connect() as conn:
|
|
# --- Issues ---
|
|
rows = conn.execute(text(
|
|
'SELECT id, photo_path, mobile_photo_paths, result_photos FROM issues'
|
|
))
|
|
for rid, photo_path, mobile_paths, result_photos in rows:
|
|
add(photo_path, f'issue:{rid}:photo_path', 'issue.photo_path')
|
|
for pth in (as_json(mobile_paths) or []):
|
|
add(pth, f'issue:{rid}:mobile_photo_paths', 'issue.mobile_photo_paths')
|
|
for pth in (as_json(result_photos) or []):
|
|
add(pth, f'issue:{rid}:result_photos', 'issue.result_photos')
|
|
|
|
# --- Inspections: form_data image fields (Inspection has NO photo_path) ---
|
|
rows = conn.execute(text('SELECT id, form_data FROM inspections'))
|
|
for rid, form_data in rows:
|
|
for s in walk_strings(as_json(form_data) or {}):
|
|
add(s, f'inspection:{rid}:form_data', 'inspection.form_data(image)')
|
|
|
|
# --- InspectionResult: per checklist-item result photo ---
|
|
rows = conn.execute(text('SELECT id, photo_path FROM inspection_results'))
|
|
for rid, photo_path in rows:
|
|
add(photo_path, f'inspection_result:{rid}:photo_path',
|
|
'inspection_result.photo_path')
|
|
|
|
# --- TenantSettings: branding logo (MT-22) ---
|
|
# The logo was written straight to disk before MT-22, so it was
|
|
# never part of this inventory and would have been left behind at
|
|
# R2 cutover — a tenant flipping to s3 would lose its logo.
|
|
# Tolerant of the table being absent: tenants provisioned before
|
|
# MT-7 have no tenant_settings.
|
|
try:
|
|
rows = conn.execute(text('SELECT id, logo_url FROM tenant_settings'))
|
|
for rid, logo_url in rows:
|
|
add(logo_url, f'tenant_settings:{rid}:logo_url',
|
|
'tenant_settings.logo_url')
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
engine.dispose()
|
|
|
|
return referenced, by_source
|
|
|
|
|
|
# ── Disk scan ─────────────────────────────────────────────────────────────────
|
|
|
|
def scan_disk(uploads_root):
|
|
"""Return { relative_uploads_key -> size_bytes } for every file under uploads/."""
|
|
on_disk = {}
|
|
if not os.path.isdir(uploads_root):
|
|
return on_disk
|
|
base = os.path.dirname(uploads_root) # .../static (so keys start with 'uploads/')
|
|
for dirpath, _dirs, files in os.walk(uploads_root):
|
|
for name in files:
|
|
full = os.path.join(dirpath, name)
|
|
rel = os.path.relpath(full, base).replace(os.sep, '/')
|
|
try:
|
|
on_disk[rel] = os.path.getsize(full)
|
|
except OSError:
|
|
on_disk[rel] = -1
|
|
return on_disk
|
|
|
|
|
|
# ── Tenant selection ──────────────────────────────────────────────────────────
|
|
|
|
def select_tenants(selector):
|
|
"""Return [(id, slug, db_uri)] for 'all' or a single id/slug."""
|
|
out = []
|
|
with control_session() as s:
|
|
q = s.query(Tenant).filter(Tenant.status.in_(AUDITABLE_STATUSES))
|
|
if selector and selector != 'all':
|
|
if str(selector).isdigit():
|
|
q = q.filter(Tenant.id == int(selector))
|
|
else:
|
|
q = q.filter(Tenant.slug == selector)
|
|
for t in q.order_by(Tenant.id).all():
|
|
out.append((t.id, t.slug, t.db_uri)) # db_uri decrypts here
|
|
return out
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────────────
|
|
|
|
def human(nbytes):
|
|
n = float(nbytes)
|
|
for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
|
|
if abs(n) < 1024.0:
|
|
return f'{n:.1f} {unit}'
|
|
n /= 1024.0
|
|
return f'{n:.1f} PB'
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description='Read-only per-tenant photo audit (R2 cutover, phase 0).')
|
|
parser.add_argument('--tenant', default='all',
|
|
help="Tenant id or slug, or 'all' (default).")
|
|
parser.add_argument('--report', default=None,
|
|
help='Path to write the JSON report (default: /tmp/jqc_photo_audit_<ts>.json)')
|
|
parser.add_argument('--list-orphans', action='store_true',
|
|
help='Also print orphan file paths to stdout')
|
|
args = parser.parse_args()
|
|
|
|
tenants = select_tenants(args.tenant)
|
|
if not tenants:
|
|
print(f'No tenants matched --tenant {args.tenant!r} '
|
|
f'(statuses: {", ".join(AUDITABLE_STATUSES)}).')
|
|
return
|
|
|
|
on_disk = scan_disk(UPLOADS_ROOT)
|
|
on_disk_keys = set(on_disk.keys())
|
|
|
|
per_tenant = []
|
|
claimed = defaultdict(list) # key -> [tenant slug, ...] (multi-claim detection)
|
|
|
|
print('=' * 72)
|
|
print(' JQC PHOTO AUDIT — MULTI-TENANT (read-only — nothing was modified)')
|
|
print('=' * 72)
|
|
print(f' uploads root : {UPLOADS_ROOT}')
|
|
print(f' tenants : {len(tenants)}')
|
|
|
|
for tid, slug, db_uri in tenants:
|
|
try:
|
|
referenced, by_source = collect_referenced(db_uri)
|
|
except Exception as exc:
|
|
print(f'\n-- tenant {tid} ({slug}) : ERROR reading DB: {exc}')
|
|
per_tenant.append({'tenant_id': tid, 'slug': slug, 'error': str(exc)})
|
|
continue
|
|
|
|
referenced_keys = set(referenced.keys())
|
|
for k in referenced_keys:
|
|
claimed[k].append(slug)
|
|
|
|
present = referenced_keys & on_disk_keys
|
|
missing = referenced_keys - on_disk_keys
|
|
present_bytes = sum(on_disk[k] for k in present if on_disk[k] > 0)
|
|
|
|
per_tenant.append({
|
|
'tenant_id': tid,
|
|
'slug': slug,
|
|
'referenced': {
|
|
'distinct_keys': len(referenced_keys),
|
|
'by_source': dict(by_source),
|
|
'present_on_disk': len(present),
|
|
'present_bytes': present_bytes,
|
|
'missing_on_disk': len(missing),
|
|
},
|
|
'missing_on_disk': sorted(
|
|
[{'key': k, 'referenced_by': sorted(set(referenced[k]))} for k in missing],
|
|
key=lambda d: d['key']
|
|
),
|
|
# The exact upload set for this tenant, consumed by
|
|
# scripts/migrate_photos_to_r2.py (which recomputes it itself).
|
|
'r2_object_prefix': f't{tid}/',
|
|
})
|
|
|
|
print('-' * 72)
|
|
print(f' tenant {tid} ({slug})')
|
|
for src, cnt in sorted(by_source.items()):
|
|
print(f' {src:<34} {cnt:>8} refs')
|
|
print(f' {"distinct keys referenced":<34} {len(referenced_keys):>8}')
|
|
print(f' {"present on disk":<34} {len(present):>8} '
|
|
f'({human(present_bytes)}) <-- must be unchanged after cutover')
|
|
print(f' {"MISSING on disk":<34} {len(missing):>8} '
|
|
f'<-- pre-existing broken refs')
|
|
print(f' {"R2 object prefix":<34} {"t" + str(tid) + "/":>8}')
|
|
|
|
all_referenced = set(claimed.keys())
|
|
orphans = on_disk_keys - all_referenced
|
|
orphan_bytes = sum(on_disk[k] for k in orphans if on_disk[k] > 0)
|
|
total_bytes = sum(v for v in on_disk.values() if v > 0)
|
|
multi_claimed = {k: v for k, v in claimed.items() if len(v) > 1}
|
|
|
|
# Per-subfolder disk breakdown
|
|
by_folder = defaultdict(lambda: [0, 0]) # folder -> [count, bytes]
|
|
for k, sz in on_disk.items():
|
|
parts = k.split('/')
|
|
folder = parts[1] if len(parts) > 2 else '(root)'
|
|
by_folder[folder][0] += 1
|
|
by_folder[folder][1] += sz if sz > 0 else 0
|
|
|
|
print('-' * 72)
|
|
print(' ON DISK (shared, unprefixed — local backend does not isolate)')
|
|
for folder, (cnt, b) in sorted(by_folder.items()):
|
|
print(f' {folder:<24} {cnt:>8} files {human(b):>12}')
|
|
print(f' {"TOTAL":<24} {len(on_disk):>8} files {human(total_bytes):>12}')
|
|
print('-' * 72)
|
|
print(' RECONCILIATION (all audited tenants)')
|
|
print(f' referenced by some tenant : {len(all_referenced):>8}')
|
|
print(f' on disk but UNREFERENCED : {len(orphans):>8} '
|
|
f'({human(orphan_bytes)}, NOT synced — no tenant owns them)')
|
|
print(f' keys claimed by >1 tenant : {len(multi_claimed):>8} '
|
|
f'<-- must be 0')
|
|
print('=' * 72)
|
|
|
|
if multi_claimed:
|
|
print('\n WARNING: the following keys are referenced by more than one tenant.')
|
|
print(' They will be copied under EVERY claiming prefix. Investigate the')
|
|
print(' source (a DB restore across tenants?) before cutover:')
|
|
for k, slugs in list(multi_claimed.items())[:10]:
|
|
print(f' - {k} ({", ".join(slugs)})')
|
|
if len(multi_claimed) > 10:
|
|
print(f' ... and {len(multi_claimed) - 10} more (see JSON report)')
|
|
|
|
if args.list_orphans and orphans:
|
|
print(f'\n Orphan files ({len(orphans)}):')
|
|
for k in sorted(orphans):
|
|
print(f' - {k} ({human(on_disk[k])})')
|
|
|
|
report = {
|
|
'generated_at': datetime.utcnow().isoformat() + 'Z',
|
|
'static_folder': STATIC_FOLDER,
|
|
'uploads_root': UPLOADS_ROOT,
|
|
'tenant_selector': args.tenant,
|
|
'tenants': per_tenant,
|
|
'disk': {
|
|
'total_files': len(on_disk),
|
|
'total_bytes': total_bytes,
|
|
'by_folder': {k: {'files': v[0], 'bytes': v[1]} for k, v in by_folder.items()},
|
|
},
|
|
'orphans': {
|
|
'count': len(orphans),
|
|
'bytes': orphan_bytes,
|
|
'files': sorted([{'key': k, 'bytes': on_disk[k]} for k in orphans],
|
|
key=lambda d: d['key']),
|
|
},
|
|
'multi_claimed': {k: sorted(v) for k, v in multi_claimed.items()},
|
|
}
|
|
|
|
report_path = args.report or os.path.join(
|
|
'/tmp', f'jqc_photo_audit_{datetime.utcnow().strftime("%Y%m%d_%H%M%S")}.json')
|
|
try:
|
|
with open(report_path, 'w') as fh:
|
|
json.dump(report, fh, indent=2)
|
|
print(f'\n Full JSON report written to: {report_path}')
|
|
except OSError as e:
|
|
print(f'\n Could not write report to {report_path}: {e}')
|
|
|
|
baseline = sum(t.get('referenced', {}).get('present_on_disk', 0) for t in per_tenant)
|
|
print(f'\n BASELINE TO REMEMBER: {baseline} photos must still resolve after cutover.\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|