"""Stage 6: receipts (F19), credit blocks (F20), deductions/disputes (F21), escalations (F22).
Stage 7: team attendance (F23), meetings (F24), action log (F25)."""
import numpy as np, pandas as pd
from common import *
import warnings; warnings.filterwarnings('ignore')

r = rng(6)
FI = load('fact_sell_in'); FI['doc_date'] = pd.to_datetime(FI.doc_date)
acc = load('dim_account').set_index('account_id'); outl = load('dim_outlet'); emp = load('dim_employee')
F = load('fact_order_fulfilment'); TS = load('fact_trade_spend')
END_ = pd.Timestamp(END)

# ---------------- invoices: one per document (free goods excluded)
inv = FI[~FI.is_foc].groupby(['doc_no', 'account_id', 'channel_id']).agg(invoice_date=('doc_date', 'min'), invoice_sar=('net_amt', 'sum')).reset_index()
inv = inv[inv.invoice_sar > 0]
inv['terms'] = inv.account_id.map(acc.payment_terms_days).fillna(0).astype(int)
inv['due_date'] = inv.invoice_date + pd.to_timedelta(inv.terms, 'D')

# real 20-Oct balances for grocery outlets (twins summed onto the main code)
cust = pd.read_parquet(B + 'dim_customer.parquet')
cust['code'] = cust.dist_cd.astype(str) + '-' + cust.cust_cd.astype(str)
gro = outl[outl.channel_id == 'GR']
c2m = dict(gro.assign(c=gro.source_codes.str.split('|')).explode('c')[['c', 'outlet_id']].values)
cust['outlet_id'] = cust.code.map(c2m)
real_bal = cust.dropna(subset=['outlet_id']).groupby('outlet_id').outstanding_bal.sum().clip(lower=0)
real_bal.index = 'GR-' + real_bal.index

# payment behaviour per account: prompt / normal / slow / chronic
beh = pd.Series(r.choice(['prompt', 'normal', 'slow', 'chronic'], len(acc), p=[0.35, 0.40, 0.18, 0.07]), index=acc.index)
for a in ['KA02', 'WS007', 'WS023', 'DS03']:        # a few large late payers so collections has something to find
    if a in beh.index: beh[a] = 'chronic'
delay_mu = {'prompt': -2, 'normal': 5, 'slow': 20, 'chronic': 55}

rec, open_rows = [], []
for a, g in inv.sort_values('invoice_date', ascending=False).groupby('account_id', sort=False):
    b = beh.get(a, 'normal'); ch = g.channel_id.iat[0]
    target = real_bal.get(a, None) if ch == 'GR' else None
    remaining = target if target is not None else None
    for row in g.itertuples():
        if row.terms == 0:   # cash on delivery
            paid = row.invoice_sar; rdate = row.invoice_date; method = r.choice(['cash', 'card'], p=[0.65, 0.35])
            rec.append((row.doc_no, a, ch, row.invoice_date, row.due_date, rdate, round(paid, 2), method, 0.0)); continue
        delay = max(-row.terms, int(r.normal(delay_mu[b], 6 + delay_mu[b] * 0.3)))
        rdate = row.due_date + pd.Timedelta(days=delay)
        if remaining is not None:                       # grocery credit: newest invoices stay open up to the real balance
            open_amt = min(row.invoice_sar, remaining); remaining -= open_amt
            rdate = min(rdate, END_)
        else:
            open_amt = row.invoice_sar if rdate > END_ else 0.0
        paid = row.invoice_sar - open_amt
        if paid > 0.01:
            method = 'bank transfer' if ch != 'GR' else r.choice(['cash', 'cheque', 'bank transfer'], p=[0.5, 0.2, 0.3])
            rec.append((row.doc_no, a, ch, row.invoice_date, row.due_date, max(rdate, row.invoice_date), round(paid, 2), method, 0.0))
        if open_amt > 0.01:
            open_rows.append((row.doc_no, a, ch, row.invoice_date, row.due_date, round(open_amt, 2)))
R = pd.DataFrame(rec, columns=['invoice_id', 'customer_id', 'channel_id', 'invoice_date', 'due_date', 'receipt_date', 'allocated_sar', 'payment_method', 'deducted_sar'])
OPEN = pd.DataFrame(open_rows, columns=['invoice_id', 'customer_id', 'channel_id', 'invoice_date', 'due_date', 'open_sar'])

