"""Stage 1: master tables (channel, account, outlet, product, employee, route, vehicle, competitor product)."""
import numpy as np, pandas as pd, re, unicodedata
from difflib import SequenceMatcher
from sklearn.neighbors import BallTree
from common import *

r = rng(1)
S = real_sales(); V2D = van_division()
S['division_van'] = S.slsman_cd.map(V2D)
sell = S[(~S.is_foc) & (S.net_amt > 0)]

# ---------------- M01 channel
ch = pd.DataFrame([
 ('GR', 'Grocery', 'van sales (direct store delivery by van)'), ('MT', 'Modern trade', 'delivery to store or account DC'),
 ('WS', 'Wholesale', 'delivery to wholesaler warehouse'), ('DS', 'Discounters', 'delivery to chain DC'),
 ('EC', 'E-commerce', 'delivery to platform fulfilment centre')], columns=['channel_id', 'channel_name', 'delivery_model'])
ch['default_payment_terms_days'] = ch.channel_id.map(P['channel_terms_days'])
ch['planned_share_of_sales'] = ch.channel_id.map(P['channel_share'])
ch['otif_target_pct'] = ch.channel_id.map(P['channel_otif_target'])
ch['basis'] = np.where(ch.channel_id == 'GR', 'Actual', 'Simulated')
save(ch, 'dim_channel')

# ---------------- M04 product (active SKUs = sold in the window)
prod = pd.read_parquet(B + 'dim_product.parquet')
sk = sell.groupby('prd_cd').agg(value=('net_amt', 'sum'), ea=('qty_base_ea', 'sum')).reset_index()
divv = sell.groupby(['prd_cd', 'division_van']).net_amt.sum().reset_index().sort_values('net_amt').drop_duplicates('prd_cd', keep='last')
sk = sk.merge(divv[['prd_cd', 'division_van']], on='prd_cd').rename(columns={'prd_cd': 'sku', 'division_van': 'division'})
px = sell.groupby('prd_cd').unit_net_price_per_ea.quantile(0.75).rename('list_price_sar')
sk = sk.merge(px, left_on='sku', right_index=True)
pm = prod[['prd_cd', 'prd_desc_en', 'l1_desc', 'l2_desc', 'l4_desc', 'l5_desc', 'packing', 'ean_no']].rename(columns={'prd_cd': 'sku'})
sk = sk.merge(pm, on='sku', how='left')
harvest = sk.l2_desc.fillna('').str.upper().str.contains('HARVEST')
sk.loc[harvest, 'division'] = 'Drinks'                     # Binzagr rule: Harvest brand belongs to Drinks
cs = pd.read_parquet(B + 'dim_uom_conversion.parquet')
case = cs[cs.conv_factor > 1].groupby('prd_cd').conv_factor.max().rename('case_size')
sk = sk.merge(case, left_on='sku', right_index=True, how='left'); sk['case_size'] = sk.case_size.fillna(12).clip(2, 200).astype(int)
sk = sk[sk.list_price_sar > 0]
sk['cluster_code'] = sk.division.map(P['division_cluster'])
sk['cost_price_sar'] = (sk.list_price_sar * (1 - sk.division.map(P['division_margin']) * r.uniform(0.8, 1.2, len(sk)))).round(3)
sk['shelf_life_days'] = (sk.division.map(P['division_shelf_life_days']) * r.uniform(0.7, 1.2, len(sk))).round().astype(int)
sk['temperature_class'] = sk.division.map(P['division_temp'])
sk = sk.sort_values('value', ascending=False)
sk['cum_share_div'] = sk.groupby('division').value.cumsum() / sk.groupby('division').value.transform('sum')
sk['must_sell_GR'] = sk.cum_share_div <= 0.60
sk['must_sell_MT'] = sk.cum_share_div <= 0.75
sk['allowed_channels'] = 'GR|MT|WS|DS|EC'
first = sell.groupby('prd_cd').doc_date.min()
sk['launch_date'] = sk.sku.map(first).dt.date
sk['channel_variant_of'] = None
sk['basis'] = 'Inferred'
# discounter-only packs: 30 top SKUs get a DS variant (task 11)
top = sk.head(30).copy()
top['channel_variant_of'] = top.sku; top['sku'] = 'DSV-' + top.sku.astype(str)
top['allowed_channels'] = 'DS'; top['must_sell_GR'] = False; top['must_sell_MT'] = False
top['prd_desc_en'] = top.prd_desc_en.fillna('') + ' (discounter pack)'; top['basis'] = 'Simulated'
top['list_price_sar'] = (top.list_price_sar * 0.93).round(3)
sk = pd.concat([sk, top], ignore_index=True)
sk = sk.rename(columns={'prd_desc_en': 'sku_name', 'l1_desc': 'principal', 'l2_desc': 'brand', 'l4_desc': 'category', 'l5_desc': 'sub_category',
                        'value': 'real_gr_value_sar', 'ea': 'real_gr_units'})
