server.py

#!/usr/bin/env python3
"""Winning Structures — local app server.

Paste a TikTok/YouTube/Instagram URL in the UI; the server downloads it (yt-dlp),
transcribes with word timestamps (Whisper), asks Claude to segment the script
into structural beats, and adds it to the library.

Run:  python3 server.py   →  http://localhost:8756
Needs on PATH: yt-dlp, ffmpeg, and one of mlx_whisper (Apple Silicon) or
whisper (openai-whisper). Beat mapping uses the claude CLI if installed,
otherwise the Anthropic API via ANTHROPIC_API_KEY.
"""
import json, os, re, shutil, subprocess, threading, urllib.request, uuid
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler

ROOT = os.path.dirname(os.path.abspath(__file__))
LIB = os.path.join(ROOT, 'library.json')
THUMBS = os.path.join(ROOT, 'thumbs')
WORK = os.path.join(ROOT, '.work')
PORT = 8756

JOBS = {}          # id -> {status, stage, error, video_id}
LIB_LOCK = threading.Lock()

os.makedirs(THUMBS, exist_ok=True)
os.makedirs(WORK, exist_ok=True)


def load_lib():
    with LIB_LOCK:
        return json.load(open(LIB)) if os.path.exists(LIB) else []


def add_to_lib(video):
    with LIB_LOCK:
        lib = json.load(open(LIB)) if os.path.exists(LIB) else []
        lib = [v for v in lib if v['id'] != video['id']]
        lib.insert(0, video)
        json.dump(lib, open(LIB, 'w'), ensure_ascii=False, indent=1)


def run(cmd, timeout=600):
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)


def sentence_split(words):
    sents, cur = [], []
    for w in words:
        cur.append(w)
        if re.search(r'[.!?]$', w['word'].strip()):
            sents.append(cur)
            cur = []
    if cur:
        sents.append(cur)
    return sents


ANALYSIS_PROMPT = """You are analyzing the script structure of a short-form video (TikTok-style). Winning videos retain viewers through one of two engines:
- Engine A — WITHHOLD: sentence 1 opens a debt (a withheld list/secret/number), the video pays it out on a schedule (hook -> negation -> reveal -> mechanism -> recap/CTA).
- Engine B — DEEPENING: the answer/take is given immediately, then every sentence reframes it one level deeper or more absurd. Retention = momentum.

Below are the transcript sentences, numbered, with durations. Group CONSECUTIVE sentences into "beats" — units where each beat has ONE structural job.

Return ONLY valid JSON, no markdown fences, matching exactly:
{"title": "<3-6 word title for this video>",
 "engine": "<e.g. 'Engine A — withhold funnel' or 'Engine B — deepening essay'>",
 "engineNote": "<one sentence: how this specific video retains viewers>",
 "beats": [{"label": "<SHORT UPPERCASE role: HOOK, POINT 1, CONTRAST, MECHANISM, SCRIPTS, MONEY LINE, REVEAL, NEGATION, GATE (CTA), DIAGNOSIS, ESCAPE HATCH, RECAP, CAVEAT, CTA, CLOSE...>",
            "cat": "<one of: hook|structure|value|scripts|contrast|money|cta|neutral>",
            "job": "<one short sentence: what this beat does for retention>",
            "from": <first sentence index>, "to": <last sentence index>}]}

Rules:
- Beats must cover every sentence index exactly once, in order, contiguous (beat N+1 "from" = beat N "to" + 1). First beat starts at 0, last beat ends at the final index.
- The first beat is almost always HOOK (cat "hook").
- Mark the single most quotable/screenshotable line as cat "money" (label MONEY LINE) — usually ~70-80%% through.
- cat guide: hook=opening device; structure=point names/recaps/transitions; value=mechanism/explanation/definition; scripts=literal words the viewer could say; contrast=flawed-majority/diagnosis/negation; money=the quotable; cta=any ask (follow/comment/buy), wherever it appears; neutral=concession/caveat/reassurance.
- 3 to 30 beats depending on video length. Merge tiny fragments into their beat.

Sentences:
%s"""


