"""Reproduce metadata-only audit offline. Python3.11 standard library, no network."""
import collections, csv, datetime, hashlib, html, json, pathlib, re, tarfile, xml.etree.ElementTree as ET

B = pathlib.Path(__file__).resolve().parent
NS = {'pg':'http://www.gutenberg.org/2009/pgterms/', 'dc':'http://purl.org/dc/terms/', 'rdf':'http://www.w3.org/1999/02/22-rdf-syntax-ns#'}
ABOUT = '{'+NS['rdf']+'}about'
FLAGS = ['html','epub','plain_text','utf8_plain_text','pdf','all_three']
def vals(e,path):
    return sorted({(x.text or '').strip() for x in e.findall(path,NS) if (x.text or '').strip()})
def writecsv(name,rows,fields=None):
    rows=list(rows)
    with (B/name).open('w',newline='',encoding='utf-8') as f:
        w=csv.DictWriter(f,fieldnames=fields or list(rows[0]));w.writeheader();w.writerows(rows)
def pct(a,b): return round(100*a/b,4) if b else None
def aggregate(rows,key,label):
    groups=collections.defaultdict(list)
    for r in rows:
        keys=r[key].split('|') if r[key] else ['[missing]']
        for k in keys:groups[k].append(r)
    out=[]
    for k,rr in groups.items():
        n=len(rr);o={label:k,'text_entries':n}
        for flag in FLAGS:o[flag+'_entries']=sum(r[flag] for r in rr);o[flag+'_pct']=pct(o[flag+'_entries'],n)
        o['missing_any_three_entries']=n-o['all_three_entries'];o['missing_any_three_pct']=pct(o['missing_any_three_entries'],n)
        out.append(o)
    return sorted(out,key=lambda x:(-x['text_entries'],x[label]))

manifest=json.loads((B/'source-manifest.json').read_text())
source=B/pathlib.Path(manifest['local_file'].replace(chr(92), '/'))
assert hashlib.file_digest(source.open('rb'),'sha256').hexdigest()==manifest['sha256']
rows=[]; errors=[]; archive_files=0; mimecounts=collections.Counter(); duplicate_ids=[]; seen=set();sample=[]
with tarfile.open(source,'r|bz2') as tar:
    for m in tar:
        if not m.isfile():continue
        archive_files+=1
        if not m.name.endswith('.rdf'):errors.append({'member':m.name,'error':'not RDF'});continue
        raw=tar.extractfile(m).read()
        try:root=ET.fromstring(raw)
        except ET.ParseError as ex:errors.append({'member':m.name,'error':str(ex)});continue
        ebooks=root.findall('pg:ebook',NS)
        if len(ebooks)!=1:errors.append({'member':m.name,'error':f'{len(ebooks)} ebook elements'});continue
        e=ebooks[0];ident=e.get(ABOUT,'').rsplit('/',1)[-1]
        if ident in seen:duplicate_ids.append({'id':ident,'member':m.name});continue
        seen.add(ident)
        types=vals(e,'dc:type/rdf:Description/rdf:value');languages=vals(e,'dc:language/rdf:Description/rdf:value')
        files=e.findall('dc:hasFormat/pg:file',NS);mimes=set();file_urls=[];missing_mime=0;duplicate_file_urls=0;urls_seen=set()
        for file in files:
            url=file.get(ABOUT,'');file_urls.append(url)
            if url in urls_seen:duplicate_file_urls+=1
            urls_seen.add(url)
            mm=vals(file,'dc:format/rdf:Description/rdf:value')
            if not mm:missing_mime+=1
            mimes.update(x.lower() for x in mm)
        normalized={v.split(';',1)[0].strip() for v in mimes}
        flags={'html':bool(normalized&{'text/html','application/xhtml+xml'}),'epub':'application/epub+zip' in normalized,'plain_text':'text/plain' in normalized,
            'utf8_plain_text':any(v.split(';',1)[0].strip()=='text/plain' and 'charset=utf-8' in re.sub(r'[\s\"\x27]','',v) for v in mimes),'pdf':'application/pdf' in normalized}
        flags['all_three']=all(flags[x] for x in ['html','epub','plain_text'])
        issued=e.findtext('dc:issued',default='',namespaces=NS)
        period='[missing]' if not issued else ('before2004' if issued<'2004-01-01' else '2004-2021' if issued<'2022-01-01' else '2022-2025' if issued<'2026-01-01' else '2026_or_later')
        creators=vals(e,'dc:creator/pg:agent/pg:name')
        row={'id':ident,'source_member':m.name,'source_member_sha256':hashlib.sha256(raw).hexdigest(),'title':e.findtext('dc:title',default='',namespaces=NS),'creators':'|'.join(creators),'types':'|'.join(types),'languages':'|'.join(languages),'language_count':len(languages),'issued':issued,'release_period':period,'included_text':types==['Text'],'listed_files':len(files),'unique_file_urls':len(urls_seen),'duplicate_file_urls':duplicate_file_urls,'files_missing_mime':missing_mime,'mime_types':'|'.join(sorted(mimes)), **flags}
        rows.append(row)
        for mime in normalized:mimecounts[mime]+=1
        if len(sample)<12 and types==['Text'] and not flags['all_three']:
            sample.append({'id':ident,'member':m.name,'types':types,'languages':languages,'mime_types':sorted(mimes),'listed_file_urls':file_urls})
        if len(rows)%20000==0:print('parsed',len(rows),flush=True)