sk = sk.drop(columns=['cum_share_div'])
save(sk, 'dim_product_ext')

# ---------------- M05 employees
sm = pd.read_parquet(B + 'dim_salesman.parquet')
reps = pd.DataFrame({'employee_id': sm.slsman_cd.astype(str), 'role': 'Van rep', 'division': sm.slsman_cd.map(V2D),
                     'status': np.where(sm.is_active, 'active', 'left'), 'basis': 'Actual'})
divs = list(P['division_margin'])
sup = pd.DataFrame({'employee_id': [f'SUP{i+1:02d}' for i in range(12)], 'role': 'Sales supervisor', 'division': [divs[i % 6] for i in range(12)]})
merch = pd.DataFrame({'employee_id': [f'MER{i+1:03d}' for i in range(40)], 'role': 'Merchandiser', 'division': 'All'})
kam = pd.DataFrame({'employee_id': [f'KAM{i+1:02d}' for i in range(8)], 'role': 'Key account manager', 'division': 'All'})
oth = pd.DataFrame({'employee_id': ['CC01', 'CC02', 'CC03', 'DP01', 'DP02', 'MGR01', 'MGR02', 'MGR03'],
                    'role': ['Credit controller'] * 3 + ['Demand planner'] * 2 + ['Channel manager GR', 'Channel manager MT/DS', 'Channel manager WS/EC'],
                    'division': 'All'})
emp = pd.concat([reps, sup, merch, kam, oth], ignore_index=True)
emp['basis'] = emp.basis.fillna('Simulated'); emp['status'] = emp.status.fillna('active')
# supervisors manage reps of their division; merch/KAM report to MGR02; reps' manager = a supervisor of the division
sup_by_div = sup.groupby('division').employee_id.apply(list).to_dict()
emp['manager_id'] = None
m = emp.role == 'Van rep'
emp.loc[m, 'manager_id'] = [sup_by_div.get(d, ['SUP01'])[i % 2] for i, d in enumerate(emp.loc[m, 'division'])]
emp.loc[emp.role == 'Sales supervisor', 'manager_id'] = 'MGR01'
emp.loc[emp.role.isin(['Merchandiser', 'Key account manager']), 'manager_id'] = 'MGR02'
emp.loc[emp.role.isin(['Credit controller', 'Demand planner']), 'manager_id'] = 'MGR03'
# territories: reps from their outlets' k-means zone later; merchandisers by city
emp['territory'] = None
emp.loc[emp.role == 'Merchandiser', 'territory'] = ['Riyadh'] * 20 + ['Jeddah'] * 12 + ['Dammam'] * 8
# 3 merchandiser vacancies for task 27
emp.loc[emp.employee_id.isin(['MER007', 'MER025', 'MER038']), 'status'] = 'vacant'
save(emp, 'dim_employee')

# ---------------- grocery outlets (real) + twin merge
cust = pd.read_parquet(B + 'dim_customer.parquet'); cust['outlet_code'] = cust.dist_cd.astype(str) + '-' + cust.cust_cd.astype(str)
buy = sell.groupby('outlet_id').agg(value=('net_amt', 'sum'), first=('doc_date', 'min'), last=('doc_date', 'max'),
                                    main_van=('slsman_cd', lambda x: x.value_counts().index[0])).reset_index()
g = cust[cust.outlet_code.isin(buy.outlet_id)].merge(buy, left_on='outlet_code', right_on='outlet_id')


