"""Apply media review decisions made in the Speech app to the content workspace trackers. The app records who approved or rejected each imported file (speech:export-review-decisions). This script fetches that export over SSH, or reads a saved file, and updates: language_mastery_2026-09-12/05_tracking/production_queue.json (+ production_history.jsonl) content_factory/tracking/legacy_media_jobs.json (mirror of the queue) ixl_benchmark_and_expansion_2026-09-12/tracking/media_queue.json content_factory/assets/**/job.json and assets/diagram_registry.json A picture is approved when its file is approved and goes to rework when rejected. An audio pack or image pair is approved only when every file in it is approved, and goes to rework as soon as one is rejected. Only review states are changed; a status someone set by hand for another reason (for example real_object_alternative) is reported and left alone. Dry run unless --apply is given. python scripts/sync_review_decisions.py # fetch from the server, show changes python scripts/sync_review_decisions.py --apply # and write them python scripts/sync_review_decisions.py --file decisions.json --apply """ import argparse, collections, json, posixpath, shutil, subprocess, sys from datetime import datetime, timezone from pathlib import Path LM, IX, CF = 'language_mastery_2026-09-12', 'ixl_benchmark_and_expansion_2026-09-12', 'content_factory' STAMP = datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S') def load_decisions(args): if args.file: return json.loads(Path(args.file).read_text(encoding='utf-8')) cmd = f"cd {args.app} && php artisan speech:export-review-decisions --all" out = subprocess.run(['ssh', '-o', 'BatchMode=yes', args.ssh, cmd], capture_output=True, text=True, encoding='utf-8', check=True).stdout return json.loads(out[out.index('{'):]) class Writer: def __init__(self, apply): self.apply, self.changes, self.backed_up = apply, collections.Counter(), set() def save(self, path, data): if not self.apply: return if path not in self.backed_up: shutil.copy2(path, path.with_name(f'{path.name}.bak-sync-{STAMP}')) self.backed_up.add(path) tmp = path.with_suffix(path.suffix + '.tmp') tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') tmp.replace(path) def who(d): dec = d.get('decision') or {} by = dec.get('by_name') or (f"user {dec['by']}" if dec.get('by') else 'the app') at = (dec.get('at') or d.get('updated_at') or '')[:10] return f"{by} on {at}" + (f": {dec['note']}" if dec.get('note') else '') def pack_outcome(files, expected): statuses = collections.Counter(f['status'] for f in files) rejected = [f.get('item') or f.get('source_file') for f in files if f['status'] == 'rejected'] if rejected: return 'rejected', f"{len(rejected)} of {expected} rejected in app review ({', '.join(map(str, rejected[:6]))}{'…' if len(rejected) > 6 else ''})" if statuses['approved'] == expected: return 'approved', f"all {expected} files approved in app review" return 'pending', f"{statuses['approved']} of {expected} files approved in app review" def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument('--content-root', default='C:/Git/content/chatgpt') ap.add_argument('--file', help='a saved export instead of fetching over SSH') ap.add_argument('--ssh', default='neurapyai-dotai') ap.add_argument('--app', default='~/public_html/speech.neurapy.ai') ap.add_argument('--apply', action='store_true') args = ap.parse_args() root = Path(args.content_root) export = load_decisions(args) by_file = {d['source_file']: d for d in export['decisions'] if d.get('source_file')} w = Writer(args.apply) now = datetime.now(timezone.utc).isoformat() report = [] # ---- Language to Mastery production queue (pictures one by one, audio by pack) qpath = root / LM / '05_tracking/production_queue.json' queue = json.loads(qpath.read_text(encoding='utf-8')) history = [] review_states = {'needs_review', 'approved', 'rework'} for job in queue: if job['type'] in ('image_or_scene_card', 'diagram_or_typeset_card'): d = by_file.get(f"{LM}/{job.get('actual_output', '')}") if not d: continue if d['status'] in ('review', 'draft') and not d.get('decision') and job['status'] == 'needs_review': continue # nobody has decided yet: nothing to record target = {'approved': 'approved', 'rejected': 'rework'}.get(d['status'], 'needs_review') summary = f"app review: {d['status']} by {who(d)}" elif job['type'] == 'audio_pack': # A pack's clips may come from either recording set (earlier in briefs/, slower in outputs/). files = [d for f, d in by_file.items() if f.startswith((f"{LM}/04_media/briefs/{job['id']}/", f"{LM}/04_media/outputs/{job['id']}/")) and f.endswith(('.mp3', '.wav'))] manifest = root / LM / '04_media/briefs' / job['id'] / 'manifest.json' if not files or not manifest.exists(): continue expected = len(json.loads(manifest.read_text(encoding='utf-8'))['tracks']) outcome, summary = pack_outcome(files, expected) if outcome == 'pending' and job['status'] == 'needs_review' and not any(f.get('decision') for f in files): continue target = {'approved': 'approved', 'rejected': 'rework'}.get(outcome, 'needs_review') summary = 'app review: ' + summary else: continue if job['status'] not in review_states: report.append(f"left alone {job['id']}: status {job['status']} was set outside review") continue marker = f" | {summary}" if job['status'] == target and marker in (job.get('qa_note') or ''): continue base_note = (job.get('qa_note') or '').split(' | app review:')[0] history.append({'date': now, 'id': job['id'], 'from': job['status'], 'to': target, 'note': summary}) job.update(status=target, qa_note=base_note + marker, updated_at=now) w.changes[f'production_queue → {target}'] += 1 if history: w.save(qpath, queue) if args.apply: with (root / LM / '05_tracking/production_history.jsonl').open('a', encoding='utf-8') as f: for h in history: f.write(json.dumps(h, ensure_ascii=False) + '\n') lpath = root / CF / 'tracking/legacy_media_jobs.json' legacy = json.loads(lpath.read_text(encoding='utf-8')) src = {j['id']: j for j in queue} mirrored = 0 for j in legacy: s = src.get(j['id']) if s and (j['status'], j.get('qa_note')) != (s['status'], s.get('qa_note')): j.update(status=s['status'], qa_note=s.get('qa_note', ''), updated_at=now) mirrored += 1 if mirrored: w.save(lpath, legacy) w.changes['legacy_media_jobs mirrored'] += mirrored # ---- Spatial pilot media queue (image pairs and the audio pack) mpath = root / IX / 'tracking/media_queue.json' mq = json.loads(mpath.read_text(encoding='utf-8')) changed = False for job in mq: paths = [posixpath.normpath(f"{IX}/tracking/{p}") for p in job.get('output_paths', [])] files = [by_file[p] for p in paths if p in by_file] files = [f for f in files if not f['source_file'].endswith(('.html', '.json'))] expected = len([p for p in job.get('output_paths', []) if not p.endswith(('.html', '.json'))]) if not files or job['status'] not in ('generated', 'review', 'approved', 'creating'): continue outcome, summary = pack_outcome(files, expected) if outcome == 'pending' and not any(f.get('decision') for f in files): continue target = {'approved': 'approved', 'rejected': 'creating'}.get(outcome, 'review') reviewers = sorted({who(f) for f in files if f.get('decision')}) note = f"{datetime.now().date()} app review: {summary}" if job['status'] == target and job.get('review_note', '').endswith(summary): continue job.update(status=target, review_note=note, reviewer='; '.join(reviewers)[:500] or job.get('reviewer'), updated_at=now) w.changes[f'media_queue → {target}'] += 1 changed = True if changed: w.save(mpath, mq) # ---- Content factory assets: scene images, location pilot, diagrams for job_path in sorted((root / CF / 'assets').glob('*/*/v1/job.json')): job = json.loads(job_path.read_text(encoding='utf-8')) out = job.get('actual_output') or job.get('output_path') d = by_file.get(f"{CF}/{out}") if out else None if not d or (d['status'] in ('review', 'draft') and not d.get('decision')): continue target = {'approved': 'approved', 'rejected': 'rejected'}.get(d['status'], 'pending') note = f"app review: {d['status']} by {who(d)}" if job.get('review_status') == target and job.get('app_review') == note: continue job.update(review_status=target, app_review=note) w.save(job_path, job) w.changes[f'asset job → {target}'] += 1 rpath = root / CF / 'assets/diagram_registry.json' registry = json.loads(rpath.read_text(encoding='utf-8')) changed = False for entry in registry: d = by_file.get(f"{CF}/{entry['output_path']}") if not d or (d['status'] in ('review', 'draft') and not d.get('decision')): continue target = {'approved': 'approved', 'rejected': 'rejected'}.get(d['status'], 'pending') note = f"app review: {d['status']} by {who(d)}" if entry.get('review_status') == target and entry.get('app_review') == note: continue entry.update(review_status=target, app_review=note) w.changes[f'diagram → {target}'] += 1 changed = True if changed: w.save(rpath, registry) decided = collections.Counter(d['status'] for d in export['decisions']) print(f"export {export.get('exported_at', '')[:19]}: {len(export['decisions'])} files ({dict(decided)})") print(('APPLIED' if args.apply else 'DRY RUN') + ':', dict(w.changes) or 'no tracker changes') for line in report[:20]: print(' ' + line) if args.apply and w.changes: print('Backups: *.bak-sync-' + STAMP + '. Refresh the content factory dashboard with: python tools/content.py status') if __name__ == '__main__': sys.exit(main())