"""Exercise the staged application against local media fixtures, never production data."""
import hashlib,json,sqlite3,subprocess,threading,time,urllib.request,urllib.error,os
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from pathlib import Path
ROOT=Path(__file__).resolve().parent.parent
TEST=ROOT/'integration-work'/os.environ.get('HUOBAO_TEST_FOLDER','hd-test')
assert TEST.resolve().is_relative_to(ROOT/'integration-work')
FIXTURE=json.loads((TEST/'fixture.json').read_text())
BASE='http://127.0.0.1:'+os.environ.get('HUOBAO_TEST_PORT','18879')+'/api/v1'
BIN=ROOT/'integration-work/stage-4.0.11/app/resources/bin'
def api(path,body=None):
    req=urllib.request.Request(BASE+path,data=json.dumps(body).encode() if body is not None else None,headers={'Content-Type':'application/json'})
    try:
        with urllib.request.urlopen(req,timeout=30) as r: return json.load(r)['data']
    except urllib.error.HTTPError as e: raise RuntimeError(e.read().decode())
def connect(): return sqlite3.connect(TEST/'huobao.sqlite3')
def probe(path):
    meta=json.loads(subprocess.check_output([str(BIN/'ffprobe.exe'),'-v','error','-show_streams','-of','json',str(path)],creationflags=0x08000000))
    s=next(s for s in meta['streams'] if s['codec_type']=='video')
    return s['width'],s['height']
for label,size in [('2K','2560x1440'),('4K','3840x2176')]:
    target=TEST/(label+'.mp4')
    if not target.exists() or probe(target)!=tuple(map(int,size.split('x'))):
        subprocess.run([str(BIN/'ffmpeg.exe'),'-y','-hide_banner','-loglevel','error','-f','lavfi','-i',f'color=c=0x456789:s={size}:r=24:d=1',
            '-c:v','libx264','-preset','ultrafast','-pix_fmt','yuv420p',str(target)],check=True,creationflags=0x08000000)
calls=[]
bad=False
class Handler(BaseHTTPRequestHandler):
    def log_message(self,*args):pass
    def answer(self,data):
        self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(json.dumps(data).encode())
    def do_POST(self):
        body=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
        calls.append(body)
        self.answer({'status':'succeeded','video_url':'http://127.0.0.1:18880/'+('2K' if bad else body['resolution'])+'.mp4','duration':4})
    def do_GET(self):
        if self.path=='/v1/models': return self.answer({'data':[{'id':'MiniMax-H3-VDN-H3'}]})
        name=self.path.lstrip('/')
        if name not in ('2K.mp4','4K.mp4'): self.send_error(404);return
        data=(TEST/name).read_bytes()
        self.send_response(200);self.send_header('Content-Type','video/mp4');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
server=ThreadingHTTPServer(('127.0.0.1',18880),Handler)
threading.Thread(target=server.serve_forever,daemon=True).start()
with connect() as db:
    cfgid=db.execute("SELECT id FROM ai_service_configs WHERE model LIKE '%VDN-H3%' ORDER BY id DESC LIMIT 1").fetchone()[0]
    db.execute("UPDATE ai_service_configs SET is_active=0 WHERE model LIKE '%VDN-H3%' AND id!=?",(cfgid,))
    db.execute('UPDATE ai_service_configs SET base_url=?,is_active=1 WHERE id=?',('http://127.0.0.1:18880',cfgid))
    assert db.execute("SELECT COUNT(*) FROM ai_service_configs WHERE is_active=1 AND model LIKE '%VDN-H3%' AND base_url!='http://127.0.0.1:18880'").fetchone()[0]==0
sid=FIXTURE['storyboard']; eid=FIXTURE['episode']; original=FIXTURE['original']
sha=hashlib.sha256((TEST/original).read_bytes()).hexdigest()
def wait_task(tid):
    for _ in range(60):
        data=api('/tasks/'+str(tid)+'/diagnostics')
        if data['status']!='processing':return data
        time.sleep(0.5)
    raise AssertionError('task timed out')
try:
    complete=[]
    for resolution,size in [('2K',(2560,1440)),('4K',(3840,2160))]:
        tid=api(f'/hd/storyboards/{sid}',{'resolution':resolution})['id']
        result=wait_task(tid)
        assert result['status']=='completed',result
        with connect() as db:
            assert db.execute('SELECT video_url FROM storyboards WHERE id=?',(sid,)).fetchone()[0]==original
            video=db.execute('SELECT local_path FROM sys_task WHERE id=?',(tid,)).fetchone()[0]
        assert probe(TEST/video)==size
        complete.append((tid,video))
    listing=api(f'/hd/episodes/{eid}')
    shot=next(s for s in listing['items'] if s['id']==sid)
    assert shot['original']==original and shot['current']==original
    assert all(v['error'] is None for v in shot['versions'] if v['status']=='completed')
    api(f'/hd/storyboards/{sid}/select',{'task_id':complete[1][0]})
    with connect() as db: assert db.execute('SELECT video_url FROM storyboards WHERE id=?',(sid,)).fetchone()[0]==complete[1][1]
    api(f'/hd/storyboards/{sid}/select',{'original':True})
    bad=True
    tid=api(f'/hd/storyboards/{sid}',{'resolution':'4K'})['id']
    assert wait_task(tid)['status']=='failed'
    try: api(f'/hd/storyboards/{sid}/select',{'task_id':tid})
    except RuntimeError:pass
    else:raise AssertionError('failed low-resolution output was selectable')
    assert hashlib.sha256((TEST/original).read_bytes()).hexdigest()==sha
    assert all(c['model']=='MiniMax-H3-VDN-H3' and c['content'][1]['role']=='reference_image' and c['content'][1]['image_url']['url'].startswith('data:') for c in calls)
    with connect() as db:
        second=db.execute('SELECT id FROM storyboards WHERE episode_id=? AND id!=? ORDER BY id LIMIT 1',(eid,sid)).fetchone()[0]
        db.execute('UPDATE storyboards SET video_url=? WHERE id=?',(complete[1][1],second))
    merge=api(f'/merge/episodes/{eid}/merge',{'storyboard_ids':[sid,second]})
    merge_id=merge.get('merge_id') or merge.get('id')
    for _ in range(90):
        with connect() as db: row=db.execute('SELECT status,merged_url,error_msg FROM video_merges WHERE id=?',(merge_id,)).fetchone()
        if row and row[0]!='processing':break
        time.sleep(1)
    assert row and row[0]=='completed',row
    assert probe(TEST/row[1])==(3840,2160)
    (TEST/'test-results.json').write_text(json.dumps({'passed':True,'checks':['2K delivery','4K delivery','original remains unchanged','select and restore export version','wrong output size rejected','mixed resolution export keeps 4K'],'complete':complete,'original_sha256':sha},indent=2))
    print('PASS: 2K/4K delivery, preserved original, version switching, wrong-size rejection and mixed-size 4K export')
finally:
    with connect() as db: db.execute('UPDATE ai_service_configs SET base_url=? WHERE id=?',('http://192.168.110.111:18794',cfgid))
    server.shutdown()