#!/usr/bin/env python3
"""Python 3, standard library only. Credentials stay in environment, never checkpoints.
DATATAP_URL=https://your-datatap-host DATATAP_INGEST_KEY=... python ingest.py DATASET FILE --format ndjson
Run again with the same file to resume; delete FILE.datatap-upload.json for a new run.
Use DATASET --no-changes after a successful source check that found no changes.
"""
import argparse,json,os,time,uuid,urllib.request
from pathlib import Path
p=argparse.ArgumentParser();p.add_argument('dataset');p.add_argument('file',nargs='?');p.add_argument('--format',choices=['csv','ndjson','json','parquet'],default='ndjson');p.add_argument('--no-changes',action='store_true');a=p.parse_args()
origin=os.environ['DATATAP_URL'].rstrip('/');key=os.environ['DATATAP_INGEST_KEY']
if not origin.startswith('https://') and origin not in ['http://localhost:3017','http://127.0.0.1:3017']:raise ValueError('Use HTTPS')
def api(path,body=None):
 r=urllib.request.Request(origin+path,data=json.dumps(body).encode() if body is not None else None,headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'})
 with urllib.request.urlopen(r,timeout=60) as response:return json.load(response)
if a.no_changes:
 print(api('/api/v1/ingest/'+a.dataset+'/check',{'key':str(uuid.uuid4()),'result':'no_changes'}));raise SystemExit
if not a.file:p.error('FILE is required unless --no-changes')
file=Path(a.file);stat=file.stat();checkpoint=Path(str(file)+'.datatap-upload.json');fingerprint=[str(file.resolve()),stat.st_size,stat.st_mtime_ns,a.dataset,origin,a.format]
state=json.loads(checkpoint.read_text()) if checkpoint.exists() else {'fingerprint':fingerprint,'key':str(uuid.uuid4()),'parts':[]}
if state['fingerprint']!=fingerprint:raise ValueError('File or target changed; use a new checkpoint')
def save():
 tmp=Path(str(checkpoint)+'.tmp');tmp.write_text(json.dumps(state));tmp.replace(checkpoint)
save();path='/api/v1/seller/datasets/'+a.dataset+'/uploads';upload=api(path,{'action':'start','bytes':stat.st_size,'format':a.format,'key':state['key']})
if upload.get('status','uploading')=='uploading':
 with file.open('rb') as source:
  total=(stat.st_size+upload['part_bytes']-1)//upload['part_bytes']
  for part in range(1,total+1):
   if any(v['part']==part for v in state['parts']):continue
   source.seek((part-1)*upload['part_bytes']);chunk=source.read(upload['part_bytes']);url=api(path,{'action':'part','upload':upload['id'],'part':part})['url']
   # Signed storage requests must not include the ingestion credential.
   with urllib.request.urlopen(urllib.request.Request(url,data=chunk,method='PUT'),timeout=180) as r:etag=r.headers.get('ETag')
   if not etag:raise ValueError('Storage did not return an ETag')
   state['parts'].append({'part':part,'etag':etag});save();print('Uploaded part',part,'of',total)
after=file.stat()
if after.st_size!=stat.st_size or after.st_mtime_ns!=stat.st_mtime_ns:raise ValueError('File changed during upload; use a new checkpoint')
job=api(path,{'action':'complete','upload':upload['id'],'parts':sorted(state['parts'],key=lambda v:v['part'])});state['job_id']=job['job_id'];save()
while True:
 result=api('/api/v1/ingest/'+a.dataset+'/jobs/'+job['job_id'])
 if result['status'] in ['succeeded','failed','cancelled']:print(json.dumps(result));raise SystemExit(0 if result['status']=='succeeded' else 1)
 time.sleep(3)