# ---------------- F21 deductions: MT / DS / WS short-pay on ~6% of invoices, plus disputed trade-spend claims
ded_idx = R.index[R.channel_id.isin(['MT', 'DS', 'WS']) & (r.random(len(R)) < 0.06)]
reasons = ['promo claim offset', 'price difference', 'shortage / OTIF penalty', 'damaged goods', 'listing / display fee', 'unidentified']
rs = r.choice(reasons, len(ded_idx), p=[0.30, 0.22, 0.20, 0.12, 0.08, 0.08])
pct = r.uniform(0.02, 0.15, len(ded_idx))
R.loc[ded_idx, 'deducted_sar'] = (R.loc[ded_idx, 'allocated_sar'] * pct).round(2)
R.loc[ded_idx, 'allocated_sar'] = (R.loc[ded_idx, 'allocated_sar'] - R.loc[ded_idx, 'deducted_sar']).round(2)
valid_p = {'promo claim offset': 0.7, 'price difference': 0.45, 'shortage / OTIF penalty': 0.6, 'damaged goods': 0.65, 'listing / display fee': 0.8, 'unidentified': 0.2}
D = R.loc[ded_idx, ['customer_id', 'invoice_id', 'receipt_date', 'deducted_sar']].rename(columns={'receipt_date': 'opened_date', 'deducted_sar': 'deduction_sar'})
D['reason'] = rs; D['source'] = 'short payment'
tsd = TS[TS.status.isin(['disputed', 'rejected'])]
D2 = pd.DataFrame({'customer_id': tsd.account_id, 'invoice_id': tsd.claim_id, 'opened_date': pd.to_datetime(tsd.claim_date),
                   'deduction_sar': (tsd.claim_sar - tsd.approved_sar.fillna(0)).round(2), 'reason': 'promo claim above accrual', 'source': 'trade spend claim'})
D = pd.concat([D, D2], ignore_index=True)
age = (r.gamma(2, 18, len(D))).astype(int)
D['resolved_date'] = D.opened_date + pd.to_timedelta(age, 'D')
open_case = D.resolved_date > END_
valid = r.random(len(D)) < D.reason.map(valid_p).fillna(0.5)
D['status'] = np.where(open_case, 'open', np.where(valid, 'accepted (credit note)', 'rejected (recovered)'))
D['recovered_sar'] = np.where(D.status == 'rejected (recovered)', (D.deduction_sar * r.uniform(0.6, 1.0, len(D))).round(2), 0.0)
D.loc[open_case, 'resolved_date'] = pd.NaT
D.insert(0, 'case_id', [f'DD{i+1:05d}' for i in range(len(D))])
D['days_open'] = ((D.resolved_date.fillna(END_) - D.opened_date).dt.days)
save(D, 'fact_deduction_dispute')

R.insert(0, 'receipt_id', [f'RC{i+1:07d}' for i in range(len(R))])
R['unapplied_sar'] = 0.0
ua = r.random(len(R)) < 0.004   # on-account cash not matched to an invoice
R.loc[ua, 'unapplied_sar'] = (R.loc[ua, 'allocated_sar'] * r.uniform(0.05, 0.3, ua.sum())).round(2)
R['days_late'] = (R.receipt_date - R.due_date).dt.days.clip(lower=0)
save(R, 'fact_receipt')
OPEN['days_overdue'] = (END_ - OPEN.due_date).dt.days.clip(lower=0)
OPEN['aging_bucket'] = pd.cut(OPEN.days_overdue, [-1, 0, 30, 60, 90, 10_000], labels=['not due', '1-30', '31-60', '61-90', '90+']).astype(str)
save(OPEN, 'fact_open_invoice')
print('  open AR SAR', round(OPEN.open_sar.sum()), '| GR open vs real balance', round(OPEN[OPEN.channel_id == 'GR'].open_sar.sum()), round(real_bal.sum()),
      '| overdue share', round(OPEN.loc[OPEN.days_overdue > 0, 'open_sar'].sum() / OPEN.open_sar.sum(), 3))

