"""CME DataMine historical API/flat-file adapter. No orders, broker ledger writes or live-stream claim."""
import argparse,base64,csv,gzip,hashlib,io,json,os,re,ssl,urllib.parse,urllib.request
from pathlib import Path
from datetime import date,datetime
from infinbot_common import Rejected,strict_json,private_read,private_dir,number,timestamp

HOST='datamine.new.cmegroup.com'
LIST_URL='https://'+HOST+'/api/list_entitlements_files'
DOWNLOAD_URL='https://'+HOST+'/cme/api/v2/download'
SCHEMA_URL='https://cmegroupclientsite.atlassian.net/wiki/spaces/EPICSANDBOX/pages/457223239/Time+and+Sales'
SCHEMA='CME_TIMESALES_V2_19_FIELDS'
MAX_FILE=32*1024*1024;MAX_EXPANDED=128*1024*1024
FIELDS=('Transaction_Date','Transaction_Time','Sequence_Number','Ticker_Symbol','Instrument_Symbol',
        'Security_ID','Future_Option_Index_Indicator','Delivery_Date','Put_Call_Indicator','Strike_Price',
        'Market_Data_Report_Type','Trade_Price_Formatted','Trade_Price_Decimal','Trade_Quantity',
        'Vol_Quantity','Trade_Aggressor','Close_Open_Type','User_Defined','Exchange_Code')

def file_id(value):
    if not isinstance(value,str) or not re.fullmatch(r'\d{8}-[A-Za-z0-9_.-]{1,200}',value):raise Rejected('Invalid DataMine file ID')
    datetime.strptime(value[:8],'%Y%m%d');return value

def exact_uint(value,positive=False):
    if not isinstance(value,str) or not re.fullmatch(r'\d{1,18}',value):raise Rejected('Unsigned integer field required')
    n=int(value)
    if positive and not n:raise Rejected('Positive integer required')
    return n

def manifest_entry(payload,fid):
    file_id(fid)
    if not isinstance(payload,dict) or not isinstance(payload.get('data'),list) or not isinstance(payload.get('paging'),dict):raise Rejected('DataMine List API schema required')
    matches=[]
    for dataset in payload['data']:
        if not isinstance(dataset,dict) or not isinstance(dataset.get('files'),list):raise Rejected('Malformed List API dataset')
        for item in dataset['files']:
            if not isinstance(item,dict):raise Rejected('Malformed List API file')
            if item.get('file_id')!=fid:continue
            parsed=urllib.parse.urlsplit(item.get('api_download_link',''))
            if (parsed.scheme!='https' or parsed.hostname!=HOST or parsed.netloc!=HOST or parsed.path!='/cme/api/v2/download'
                or parsed.fragment or urllib.parse.parse_qs(parsed.query)!= {'fid':[fid]}):raise Rejected('Unexpected DataMine download route')
            if (not isinstance(item.get('file_name'),str) or not item['file_name'] or item['file_name'] in ('.','..')
                or any(c in item['file_name'] for c in '\x00\r\n/\\') or Path(item['file_name']).name!=item['file_name']):raise Rejected('Invalid DataMine filename')
            if type(item.get('size')) is not int or not 0<item['size']<=MAX_FILE:raise Rejected('Invalid file size')
            if not isinstance(dataset.get('dataset_code'),str) or not dataset['dataset_code']:raise Rejected('Explicit dataset code required')
            if not isinstance(dataset.get('period_date'),str) or dataset['period_date']!=fid[:8]:raise Rejected('File date mismatch')
            matches.append((dataset,item))
    if len(matches)!=1:raise Rejected('Exactly one explicit file ID required; no guessed entitlement')
    return matches[0]