def norm(t):
    t = unicodedata.normalize('NFKC', str(t or '')).lower()
    t = re.sub(r'[ً-ْـ]', '', t).replace('أ', 'ا').replace('إ', 'ا').replace('آ', 'ا').replace('ة', 'ه').replace('ى', 'ي')
    return ' '.join(re.sub(r'[^0-9a-z؀-ۿ ]', ' ', t).split())
g['n1'] = g.cust_name.map(norm); g['n2'] = g.cust_name2.map(norm)
base = g[(g.dist_cd != '8000') & g.latitude.notna()].reset_index(drop=True)
tree = BallTree(np.radians(base[['latitude', 'longitude']].astype(float).values), metric='haversine')
twin = {}
for _, row in g[(g.dist_cd == '8000') & g.latitude.notna()].iterrows():
    idx = tree.query_radius(np.radians([[float(row.latitude), float(row.longitude)]]), r=50 / 6371000)[0]
    best, bs = None, 0
    for i in idx:
        q = base.iloc[i]
        sc = max(SequenceMatcher(None, row.n1, q.n1).ratio() if row.n1 and q.n1 else 0, SequenceMatcher(None, row.n2, q.n2).ratio() if row.n2 and q.n2 else 0)
        if sc > bs: bs, best = sc, q.outlet_code
    if best is not None and bs >= 0.8: twin[row.outlet_code] = best
g['main_outlet_id'] = g.outlet_code.map(twin).fillna(g.outlet_code)
src = g.groupby('main_outlet_id').outlet_code.apply(lambda x: '|'.join(sorted(x)))
go = g[g.outlet_code == g.main_outlet_id].copy()
go['source_codes'] = go.outlet_code.map(src)
go['spend_year'] = go.value * 365 / ((END - START).days + 1)
go['size_band'] = pd.qcut(go.spend_year.rank(method='first'), 5, labels=['XS', 'S', 'M', 'L', 'XL']).astype(str)
go['outlet_type'] = np.select([go.size_band.isin(['XL']), go.size_band.isin(['L', 'M'])], ['Mini-market', 'Grocery (baqala)'], 'Small baqala')
days_since = (END - go['last']).dt.days
go['status'] = np.where(days_since > 90, 'lapsed', np.where(go['first'] >= START + pd.Timedelta(days=90), 'new', 'active'))
# closure confirmations for ~35% of lapsed outlets (simulated, task 13)
lap = go.status == 'lapsed'
close = lap & (r.random(len(go)) < 0.35)
go['close_date'] = pd.NaT
go.loc[close, 'close_date'] = go.loc[close, 'last'] + pd.to_timedelta(r.integers(20, 80, close.sum()), unit='D')
go.loc[close, 'status'] = 'closed'
gro = pd.DataFrame({'outlet_id': go.outlet_code, 'source_codes': go.source_codes, 'account_id': 'GR-' + go.outlet_code, 'channel_id': 'GR',
                    'outlet_type': go.outlet_type, 'size_band': go.size_band, 'city': 'Riyadh', 'latitude': go.latitude.astype(float),
                    'longitude': go.longitude.astype(float), 'main_van': go.main_van.astype(str), 'merchandiser_id': None,
                    'open_date': go['first'].dt.date, 'close_date': go.close_date.dt.date, 'status': go.status,
                    'basis': np.where(close, 'Simulated', 'Actual')})
print('  twin codes merged:', len(twin), '| grocery outlets', len(gro))

# ---------------- M02 accounts + store outlets
acc, stores = [], []
riy = gro[['latitude', 'longitude']].dropna().values


def place(city, n):
    if city == 'Riyadh':
        pts = riy[r.integers(0, len(riy), n)] + r.normal(0, 0.004, (n, 2))
    else:
        c = CITY_XY[city]; pts = np.column_stack([c[0] + r.normal(0, 0.06, n), c[1] + r.normal(0, 0.06, n)])
    return pts
cities = list(P['city_mix_stores']); cw = list(P['city_mix_stores'].values())
size_cycle = ['XL', 'L', 'L', 'M', 'M', 'M', 'S', 'S']
for i, n in enumerate(P['mt_accounts_stores']):
    a = f'KA{i+1:02d}'
    acc.append((a, f'Key account {i+1:02d}', 'MT', None, 'A' if i < 2 else ('B' if i < 5 else 'C'), f'KAM{i+1:02d}', 60 + 15 * (i % 3),
                [0.98, 0.98, 0.95, 0.95, 0.95, 0.95, 0.93, 0.93][i], [0.03, 0.03, 0.02, 0.02, 0.02, 0.015, 0.01, 0.01][i], int(600 - 50 * i)))
    cty = r.choice(cities, n, p=cw)
    for j in range(n):
        lat, lon = place(cty[j], 1)[0]
        stores.append((f'ST{len(stores)+1:04d}', a, 'MT', ['Hypermarket', 'Supermarket', 'Supermarket'][j % 3], size_cycle[(i + j) % 8], cty[j], lat, lon))