# ---------------- F20 credit blocks: weekly check on overdue > 30 days or balance over limit
allinv = pd.concat([R[['invoice_id', 'customer_id', 'invoice_date', 'due_date', 'receipt_date']].assign(amt=R.allocated_sar + R.deducted_sar),
                    OPEN[['invoice_id', 'customer_id', 'invoice_date', 'due_date']].assign(receipt_date=pd.NaT, amt=OPEN.open_sar)])
allinv = allinv[allinv.customer_id.map(acc.payment_terms_days).fillna(0) > 0]
blocks = []
for wk in pd.date_range(START, END, freq='7D')[4:]:
    op = allinv[(allinv.invoice_date <= wk) & (allinv.receipt_date.isna() | (allinv.receipt_date > wk))]
    od = op[op.due_date < wk - pd.Timedelta(days=60)].groupby('customer_id').amt.sum()
    bal = op.groupby('customer_id').amt.sum()
    lim = acc.credit_limit_sar.reindex(bal.index)
    od0 = op[op.due_date < wk].groupby('customer_id').amt.sum()
    over = bal[(bal > lim * 1.3) & (bal - lim > 3000)].index
    hit = set(od[od > 1500].index) | set(over.intersection(od0[od0 > 1000].index))
    for c in hit:
        blocks.append((c, wk, 'overdue > 60 days' if c in od.index and od[c] > 1500 else 'over credit limit', bal[c], lim.get(c)))
BL = pd.DataFrame(blocks, columns=['customer_id', 'check_date', 'reason', 'balance_sar', 'credit_limit_sar'])
BL = BL.sort_values(['customer_id', 'check_date'])
BL['new'] = BL.groupby('customer_id').check_date.diff().ne(pd.Timedelta(days=7))
BL['grp'] = BL.new.cumsum()
CB = BL.groupby('grp').agg(customer_id=('customer_id', 'first'), block_date=('check_date', 'min'), last_week=('check_date', 'max'),
                           reason=('reason', 'first'), balance_sar=('balance_sar', 'max'), credit_limit_sar=('credit_limit_sar', 'first')).reset_index(drop=True)
CB['release_date'] = CB.last_week + pd.to_timedelta(r.integers(1, 7, len(CB)), 'D')
CB.loc[CB.last_week >= END_ - pd.Timedelta(days=7), 'release_date'] = pd.NaT
CB['release_type'] = np.where(CB.release_date.isna(), 'still blocked', np.where(r.random(len(CB)) < 0.3, 'manager override', 'payment received'))
# value of orders held while blocked (orders placed by that customer in the block window)
ordv = FI.groupby(['account_id', 'doc_date']).net_amt.sum()
CB['blocked_order_value_sar'] = [round(ordv.get(c, pd.Series(dtype=float)).loc[s: (e if pd.notna(e) else END_)].sum() * r.uniform(0.25, 0.6), 2)
                                 if c in ordv.index.get_level_values(0) else 0.0 for c, s, e in zip(CB.customer_id, CB.block_date, CB.release_date)]
CB.insert(0, 'block_id', [f'CB{i+1:05d}' for i in range(len(CB))])
save(CB.drop(columns='last_week'), 'fact_credit_block')
print('  credit blocks', len(CB), '| customers', CB.customer_id.nunique(), '| override share', round((CB.release_type == 'manager override').mean(), 2))

# ---------------- F22 escalations: triggered by OTIF misses, price gaps, disputes, blocks
kam = acc.kam_employee_id
F['month'] = pd.to_datetime(F.order_date).dt.to_period('M')
otif = F[F.channel_id.isin(['MT', 'DS', 'EC'])].groupby(['account_id', 'month']).otif_flag.mean()
tgt = otif.index.get_level_values(0).map(acc.otif_target_pct).values / (100 if acc.otif_target_pct.max() > 1 else 1)
esc = []
for (a, m), v in otif[otif < tgt - 0.05].items():
    esc.append((a, 'service failure (OTIF)', m.to_timestamp() + pd.Timedelta(days=int(r.integers(32, 40))), round(float(FI[(FI.account_id == a)].net_amt.sum() / 8 * 0.05), 2)))