assert not errors, errors[:10]
text=[r for r in rows if r['included_text']]
lang=aggregate(text,'languages','language');period=aggregate(text,'release_period','release_period')
types=collections.Counter(r['types'] or '[missing]' for r in rows)
formats=[]
for f in FLAGS:
    count=sum(r[f] for r in text);formats.append({'format':f,'text_entries_with_format':count,'denominator_text_entries':len(text),'pct':pct(count,len(text)),'not_listed_entries':len(text)-count})
proxy=collections.defaultdict(list)
for r in text:
    key=(re.sub(r'\s+',' ',r['title'].casefold()).strip(),r['creators'].casefold(),r['languages'])
    if key[0]:proxy[key].append(r['id'])
dupes=[{'normalized_title':k[0],'creators':k[1],'languages':k[2],'catalogue_ids':'|'.join(v),'entry_count':len(v)} for k,v in proxy.items() if len(v)>1]
dupes.sort(key=lambda r:-r['entry_count'])
summary={'analyzed_at_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'source_sha256':manifest['sha256'],'source_last_modified':manifest['response_headers'].get('last-modified'),
 'archive_regular_files':archive_files,'distinct_ids':len(rows),'duplicate_id_records':duplicate_ids,'parse_errors':errors,'type_combinations':dict(types),'text_entries':len(text),'excluded_entries':len(rows)-len(text),'language_tags':len([x for x in lang if x['language']!='[missing]']),
 'text_entries_missing_language':sum(not r['languages'] for r in text),'multilingual_text_entries':sum(r['language_count']>1 for r in text),'language_memberships':sum(r['language_count'] for r in text),
 'text_entries_with_duplicate_file_urls':sum(r['duplicate_file_urls']>0 for r in text),'text_entries_with_files_missing_mime':sum(r['files_missing_mime']>0 for r in text),
 'exact_title_creator_language_proxy_groups':len(dupes),'entries_in_proxy_groups':sum(r['entry_count'] for r in dupes),'proxy_caveat':'Candidate matching only; neither confirmed duplicate works nor safe work-level deduplication. All IDs retained.',
 'formats':formats,'language_table':lang,'release_period_table':period,'all_entry_mime_presence':dict(mimecounts),
 'validations':{'all_regular_files_parsed':archive_files==len(rows)+len(duplicate_ids),'all_ids_unique_after_logged_dedup':len(seen)==len(rows),'type_counts_sum_to_all':sum(types.values())==len(rows),'language_membership_sum':sum(x['text_entries'] for x in lang)==sum(r['language_count'] or 1 for r in text)}}
writecsv('catalogue-record-audit.csv',rows)
writecsv('language-format-coverage.csv',lang)
writecsv('release-period-format-coverage.csv',period)
writecsv('format-summary.csv',formats)
writecsv('type-audit.csv',({'type_combination':k,'entries':v,'included':k=='Text'} for k,v in types.most_common()))
writecsv('possible-title-creator-language-matches.csv',dupes,['normalized_title','creators','languages','catalogue_ids','entry_count'])
(B/'analysis.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2),encoding='utf-8')
(B/'anomaly-samples.json').write_text(json.dumps(sample,ensure_ascii=False,indent=2),encoding='utf-8')
print(json.dumps({k:v for k,v in summary.items() if k not in ['language_table','all_entry_mime_presence']},ensure_ascii=False,indent=2))