def parse_timesales(raw,product,*,header=False):
    if not re.fullmatch(r'[A-Z0-9]{1,12}',product):raise Rejected('Explicit Globex product code required')
    if len(raw)>MAX_FILE:raise Rejected('Oversized source')
    source=gzip.GzipFile(fileobj=io.BytesIO(raw)) if raw.startswith(b'\x1f\x8b') else io.BytesIO(raw)
    def lines(stream):
        total=0
        for line in stream:
            total+=len(line.encode('utf-8'))
            if total>MAX_EXPANDED or len(line)>8192:raise Rejected('Expanded file/line exceeds bounds')
            yield line
    count=trades=volumes=indicative=spreads=unsequenced_volumes=0;prior_by_security={};first=None;last=None;contracts=set()
    months=dict(zip('FGHJKMNQUVXZ',range(1,13)))
    with source,io.TextIOWrapper(source,encoding='utf-8-sig',newline='') as text:
        rows=csv.reader(lines(text),strict=True)
        if header:
            actual=tuple(next(rows,()))
            documented=('Transaction__Date',)+FIELDS[1:]
            if actual not in (FIELDS,documented):raise Rejected('Unrecognized exact v2 header')
        for row in rows:
            if len(row)!=19 or any(len(v)>256 for v in row):raise Rejected('Exactly 19 bounded v2 fields required')
            d=dict(zip(FIELDS,row))
            if not re.fullmatch(r'\d{4}-\d{2}-\d{2}',d['Transaction_Date']):raise Rejected('ISO transaction date required')
            day=date.fromisoformat(d['Transaction_Date']);t=d['Transaction_Time']
            if not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{9}Z',t):raise Rejected('UTC nanosecond v2 timestamp required')
            datetime.fromisoformat(t.replace('Z','+00:00'))
            if t[:10]!=day.isoformat():raise Rejected('Transaction date/timestamp mismatch')
            security=exact_uint(d['Security_ID'],True);key=(day,security)
            if d['Sequence_Number']:
                seq=exact_uint(d['Sequence_Number'],True);prior=prior_by_security.get(key)
                if prior is not None and seq<=prior:raise Rejected('Duplicate or decreasing instrument sequence; never silently sort')
                prior_by_security[key]=seq
            elif d['Market_Data_Report_Type']=='VOL':unsequenced_volumes+=1
            else:raise Rejected('Trade record sequence required')
            if d['Ticker_Symbol']!=product or d['Future_Option_Index_Indicator']!='FUT' or d['Exchange_Code']!='XCME':raise Rejected('Explicit CME futures product mismatch')
            instrument=d['Instrument_Symbol']
            pattern=re.escape(product)+r'[FGHJKMNQUVXZ]\d{1,4}'
            if not re.fullmatch(pattern+r'(?:-'+pattern+r')?',instrument):raise Rejected('Unsupported outright/calendar instrument; never guess mapping')
            parts=instrument.split('-');spreads+=int(len(parts)>1)
            exact_uint(d['Security_ID'],True)
            if not re.fullmatch(r'\d{6}',d['Delivery_Date']):raise Rejected('Explicit delivery month required')
            datetime.strptime(d['Delivery_Date'],'%Y%m')
            contract=parts[0][len(product):]
            if months[contract[0]]!=int(d['Delivery_Date'][-2:]) or not d['Delivery_Date'][:4].endswith(contract[1:]):raise Rejected('Contract/month/year mismatch')
            if d['Put_Call_Indicator'] or d['Strike_Price']:raise Rejected('Option fields must be empty for futures')
            if d['Trade_Aggressor'] not in ('','B','A') or d['Close_Open_Type'] not in ('','O') or d['User_Defined'] not in ('H','U'):raise Rejected('Unsupported v2 enumeration')
            if d['Vol_Quantity']:exact_uint(d['Vol_Quantity'])
            if d['Market_Data_Report_Type']=='TRD':
                number(d['Trade_Price_Decimal'])
                if d['Trade_Quantity'] and exact_uint(d['Trade_Quantity'])>0:trades+=1
                else:indicative+=1
            elif d['Market_Data_Report_Type']=='VOL':
                if d['Trade_Price_Formatted'] or d['Trade_Price_Decimal'] or d['Trade_Quantity']:raise Rejected('VOL rows cannot be invented trades')
                volumes+=1
            else:raise Rejected('Unsupported report type')
            count+=1;contracts.add(instrument);first=t if first is None or t<first else first;last=t if last is None or t>last else last
    if not count:raise Rejected('Empty file unavailable')
    return {'rows_validated':count,'trade_records':trades,'volume_records':volumes,'non_execution_price_records':indicative,
            'calendar_spread_rows':spreads,'unsequenced_volume_records':unsequenced_volumes,'contract_count':len(contracts),
            'earliest_source_timestamp':first,'latest_source_timestamp':last,'nanosecond_text_preserved':True,
            'sequence_order_verified':True,'sequence_scope':'Per transaction date and explicit security ID; no global product ordering asserted',
            'clearing_prices_parsed_with_decimal':True,'display_factor_applied':False,
            'position_allocation_verified':False,'live_stream_verified':False}

