import random,duckdb,datetime as dt
random.seed(7)
L=['2026-01-05','2026-03-02','2026-05-04']   # three weekly-ish batch loads
cities=['Zürich','Bern','Basel','Luzern','St. Gallen','Winterthur','Aarau','Chur','Thun','Zug','Biel','Olten']
names=['Alpenblick Gastro AG','Bergquell Hotel GmbH','Café Lindenhof','Seeblick Bistro','Rheinufer Kantine AG','Gipfelstube Sàrl','Talwind Catering','Sonnenhang Pflege AG','Ahornweg Bäckerei','Felsenkeller Bar','Uferweg Restaurant','Moosalp Hütte']
cust=[dict(customer_no=f'C-{1001+i}',name=n,city=cities[i],segment=random.choice(['Gastro','Hotel','Care']),credit_limit=random.choice([5000,10000,20000])) for i,n in enumerate(names)]
prods=[('P-100','Espresso beans 1 kg','Coffee',24.5),('P-101','Decaf beans 1 kg','Coffee',26.0),('P-110','Oat milk 1 l','Dairy alt',2.9),('P-111','Whole milk 1 l','Dairy',1.8),('P-120','Croissant frozen x50','Bakery',39.0),('P-121','Bread rolls frozen x40','Bakery',22.0),('P-130','Paper cups 1000','Supplies',58.0),('P-131','Napkins 2000','Supplies',31.0),('P-140','Mineral water 24x0.5 l','Drinks',14.4),('P-141','Apple juice 6x1 l','Drinks',13.2),('P-150','Dish soap 5 l','Cleaning',18.9),('P-151','Hand soap 5 l','Cleaning',21.5)]
prod=[dict(sku=s,product_name=n,category=c,list_price=p) for s,n,c,p in prods]
sc,sp,so,sl=[],[],[],[]
orders=[];on=5000
for li,ld in enumerate(L):
    if li==1:
        cust[2]['city']='Kloten'; cust[5]['segment']='Hotel'; cust[0]['credit_limit']=30000
        prod[0]['list_price']=26.9; prod[4]['list_price']=41.0
    if li==2:
        cust[2]['credit_limit']=15000; cust[8]['city']='Wil'; prod[0]['list_price']=27.5; prod[9]['product_name']='Apple juice organic 6x1 l'
        cust.append(dict(customer_no='C-1013',name='Nebelhorn Lodge AG',city='Davos',segment='Hotel',credit_limit=10000))
    for c in cust: sc.append(dict(c,load_date=ld,record_source='bexio.contacts'))
    for p in prod: sp.append(dict(p,load_date=ld,record_source='erp.articles'))
    d0=dt.date.fromisoformat(ld)-dt.timedelta(days=56)
    for o in orders:  # status progression of older orders
        if o['status']!='invoiced': o['status']={'open':'shipped','shipped':'invoiced'}[o['status']]; so.append(dict(o,load_date=ld,record_source='shop.orders'))
    for k in range(14):
        on+=1;c=random.choice(cust);od=d0+dt.timedelta(days=random.randint(0,55))
        o=dict(order_no=f'O-{on}',customer_no=c['customer_no'],order_date=str(od),status=random.choice(['open','shipped']),channel=random.choice(['webshop','phone','rep']))
        orders.append(o);so.append(dict(o,load_date=ld,record_source='shop.orders'))
        for p in random.sample(prod,random.randint(1,4)):
            sl.append(dict(order_no=o['order_no'],sku=p['sku'],quantity=random.randint(1,20),unit_price=p['list_price'],load_date=ld,record_source='shop.orders'))
# a messy key on purpose: whitespace/lowercase must hash to the same hub key
sl[3]['sku']=' '+sl[3]['sku'].lower()+' '
import pandas as pd
con=duckdb.connect()
for n,r in [('src_customer',sc),('src_product',sp),('src_order',so),('src_order_line',sl)]:
    df=pd.DataFrame(r);con.register('df',df);con.execute(f"CREATE TABLE {n} AS SELECT * EXCLUDE(load_date), CAST(load_date AS TIMESTAMP) load_date FROM df");con.unregister('df')
con.execute(open('vault.sql').read())
T=['stg_customer','stg_product','stg_order','stg_order_line','hub_customer','hub_product','hub_order','link_order_customer','link_order_product','sat_customer_details','sat_product_details','sat_order_details','sat_order_line_details','pit_customer','bridge_order_line','dim_customer','dim_product','fact_order_line']
for t in T:
    con.execute(f"COPY {t} TO 'site/data/{t}.parquet' (FORMAT parquet)")
    print(t,con.execute(f"select count(*) from {t}").fetchone()[0])