for i, n in enumerate(P['ds_chains_stores']):
    a = f'DS{i+1:02d}'
    acc.append((a, f'Discounter {i+1:02d}', 'DS', None, 'B', f'KAM0{(i % 8) + 1}', 45, 0.95, 0.02, 220))
    cty = r.choice(cities, n, p=cw)
    for j in range(n):
        lat, lon = place(cty[j], 1)[0]
        stores.append((f'ST{len(stores)+1:04d}', a, 'DS', 'Discount store', ['M', 'S'][j % 2], cty[j], lat, lon))
for i in range(P['ws_count']):
    a = f'WS{i+1:03d}'; cty = r.choice(cities, p=[0.55, 0.3, 0.15])
    acc.append((a, f'Wholesaler {i+1:03d}', 'WS', None, 'A' if i < 8 else ('B' if i < 25 else 'C'), None, 30, 0.90, 0.0, 0))
    lat, lon = place(cty, 1)[0]
    stores.append((f'WH{i+1:03d}', a, 'WS', 'Wholesale warehouse', 'L' if i < 8 else 'M', cty, lat, lon))
for i in range(P['ec_platforms']):
    a = f'EC{i+1:02d}'
    acc.append((a, f'Platform {"ABCD"[i]}', 'EC', None, 'A' if i < 2 else 'B', 'KAM08', 30, 0.97, 0.02, 0))
    for cty in ['Riyadh', 'Jeddah']:
        lat, lon = place(cty, 1)[0]
        stores.append((f'FC{len(stores)+1:04d}', a, 'EC', 'Fulfilment centre', 'L', cty, lat, lon))
A = pd.DataFrame(acc, columns=['account_id', 'account_name', 'channel_id', 'parent_account_id', 'account_tier', 'kam_employee_id', 'payment_terms_days',
                               'otif_target_pct', 'otif_penalty_pct', 'listed_sku_count'])
St = pd.DataFrame(stores, columns=['outlet_id', 'account_id', 'channel_id', 'outlet_type', 'size_band', 'city', 'latitude', 'longitude'])
St['source_codes'] = St.outlet_id; St['main_van'] = None
St['open_date'] = START.date(); St['close_date'] = None; St['status'] = 'active'; St['basis'] = 'Simulated'
# a few new and closed stores in the window
ix = r.choice(len(St), 12, replace=False)
St.loc[ix[:7], 'open_date'] = [(START + pd.Timedelta(days=int(d))).date() for d in r.integers(30, 200, 7)]; St.loc[ix[:7], 'status'] = 'new'
St.loc[ix[7:], 'close_date'] = [(START + pd.Timedelta(days=int(d))).date() for d in r.integers(60, 220, 5)]; St.loc[ix[7:], 'status'] = 'closed'
# merchandisers: MT and DS stores by city
mer = emp[(emp.role == 'Merchandiser')]
for cty in cities:
    pool = mer[mer.territory == cty].employee_id.tolist()
    m = St.channel_id.isin(['MT', 'DS']) & (St.city == cty)
    St.loc[m, 'merchandiser_id'] = [pool[i % len(pool)] for i in range(m.sum())]
# grocery accounts: one per outlet
GA = pd.DataFrame({'account_id': gro.account_id, 'account_name': 'Grocery outlet ' + gro.outlet_id, 'channel_id': 'GR', 'parent_account_id': None,
                   'account_tier': gro.size_band.map({'XL': 'A', 'L': 'B', 'M': 'B', 'S': 'C', 'XS': 'C'}), 'kam_employee_id': None})
