"""Local staging/checkpoint demonstration. Trusted tenant fixture; no HTTP server."""
import csv,hashlib,io,json,sqlite3,tempfile
from pathlib import Path
SCHEMA='''
CREATE TABLE IF NOT EXISTS imports(tenant TEXT,id TEXT,hash TEXT,checkpoint INTEGER DEFAULT 0,PRIMARY KEY(tenant,id));
CREATE TABLE IF NOT EXISTS staged(tenant TEXT,id TEXT,row_no INTEGER,label TEXT,quantity INTEGER,error TEXT,PRIMARY KEY(tenant,id,row_no));
CREATE TABLE IF NOT EXISTS results(tenant TEXT,id TEXT,row_no INTEGER,label TEXT,quantity INTEGER,PRIMARY KEY(tenant,id,row_no));
'''
def connect(path):
    db=sqlite3.connect(path);db.executescript(SCHEMA);return db

def stage(db,tenant,batch,raw):
    digest=hashlib.sha256(raw).hexdigest()
    reader=csv.DictReader(io.StringIO(raw.decode('utf-8'),newline=''),strict=True)
    if reader.fieldnames!=['label','quantity']:raise ValueError('unexpected header')
    parsed=[]
    for n,row in enumerate(reader,1):
        error=None;qty=None;label=row.get('label') or ''
        try:
            if None in row or any(v is None for v in row.values()):raise ValueError('field count')
            qty=int(row['quantity'])
            if not label.strip() or qty<=0:raise ValueError('value')
        except ValueError:error='invalid label, quantity or field count'
        parsed.append((tenant,batch,n,label,qty,error))
    with db:
        db.execute('BEGIN IMMEDIATE')
        old=db.execute('SELECT hash FROM imports WHERE tenant=? AND id=?',(tenant,batch)).fetchone()
        if old:
            if old[0]!=digest:raise ValueError('import id reused with different bytes')
            return digest
        db.execute('INSERT INTO imports(tenant,id,hash) VALUES(?,?,?)',(tenant,batch,digest))
        db.executemany('INSERT INTO staged VALUES(?,?,?,?,?,?)',parsed)
    return digest

def apply_chunk(db,tenant,batch,limit=2,fail_after_write=False):
    if limit<1:raise ValueError('positive chunk size required')
    with db:
        db.execute('BEGIN IMMEDIATE')
        item=db.execute('SELECT checkpoint FROM imports WHERE tenant=? AND id=?',(tenant,batch)).fetchone()
        if item is None:raise KeyError('unknown import')
        rows=db.execute('SELECT row_no,label,quantity,error FROM staged WHERE tenant=? AND id=? AND row_no>? ORDER BY row_no LIMIT ?',(tenant,batch,item[0],limit)).fetchall()
        for n,label,qty,error in rows:
            if error is None:
                db.execute('INSERT INTO results VALUES(?,?,?,?,?)',(tenant,batch,n,label,qty))
                if fail_after_write:raise RuntimeError('injected before checkpoint commit')
        if rows:db.execute('UPDATE imports SET checkpoint=? WHERE tenant=? AND id=?',(rows[-1][0],tenant,batch))
    return len(rows)

def snapshot(db):
    return dict(checkpoint=db.execute('SELECT checkpoint FROM imports').fetchone()[0],applied=db.execute('SELECT count(*) FROM results').fetchone()[0],rejected=db.execute('SELECT count(*) FROM staged WHERE error IS NOT NULL').fetchone()[0])

def main():
    raw=b'label,quantity\nAlpha,2\n"Two\nlines",3\nBad,zero\nDelta,4\n'
    history=[]
    with tempfile.TemporaryDirectory() as temp:
        path=Path(temp)/'import.db';db=connect(path);stage(db,'tenant-a','import-1',raw)
        assert db.execute('SELECT count(*) FROM staged').fetchone()[0]==4
        apply_chunk(db,'tenant-a','import-1');history.append(dict(step='first chunk',**snapshot(db)))
        try:apply_chunk(db,'tenant-a','import-1',fail_after_write=True)
        except RuntimeError:pass
        else:raise AssertionError('failure was not injected')
        history.append(dict(step='rolled back second chunk',**snapshot(db)));db.close();db=connect(path)
        stage(db,'tenant-a','import-1',raw);apply_chunk(db,'tenant-a','import-1');history.append(dict(step='reopen and resume',**snapshot(db)))
        assert apply_chunk(db,'tenant-a','import-1')==0
        history.append(dict(step='repeat finished import',**snapshot(db)))
        try:stage(db,'tenant-a','import-1',raw.replace(b'Delta',b'Changed'))
        except ValueError:changed=True
        else:changed=False
        try:stage(db,'tenant-a','broken',b'label,quantity\n"unterminated,2')
        except csv.Error:malformed=True
        else:malformed=False
        assert db.execute('SELECT count(*) FROM imports').fetchone()[0]==1
        try:apply_chunk(db,'tenant-b','import-1')
        except KeyError:isolated=True
        else:isolated=False
        assert [r['checkpoint'] for r in history]==[2,2,4,4]
        assert [r['applied'] for r in history]==[2,2,3,3]
        assert changed and malformed and isolated
        out=dict(scope='local SQLite transactions and orderly reopen, not power-loss or multi-worker testing',records=4,history=history,changed_bytes_rejected=changed,malformed_file_rejected=malformed,other_tenant_import_unavailable=isolated)
        db.close()
    target=Path(__file__).parent/'import-results.json';target.write_text(json.dumps(out,indent=2)+'\n');print(json.dumps(out,indent=2))
if __name__=='__main__':main()