PO = load('fact_price_observation')
for row in PO[PO.anomaly_label == 'ds_below_mt_price'].sample(frac=0.4, random_state=1).itertuples():
    mt = 'KA0' + str(int(r.integers(1, 9)))
    esc.append((mt, 'price gap vs discounter', pd.Timestamp(row.obs_date) + pd.Timedelta(days=int(r.integers(2, 10))), round(float(r.uniform(20e3, 150e3)), 2)))
for row in D[(D.deduction_sar > D.deduction_sar.quantile(0.97))].itertuples():
    esc.append((row.customer_id, 'deduction dispute', row.opened_date + pd.Timedelta(days=int(r.integers(10, 30))), row.deduction_sar))
for row in CB[~CB.customer_id.str.startswith('GR-')].itertuples():
    esc.append((row.customer_id, 'credit block on orders', row.block_date + pd.Timedelta(days=2), row.blocked_order_value_sar))
for _ in range(25):
    a = r.choice(['KA0' + str(i) for i in range(1, 9)] + ['DS0' + str(i) for i in range(1, 5)])
    esc.append((a, r.choice(['delisting threat', 'listing / range review', 'quality complaint']), pd.Timestamp(START) + pd.Timedelta(days=int(r.integers(10, 225))), round(float(r.uniform(50e3, 400e3)), 2)))
E = pd.DataFrame(esc, columns=['account_id', 'type', 'opened_date', 'value_at_risk_sar'])
E = E[E.opened_date <= END_]
E['severity'] = pd.cut(E.value_at_risk_sar, [-1, 25e3, 100e3, 1e12], labels=['low', 'medium', 'high']).astype(str)
E['owner_id'] = E.account_id.map(kam).fillna(E.account_id.str[:2].map({'GR': None}))
sup = emp[emp.role == 'Sales supervisor'].employee_id.values
E['owner_id'] = E.owner_id.fillna(pd.Series(r.choice(sup, len(E)), index=E.index))
E['due_date'] = E.opened_date + pd.to_timedelta(E.severity.map({'high': 5, 'medium': 10, 'low': 20}), 'D')
E['resolved_date'] = E.opened_date + pd.to_timedelta(r.gamma(2, 6, len(E)).astype(int) + 1, 'D')
E.loc[E.resolved_date > END_, 'resolved_date'] = pd.NaT
E['status'] = np.where(E.resolved_date.isna(), np.where(E.due_date < END_, 'open - overdue', 'open'), np.where(E.resolved_date > E.due_date, 'closed late', 'closed on time'))
E = E.sort_values('opened_date').reset_index(drop=True)
E.insert(0, 'escalation_id', [f'ES{i+1:04d}' for i in range(len(E))])
save(E, 'fact_account_escalation')
print('  escalations', len(E), E.type.value_counts().to_dict())

# ================= Stage 7: team
cal = calendar(); wd = pd.to_datetime(cal.date)
work = wd[wd.dt.day_name() != 'Friday']
RV = load('fact_route_visit'); RV['visit_date'] = pd.to_datetime(RV.visit_date)
first = RV[RV.outcome != 'not visited'].groupby(['van_id', 'visit_date']).arrival_time.min()
last = RV[RV.outcome != 'not visited'].groupby(['van_id', 'visit_date']).departure_time.max()
MA = pd.read_parquet(OUT + '/_tmp_merch_absence.parquet'); MA['date'] = pd.to_datetime(MA.date)
ma = MA.set_index(['employee_id', 'date']).status
span = first.reset_index().groupby('van_id').visit_date.agg(a='min', b='max')
rows = []
reps = emp[emp.role == 'Van rep']; sup_of = reps.set_index('employee_id').manager_id
for e in emp.itertuples():
    for d in work:
        if e.role == 'Van rep':
            k = (e.employee_id, d)
            if d < span.loc[e.employee_id, 'a'] or d > span.loc[e.employee_id, 'b']: continue
            if k in first.index:
                rows.append((e.employee_id, e.role, d, 'present', first[k], last[k], None))
            else:
                st = r.choice(['annual leave', 'sick', 'absent', 'off (no route)'], p=[0.30, 0.15, 0.10, 0.45])
                rows.append((e.employee_id, e.role, d, st, None, None, None))
        elif e.role == 'Merchandiser':
            st = ma.get((e.employee_id, d), 'present')
            ci = f'{int(r.integers(7, 9)):02d}:{int(r.integers(0, 60)):02d}' if st == 'present' else None
            co = f'{int(r.integers(15, 18)):02d}:{int(r.integers(0, 60)):02d}' if st == 'present' else None
            rows.append((e.employee_id, e.role, d, st, ci, co, None))
        else:
            st = 'present' if r.random() > 0.06 else r.choice(['annual leave', 'sick', 'training'], p=[0.6, 0.25, 0.15])
            rows.append((e.employee_id, e.role, d, st, '08:00' if st == 'present' else None, '17:00' if st == 'present' else None, None))
