#!/usr/bin/env python3
"""Prepare the official OBF TSV gzip for a source-separated, idempotent import.

Data: Open Beauty Facts, ODbL-1.0 / DbCL-1.0. No images or contributor IDs.
No model-generated product facts. Outputs are deployment-independent.
"""
import argparse
import collections
import csv
import gzip
import hashlib
import json
import re
from datetime import datetime, timezone
from pathlib import Path

SOURCE_URL = 'https://static.openbeautyfacts.org/data/en.openbeautyfacts.org.products.csv.gz'
LICENSE_URL = 'https://opendatacommons.org/licenses/odbl/1-0/'
COSMETIC_TAGS = {
    'en:cosmetic-products','en:face','en:body','en:hair','en:makeup',
    'en:suncare','en:sunscreen','en:in-sun-protections','en:facial-creams',
    'en:body-creams','en:hand-creams','en:lip-balms','en:hair-care',
    'en:shampoos','en:cleansers','en:hair-conditioners','en:perfumes',
    'en:shower-gels','en:soaps','en:deodorants','en:showers-and-baths',
    'en:toothpastes','en:toothpaste','en:shaving','en:face-masks',
    'en:body-oils','en:liquid-soaps','en:nail-polish','en:lip-cosmetics',
}

def gtin(raw):
    code = raw.strip()
    if not re.fullmatch(r'(?:[0-9]{8}|[0-9]{12}|[0-9]{13}|[0-9]{14})', code):
        return None
    total = sum(int(n) * (3 if i % 2 == 0 else 1) for i, n in enumerate(code[-2::-1]))
    return code.zfill(14) if (10-total % 10) % 10 == int(code[-1]) else None

def tags(value):
    return sorted(set(item.strip() for item in value.split(',') if item.strip()))

def transform(row):
    code = row.get('code', '').strip()
    normalized = gtin(code)
    name = row.get('product_name','').strip()
    brands = row.get('brands','').strip()
    if not normalized: return None, 'invalid_gtin'
    if not name: return None, 'missing_name'
    if not brands: return None, 'missing_brand'
    fields = {key: row.get(key,'').strip() for key in ('quantity','generic_name','ingredients_text')}
    if len(name)>500 or len(brands)>500 or len(fields['ingredients_text'])>30000 or len(fields['generic_name'])>5000 or len(fields['quantity'])>200:
        return None, 'oversized_field'
    categories=tags(row.get('categories_tags',''))
    countries=tags(row.get('countries_tags',''))
    issues=[]
    if not COSMETIC_TAGS.intersection(categories): issues.append('cosmetic_category_unconfirmed')
    if int(normalized)<1000000: issues.append('low_value_code_needs_review')
    if row.get('data_quality_errors_tags'): issues.append('upstream_data_quality_errors')
    if re.search(r'\b(airpods|iphone|laptop|chocolate|pizza|pasta|protein powder)\b',name,re.I): issues.append('possible_non_cosmetic')
    modified=row.get('last_modified_datetime','').strip() or None
    if modified:
        try: datetime.fromisoformat(modified.replace('Z','+00:00'))
        except ValueError: modified=None;issues.append('invalid_source_date')
    result={'source_code':code,'gtin14':normalized,'product_name':name,'brands':brands,
            **fields,'categories':categories,'countries':countries,
            'source_modified_at':modified,'quality_status':'review' if issues else 'published',
            'quality_issues':issues,'source_url':f'https://world.openbeautyfacts.org/product/{code}'}
    result['record_hash']=hashlib.sha256(json.dumps(result,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).hexdigest()
    return result,None

def main():
    p=argparse.ArgumentParser(description=__doc__)
    p.add_argument('archive',type=Path);p.add_argument('output',type=Path)
    args=p.parse_args();args.output.mkdir(parents=True,exist_ok=True)
    sha=hashlib.sha256(args.archive.read_bytes()).hexdigest()
    counts=collections.Counter();records={};rejected=[]
    with gzip.open(args.archive,'rt',encoding='utf-8-sig',newline='') as stream:
        reader=csv.DictReader(stream,delimiter='\t')
        required={'code','product_name','brands','categories_tags','ingredients_text'}
        if not required.issubset(reader.fieldnames or []): raise ValueError('Unexpected upstream schema')
        for row in reader:
            counts['source_rows']+=1
            item,reason=transform(row)
            if reason:
                counts[reason]+=1;rejected.append({'source_code':row.get('code',''),'reason':reason});continue
            old=records.get(item['source_code'])
            if old:
                counts['duplicate_source_code']+=1
                if (old['source_modified_at'] or '')>(item['source_modified_at'] or ''):continue
            records[item['source_code']]=item
    values=sorted(records.values(),key=lambda r:r['source_code'])
    for r in values:
        counts[r['quality_status']]+=1
        if r['ingredients_text']:counts[r['quality_status']+'_with_ingredients']+=1
        for country in ['en:hong-kong','en:japan','en:south-korea']:
            if country in r['countries']:counts[country]+=1
    # Separate compact read chunks from transactional import batch size.
    for i in range(0,len(values),40):
        (args.output/f'chunk-{i//40:05d}.json').write_text(json.dumps(values[i:i+40],ensure_ascii=False,separators=(',',':')))
    with gzip.open(args.output/'catalog-derivative.jsonl.gz','wt',encoding='utf-8') as f:
        for row in values:f.write(json.dumps(row,ensure_ascii=False,separators=(',',':'))+'\n')
    (args.output/'rejected.json').write_text(json.dumps(rejected,separators=(',',':')))
    manifest={'source':'open-beauty-facts','source_url':SOURCE_URL,'archive_sha256':sha,
              'prepared_at':datetime.now(timezone.utc).isoformat(),'license':'ODbL-1.0',
              'license_url':LICENSE_URL,'contents_license':'DbCL-1.0','counts':dict(counts),
              'prepared_rows':len(values),'chunks':(len(values)+39)//40,
              'instructions_present':False,'language_present':False,'image_data_imported':False,
              'transform_version':'1.0.0'}
    (args.output/'manifest.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2))
    print(json.dumps(manifest,ensure_ascii=False))

if __name__=='__main__':main()