def ask_claude(prompt):
    if shutil.which('claude'):
        r = run(['claude', '-p', prompt], timeout=300)
        if r.returncode != 0:
            raise RuntimeError(f'claude CLI failed: {r.stderr[:300]}')
        return r.stdout
    api_key = os.environ.get('ANTHROPIC_API_KEY')
    if not api_key:
        raise RuntimeError('need the claude CLI on PATH or ANTHROPIC_API_KEY set')
    req = urllib.request.Request(
        'https://api.anthropic.com/v1/messages',
        data=json.dumps({
            'model': 'claude-opus-4-8',
            'max_tokens': 16000,
            'messages': [{'role': 'user', 'content': prompt}],
        }).encode(),
        headers={'Content-Type': 'application/json', 'x-api-key': api_key,
                 'anthropic-version': '2023-06-01'})
    with urllib.request.urlopen(req, timeout=300) as resp:
        body = json.load(resp)
    return ''.join(b.get('text', '') for b in body['content'] if b.get('type') == 'text')


def analyze_structure(sents):
    lines = []
    for i, s in enumerate(sents):
        text = ''.join(w['word'] for w in s).strip()
        dur = s[-1]['end'] - s[0]['start']
        lines.append(f"{i} ({dur:.1f}s): {text}")
    out = ask_claude(ANALYSIS_PROMPT % '\n'.join(lines))
    m = re.search(r'\{.*\}', out, re.DOTALL)
    if not m:
        raise RuntimeError(f'no JSON in claude output: {out[:300]}')
    plan = json.loads(m.group(0))
    # validate coverage; fall back to 1 beat per sentence if broken
    idx = 0
    for b in plan['beats']:
        if b['from'] != idx or b['to'] < b['from']:
            raise RuntimeError(f'non-contiguous beats from claude: {plan["beats"]}')
        idx = b['to'] + 1
    if idx != len(sents):
        raise RuntimeError(f'beats cover {idx} of {len(sents)} sentences')
    return plan


def build_beats(plan, sents):
    beats = []
    for b in plan['beats']:
        seg = [w for s in sents[b['from']:b['to'] + 1] for w in s]
        beats.append({
            'label': b['label'], 'cat': b.get('cat', 'value'), 'job': b.get('job', ''),
            'text': ''.join(w['word'] for w in seg).strip(),
            'start': round(seg[0]['start'], 2), 'end': round(seg[-1]['end'], 2),
            'dur': round(seg[-1]['end'] - seg[0]['start'], 2),
            'words': len(seg),
        })
    return beats


def process(job_id, url):
    job = JOBS[job_id]
    wdir = os.path.join(WORK, job_id)
    os.makedirs(wdir, exist_ok=True)
    try:
        job['stage'] = 'downloading'
        # Some TikTok formats claim aac audio in metadata but ship video-only files,
        # so probe after download and retry with h264 formats (which carry real audio).
        vpath = None
        for fmt in ['bv*+ba/b', 'b[vcodec^=h264]/b[format_id^=h264]/download/b']:
            for f in os.listdir(wdir):
                if f.startswith('video.'):
                    os.remove(os.path.join(wdir, f))
            r = run(['yt-dlp', '--no-playlist', '-o', os.path.join(wdir, 'video.%(ext)s'),
                     '--print-to-file', '%(uploader)s\t%(id)s\t%(webpage_url)s', os.path.join(wdir, 'meta.txt'),
                     '--no-simulate', '-f', fmt, '--merge-output-format', 'mp4', url], timeout=300)
            vids = [f for f in os.listdir(wdir) if f.startswith('video.')]
            if r.returncode != 0 or not vids:
                continue
            candidate = os.path.join(wdir, vids[0])
            streams = run(['ffprobe', '-v', 'error', '-show_entries', 'stream=codec_type',
                           '-of', 'csv=p=0', candidate]).stdout
            if 'audio' in streams:
                vpath = candidate
                break
        if not vpath:
            raise RuntimeError(f'download failed or no audio track: {(r.stderr or r.stdout)[-300:]}')
        uploader, vid_id, page_url = open(os.path.join(wdir, 'meta.txt')).read().strip().splitlines()[-1].split('\t')
        handle = '@' + re.sub(r'^@', '', uploader or 'unknown')
        video_id = re.sub(r'[^a-z0-9_]', '', (uploader or 'v').lower()) + '_' + vid_id

        job['stage'] = 'transcribing'
        if shutil.which('mlx_whisper'):
            r = run(['mlx_whisper', vpath, '--model', 'mlx-community/whisper-large-v3-turbo',
                     '--word-timestamps', 'True', '--output-format', 'json', '--output-dir', wdir], timeout=900)
        elif shutil.which('whisper'):
            r = run(['whisper', vpath, '--model', 'turbo', '--word_timestamps', 'True',
                     '--output_format', 'json', '--output_dir', wdir], timeout=1800)
        else:
            raise RuntimeError('no whisper found: install mlx-whisper (Apple Silicon) or openai-whisper')
        jsons = [f for f in os.listdir(wdir) if f.endswith('.json')]
        if not jsons:
            raise RuntimeError(f'transcription failed: {r.stderr[-300:]}')
        d = json.load(open(os.path.join(wdir, jsons[0])))
        words = [w for s in d['segments'] for w in s.get('words', [])]
        if not words:
            raise RuntimeError('no words in transcript')
        sents = sentence_split(words)

        job['stage'] = 'analyzing structure'
        plan = analyze_structure(sents)
        beats = build_beats(plan, sents)

        job['stage'] = 'finishing'
        run(['ffmpeg', '-y', '-loglevel', 'error', '-ss', '1.5', '-i', vpath,
             '-frames:v', '1', '-vf', 'scale=-2:480', '-q:v', '4',
             os.path.join(THUMBS, video_id + '.jpg')])
        video = {
            'id': video_id, 'creator': handle, 'url': page_url,
            'title': plan.get('title', 'Untitled'),
            'engine': plan.get('engine', ''), 'engineNote': plan.get('engineNote', ''),
            'total': round(beats[-1]['end'] - beats[0]['start'], 1),
            'beats': beats,
        }
        add_to_lib(video)
        job['video_id'] = video_id
        job['status'] = 'done'
    except Exception as e:
        job['status'] = 'error'
        job['error'] = str(e)[:500]
    finally:
        job['stage'] = job.get('stage', '')
        shutil.rmtree(wdir, ignore_errors=True)