TA = pd.DataFrame(rows, columns=['employee_id', 'role', 'date', 'status', 'check_in', 'check_out', 'cover_employee_id'])
# cover: merchandiser absence covered by another merchandiser 55% of time; uncovered stores flagged by the gap
merch = emp[emp.role == 'Merchandiser'].employee_id.values
absm = TA.index[(TA.role == 'Merchandiser') & (TA.status != 'present')]
cov = r.random(len(absm)) < 0.55
TA.loc[absm[cov], 'cover_employee_id'] = r.choice(merch, cov.sum())
absr = TA.index[(TA.role == 'Van rep') & TA.status.isin(['annual leave', 'sick', 'absent'])]
TA['route_uncovered'] = False
TA.loc[absr, 'route_uncovered'] = r.random(len(absr)) < 0.4
TA['date'] = TA.date.dt.date
save(TA, 'fact_team_attendance')
print('  field attendance rows', len(TA), '| present share', round((TA.status == 'present').mean(), 3))

# warehouse attendance: real Jeddah scans (names never held; employee number only)
AT = pd.read_parquet('/home/claude/binzagr/emails/att.parquet').rename(columns={'Employee ID': 'payroll_no', 'Date': 'date', 'POSITION_TITLE': 'position',
                                                                               'Number of Minute': 'minutes', 'Number of Hour': 'hours', 'Over Time Hour': 'overtime_min', 'Absent Hour': 'absent_min'})
AT = AT.drop(columns=[c for c in AT.columns if 'name' in c.lower()])
save(AT, 'fact_warehouse_attendance_jeddah', label='Actual')

# ---------------- F24 meetings
kams = emp[emp.role == 'Key account manager'].employee_id.values; mgr = emp[emp.role.str.startswith('Channel manager')]
mt_accts = acc.index[acc.channel_id.isin(['MT', 'DS', 'WS', 'EC'])]
M = []
for a in mt_accts:
    owner = acc.kam_employee_id.get(a) or r.choice(kams)
    tier_n = {'MT': 2, 'DS': 1.5, 'EC': 1.5, 'WS': 0.6}[acc.channel_id[a]]
    for d in pd.date_range(START, END, freq='MS'):
        for _ in range(r.poisson(tier_n)):
            t = r.choice(['business review', 'promotion negotiation', 'range / listing', 'collections call', 'joint business plan'], p=[0.3, 0.3, 0.15, 0.15, 0.1])
            M.append((d + pd.Timedelta(days=int(r.integers(0, 27))), owner, a, t, 60 if t != 'joint business plan' else 180))
for e in E.itertuples():
    M.append((e.opened_date + pd.Timedelta(days=int(r.integers(0, 4))), e.owner_id, e.account_id, 'escalation meeting', 60))
for s in sup:
    for d in work[::7]:
        M.append((d, s, None, 'route ride-along', 240))
        M.append((d + pd.Timedelta(days=1), s, None, 'team huddle', 45))
for m in mgr.employee_id:
    for d in work[::7]:
        M.append((d, m, None, 'weekly sales review', 90))
M = pd.DataFrame(M, columns=['date', 'organiser_id', 'account_id', 'type', 'duration_min'])
M = M[M.date <= END_]
neg = M.type.isin(['promotion negotiation', 'range / listing', 'joint business plan', 'escalation meeting'])
M['value_at_stake_sar'] = np.where(neg, r.lognormal(11.3, 0.8, len(M)).round(-2), np.nan)
M['outcome'] = np.where(neg, r.choice(['agreed', 'partly agreed', 'no agreement', 'follow-up needed'], len(M), p=[0.4, 0.25, 0.1, 0.25]), 'held')
M.loc[r.random(len(M)) < 0.07, 'outcome'] = 'cancelled'
M = M.sort_values('date').reset_index(drop=True); M.insert(0, 'meeting_id', [f'MT{i+1:05d}' for i in range(len(M))])
M['date'] = M.date.dt.date
save(M, 'fact_meeting')

