"""Local supplier-catalog review application. Python 3.12+, standard library only."""
import argparse
import csv
import hashlib
import io
import json
import re
import secrets
import sqlite3
import time
import uuid
from contextlib import contextmanager
from decimal import Decimal
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

ROOT = Path(__file__).resolve().parent
MAX_BYTES, MAX_ROWS = 16 * 1024 * 1024, 100_000
ALIASES = {
    'part_number': ('part_number', 'part', 'sku', 'item'),
    'description': ('description', 'name'),
    'price': ('price', 'unit_price', 'unit price'),
    'currency': ('currency',),
    'unit': ('unit', 'uom'),
}
UNITS = {'ea': 'EA', 'each': 'EA', 'box': 'BOX', 'kg': 'KG', 'm': 'M'}
ACCOUNTS = {'importer': ('import-demo', 'importer'), 'reviewer': ('review-demo', 'reviewer'), 'viewer': ('view-demo', 'viewer')}

class DomainError(Exception):
    def __init__(self, message, status=422):
        self.status = status
        super().__init__(message)

class Catalog:
    def __init__(self, path):
        self.path = str(path)
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as db:
            version = db.execute('PRAGMA user_version').fetchone()[0]
            if version not in (0, 1):
                raise RuntimeError('Unsupported schema version')
            db.executescript((ROOT / 'schema.sql').read_text())
    def connect(self):
        db = sqlite3.connect(self.path, timeout=15, isolation_level=None)
        db.row_factory = sqlite3.Row
        db.execute('PRAGMA foreign_keys=ON')
        db.execute('PRAGMA journal_mode=WAL')
        db.execute('PRAGMA synchronous=FULL')
        return db
    @contextmanager
    def transaction(self):
        db = self.connect()
        try:
            db.execute('BEGIN IMMEDIATE')
            yield db
            db.commit()
        except BaseException:
            db.rollback()
            raise
        finally:
            db.close()
    def seed(self):
        with self.transaction() as db:
            db.execute('INSERT OR IGNORE INTO suppliers VALUES(?,0)', ('ACME',))
            if db.execute('SELECT COUNT(*) FROM items').fetchone()[0] == 0:
                for part, desc, cents in [('BRG-10','Bearing, 10 mm',1250),('BELT-20','Drive belt',3200),('SEAL-5','Shaft seal',475)]:
                    db.execute('INSERT INTO items VALUES(?,?,?,?,?,?)', ('ACME',part,desc,cents,'USD','EA'))
    @staticmethod
    def parse(source, locale):
        if locale not in ('en-US', 'de-DE'):
            raise DomainError('Price format must be en-US or de-DE')
        try:
            valid = isinstance(source, str) and len(source.encode('utf-8')) <= MAX_BYTES
        except UnicodeEncodeError:
            valid = False
        if not valid:
            raise DomainError('CSV must be UTF-8 text, at most 16 MiB')
        mapping = 'catalog-v1-' + locale
        try:
            reader = csv.reader(io.StringIO(source.lstrip('\ufeff'), newline=''), strict=True)
            headers = next(reader)
            headers = [x.strip().lower() for x in headers]
            if len(headers) != len(set(headers)):
                raise DomainError('Duplicate header names')
            indexes = {}
            for key, aliases in ALIASES.items():
                found = [i for i,h in enumerate(headers) if h in aliases]
                if len(found) != 1:
                    raise DomainError('Exactly one mapped column required: ' + key)
                indexes[key] = found[0]
            rows, errors, seen = [], [], {}
            for number, values in enumerate(reader, 2):
                if not any(v.strip() for v in values):
                    continue
                if len(rows) + len(errors) >= MAX_ROWS:
                    raise DomainError('At most 100,000 input rows are supported')
                error = None
                item = None
                try:
                    if len(values) != len(headers):
                        raise ValueError('Column count differs from header')
                    item = {key: values[index].strip() for key,index in indexes.items()}
                    part = item['part_number'].upper()
                    if not re.fullmatch(r'[A-Z0-9][A-Z0-9._/-]{0,63}', part):
                        raise ValueError('Part number must be 1-64 letters, digits, dots, slashes, underscores or hyphens')
                    if part in seen:
                        errors.append({'row': seen[part], 'part_number': part, 'error': 'Duplicate normalized part number'})
                        raise ValueError('Duplicate normalized part number')
                    seen[part] = number
                    if not item['description'] or len(item['description']) > 300:
                        raise ValueError('Description must contain 1-300 characters')
                    currency = item['currency'].upper()
                    if currency not in ('USD','EUR','GBP'):
                        raise ValueError('Currency must be USD, EUR or GBP')
                    unit = UNITS.get(item['unit'].lower())
                    if not unit:
                        raise ValueError('Unit must be EA/each, BOX, KG or M')
                    price = item['price']
                    pattern = r'(?:\d+|\d{1,3}(?:,\d{3})+)(?:\.\d{1,2})?' if locale == 'en-US' else r'(?:\d+|\d{1,3}(?:\.\d{3})+)(?:,\d{1,2})?'
                    if not re.fullmatch(pattern, price):
                        raise ValueError('Price does not match '+locale+' decimal format (maximum two decimal places)')
                    canonical = price.replace(',','') if locale=='en-US' else price.replace('.','').replace(',','.')
                    cents = int(Decimal(canonical) * 100)
                    if cents > 1_000_000_000:
                        raise ValueError('Price exceeds 10,000,000 currency units')
                    rows.append({'row':number,'part_number':part,'description':item['description'],'price_cents':cents,'currency':currency,'unit':unit})
                except ValueError as exc:
                    error = str(exc)
                if error:
                    errors.append({'row':number,'part_number':(item or {}).get('part_number',''),'error':error})
            if not rows and not errors:
                errors.append({'row':1,'part_number':'','error':'The file contains no items'})
            invalid_lines = {x['row'] for x in errors}
            rows = [r for r in rows if r['row'] not in invalid_lines]
            return rows, errors, mapping
        except (csv.Error, StopIteration, DomainError) as exc:
            return [], [{'row':1,'part_number':'','error':str(exc) or 'CSV header missing'}], mapping
    @staticmethod
    def _diff(db, supplier, rows):
        existing = {r['part_number']:dict(r) for r in db.execute('SELECT * FROM items WHERE supplier=?',(supplier,))}
        changes = []
        for row in rows:
            part = row['part_number']
            new = {k:v for k,v in row.items() if k != 'row'}
            previous = existing.pop(part, None)
            if previous:
                previous.pop('supplier')
            action = 'added' if previous is None else 'unchanged' if previous == new else 'changed'
            changes.append({'row':row['row'],'part_number':part,'action':action,'before':previous,'after':new})
        for part, previous in sorted(existing.items()):
            previous.pop('supplier')
            changes.append({'row':None,'part_number':part,'action':'missing','before':previous,'after':None})
        return changes
    @staticmethod
    def _digest(changes, version):
        return hashlib.sha256(json.dumps([version,changes],sort_keys=True,separators=(',',':')).encode()).hexdigest()
    @staticmethod
    def _audit(db, batch, action, actor, detail):
        db.execute('INSERT INTO audit(batch_id,action,actor,created,detail) VALUES(?,?,?,?,?)',(batch,action,actor,time.time(),json.dumps(detail,sort_keys=True)))
    def stage(self, supplier, source, locale='en-US', actor='importer'):
        if not isinstance(supplier,str) or not re.fullmatch(r'[A-Z0-9_-]{1,32}',supplier):
            raise DomainError('Supplier must be 1-32 uppercase letters, digits, underscores or hyphens')
        rows, errors, mapping = self.parse(source,locale)
        digest = hashlib.sha256(source.encode()).hexdigest()
        with self.transaction() as db:
            old=db.execute('SELECT id FROM batches WHERE supplier=? AND digest=? AND mapping=?',(supplier,digest,mapping)).fetchone()
            if old:
                return {'id':old['id'],'reused':True}
            db.execute('INSERT OR IGNORE INTO suppliers VALUES(?,0)',(supplier,))
            version=db.execute('SELECT version FROM suppliers WHERE id=?',(supplier,)).fetchone()[0]
            changes=self._diff(db,supplier,rows)
            batch=str(uuid.uuid4())
            db.execute('INSERT INTO batches(id,supplier,digest,mapping,source,base_version,status,review_digest,rows_json,changes_json,errors_json,created) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
                (batch,supplier,digest,mapping,source,version,'blocked' if errors else 'staged',self._digest(changes,version),json.dumps(rows),json.dumps(changes),json.dumps(errors),time.time()))
            self._audit(db,batch,'staged',actor,{'base_version':version,'errors':len(errors),'source_sha256':digest})
            return {'id':batch,'reused':False}
    def refresh(self, batch, actor='importer'):
        with self.transaction() as db:
            row=db.execute('SELECT * FROM batches WHERE id=?',(batch,)).fetchone()
            if not row: raise DomainError('Batch not found',404)
            if row['status'] != 'staged': raise DomainError('Only staged batches can be refreshed',409)
            version=db.execute('SELECT version FROM suppliers WHERE id=?',(row['supplier'],)).fetchone()[0]
            changes=self._diff(db,row['supplier'],json.loads(row['rows_json']))
            db.execute('UPDATE batches SET base_version=?,changes_json=?,review_digest=? WHERE id=?',(version,json.dumps(changes),self._digest(changes,version),batch))
            self._audit(db,batch,'review_refreshed',actor,{'base_version':version})
        return self.batch(batch)
    def approve(self,batch,expected_version,review_digest,actor='reviewer',fail_after=None):
        # The injected fault hook is for transaction tests. No HTTP route exposes it.
        with self.transaction() as db:
            row=db.execute('SELECT * FROM batches WHERE id=?',(batch,)).fetchone()
            if not row: raise DomainError('Batch not found',404)
            if row['status']=='applied': return {'id':batch,'version':row['applied_version'],'replayed':True}
            if row['status']!='staged': raise DomainError('Correct the rejected rows and stage a new file before approval',409)
            version=db.execute('SELECT version FROM suppliers WHERE id=?',(row['supplier'],)).fetchone()[0]
            if type(expected_version) is not int or expected_version!=row['base_version'] or expected_version!=version or review_digest!=row['review_digest']:
                raise DomainError('Review is stale. Refresh and review the new diff before approving.',409)
            written=0
            for change in json.loads(row['changes_json']):
                if change['action'] not in ('added','changed'): continue
                r=change['after']
                db.execute('INSERT INTO items VALUES(?,?,?,?,?,?) ON CONFLICT(supplier,part_number) DO UPDATE SET description=excluded.description,price_cents=excluded.price_cents,currency=excluded.currency,unit=excluded.unit',
                    (row['supplier'],r['part_number'],r['description'],r['price_cents'],r['currency'],r['unit']))
                written+=1
                if fail_after is not None and written>=fail_after: raise RuntimeError('Injected interruption before commit')
            db.execute('UPDATE suppliers SET version=version+1 WHERE id=?',(row['supplier'],))
            db.execute('UPDATE batches SET status=?,applied_version=?,reviewer=?,applied_at=? WHERE id=?',('applied',version+1,actor,time.time(),batch))
            self._audit(db,batch,'applied',actor,{'base_version':version,'version':version+1,'written_rows':written})
            return {'id':batch,'version':version+1,'replayed':False,'written_rows':written}
    def batch(self,batch,page=1,size=100):
        with self.connect() as db:
            r=db.execute('SELECT * FROM batches WHERE id=?',(batch,)).fetchone()
            if not r: raise DomainError('Batch not found',404)
            r=dict(r);changes=json.loads(r.pop('changes_json'));errors=json.loads(r.pop('errors_json'))
            r.pop('source');r.pop('rows_json')
            r['summary']={a:sum(x['action']==a for x in changes) for a in ['added','changed','unchanged','missing']}
            r['summary']['rejected']=len(errors)
            r['changes']=changes if size is None else changes[(page-1)*size:page*size];r['errors']=errors if size is None else errors[(page-1)*size:page*size]
            r['page']=page;r['page_size']=size;r['total']=max(len(changes),len(errors))
            r['current_version']=db.execute('SELECT version FROM suppliers WHERE id=?',(r['supplier'],)).fetchone()[0]
            r['audit']=[{**dict(x),'detail':json.loads(x['detail'])} for x in db.execute('SELECT * FROM audit WHERE batch_id=? ORDER BY id',(batch,))]
            return r
    def report(self,batch):
        return self.batch(batch,size=None)
    def source(self,batch):
        with self.connect() as db:
            r=db.execute('SELECT source FROM batches WHERE id=?',(batch,)).fetchone()
            if not r: raise DomainError('Batch not found',404)
            return r[0]
    def summary(self):
        with self.connect() as db:
            return {'suppliers':[dict(r) for r in db.execute('SELECT * FROM suppliers ORDER BY id')],
                'batches':[dict(r) for r in db.execute('SELECT id,supplier,base_version,status,created,applied_version FROM batches ORDER BY created DESC LIMIT 200')]}
    def items(self,supplier,page=1,size=100):
        with self.connect() as db:
            return {'items':[dict(r) for r in db.execute('SELECT * FROM items WHERE supplier=? ORDER BY part_number LIMIT ? OFFSET ?',(supplier,size,(page-1)*size))],
                'total':db.execute('SELECT COUNT(*) FROM items WHERE supplier=?',(supplier,)).fetchone()[0]}

