"""RetailFlow local reference: stdlib SQLite, synthetic data, no cloud/API calls.
Run: python3 pipeline.py --output result.json
This validates the teaching logic, not the planned Snowflake/dbt implementation.
"""
import argparse
import csv
import json
import sqlite3
from pathlib import Path

ROOT = Path(__file__).resolve().parent


def execute():
    db = sqlite3.connect(':memory:')
    db.row_factory = sqlite3.Row
    db.executescript((ROOT / 'models.sql').read_text())

    def load_orders(filename):
        for row in csv.DictReader((ROOT / filename).open()):
            db.execute('''INSERT INTO orders VALUES (:order_id,:status,:updated_at)
                ON CONFLICT(order_id) DO UPDATE SET status=excluded.status,
                updated_at=excluded.updated_at WHERE excluded.updated_at > orders.updated_at''', row)

    def load_lines(filename):
        for row in csv.DictReader((ROOT / filename).open()):
            db.execute('''INSERT INTO lines VALUES (:order_id,:line_id,:product_id,:quantity,:unit_price)
                ON CONFLICT(order_id,line_id) DO NOTHING''', row)

    def validate():
        missing = db.execute('''SELECT l.order_id,l.line_id,l.product_id FROM lines l
            LEFT JOIN products p USING(product_id) WHERE p.product_id IS NULL''').fetchall()
        if missing:
            raise ValueError('Missing product reference: ' + json.dumps([dict(r) for r in missing]))
        invalid = db.execute('SELECT COUNT(*) FROM lines WHERE quantity<=0 OR unit_price<0').fetchone()[0]
        if invalid:
            raise ValueError('Invalid quantity or unit price')

    def snapshot():
        validate()
        row = dict(db.execute('SELECT * FROM daily_sales').fetchone())
        row['order_count'] = int(row['order_count'])
        return row

    db.executemany('INSERT INTO products VALUES (?,?)', [('P100','Accessories'),('P101','Stationery'),('P102','Home')])
    load_orders('orders-batch-1.csv')
    load_lines('lines-batch-1.csv')
    first = snapshot()
    assert first['revenue'] == 150 and first['order_count'] == 2 and first['units'] == 4
    load_orders('orders-batch-2.csv')
    load_lines('lines-batch-2.csv')
    second = snapshot()
    assert second['revenue'] == 130 and second['aov'] == 65
    load_orders('orders-batch-2.csv')
    load_lines('lines-batch-2.csv')
    repeated = snapshot()
    assert second == repeated
    load_orders('orders-late.csv')
    assert snapshot() == second, 'Older paid status must not overwrite latest cancellation'
    db.execute('SAVEPOINT invalid_batch')
    db.execute("INSERT INTO lines VALUES ('O102','L2','P999',1,99)")
    try:
        snapshot()
    except ValueError as error:
        failure = str(error)
    else:
        raise AssertionError('Missing product should block result publication')
    db.execute('ROLLBACK TO invalid_batch')
    restored = snapshot()
    assert restored == second
    return {
        'evidence': 'Executed local SQLite reference; synthetic data; no Snowflake/dbt or scheduler validation',
        'business_date': '2026-10-01',
        'currency': 'AUD',
        'stages': [
            {'id':'first','label':'第一批：完成去重','status':'passed','result':first,
             'explanation':'重复的 O100-L1 只保留一次，三条有效明细对应两个订单。'},
            {'id':'second','label':'第二批：更新与取消','status':'passed','result':second,
             'explanation':'O101 最新状态变为 cancelled，移出 A$50；O102 新增 A$30。'},
            {'id':'repeat','label':'同批重跑','status':'passed','result':repeated,
             'explanation':'第二批再次执行，业务结果与上次完全一致；更旧的 paid 状态也不会覆盖取消。'},
            {'id':'failure','label':'异常批次：阻止发布','status':'blocked','result':restored,
             'explanation':'P999 不存在，检查失败，异常批次回滚；这里显示上一次有效结果，不是本次新结果。','error':failure},
        ],
        'checks': [
            {'name':'Duplicate line removed','status':'passed'},
            {'name':'Cancellation applied','status':'passed'},
            {'name':'Repeated batch unchanged','status':'passed'},
            {'name':'Older status ignored','status':'passed'},
            {'name':'Missing product blocks publication','status':'passed'},
            {'name':'Rollback retains last valid result','status':'passed'},
        ],
        'limitations': ['One business date; fixed AUD; no tax/refund accounting',
                         'Line changes with the same key are outside this reference implementation',
                         'Equal update timestamps require a defined tie-breaker in the course implementation',
                         'No cloud, AI execution, scheduling, or production deployment demonstrated'],
    }


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--output', type=Path, default=ROOT / 'result.json')
    args = parser.parse_args()
    args.output.write_text(json.dumps(execute(), ensure_ascii=False, indent=2) + '\n')
    print('Local reference executed: 6 checks passed; wrote', args.output)