cterms = cust.set_index('outlet_code').invterm_cd
pt = pd.read_parquet(B + 'dim_payment_terms.parquet').set_index('invterm_cd').net_days
GA['payment_terms_days'] = gro.outlet_id.map(cterms).map(pt).fillna(0).astype(int).values
GA['otif_target_pct'] = None; GA['otif_penalty_pct'] = None; GA['listed_sku_count'] = None
A['basis'] = 'Simulated'; GA['basis'] = 'Actual'
acc_all = pd.concat([A, GA], ignore_index=True)
# credit limit: 2× average monthly purchase (real for grocery from sales; stores set later from sell-in) — placeholder fill here
mon = sell.groupby('outlet_id').net_amt.sum() / 7.7
acc_all['credit_limit_sar'] = acc_all.account_id.str.replace('GR-', '', regex=False).map(mon * 2).round(-2)
acc_all['city'] = acc_all.account_id.map(St.groupby('account_id').city.agg(lambda x: x.mode()[0])).fillna('Riyadh')
acc_all['status'] = 'active'
save(acc_all, 'dim_account')
outlets = pd.concat([gro, St], ignore_index=True)
save(outlets, 'dim_outlet')

# ---------------- M06 route + M07 vehicle
s90 = sell[sell.doc_date > END - pd.Timedelta(days=90)]
rows = []
for van, d in sell.groupby('slsman_cd'):
    div = V2D.get(van, 'Food')
    days = d.drop_duplicates(['doc_date']).doc_date.dt.day_name()
    share = days.value_counts(normalize=True)
    planned = '|'.join([x for x in ['Saturday', 'Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday'] if share.get(x, 0) >= 0.10])
    t1, t2, t3 = P['breakevens'][div]
    rows.append((f'R-{van}', str(van), str(van), div, P['division_cluster'][div], planned, s90[s90.slsman_cd == van].outlet_id.nunique(),
                 d.doc_date.max().date(), t1, t2, t3))
R = pd.DataFrame(rows, columns=['route_id', 'van_id', 'rep_employee_id', 'division', 'cluster_code', 'planned_days', 'planned_outlets_90d',
                                'last_sale_date', 't1_sar_month', 't2_sar_month', 't3_sar_month'])
R['status'] = np.where(pd.to_datetime(R.last_sale_date) >= END - pd.Timedelta(days=20), 'active', 'inactive')
R['basis'] = 'Inferred'
save(R, 'dim_route')
cap = {'ambient': 650, 'chilled': 380, 'frozen': 320}
Vh = pd.DataFrame({'van_id': R.van_id, 'division': R.division, 'temperature_class': R.division.map(P['division_temp'])})
Vh['capacity_cases'] = Vh.temperature_class.map(cap)
Vh['cost_per_day_sar'] = (R.t1_sar_month * P['van_cost_margin_for_T1'] / 26).round(0)
Vh['status'] = R.status; Vh['basis'] = 'Simulated'
save(Vh, 'dim_vehicle')
# rep territory = route's zone (division + main area)
emp = emp.merge(R[['rep_employee_id', 'route_id']], left_on='employee_id', right_on='rep_employee_id', how='left').drop(columns='rep_employee_id')
emp.loc[emp.role == 'Van rep', 'territory'] = emp.loc[emp.role == 'Van rep', 'route_id']
save(emp.drop(columns='route_id'), 'dim_employee')

# ---------------- M08 competitor products (2 per top-300 SKU)
t300 = sk[sk.channel_variant_of.isna()].head(300)
cp = t300.loc[t300.index.repeat(2), ['sku', 'category', 'division', 'list_price_sar']].reset_index(drop=True)
cp['competitor_sku_id'] = [f'CP{i+1:04d}' for i in range(len(cp))]
cp['competitor_brand'] = [f'Competitor {chr(65 + (i % 12))}' for i in range(len(cp))]
cp['base_price_gap_pct'] = np.clip(r.normal(-0.03, 0.08, len(cp)), -0.25, 0.15).round(3)
cp['competitor_list_price_sar'] = (cp.list_price_sar * (1 + cp.base_price_gap_pct)).round(3)
cp['launch_date'] = None
new = r.choice(len(cp), 24, replace=False)
cp.loc[new, 'launch_date'] = [(START + pd.Timedelta(days=int(d))).date() for d in r.integers(10, 220, 24)]
cp = cp.rename(columns={'sku': 'mapped_our_sku'})
save(cp, 'dim_competitor_product')
write_params()