class Application(ThreadingHTTPServer):
    daemon_threads=True
    def __init__(self,address,catalog):
        self.catalog=catalog;self.sessions={}
        super().__init__(address,Handler)

class Handler(BaseHTTPRequestHandler):
    def log_message(self,*args): pass
    def setup(self):
        super().setup();self.connection.settimeout(15)
    def send(self,status,data,kind='application/json',extra=None):
        body=json.dumps(data).encode() if kind=='application/json' else data.encode() if isinstance(data,str) else data
        self.send_response(status)
        for k,v in {'Content-Type':kind,'Content-Length':str(len(body)),'Cache-Control':'no-store','X-Content-Type-Options':'nosniff','Referrer-Policy':'no-referrer',
            'Content-Security-Policy':"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'",**(extra or {})}.items(): self.send_header(k,v)
        self.end_headers();self.wfile.write(body)
    def session(self):
        cookies=dict(p.strip().split('=',1) for p in self.headers.get('Cookie','').split(';') if '=' in p)
        session=self.server.sessions.get(cookies.get('session'))
        if not session or session['expires']<time.time(): raise DomainError('Sign in to a local demo account',401)
        return session
    def origin(self):
        host=self.headers.get('Host','')
        if host not in [f'127.0.0.1:{self.server.server_port}',f'localhost:{self.server.server_port}']:
            raise DomainError('Localhost access only',403)
        origin=self.headers.get('Origin')
        if origin and origin!='http://'+host: raise DomainError('Cross-origin request rejected',403)
    def body(self):
        if self.headers.get_content_type()!='application/json': raise DomainError('Expected application/json',415)
        try: size=int(self.headers.get('Content-Length','0'))
        except ValueError: raise DomainError('Invalid content length')
        if not 0<size<=MAX_BYTES+4096: raise DomainError('Request size exceeds limit',413)
        try:
            b=json.loads(self.rfile.read(size))
            if not isinstance(b,dict): raise ValueError()
            return b
        except (ValueError,UnicodeDecodeError): raise DomainError('Invalid JSON object')
    def do_GET(self):
        try:
            self.origin();u=urlparse(self.path);path=u.path;q=parse_qs(u.query)
            assets={'/':('index.html','text/html; charset=utf-8'),'/app.js':('app.js','text/javascript'),'/style.css':('style.css','text/css')}
            if path in assets:
                name,kind=assets[path];return self.send(200,(ROOT/'static'/name).read_bytes(),kind)
            session=self.session()
            if path=='/api/session': return self.send(200,{k:session[k] for k in ['username','role','csrf']})
            page=max(1,min(100001,int(q.get('page',['1'])[0])));size=max(1,min(200,int(q.get('size',['100'])[0])))
            if path=='/api/summary': return self.send(200,self.server.catalog.summary())
            if path=='/api/catalog': return self.send(200,self.server.catalog.items(q.get('supplier',['ACME'])[0],page,size))
            m=re.fullmatch(r'/api/batches/([a-f0-9-]+)(/report|/source)?',path)
            if m:
                bid,suffix=m.groups()
                if suffix=='/report': return self.send(200,self.server.catalog.report(bid),extra={'Content-Disposition':'attachment; filename="import-report.json"'})
                if suffix=='/source': return self.send(200,self.server.catalog.source(bid),'text/csv; charset=utf-8',{'Content-Disposition':'attachment; filename="original-source.csv"'})
                return self.send(200,self.server.catalog.batch(bid,page,size))
            raise DomainError('Not found',404)
        except DomainError as exc: self.send(exc.status,{'error':str(exc)})
        except ValueError: self.send(422,{'error':'Invalid query parameter'})
    def do_POST(self):
        try:
            self.origin();path=urlparse(self.path).path;body=self.body()
            if path=='/api/login':
                username=body.get('username','');account=ACCOUNTS.get(username) if isinstance(username,str) else None
                if not account or not secrets.compare_digest(str(body.get('password','')),account[0]): raise DomainError('Invalid demo credentials',401)
                token=secrets.token_urlsafe(32);s={'username':username,'role':account[1],'csrf':secrets.token_urlsafe(24),'expires':time.time()+3600}
                self.server.sessions[token]=s
                return self.send(200,{k:s[k] for k in ['username','role','csrf']},extra={'Set-Cookie':'session='+token+'; HttpOnly; SameSite=Strict; Path=/; Max-Age=3600'})
            session=self.session()
            if not secrets.compare_digest(self.headers.get('X-CSRF-Token',''),session['csrf']): raise DomainError('Invalid CSRF token',403)
            if path=='/api/logout':
                session['expires']=0;return self.send(200,{'ok':True},extra={'Set-Cookie':'session=; HttpOnly; SameSite=Strict; Path=/; Max-Age=0'})
            if path=='/api/stage':
                if session['role'] not in ('importer','reviewer'): raise DomainError('Import permission required',403)
                return self.send(201,self.server.catalog.stage(body.get('supplier'),body.get('csv'),body.get('locale','en-US'),session['username']))
            m=re.fullmatch(r'/api/batches/([a-f0-9-]+)/(approve|refresh)',path)
            if m:
                bid,action=m.groups()
                if action=='approve':
                    if session['role']!='reviewer': raise DomainError('Reviewer permission required',403)
                    result=self.server.catalog.approve(bid,body.get('expected_version'),body.get('review_digest'),session['username'])
                else:
                    if session['role'] not in ('importer','reviewer'): raise DomainError('Import permission required',403)
                    result=self.server.catalog.refresh(bid,session['username'])
                return self.send(200,result)
            raise DomainError('Not found',404)
        except DomainError as exc: self.send(exc.status,{'error':str(exc)})
        except sqlite3.Error: self.send(503,{'error':'Storage is unavailable; retry using the same batch identity'})

if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('--db',default='data/catalog.db');p.add_argument('--port',type=int,default=8011);p.add_argument('--seed',action='store_true');args=p.parse_args()
    catalog=Catalog(args.db)
    if args.seed: catalog.seed()
    server=Application(('127.0.0.1',args.port),catalog)
    print(f'Catalog review: http://127.0.0.1:{server.server_port}',flush=True)
    try: server.serve_forever()
    except KeyboardInterrupt: pass
    finally: server.server_close()