# manager diary: how channel managers spend the week (task 30)
cats = ['key account meetings', 'internal reviews / reporting', 'field visits', 'approvals & escalations', 'planning & forecasting', 'admin / email']
base_w = np.array([0.18, 0.24, 0.12, 0.16, 0.10, 0.20])
TD = []
for m in mgr.employee_id:
    for d in work:
        w = r.dirichlet(base_w * 40); hrs = r.normal(9.5, 0.8)
        for c, x in zip(cats, w):
            TD.append((m, d.date(), c, round(hrs * x, 2)))
save(pd.DataFrame(TD, columns=['employee_id', 'date', 'activity', 'hours']), 'fact_manager_time')

# ---------------- F25 action log
A = []
for e in E.itertuples():
    A.append(('escalation', e.escalation_id, e.owner_id, None, e.opened_date, e.due_date, e.resolved_date, e.value_at_risk_sar))
for c in CB[CB.release_type == 'manager override'].itertuples():
    A.append(('credit override approval', c.block_id, r.choice(emp[emp.role == 'Credit controller'].employee_id.values), mgr.employee_id.iat[0],
              c.block_date, c.block_date + pd.Timedelta(days=2), c.release_date, c.blocked_order_value_sar))
for d in D[D.deduction_sar > 2000].itertuples():
    A.append(('deduction approval', d.case_id, r.choice(emp[emp.role == 'Credit controller'].employee_id.values), mgr.employee_id.iat[1],
              d.opened_date, d.opened_date + pd.Timedelta(days=14), d.resolved_date, d.deduction_sar))
PR = load('fact_promotion').drop_duplicates('promotion_id')
for p in PR.itertuples():
    sd = pd.Timestamp(p.start_date)
    A.append(('promotion approval', p.promotion_id, r.choice(kams), mgr.employee_id.iat[1], sd - pd.Timedelta(days=int(r.integers(14, 30))), sd - pd.Timedelta(days=7), None, getattr(p, 'budget_sar', np.nan)))
REC = ['raise route frequency', 'cross-division pair (lead)', 'next-best SKU', 'price below floor', 'returns hotspot', 'collections chase',
       'churn risk outlet', 'stock-out risk', 'near-expiry stock', 'van below breakeven', 'listing inactive (e-com)', 'uncovered route']
for _ in range(1400):
    t = pd.Timestamp(START) + pd.Timedelta(days=int(r.integers(0, 230)))
    A.append(('recommendation: ' + r.choice(REC), None, r.choice(np.r_[sup, kams]), None, t, t + pd.Timedelta(days=7), None, round(float(r.lognormal(8, 1.1)), 2)))
AL = pd.DataFrame(A, columns=['source', 'source_ref', 'owner_id', 'approver_id', 'created_date', 'due_date', 'closed_date', 'value_sar'])
AL = AL[AL.created_date <= END_]
# approvals wait: closed = created + waiting; some still pending
need = AL.closed_date.isna()
wait = r.gamma(1.6, 3.5, need.sum()).astype(int)
stay = r.random(need.sum()) < 0.12
cl = AL.loc[need, 'created_date'] + pd.to_timedelta(wait, 'D')
AL.loc[need, 'closed_date'] = cl.where(~stay & (cl <= END_), pd.NaT)
rec_ = AL.source.str.startswith('recommendation')
AL['status'] = np.where(AL.closed_date.isna(), np.where(AL.due_date < END_, 'overdue', 'open'),
                        np.where(rec_ & (r.random(len(AL)) < 0.3), 'rejected', np.where(AL.approver_id.notna(), 'approved', 'done')))
AL['waiting_days'] = (AL.closed_date.fillna(END_) - AL.created_date).dt.days
AL = AL.sort_values('created_date').reset_index(drop=True); AL.insert(0, 'action_id', [f'AC{i+1:05d}' for i in range(len(AL))])
save(AL, 'fact_action_log')
print('  actions', len(AL), AL.status.value_counts().to_dict())