"""Bounded CPU-only clipping with decoded-source PTS evidence and no resampling."""
import argparse
import bisect
import csv
import datetime
from decimal import Decimal
import hashlib
import json
import math
import os
from pathlib import Path
import re
import shutil
import signal
import subprocess
import time
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parent.parent
PREP = ROOT.parent / '2026-10-28_contextra-self-filming-original01-review_r01'
SOURCE = PREP / 'clean_joined.mp4'
SOURCE_MAP = PREP / '아이서티_컨텍스트라_싸움_쓰러짐_자체촬영_원본01_검토자료/frame_map.csv'
REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor')
OBS = REPO / 'docs/profiles/ltb1021/topics/runtime-lifecycle/analysis/20261029T125858+0900-analysis-self-filming-original01-user-observations.md'
LABEL = '아이서티_컨텍스트라_싸움_쓰러짐_자체촬영_원본01_클립자료'
DOWNLOAD = ROOT / LABEL
META = DOWNLOAD / 'metadata'
REPORT = ROOT / 'reports'
CPUS = sorted(os.sched_getaffinity(0))[-2:]
OBS_RANGES = [
    (3094,3341),(8241,8396),(11963,12098),(12587,12974),(13098,13383),
    (13395,13538),(13839,14128),(14313,14501),(14502,15108),(15220,15468),
    (16168,16388),(16389,17083),(17084,18024),(18025,18730),(18731,19645),
    (19663,20470),(20525,20930),(20931,27655),(27700,30249),(30980,31937),
]
GROUPS = [
    ('C01','착석후기립_좌측퇴장',3094,3341,['O01']),
    ('C02','자리에서접근_겹쳐서기',8241,8396,['O02']),
    ('C03','걷기와달리기_멀어짐',11963,12098,['O03']),
    ('C04','문출입_원거리인물등장',12587,12974,['O04']),
    ('C05','원거리교차서기_혼자정권지르기',13098,13538,['O05','O06']),
    ('C06','같은문에서동행_한명하단퇴장',13839,14128,['O07']),
    ('C07','상대지나달리기_문으로퇴장',14313,14504,['O08','O09']),
    ('C08','대기통행_손짓접근_가림과퇴장',14505,15468,['O09','O10']),
    ('C09','문서굽힘_근거리공방_함께굽힘',16168,17083,['O11','O12']),
    ('C10','가슴밀치기_어깨흔들기_대기',17084,18024,['O13']),
    ('C11','대각공방_껴안는태클_거리벌리기',18025,18730,['O14']),
    ('C12','측면공방_태클_인사굽힘',18731,19645,['O15']),
    ('C13','대화악수_어깨동무_하이파이브_쪼그림',19663,20930,['O16','O17']),
    ('C14','원거리공방_밀치기흔들기_태클',20931,22046,['O18']),
    ('C15','악수하이파이브_접근_쪼그림',22110,22669,['O18']),
    ('C16','손잡기팔짱_동행왕복보행',22884,23781,['O18']),
    ('C17','근거리교차보행_자연통행_제자리운동',24075,25317,['O18']),
    ('C18','문서집기_쪼그림_상반신굽힘',25659,26433,['O18']),
    ('C19','근거리측면_쪼그림굽힘_벽기대기',27700,28551,['O19']),
    ('C20','원거리측면_벽기대기_쪼그림굽힘',28740,29514,['O19']),
    ('C21','정면비틀거림_하단잘림_회전쪼그림',29783,30249,['O19']),
    ('C22','설명미제공_사용자판단용',30980,31937,['O20']),
    ('L01','연속검토_걷기달리기부터끝까지',11963,31937,[f'O{i:02d}' for i in range(3,21)]),
]


def digest(path):
    h = hashlib.sha256()
    with path.open('rb') as f:
        for block in iter(lambda: f.read(1024*1024), b''):
            h.update(block)
    return h.hexdigest()