def inspect_saved(config):
    if config.get('provider')!='CME DataMine' or config.get('schema')!=SCHEMA:raise Rejected('Explicit supported CME DataMine schema required')
    if type(config.get('header',False)) is not bool:raise Rejected('Explicit boolean header setting required')
    payload=strict_json(Path(config['list_file']).read_bytes())
    dataset,item=manifest_entry(payload,config['file_id']);path=Path(config['data_file'])
    if path.is_symlink() or not path.is_file() or path.stat().st_size!=item['size']:raise Rejected('Saved source size/path mismatch')
    if path.name!=item['file_name']:raise Rejected('Saved source filename mismatch')
    raw=path.read_bytes();result=parse_timesales(raw,config['product'],header=config.get('header',False))
    return dict(result,status='DATAMINE_SAVED_SCHEMA_VALIDATED',provider='CME DataMine',schema=SCHEMA,
                source_sha256=hashlib.sha256(raw).hexdigest(),file_id=config['file_id'],dataset_code=dataset['dataset_code'],
                provenance='Owner-selected saved file matched to saved DataMine API metadata; origin not independently authenticated offline',
                current_entitlement_verified=False,network_used=False,ledger_modified=False,existing_provider_configuration_changed=False)

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self,*args,**kwargs):raise Rejected('Redirect blocked; credentials never forwarded')

def api_read(url,credentials_file):
    p=urllib.parse.urlsplit(url)
    if p.scheme!='https' or p.netloc!=HOST or p.path not in ('/api/list_entitlements_files','/cme/api/v2/download') or p.fragment:raise Rejected('Non-DataMine route rejected')
    credentials=strict_json(private_read(credentials_file,8192))
    if set(credentials)=={'CME_DATAMINE_ACCESS_TOKEN'}:
        token=credentials['CME_DATAMINE_ACCESS_TOKEN']
        if not isinstance(token,str) or not token or any(c.isspace() for c in token):raise Rejected('Invalid bearer token')
        auth='Bearer '+token
    elif set(credentials)=={'CME_DATAMINE_API_ID','CME_DATAMINE_API_PASSWORD'}:
        user,password=credentials['CME_DATAMINE_API_ID'],credentials['CME_DATAMINE_API_PASSWORD']
        if not all(isinstance(v,str) and v and not any(c in v for c in '\r\n') for v in (user,password)) or ':' in user:raise Rejected('Invalid Basic credentials')
        auth='Basic '+base64.b64encode((user+':'+password).encode()).decode()
    else:raise Rejected('Explicit DataMine credentials required; no broker-key aliases')
    opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect(),urllib.request.HTTPSHandler(context=ssl.create_default_context()))
    with opener.open(urllib.request.Request(url,headers={'Authorization':auth,'Accept-Encoding':'identity'},method='GET'),timeout=15) as response:
        raw=response.read(MAX_FILE+1)
        if response.status!=200 or len(raw)>MAX_FILE:raise Rejected('Download unavailable or oversized')
        return raw

def download_saved(credentials_file,fid,directory):
    fid=file_id(fid);directory=private_dir(directory)
    payload=api_read(LIST_URL+'?'+urllib.parse.urlencode({'limit':1000,'period_date':fid[:8]}),credentials_file)
    metadata=strict_json(payload);dataset,item=manifest_entry(metadata,fid)
    raw=api_read(DOWNLOAD_URL+'?'+urllib.parse.urlencode({'fid':fid}),credentials_file)
    if len(raw)!=item['size']:raise Rejected('Downloaded size mismatch')
    outputs=[]
    entries=((item['file_name'],raw),('datamine-list-'+hashlib.sha256(fid.encode()).hexdigest()[:16]+'.json',payload))
    if len({n for n,_ in entries})!=len(entries) or any((directory/n).exists() or (directory/n).is_symlink() for n,_ in entries):raise Rejected('Existing downloaded files preserved; choose new destination')
    for name,content in entries:
        path=directory/name
        fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
        with os.fdopen(fd,'wb') as f:f.write(content);f.flush();os.fsync(f.fileno())
        outputs.append(str(path))
    return {'status':'DATAMINE_FILES_RETRIEVED_SCHEMA_NOT_YET_VALIDATED','data_file':outputs[0],'list_file':outputs[1],
            'observed_at':timestamp(),'source_sha256':hashlib.sha256(raw).hexdigest(),'broker_orders_submitted':0}

if __name__=='__main__':
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--config');p.add_argument('--download',action='store_true')
    p.add_argument('--credentials-file');p.add_argument('--file-id');p.add_argument('--output-dir');a=p.parse_args()
    try:
        if a.download:
            if not all((a.credentials_file,a.file_id,a.output_dir)) or a.config:p.error('Explicit credentials/file-id/output-dir required for download')
            result=download_saved(a.credentials_file,a.file_id,a.output_dir)
        else:
            if not a.config:p.error('Saved explicit DataMine config required')
            result=inspect_saved(strict_json(private_read(a.config)))
        print(json.dumps(result,indent=2))
    except Exception:print(json.dumps({'status':'DATAMINE_VALIDATION_BLOCKED','reason':'Missing, unsupported or mismatched schema/metadata/file; no fallback, no secret output','live_stream_verified':False}));raise SystemExit(1)