`poll.py`

223 lines
#!/usr/bin/env python3
"""
videos.agentabrams.com chat → auto-regenerate videos.

Polls /api/requests, processes new entries, edits flow JSON via local Ollama
(qwen3:14b on Mac1), re-runs ship-video which re-publishes to Kamatera.

Cursor (last-processed timestamp) lives at cursor.txt to avoid re-processing.
First run starts cursor at "now" so legacy/test entries are skipped.
"""
import os, sys, json, time, base64, subprocess, urllib.request, urllib.error, re
from pathlib import Path
from datetime import datetime, timezone

ROOT = Path.home() / 'Projects' / 'videos-agentabrams-worker'
ROOT.mkdir(parents=True, exist_ok=True)
CURSOR = ROOT / 'cursor.txt'
LOG = ROOT / 'log.jsonl'
MAP_FILE = ROOT / 'flow-map.json'
FLOWS_DIR = Path.home() / '.claude' / 'skills' / 'app-demo-video' / 'flows'
SHIP_VIDEO = Path.home() / 'bin' / 'ship-video'
OLLAMA_URL = os.environ.get('OLLAMA_URL', 'http://localhost:11434')
OLLAMA_MODEL = os.environ.get('OLLAMA_MODEL', 'gemma3:4b')  # fast + no-thinking for flow JSON edits

API_BASE = 'https://videos.agentabrams.com'
API_AUTH = ('admin', 'DWSecure2024!')


def log_event(entry):
    entry['ts'] = datetime.now(timezone.utc).isoformat()
    with LOG.open('a') as f:
        f.write(json.dumps(entry) + '\n')


def http_get(url):
    req = urllib.request.Request(url)
    token = base64.b64encode(f'{API_AUTH[0]}:{API_AUTH[1]}'.encode()).decode()
    req.add_header('Authorization', f'Basic {token}')
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read())


def call_ollama(system, user, timeout=360):
    body = json.dumps({
        'model': OLLAMA_MODEL,
        'prompt': system + '\n\n' + user + '\n\nOutput only the updated JSON object, no explanation, no markdown.',
        'stream': False,
        'keep_alive': '30m',
        'options': {'temperature': 0.4, 'num_predict': 1500, 'num_ctx': 4096},
    }).encode()
    req = urllib.request.Request(
        f'{OLLAMA_URL}/api/generate',
        data=body,
        headers={'Content-Type': 'application/json'},
    )
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())['response']


def vid_id_to_slug(vid_id):
    """'Animals-Pitch-2026-05-06.mp4' → 'Animals-Pitch'"""
    base = vid_id[:-4] if vid_id.endswith('.mp4') else vid_id
    return re.sub(r'-\d{4}-\d{2}-\d{2}$', '', base)


def get_cursor():
    if CURSOR.exists():
        return CURSOR.read_text().strip()
    now = datetime.now(timezone.utc).isoformat()
    CURSOR.write_text(now)
    return now


def set_cursor(ts):
    CURSOR.write_text(ts)


VALID_ACTIONS = {'wait', 'scroll', 'click', 'hover'}


def validate_flow(flow):
    if not isinstance(flow, dict): return False, 'not a dict'
    if 'steps' not in flow or not isinstance(flow['steps'], list): return False, 'no steps[]'
    if not flow['steps']: return False, 'empty steps'
    if len(flow['steps']) > 16: return False, f'too many steps ({len(flow["steps"])})'
    for i, s in enumerate(flow['steps']):
        if not isinstance(s, dict): return False, f'step {i} not dict'
        if s.get('action') not in VALID_ACTIONS:
            return False, f'step {i} invalid action: {s.get("action")}'
        if s['action'] in ('click', 'hover') and not s.get('selector'):
            return False, f'step {i} ({s["action"]}) missing selector'
    return True, 'ok'


