#!/usr/bin/env python3
"""Download every shard of one declared snapshot; resume verified files, verify every line.
Public source records only. Python standard library. Never executes downloaded contents.
"""
from pathlib import Path
from urllib.parse import urlsplit,urljoin
import argparse,json,hashlib,gzip,urllib.request,time,os,re

def canonical(x):return json.dumps(x,ensure_ascii=False,sort_keys=True,separators=(',',':'),allow_nan=False).encode()
def sha(b):return hashlib.sha256(b).hexdigest()
def fsha(p):
    h=hashlib.sha256()
    with p.open('rb') as f:
        for b in iter(lambda:f.read(1024*1024),b''):h.update(b)
    return h.hexdigest()
def fetch(url,maxbytes,output=None):
    class NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self,*a,**k):raise RuntimeError('REDIRECT_REFUSED')
    opener=urllib.request.build_opener(NoRedirect())
    with opener.open(url,timeout=60) as r:
        count=0;parts=[]
        while True:
            b=r.read(1024*1024)
            if not b:break
            count+=len(b)
            if count>maxbytes:raise RuntimeError('RESPONSE_SIZE_LIMIT')
            if output:output.write(b)
            else:parts.append(b)
        return b''.join(parts)
def download(url,root,allow_local=False):
    u=urlsplit(url)
    if u.scheme!='https' and not (allow_local and u.scheme=='http' and u.hostname in ('127.0.0.1','localhost')):raise ValueError('HTTPS_REQUIRED')
    root=Path(root).resolve();root.mkdir(parents=True,exist_ok=True);(root/'shards').mkdir(exist_ok=True)
    raw=fetch(url,4*1024*1024);m=json.loads(raw);assert m['schema']=='hsimc.reality.corpus-manifest.v1'
    if (root/'manifest.json').exists() and (root/'manifest.json').read_bytes()!=raw:raise RuntimeError('TARGET_HAS_DIFFERENT_MANIFEST')
    seen=set();prior_id='';total=0;seq=0
    for s in m['shards']:
        name=s['name'];assert re.fullmatch(r'part-\d{5}\.jsonl\.gz',name) and name not in seen;seen.add(name)
        v=urlsplit(s['url']);assert v.scheme=='https' and v.netloc=='hsimc.tech' and '/shards/'+name==v.path[v.path.rfind('/shards/'):]
        # Manifest shard paths are relative to this manifest, preserving test/portable origin.
        link=urljoin(url,'shards/'+name);target=root/'shards'/name
        if target.is_symlink():raise ValueError('SYMLINK_REFUSED')
        if not target.exists() or fsha(target)!=s['sha256']:
            tmp=target.with_suffix('.part')
            if tmp.is_symlink():raise ValueError('SYMLINK_REFUSED')
            with tmp.open('wb') as f:fetch(link,s['bytes'],f);f.flush();os.fsync(f.fileno())
            if tmp.stat().st_size!=s['bytes'] or fsha(tmp)!=s['sha256']:raise RuntimeError('SHARD_DIGEST_FAILED')
            os.replace(tmp,target)
        h=hashlib.sha256();n=0;size=0;first=None;last=None
        with gzip.open(target,'rb') as f:
            while True:
                line=f.readline(256*1024+1)
                if not line:break
                if len(line)>256*1024 or not line.endswith(b'\n'):raise ValueError('LINE_SIZE_OR_FRAMING')
                h.update(line);size+=len(line);n+=1;e=json.loads(line);r=e['source_record'];rid=r['runtime_id']
                assert rid>prior_id and e['source_registry_sha256']==m['source_registry_sha256'] and sha(canonical(r))==e['source_record_sha256']
                if first is None:first=rid
                last=rid;prior_id=rid
        assert (n,size,h.hexdigest(),first,last)==(s['count'],s['raw_bytes'],s['raw_sha256'],s['first_id'],s['last_id'])
        total+=n;seq+=1;print(f'VERIFIED_SHARDS={seq}/{len(m["shards"])} RECORDS={total}',flush=True)
    assert total==m['record_count'] and len(seen)==m['shard_count']
    (root/'manifest.json').write_bytes(raw)
    result={'state':'PASS','snapshot_id':m['snapshot_id'],'records':total,'manifest_download_sha256':sha(raw),'shards':seq,'source_truth_verification':'NOT_IMPLIED_BY_CHECKSUM'}
    (root/'DOWNLOAD-RECEIPT.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');return result
if __name__=='__main__':
    ap=argparse.ArgumentParser();ap.add_argument('manifest_url');ap.add_argument('output');a=ap.parse_args();print(json.dumps(download(a.manifest_url,a.output),ensure_ascii=False,indent=2))