#!/usr/bin/env python3
"""CDN capability probe. Requires Python 3 and curl (HTTP/2 build for HTTPS)."""
import argparse
import json
import shutil
import subprocess
import tempfile
import threading
import time
import urllib.parse


def main():
    p=argparse.ArgumentParser(description='CDN capability probe; no third-party Python packages required')
    p.add_argument('base',help='Origin or CDN URL, e.g. https://example.com')
    p.add_argument('--http1',action='store_true',help='Force client HTTP/1.1 instead of requesting HTTP/2')
    p.add_argument('--content-type',choices=['application/octet-stream','application/grpc'],default='application/octet-stream')
    p.add_argument('--response',choices=['sse','raw'],default='sse')
    args=p.parse_args()
    url=urllib.parse.urlsplit(args.base)
    if url.scheme not in ('http','https') or not url.hostname or url.username or url.password:
        p.error('Use an http(s) origin URL without credentials')
    if not shutil.which('curl'):p.error('curl is required')
    base=urllib.parse.urlunsplit((url.scheme,url.netloc,'/cdn-test/api','',''))
    common=['curl','--silent','--show-error','--no-buffer','--connect-timeout','8','--max-time','25',
            '--http1.1' if args.http1 else '--http2','--header','Cache-Control: no-cache']
    if url.scheme=='http' and not args.http1:
        common[common.index('--http2')]='--http1.1'
    results=[]

    def probe(kind):
        begin=time.monotonic();arrivals=[];state={'ended':None,'error':None};workers=[]
        nonce=str(time.time_ns())
        if kind=='down':target=base+'/down?seconds=6&nonce='+nonce+'&mime='+args.response
        elif kind=='duplex':target=base+'/duplex?mime='+args.response
        else:
            out=subprocess.run(common+['--request','POST','--url',base+'/session'],capture_output=True,text=True,timeout=30)
            if out.returncode:raise RuntimeError(out.stderr.strip())
            token=json.loads(out.stdout)['id']
            target=base+'/events/'+token+'?mime='+args.response
        with tempfile.TemporaryFile() as err:
            with tempfile.NamedTemporaryFile(prefix='cdn-headers-',suffix='.txt') as header_file:
                cmd=common+['--dump-header',header_file.name,'--url',target]
                if kind=='duplex':cmd+=['--request','POST','--upload-file','-','--header','Expect:', '--header','Content-Type: '+args.content_type]
                proc=subprocess.Popen(cmd,stdin=subprocess.PIPE if kind=='duplex' else subprocess.DEVNULL,stdout=subprocess.PIPE,stderr=err)
                upload=None
                if kind=='split':
                    upload=subprocess.Popen(common+['--request','POST','--upload-file','-','--header','Expect:', '--header','Content-Type: '+args.content_type,'--url',base+'/upload/'+token],stdin=subprocess.PIPE,stdout=subprocess.DEVNULL,stderr=err)
                if kind!='down':
                    writer=proc if kind=='duplex' else upload
                    def produce():
                        try:
                            for seq in range(1,7):
                                line=json.dumps({'seq':seq,'padding':'a'*1000}).encode()+b'\n'
                                writer.stdin.write(line);writer.stdin.flush();time.sleep(1)
                        except (BrokenPipeError,OSError) as e:state['error']=str(e)
                        finally:
                            state['ended']=time.monotonic()
                            try:writer.stdin.close()
                            except OSError:pass
                    thread=threading.Thread(target=produce,daemon=True);thread.start();workers.append(thread)
                for raw in proc.stdout:
                    line=raw.decode('utf-8','replace').strip()
                    if not line.startswith('data: '):continue
                    try:event=json.loads(line[6:])
                    except ValueError:continue
                    now=time.monotonic()
                    arrivals.append((now,event))
                code=proc.wait(timeout=30)
                for thread in workers:thread.join(timeout=10)
                uploadcode=upload.wait(timeout=30) if upload else 0
                err.seek(0);error=err.read().decode('utf-8','replace')[:1500]
                header_file.seek(0);headers=header_file.read().decode('latin1')
        statuses=[line for line in headers.splitlines() if line.startswith('HTTP/')]
        records=[(t,e) for t,e in arrivals if e.get('event')==('tick' if kind=='down' else 'ack')]
        done=any(e.get('event')=='done' for _,e in arrivals)
        span=records[-1][0]-records[0][0] if len(records)>1 else 0
        early=sum(1 for t,e in records if state['ended'] is None or t<state['ended']) if kind!='down' else None
        valid=(code==0 and uploadcode==0 and done and len(records)==6)
        if kind=='down':valid=valid and span>=3 and all(e.get('nonce')==nonce for _,e in records)
        else:valid=valid and early>=2 and span>=3 and [e.get('seq') for _,e in records]==list(range(1,7))
        return {'test':kind,'result':'PASS' if valid else 'NOT_CONFIRMED','http':statuses[-1] if statuses else 'unknown',
                'records':len(records),'arrival_span_s':round(span,3),'origin_receive_span_s':round(records[-1][1].get('elapsed',0)-records[0][1].get('elapsed',0),3) if len(records)>1 else 0,'acks_before_upload_end':early,
                'elapsed_s':round(time.monotonic()-begin,3),'curl_exit':code,'upload_exit':uploadcode,
                'error':error or state['error']}

    for kind in ['down','split','duplex']:
        try:result=probe(kind)
        except Exception as e:result={'test':kind,'result':'ERROR','error':str(e)}
        results.append(result)
        print(json.dumps(result,ensure_ascii=False),flush=True)
    print(json.dumps({'base':args.base,'content_type':args.content_type,'response':args.response,
                      'note':'Only the tested route, HTTP version and short observation window are covered; not a long-term availability guarantee.',
                      'results':results},ensure_ascii=False))

if __name__=='__main__':main()