def edit_flow_via_llm(current_flow, user_msg, entry):
    system = (
        "Edit a JSON flow for a demo recorder. Action types: wait{ms,narration}, "
        "scroll{y,duration,narration}, click{selector,narration}, hover{selector,narration}. "
        "6-14 steps. Narration 5-10 words. Output ONE JSON object, nothing else."
    )
    user = (
        f"FLOW:\n{json.dumps(current_flow)}\n\n"
        f"SELECTORS: {entry.get('selectors_hint', '(none)')}\n\n"
        f"REQUEST: {user_msg}\n\n"
        "Output the updated flow JSON."
    )
    try:
        resp = call_ollama(system, user)
    except Exception as e:
        log_event({'event': 'ollama_error', 'error': str(e)})
        return None

    log_event({'event': 'llm_response_raw', 'head': (resp or '')[:400]})
    new_flow = None
    try:
        new_flow = json.loads(resp)
    except json.JSONDecodeError:
        m = re.search(r'\{[\s\S]*\}', resp)
        if m:
            try:
                new_flow = json.loads(m.group(0))
            except json.JSONDecodeError:
                pass
    if new_flow is None:
        log_event({'event': 'llm_invalid_json', 'response_head': resp[:400]})
        return None

    ok, why = validate_flow(new_flow)
    if not ok:
        log_event({'event': 'llm_invalid_schema', 'reason': why, 'flow': new_flow})
        return None

    if new_flow == current_flow:
        return None
    return new_flow


def process_request(req, flow_map):
    msg = req.get('message', '')
    ids = req.get('video_ids', []) or []
    log_event({'event': 'process_start', 'request_ts': req['ts'], 'video_ids': ids, 'message': msg})

    for vid_id in ids:
        slug = vid_id_to_slug(vid_id)
        if slug not in flow_map:
            log_event({'event': 'unknown_slug', 'video_id': vid_id, 'slug': slug})
            continue
        entry = flow_map[slug]
        flow_path = FLOWS_DIR / entry['flow_file']
        if not flow_path.exists():
            log_event({'event': 'flow_file_missing', 'video_id': vid_id, 'path': str(flow_path)})
            continue

        current = json.loads(flow_path.read_text())
        new_flow = edit_flow_via_llm(current, msg, entry)
        if not new_flow:
            log_event({'event': 'no_change', 'video_id': vid_id})
            continue

        # Backup + write
        backup = flow_path.with_suffix(f'.{int(time.time())}.bak.json')
        backup.write_text(flow_path.read_text())
        flow_path.write_text(json.dumps(new_flow, indent=2) + '\n')
        log_event({
            'event': 'flow_updated',
            'video_id': vid_id,
            'flow_file': str(flow_path),
            'backup': str(backup),
            'old_steps': len(current.get('steps', [])),
            'new_steps': len(new_flow.get('steps', [])),
        })

        # Re-record
        env = os.environ.copy()
        try:
            result = subprocess.run(
                [str(SHIP_VIDEO), entry['app_name'], entry['url']],
                capture_output=True, text=True, env=env, timeout=300,
            )
            log_event({
                'event': 'ship_video_done',
                'video_id': vid_id,
                'app_name': entry['app_name'],
                'returncode': result.returncode,
                'stdout_tail': (result.stdout or '')[-500:],
                'stderr_tail': (result.stderr or '')[-500:],
            })
        except subprocess.TimeoutExpired:
            log_event({'event': 'ship_video_timeout', 'video_id': vid_id})


def main():
    if not MAP_FILE.exists():
        print(f'flow-map.json missing at {MAP_FILE}', file=sys.stderr)
        sys.exit(2)
    flow_map = json.loads(MAP_FILE.read_text())

    cursor = get_cursor()
    try:
        data = http_get(f'{API_BASE}/api/requests')
    except Exception as e:
        log_event({'event': 'fetch_failed', 'error': str(e)})
        return

    requests = data.get('requests', [])
    new = [r for r in requests if r.get('ts', '') > cursor]
    if not new:
        return

    new.sort(key=lambda r: r['ts'])  # oldest first
    log_event({'event': 'tick', 'new_count': len(new), 'cursor': cursor})

    for req in new:
        try:
            process_request(req, flow_map)
        except Exception as e:
            log_event({'event': 'process_exception', 'request_ts': req['ts'], 'error': str(e)})
        set_cursor(req['ts'])


if __name__ == '__main__':
    main()