class Handler(SimpleHTTPRequestHandler):
    def __init__(self, *a, **kw):
        super().__init__(*a, directory=os.path.join(ROOT, 'static'), **kw)

    def log_message(self, *a):
        pass

    def send_json(self, obj, code=200):
        body = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == '/api/videos':
            return self.send_json(load_lib())
        if self.path.startswith('/api/job/'):
            job = JOBS.get(self.path.rsplit('/', 1)[1])
            return self.send_json(job or {'status': 'unknown'}, 200 if job else 404)
        if self.path.startswith('/thumbs/'):
            name = os.path.basename(self.path)
            fp = os.path.join(THUMBS, name)
            if os.path.exists(fp):
                self.send_response(200)
                self.send_header('Content-Type', 'image/jpeg')
                self.send_header('Content-Length', str(os.path.getsize(fp)))
                self.end_headers()
                self.wfile.write(open(fp, 'rb').read())
                return
            return self.send_json({'error': 'not found'}, 404)
        return super().do_GET()

    def do_POST(self):
        if self.path == '/api/add':
            length = int(self.headers.get('Content-Length', 0))
            try:
                data = json.loads(self.rfile.read(length))
                url = data['url'].strip()
                assert re.match(r'^https?://', url)
            except Exception:
                return self.send_json({'error': 'bad url'}, 400)
            job_id = uuid.uuid4().hex[:12]
            JOBS[job_id] = {'status': 'running', 'stage': 'queued', 'url': url}
            threading.Thread(target=process, args=(job_id, url), daemon=True).start()
            return self.send_json({'job': job_id})
        return self.send_json({'error': 'not found'}, 404)


def check_deps():
    missing = [t for t in ('yt-dlp', 'ffmpeg') if not shutil.which(t)]
    if not (shutil.which('mlx_whisper') or shutil.which('whisper')):
        missing.append('mlx_whisper or whisper (openai-whisper)')
    if not (shutil.which('claude') or os.environ.get('ANTHROPIC_API_KEY')):
        missing.append('claude CLI or ANTHROPIC_API_KEY')
    if missing:
        print('Missing (needed to add new videos; browsing the library works without them):')
        for m in missing:
            print(f'  - {m}')


if __name__ == '__main__':
    check_deps()
    print(f'Winning Structures → http://localhost:{PORT}')
    ThreadingHTTPServer(('127.0.0.1', PORT), Handler).serve_forever()