def save(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open('x', encoding='utf-8') as f:
        json.dump(value, f, ensure_ascii=False, indent=2)
        f.write('\n')


def load(path):
    return json.loads(path.read_text(encoding='utf-8'))


def resources():
    available = next(int(s.split()[1])*1024 for s in Path('/proc/meminfo').read_text().splitlines() if s.startswith('MemAvailable:'))
    state = dict(root_free=shutil.disk_usage('/').free, data_free=shutil.disk_usage(ROOT).free,
                 ram_available=available, load=os.getloadavg(), cpus=CPUS)
    assert state['root_free'] >= 15*2**30, state
    assert state['data_free'] >= 3*2**30 and available >= 4*2**30, state
    return state


def run(cmd, stage, timeout=600, allow_info=False):
    REPORT.mkdir(exist_ok=True)
    attempt = 1
    while (REPORT/f'{stage}_{attempt:02d}.command.json').exists():
        attempt += 1
    prefix = REPORT/f'{stage}_{attempt:02d}'
    save(Path(str(prefix)+'.command.json'), cmd)
    outpath, errpath = Path(str(prefix)+'.stdout'), Path(str(prefix)+'.stderr')
    before = resources()
    samples = []
    started = time.monotonic()
    with outpath.open('xb') as out, errpath.open('xb') as err:
        proc = subprocess.Popen(cmd, stdout=out, stderr=err, start_new_session=True)
        try:
            while True:
                try:
                    rc = proc.wait(timeout=10)
                    break
                except subprocess.TimeoutExpired:
                    if time.monotonic()-started >= timeout:
                        raise TimeoutError(stage)
                    current = resources()
                    status = Path(f'/proc/{proc.pid}/status')
                    rss = next((int(s.split()[1])*1024 for s in status.read_text().splitlines() if s.startswith('VmRSS:')),0) if status.exists() else 0
                    assert rss < 2*2**30, ('own child RSS', rss)
                    current['child_rss'] = rss
                    samples.append(current)
                    if len(samples) % 3 == 0:
                        print(stage, round(time.monotonic()-started,1), 'seconds', flush=True)
        except BaseException:
            if proc.poll() is None:
                os.killpg(proc.pid, signal.SIGTERM)
                try:
                    proc.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    os.killpg(proc.pid, signal.SIGKILL)
                    proc.wait()
            raise
    save(Path(str(prefix)+'.resources.json'), dict(before=before, samples=samples, after=resources(),
         elapsed_seconds=round(time.monotonic()-started,3), returncode=rc, child_pid=proc.pid, child_reaped=True))
    errors = errpath.read_text(errors='replace')
    if rc:
        raise RuntimeError(stage+': '+errors[-1600:])
    if not allow_info:
        assert not errors.strip(), (stage, errors[-1500:])
    else:
        assert not re.search(r'error while decoding|corrupt|invalid NAL|non.monoton|conversion failed', errors, re.I), stage
    return outpath, errpath


def probe(path, stage):
    cmd = ['ffprobe','-v','error','-threads','2','-select_streams','v:0',
           '-show_frames','-show_streams','-show_entries',
           'stream=codec_name,width,height,pix_fmt,color_range,color_space,color_transfer,color_primaries,time_base,nb_frames,duration,sample_aspect_ratio:frame=best_effort_timestamp,width,height',
           '-of','json',str(path)]
    out, _ = run(cmd, stage)
    return load(out), out


def mapping():
    with SOURCE_MAP.open(encoding='utf-8-sig', newline='') as f:
        rows = list(csv.DictReader(f))
    assert [int(x['review_frame_1based']) for x in rows] == list(range(1,31938))
    ticks = [int(Decimal(x['review_pts_seconds'])*1000000) for x in rows]
    assert all(b>a for a,b in zip(ticks,ticks[1:]))
    assert ticks[14504]-ticks[14503] == 1016668
    return rows, ticks


def prepare():
    rows, ticks = mapping()
    existing = META/'manifest.json'
    if existing.exists():
        manifest = load(existing)
        assert manifest['source_sha256'] == digest(SOURCE)
        assert manifest['source_map_sha256'] == digest(SOURCE_MAP)
        assert manifest['observations_sha256'] == digest(OBS)
        return manifest, rows, ticks
    for folder in (DOWNLOAD/'clips',DOWNLOAD/'continuous',META/'frame_maps',META/'validation',META/'observations'):
        folder.mkdir(parents=True, exist_ok=True)
    source_hash = digest(SOURCE)
    source_probe, evidence = probe(SOURCE,'source_decode')
    stream = source_probe['streams'][0]
    assert stream['time_base'] == '1/1000000'
    assert [int(f['best_effort_timestamp']) for f in source_probe['frames']] == ticks
    assert all((f['width'],f['height']) == (1280,720) for f in source_probe['frames'])
    parts = [(3094,10704),(10705,10804),(10805,14504),(14505,31937)]
    specs = []
    for ident,name,a,b,obs_ids in GROUPS:
        if ident in ('L01','C22'):
            start,end = a,b
        else:
            low,high = next((lo,hi) for lo,hi in parts if lo<=a<=b<=hi)
            start = max(low,bisect.bisect_left(ticks,ticks[a-1]-2000000)+1)
            end = min(high,bisect.bisect_right(ticks,ticks[b-1]+2000000))
        folder = 'continuous' if ident=='L01' else 'clips'
        intersected = [f'O{i:02d}' for i,(lo,hi) in enumerate(OBS_RANGES,1) if lo<=end and hi>=start]
        specs.append(dict(id=ident,name=name,output=f'{folder}/{ident}_{name}.mp4',
             core_start=a,core_end=b,start=start,end=end,frames=end-start+1,
             first_pts_ticks=ticks[start-1],last_pts_ticks=ticks[end-1],
             end_exclusive_ticks=ticks[end] if end<len(ticks) else ticks[-1]+40000,
             context_before_seconds=(ticks[a-1]-ticks[start-1])/1e6,
             context_after_seconds=(ticks[end-1]-ticks[b-1])/1e6,
             core_observation_ids=obs_ids,intersected_observation_ids=intersected,
             contains_pts_gap=start<=14504 and end>=14505,
             expectation='per_observation_not_one_clip_label'))
    assert len(specs)==23 and sum(s['contains_pts_gap'] for s in specs)==1
    assert specs[-1]['start']==11963 and specs[-1]['end']==31937
    selected = set(n for s in specs[:-1] for n in range(s['start'],s['end']+1))
    assert all(any(n in selected for n in range(a,b+1)) for a,b in OBS_RANGES)
    assert all(n in selected for i,(a,b) in enumerate(OBS_RANGES,1) if i not in (18,19) for n in range(a,b+1))
    snapshot = META/'observations'/OBS.name
    with snapshot.open('xb') as f:
        f.write(OBS.read_bytes())
    manifest = dict(schema='contextra.self-filming-clips.v1',created_at=datetime.datetime.now(ZoneInfo('Asia/Seoul')).isoformat(),
         profile='ltb1021',source_run_id=PREP.name,source_file=SOURCE.name,source_sha256=source_hash,
         source_map_sha256=digest(SOURCE_MAP),observations_sha256=digest(OBS),source_probe_evidence=evidence.name,
         source_stream=stream,source_frames=len(rows),numbering='1-based review / clean_joined, NOT original full-file decoded frame',
         individual_clips=22,continuous_clips=1,cpus=CPUS,nice=10,
         policy=dict(no_resampling=True,no_semantic_analysis=True,no_model_inference=True,
                     context_seconds=2,first_excluded_frames=[1,3093],preserve_gap=[14504,14505],
                     crf=18,preset='veryfast',b_frames=0,pixel_format='yuvj420p',time_base='1/1000000'),
         clips=specs)
    save(existing, manifest)
    save(ROOT/'checkpoint_prepared.json',dict(status='MANIFEST_LOCKED',manifest_sha256=digest(existing),clips=len(specs)))
    print('MANIFEST_LOCKED',len(specs),'source_frames',len(rows),flush=True)
    return manifest, rows, ticks


def render(spec, manifest, rows, ticks):
    ident = spec['id']
    output = DOWNLOAD/spec['output']
    result_path = META/'validation'/f'{ident}.json'
    if output.exists() and result_path.exists():
        result = load(result_path)
        assert result['sha256'] == digest(output)
        print('REUSE',ident,flush=True)
        return result
    assert not output.exists() and not result_path.exists(), ('unverified existing output',ident)
    first,last = spec['first_pts_ticks'],spec['last_pts_ticks']
    end_exclusive = spec['end_exclusive_ticks']
    attempt = 1
    while (REPORT/f'{ident}_draft_{attempt:02d}.mp4').exists():
        attempt += 1
    draft = REPORT/f'{ident}_draft_{attempt:02d}.mp4'
    vf = f'trim=start_pts={first}:end_pts={end_exclusive},showinfo,setpts=PTS-STARTPTS'
    cmd = ['ffmpeg','-nostdin','-hide_banner','-loglevel','info','-n','-copyts',
           '-threads','2','-ss',f'{max(0,first-2000000)/1e6:.6f}','-i',str(SOURCE),
           '-map','0:v:0','-an','-map_metadata','-1','-filter_threads','1','-vf',vf,
           '-c:v','libx264','-preset','veryfast','-crf','18','-threads','2','-bf','0',
           '-pix_fmt','yuvj420p','-color_range','pc','-colorspace','bt709','-color_trc','bt709','-color_primaries','bt709',
           '-fps_mode','passthrough','-enc_time_base','-1','-video_track_timescale','1000000',
           '-movflags','+faststart',str(draft)]
    print('RENDER',ident,spec['start'],spec['end'],spec['frames'],flush=True)
    _, log = run(cmd, ident+'_encode',1800 if ident=='L01' else 900,allow_info=True)
    trace = [(int(n),int(p)) for n,p in re.findall(r'n:\s*(\d+)\s+pts:\s*(-?\d+)\s+pts_time:',log.read_text())]
    expected = ticks[spec['start']-1:spec['end']]
    assert trace == list(enumerate(expected)), (ident,'source selection mismatch',len(trace),len(expected))
    decoded, probe_path = probe(draft,ident+'_decode')
    frames = decoded['frames']
    actual = [int(f['best_effort_timestamp']) for f in frames]
    assert actual == [t-first for t in expected], (ident,'output PTS mismatch',len(actual),len(expected))
    st = decoded['streams'][0]
    for key in ('width','height','pix_fmt','color_range','color_space','color_transfer','color_primaries','time_base'):
        assert st[key] == manifest['source_stream'][key], (ident,key)
    assert all((f['width'],f['height'])==(1280,720) for f in frames)
    if spec['contains_pts_gap']:
        k = 14504-spec['start']
        assert actual[k+1]-actual[k] == 1016668
    else:
        assert max((b-a for a,b in zip(actual,actual[1:])),default=0) <= 100000
    frame_map = META/'frame_maps'/f'{ident}.csv'
    with frame_map.open('x',encoding='utf-8-sig',newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['clip_id','clip_frame_1based','clip_pts_seconds','review_frame_1based',
                         'review_pts_seconds','source_pts_seconds','source_packet_1based_not_decoded_frame',
                         'retained_segment','region','observation_ids','prior_review_gap_seconds','gap_boundary'])
        for local,(row,pts) in enumerate(zip(rows[spec['start']-1:spec['end']],actual),1):
            n = int(row['review_frame_1based'])
            ids = ';'.join(f'O{i:02d}' for i,(a,b) in enumerate(OBS_RANGES,1) if a<=n<=b)
            delta = ticks[n-1]-ticks[n-2] if n>1 else 0
            writer.writerow([ident,local,f'{pts/1e6:.6f}',n,row['review_pts_seconds'],row['source_pts_seconds'],
                 row['source_packet_1based_not_decoded_frame'],row['retained_segment'],
                 'core' if spec['core_start']<=n<=spec['core_end'] else 'context',ids,f'{delta/1e6:.6f}',
                 'source_pts_gap' if n==14505 else ('excluded_gop_join' if n in (10705,10805) else '')])
    result = dict(status='PASS',id=ident,frames=len(actual),source_selection_exact=True,relative_pts_error_ticks=0,
         first_review_frame=spec['start'],last_review_frame=spec['end'],relative_last_pts_seconds=actual[-1]/1e6,
         output_duration_seconds=float(st['duration']),source_display_duration_seconds=(end_exclusive-first)/1e6,
         duration_delta_seconds=float(st['duration'])-(end_exclusive-first)/1e6,
         decode_errors='',geometry_color_preserved=True,showinfo_evidence=log.name,decode_evidence=probe_path.name,
         sha256=digest(draft),bytes=draft.stat().st_size,map_sha256=digest(frame_map))
    assert abs(result['duration_delta_seconds']) < .05, (ident,'tail duration',result['duration_delta_seconds'])
    draft.rename(output)
    save(result_path,result)
    print('PASS',ident,len(actual),'duration_delta',round(result['duration_delta_seconds'],6),flush=True)
    return result


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--only')
    args = parser.parse_args()
    os.sched_setaffinity(0,CPUS)
    os.nice(10)
    manifest,rows,ticks = prepare()
    chosen = [s for s in manifest['clips'] if not args.only or s['id']==args.only]
    assert chosen
    for spec in chosen:
        render(spec,manifest,rows,ticks)
    assert digest(SOURCE)==manifest['source_sha256']
    assert digest(SOURCE_MAP)==manifest['source_map_sha256']
    assert digest(OBS)==manifest['observations_sha256']
    print('REQUESTED_RENDER_COMPLETE',len(chosen),flush=True)


if __name__=='__main__':
    main()