"""Validate filled Binzagr data-request templates and turn them into pipeline-ready tables.

usage
  python ingest_client_templates.py FILLED.xlsx [--csv NAME=FILE.csv ...] [--out DIR] [--clean DIR]

  FILLED.xlsx    the client's copy of Binzagr_Data_Request_Templates.xlsx
  --csv NAME=F   a CSV sent instead of (or as well as) a sheet, with the sheet's column names. NAME = output name or
                 sheet number, e.g.  --csv attendance=roster_feb_sep.csv   --csv 8=open_items.csv  (repeatable)
  --out DIR      output folder (default: client_inputs/ next to the workbook)
  --clean DIR    02_Cleaned/data, used to check customer codes (default: ../../02_Cleaned/data from this script)

Per sheet (and CSV) it
  * skips the grey example row (row 3 of the blank sheets) and rows where nothing was filled in;
  * normalises keys the way 02_Cleaned does (codes kept as text, employee ids upper-case) and types dates / numbers;
  * checks required fields, formats, dropdown values, simple logic, duplicates, and that ids exist (sites, clients,
    badges from the pre-filled sheets; customers from 02_Cleaned/data/dim_customer.parquet);
  * writes client_<name>.csv (+ .parquet when pyarrow is installed) with lineage (src, src_row) and
    row_problems / row_notes columns - rows are flagged, never dropped, as in the cleaning layer;
  * writes VALIDATION_REPORT.md: rows received, problems to fix, notes, and the simulated table each one replaces.
Requirements: pandas, openpyxl (pyarrow optional).
"""
import argparse
import datetime as dt
import numbers
import os
import re
from collections import Counter

import pandas as pd
from openpyxl import load_workbook
from openpyxl.utils import column_index_from_string

HERE = os.path.dirname(os.path.abspath(__file__))

SPEC = {
    'jeddah_pnl_actual': dict(
        sheet='1 Jeddah P&L', prefilled=False,
        required=['month', 'line_item', 'amount_sar'], months=['month'], numbers=['amount_sar'],
        codes=['wh_id', 'client_code'], ids={'wh_id': 'site', 'client_code': 'client'},
        dup=(['month', 'wh_id', 'line_item', 'client_code'], 'note'),
        replaces='baseline of sim_jeddah_pnl_monthly / sim_jeddah_pnl_by_wh_month'),
    'attendance': dict(
        sheet='2 Attendance', prefilled=False,
        required=['payroll_number', 'date', 'clock_in', 'clock_out'], dates=['date'], stamps=['clock_in', 'clock_out'],
        codes=['payroll_number', 'employee_id', 'shift_code', 'home_wh_id'], ids={'payroll_number': 'payroll', 'employee_id_norm': 'badge', 'home_wh_id': 'site'},
        dup=(['payroll_number', 'clock_in'], 'problem'),
        replaces='paid shifts / hours in sim_session_attendance; badge-sharing check in inf_badge_session_risk'),
    'labour_cost_actual': dict(
        sheet='3 Labour cost', prefilled=False,
        required=['month', 'role', 'headcount', 'total_cost_sar'], months=['month'],
        numbers=['headcount', 'total_cost_sar', 'overtime_hours'], codes=['wh_id'], ids={'wh_id': 'site'},
        dup=(['month', 'wh_id', 'role', 'employment_type', 'nationality_group'], 'note'),
        replaces='sim_labour_cost_model (loaded cost per paid hour / shift)'),
    'rate_card_actual': dict(
        sheet='4 Rate card', prefilled=False,
        required=['client_code', 'service', 'unit', 'rate_sar'], dates=['effective_from', 'effective_to'],
        numbers=['rate_sar', 'min_charge_sar'], codes=['client_code'], ids={'client_code': 'client'},
        dup=(['client_code', 'service', 'unit', 'effective_from'], 'problem'),
        replaces='sim_rate_card (cost-recovery rates) and revenue in the simulated P&L'),
    'sites': dict(
        sheet='5 Sites', prefilled=True,
        inputs=['site_name', 'city_area', 'temperature_regime', 'area_sqm', 'pallet_positions', 'operating_pattern', 'status',
                'change_date', 'change_reason', 'notes'],
        dates=['change_date'], numbers=['area_sqm', 'pallet_positions'], codes=['wh_id'],
        replaces='inf_warehouse_profile (temperature regime, site changes)'),
    'clients': dict(
        sheet='6 Clients', prefilled=True, inputs=['client_name', 'client_type', 'billing_basis', 'notes'], codes=['client_code'],
        replaces='inf_client_principal_map'),
    'badges': dict(
        sheet='7 Badges', prefilled=True,
        inputs=['what_it_is', 'payroll_number', 'role', 'employment_type', 'same_person_as', 'notes'],
        codes=['employee_id_norm', 'payroll_number', 'same_person_as'], upper=['same_person_as'], ids={'same_person_as': 'badge'},
        replaces='roles in sim_session_attendance, inf_badge_session_risk, alias groups in jeddah_badges'),
    'receivables_actual': dict(
        sheet='8 Receivables', prefilled=False,
        required=['dist_cd', 'cust_cd', 'doc_no', 'doc_date', 'invoice_amount_sar'],
        dates=['doc_date', 'due_date', 'last_payment_date', 'snapshot_date'],
        numbers=['invoice_amount_sar', 'paid_amount_sar', 'open_amount_sar'], codes=['dist_cd', 'cust_cd', 'doc_no'],
        ids={('dist_cd', 'cust_cd'): 'customer'}, dup=(['dist_cd', 'cust_cd', 'doc_no', 'snapshot_date'], 'problem'),
        replaces='sim_receivables_open_items / sim_receivables_customer_ageing (FIFO allocation)'),
    'channel_codes': dict(
        sheet='9 Channel codes', prefilled=True, inputs=['official_name', 'channel', 'notes'], codes=['code'],
        replaces='inf_customer_channel_map'),
    'code_meanings': dict(
        sheet='10 Code meanings', prefilled=True, inputs=['correct_meaning', 'notes'],
        replaces='interpretations listed in CLEANING_REPORT.md / SIMULATION_REPORT.md'),
    'reexport_status': dict(
        sheet='11 Re-exports & history', prefilled=True, inputs=['status', 'expected_date', 'owner', 'notes'], dates=['expected_date'],
        replaces='tracking only'),
}

SPECIAL = {'site / wh_id': 'wh_id', 'client_code (optional)': 'client_code', 'amount_sar (excl. VAT)': 'amount_sar',
           'employee_id (as in WMS)': 'employee_id', 'headcount (month end)': 'headcount', 'total_cost_sar (all-in)': 'total_cost_sar',
           'payroll_number (as in the WMS user list)': 'payroll_number', 'employee_id (WMS badge, optional)': 'employee_id',
           'what it is': 'what_it_is', 'payroll_number (if a person)': 'payroll_number',
           'city / area': 'city_area', 'operating_pattern (shifts/days)': 'operating_pattern', 'what changed and why': 'change_reason',
           'client_type (confirm)': 'client_type', 'person_ref (pseudonym ok)': 'person_ref',
           'shared_or_system_login (Y/N)': 'shared_or_system_login', 'same_person_as (other badge id)': 'same_person_as',
           'code / field': 'code_or_field', 'our current understanding': 'our_understanding',
           'correct meaning / confirmation': 'correct_meaning'}


def clean_name(c):
    c = ' '.join(str(c).split())
    return SPECIAL.get(c) or re.sub(r'[^0-9a-z]+', '_', re.sub(r'\s*\(.*?\)', '', c).lower()).strip('_')


# ---------------------------------------------------------------- value parsers: return (value, problem or None)
def is_blank(v):
    if v is None or v is pd.NaT or v is pd.NA:
        return True
    if isinstance(v, float):
        return v != v
    return isinstance(v, str) and not v.strip()


def is_datelike(v):
    return isinstance(v, (dt.date, dt.datetime, pd.Timestamp))


def to_code(v):
    if isinstance(v, bool):
        return str(v), None
    if isinstance(v, numbers.Integral):
        return str(int(v)), (None if abs(int(v)) < 10 ** 11 else 'long code stored as a number - check no digits were lost')
    if isinstance(v, numbers.Real):
        f = float(v)
        s = str(int(f)) if f.is_integer() else repr(f)
        return s, (None if abs(f) < 1e11 else 'long code stored as a number - check no digits were lost')
    return ' '.join(str(v).split()), None


def to_number(v):
    if isinstance(v, numbers.Number) and not isinstance(v, bool):
        return float(v), None
    s = re.sub(r'(?i)sar|ر\.?س', '', str(v)).replace(',', '').replace(' ', '')
    m = re.fullmatch(r'(\()?(-?\d+(?:\.\d+)?)(\))?', s)
    if m and bool(m.group(1)) == bool(m.group(3)):
        x = float(m.group(2))
        return (-x if m.group(1) else x), None
    return None, f'not a number ("{v}")'


def to_month(v):
    if is_datelike(v):
        return pd.Timestamp(v).strftime('%Y-%m'), None
    s = str(v).strip()
    m = re.fullmatch(r'(\d{4})-(\d{1,2})(?:-\d{1,2})?(?:[ T]00:00(?::00)?)?', s)
    if m and 1 <= int(m.group(2)) <= 12:
        return f'{m.group(1)}-{int(m.group(2)):02d}', None
    return None, f'not YYYY-MM ("{s}")'


def to_date(v):
    if is_datelike(v):
        return pd.Timestamp(v).normalize(), None
    s = str(v).strip()
    if re.fullmatch(r'\d{4}-\d{1,2}-\d{1,2}(?:[ T]00:00(?::00)?)?', s):
        try:
            return pd.Timestamp(dt.datetime.strptime(re.split(r'[ T]', s)[0], '%Y-%m-%d')), None
        except ValueError:
            pass
    return None, f'not YYYY-MM-DD ("{s}")'


def to_stamp(v):
    if is_datelike(v):
        return pd.Timestamp(v), None
    s = str(v).strip().replace('T', ' ')
    for fmt in ('%Y-%m-%d %H:%M', '%Y-%m-%d %H:%M:%S'):
        try:
            return pd.Timestamp(dt.datetime.strptime(s, fmt)), None
        except ValueError:
            pass
    return None, f'not YYYY-MM-DD HH:MM ("{s}")'


# ---------------------------------------------------------------- reading
def read_sheet(xlsx, sheet, prefilled):
    raw = pd.read_excel(xlsx, sheet_name=sheet, header=1, dtype=object)
    raw.columns = [' '.join(str(c).split()) for c in raw.columns]
    raw.insert(0, 'src_row', range(3, 3 + len(raw)))           # Excel row number (header is row 2)
    raw.insert(0, 'src', f'{os.path.basename(xlsx)} [{sheet}]')
    if not prefilled and len(raw) and any(isinstance(v, str) and 'example only' in v.lower() for v in raw.iloc[0].values):
        raw = raw.iloc[1:]                                       # row 3 of a blank sheet = grey example row (kept if overwritten)
    return raw


def read_csv(path):
    raw = pd.read_csv(path, dtype=str, keep_default_na=False, encoding='utf-8-sig')
    raw.columns = [' '.join(str(c).split()) for c in raw.columns]
    raw = raw.replace({'': None})
    raw.insert(0, 'src_row', range(2, 2 + len(raw)))           # line number in the CSV (header is line 1)
    raw.insert(0, 'src', os.path.basename(path))
    return raw


def dropdown_lists(ws, header):
    """{clean column name: allowed values} from the sheet's data-validation lists."""
    out = {}
    for dv in ws.data_validations.dataValidation:
        if dv.type != 'list' or not dv.formula1 or not dv.formula1.startswith('"'):
            continue
        col = column_index_from_string(re.match(r'([A-Z]+)', str(dv.sqref).split(':')[0]).group(1))
        if col <= len(header):
            out[clean_name(header[col - 1])] = {x.strip() for x in dv.formula1.strip('"').split(',')}
    return out


def known_ids(xlsx, clean_dir):
    def col(sheet, c, upper=False):
        try:
            s = pd.read_excel(xlsx, sheet_name=sheet, header=1, dtype=object)[c].dropna().map(lambda v: to_code(v)[0])
            return set(s.str.upper() if upper else s)
        except (KeyError, ValueError):
            return None
    k = {'site': col('5 Sites', 'wh_id'), 'client': col('6 Clients', 'client_code'), 'badge': None, 'payroll': None, 'customer': None}
    # badges and payroll numbers: all of them, from 02_Cleaned (sheet 7 lists only the unexplained logins)
    for key, table, c in (('badge', 'jeddah_badges', 'employee_id_norm'), ('payroll', 'jeddah_people', 'payroll_number')):
        for path, reader in ((os.path.join(clean_dir, table + '.parquet'), lambda p, c=c: pd.read_parquet(p, columns=[c])),
                             (os.path.join(os.path.dirname(clean_dir), 'csv', table + '.csv'), lambda p, c=c: pd.read_csv(p, usecols=[c], dtype=str))):
            if k[key] is None and os.path.exists(path):
                try:
                    v = reader(path)[c].dropna().astype(str)
                    k[key] = set(v.str.upper()) if key == 'badge' else set(v)
                except ImportError:
                    pass
    if k['badge'] is None:
        k['badge'] = col('7 Badges', 'employee_id_norm', upper=True)
    p = os.path.join(clean_dir, 'dim_customer.parquet')
    if os.path.exists(p):
        try:
            c = pd.read_parquet(p, columns=['dist_cd', 'cust_cd'])
            k['customer'] = set(zip(c.dist_cd.astype(str), c.cust_cd.astype(str)))
        except ImportError:                                      # no pyarrow
            pass
    if k['customer'] is None:
        p = os.path.join(os.path.dirname(clean_dir), 'csv', 'dim_customer.csv')
        if os.path.exists(p):
            c = pd.read_csv(p, usecols=['dist_cd', 'cust_cd'], dtype=str)
            k['customer'] = set(zip(c.dist_cd, c.cust_cd))
    return k


# ---------------------------------------------------------------- validation
class Issues:
    def __init__(self, n):
        self.p = [[] for _ in range(n)]
        self.n = [[] for _ in range(n)]

    def add(self, mask, msg, kind='problem'):
        target = self.p if kind == 'problem' else self.n
        for i, flag in enumerate(mask):
            if flag:
                target[i].append(msg(i) if callable(msg) else msg)


def logic_checks(name, df, iss):
    if name == 'attendance' and len(df):
        ci, co = df['clock_in'], df['clock_out']
        both = ci.notna() & co.notna()
        hours = (co - ci).dt.total_seconds() / 3600
        iss.add(both & (hours <= 0), 'clock_out: not after clock_in')
        iss.add(both & (hours > 12), 'clock_out: presence over 12 h (Labour Law Art. 101 maximum)', 'note')
        iss.add(df['date'].notna() & ci.notna() & (ci.dt.normalize() != df['date']), 'date: differs from the clock_in date', 'note')
    if name == 'labour_cost_actual' and len(df):
        h = df['headcount']
        iss.add(h.notna() & ((h < 0) | (h % 1 != 0)), 'headcount: not a whole number >= 0')
        iss.add(df['total_cost_sar'].notna() & (df['total_cost_sar'] < 0), 'total_cost_sar: negative')
    if name == 'rate_card_actual' and len(df):
        iss.add(df['rate_sar'].notna() & (df['rate_sar'] < 0), 'rate_sar: negative')
        f, t = df['effective_from'], df['effective_to']
        iss.add(f.notna() & t.notna() & (t < f), 'effective_to: before effective_from')
    if name == 'receivables_actual' and len(df):
        iss.add(df['due_date'].notna() & df['doc_date'].notna() & (df['due_date'] < df['doc_date']), 'due_date: before doc_date')
        inv, paid, opn = df['invoice_amount_sar'], df['paid_amount_sar'], df['open_amount_sar']
        iss.add(inv.notna() & paid.notna() & opn.notna() & ((inv - paid - opn).abs() > 0.05),
                'open_amount_sar: not equal to invoice - paid (partial credit notes?)', 'note')


def process(name, spec, raw, expected, lists, known):
    sheet_p, sheet_n = [], []
    cols = [c for c in raw.columns if c not in ('src', 'src_row')]
    missing = [c for c in expected if c not in cols]
    extra = [c for c in cols if c not in expected]
    if missing:
        sheet_p.append('missing columns: ' + ', '.join(missing))
    if extra:
        sheet_n.append('extra columns kept: ' + ', '.join(extra))
    df = raw.rename(columns={c: clean_name(c) for c in cols})
    fill = [c for c in (spec['inputs'] if spec['prefilled'] else [clean_name(c) for c in cols]) if c in df.columns]
    df = df[df[fill].apply(lambda s: ~s.map(is_blank)).any(axis=1)].reset_index(drop=True) if fill else df.iloc[0:0]
    iss = Issues(len(df))
    for c in df.columns:                                        # blanks -> None, strip text
        if c not in ('src', 'src_row'):
            df[c] = pd.Series([None if is_blank(v) else (' '.join(v.split()) if isinstance(v, str) else v) for v in df[c]],
                              index=df.index, dtype=object)
    for c in spec.get('required', []):
        if c in df.columns:
            iss.add(df[c].isna(), f'{c}: missing')
    for kind, fn in (('codes', to_code), ('numbers', to_number), ('months', to_month), ('dates', to_date), ('stamps', to_stamp)):
        for c in spec.get(kind, []):
            if c not in df.columns:
                continue
            parsed = [(None, None) if is_blank(v) else fn(v) for v in df[c]]
            msgs = [m for _, m in parsed]
            level = 'note' if kind == 'codes' else 'problem'
            iss.add([m is not None for m in msgs], lambda i, c=c, msgs=msgs: f'{c}: {msgs[i]}', level)
            vals = [v for v, _ in parsed]
            if kind in ('dates', 'stamps'):
                df[c] = pd.to_datetime(pd.Series(vals, dtype=object))
            elif kind == 'numbers':
                df[c] = pd.to_numeric(pd.Series(vals, dtype=object), errors='coerce')
            else:
                df[c] = pd.Series(vals, dtype='string')
    typed = {c for k in ('codes', 'numbers', 'months', 'dates', 'stamps') for c in spec.get(k, [])}
    for c in df.columns:                                        # remaining columns: infer (pre-filled numbers) or text
        if c not in typed and c not in ('src', 'src_row'):
            s = df[c].infer_objects()
            df[c] = s.astype('string') if s.dtype == object else s
    for c in spec.get('upper', []):
        if c in df.columns:
            df[c] = df[c].str.upper()
    if name == 'attendance' and 'employee_id' in df.columns:
        df['employee_id_norm'] = df['employee_id'].str.upper()     # same rule as 02_Cleaned (upper-case WMS id)
    for c, allowed in lists.items():
        if c in df.columns:
            iss.add(df[c].notna() & ~df[c].astype(str).isin(allowed), lambda i, c=c: f'{c}: "{df[c].iloc[i]}" not in the dropdown list')
    for c, kind in spec.get('ids', {}).items():
        ref = known.get(kind)
        if isinstance(c, tuple):
            if not all(x in df.columns for x in c):
                continue
            if ref is None:
                sheet_n.append(f'{kind} codes not checked (02_Cleaned/data/dim_customer not found)')
                continue
            keys = list(zip(df[c[0]], df[c[1]]))
            iss.add([not is_blank(k[0]) and not is_blank(k[1]) and k not in ref for k in keys], f'{"+".join(c)}: not in the customer master')
            continue
        if c not in df.columns:
            continue
        if ref is None:
            sheet_n.append(f'{c} not checked (pre-filled sheet with the {kind} list not found)')
            continue
        unknown = df[c].notna() & ~df[c].isin(ref)
        if kind == 'badge' and name == 'attendance':
            iss.add(unknown, f'{c}: not one of the {len(ref)} WMS badges (no scans, or a different id format)', 'note')
            if len(df) and unknown.mean() > 0.5:
                sheet_n.append(f'{unknown.mean():.0%} of employee ids are not WMS badges - please add a column or file mapping '
                               'the HR id to the WMS badge id')
        else:
            iss.add(unknown, f'{c}: not a known {kind} code', 'problem')
    logic_checks(name, df, iss)
    if 'dup' in spec and len(df):
        keys, level = spec['dup']
        keys = [k for k in keys if k in df.columns]
        d = df.duplicated(keys, keep=False) & df[keys[0]].notna()
        iss.add(d, f'duplicate of another row on {" + ".join(keys)}', level)
    df['row_problems'] = ['; '.join(x) for x in iss.p]
    df['row_notes'] = ['; '.join(x) for x in iss.n]
    return df, sheet_p, sheet_n


def summarise(df, col, limit=8):
    c, rows = Counter(), {}
    several = df['src'].nunique() > 1
    for s, r, txt in zip(df['src'], df['src_row'], df[col]):
        for m in filter(None, txt.split('; ')):
            k = re.sub(r' \(".*"\)$', '', re.sub(r': ".*" not in', ': value not in', m))
            c[k] += 1
            rows.setdefault(k, []).append((('sheet' if s.endswith(']') else s) + ':' if several else '') + str(r))
    return [f'{n:,} × {k} (rows {", ".join(rows[k][:limit])}{" …" if len(rows[k]) > limit else ""})' for k, n in c.most_common()]


def main():
    ap = argparse.ArgumentParser(description=__doc__.split('\n')[0])
    ap.add_argument('xlsx')
    ap.add_argument('--csv', action='append', default=[], metavar='NAME=FILE')
    ap.add_argument('--out')
    ap.add_argument('--clean', default=os.path.normpath(os.path.join(HERE, '..', '..', '02_Cleaned', 'data')))
    a = ap.parse_args()
    out = a.out or os.path.join(os.path.dirname(os.path.abspath(a.xlsx)), 'client_inputs')
    os.makedirs(out, exist_ok=True)
    by_number = {v['sheet'].split()[0]: k for k, v in SPEC.items()}
    csvs = {}
    for item in a.csv:
        k, _, path = item.partition('=')
        k = by_number.get(k.strip(), k.strip())
        if k not in SPEC or not path:
            ap.error(f'--csv {item}: use NAME=FILE with NAME one of {", ".join(SPEC)} or a sheet number 1-11')
        csvs.setdefault(k, []).append(path)

    wb = load_workbook(a.xlsx, read_only=False, data_only=True)
    known = known_ids(a.xlsx, a.clean)
    try:
        import pyarrow  # noqa: F401
        have_parquet = True
    except ImportError:
        have_parquet = False
    run = dt.datetime.now().strftime('%Y-%m-%d %H:%M')
    summary = ['| Sheet | Source | Rows received | Rows with problems | Rows with notes | Output | Replaces |', '|---|---|---|---|---|---|---|']
    details = []
    for name, spec in SPEC.items():
        sheet = spec['sheet']
        if sheet not in wb.sheetnames:
            summary.append(f'| {sheet} | sheet missing | - | - | - | - | {spec["replaces"]} |')
            continue
        ws = wb[sheet]
        header = [' '.join(str(c.value).split()) for c in ws[2] if c.value is not None]
        lists = dropdown_lists(ws, header)
        parts = [read_sheet(a.xlsx, sheet, spec['prefilled'])] + [read_csv(p) for p in csvs.get(name, [])]
        frames, sp, sn = [], [], []
        for raw in parts:
            df, p_, n_ = process(name, spec, raw, header, lists, known)
            frames.append(df)
            src = raw['src'].iloc[0] if len(raw) else ''
            sp += [f'{src}: {m}' for m in p_]
            sn += [f'{src}: {m}' for m in n_]
        df = pd.concat(frames, ignore_index=True) if len(frames) > 1 else frames[0]
        sources = ' + '.join(['sheet'] + [os.path.basename(p) for p in csvs.get(name, [])])
        base = os.path.join(out, f'client_{name}')
        df.to_csv(base + '.csv', index=False, encoding='utf-8')
        if have_parquet:
            df.to_parquet(base + '.parquet', index=False)
        n_p, n_n = int((df.row_problems != '').sum()), int((df.row_notes != '').sum())
        summary.append(f'| {sheet} | {sources} | {len(df):,} | {n_p:,} | {n_n:,} | client_{name} | {spec["replaces"]} |')
        if len(df) or sp or sn:
            details += [f'### {sheet}', '']
            details += [f'- **Problem:** {m}' for m in sp] + [f'- **Note:** {m}' for m in sn]
            details += [f'- **Problem:** {m}' for m in summarise(df, 'row_problems')]
            details += [f'- Note: {m}' for m in summarise(df, 'row_notes')]
            if not (n_p or n_n or sp or sn):
                details.append('- all rows passed')
            details.append('')
    md = ['# Validation report - client data templates', '',
          f'Source: `{os.path.basename(a.xlsx)}`' + (f' + {sum(len(v) for v in csvs.values())} CSV file(s)' if csvs else '') + f' | run {run}', '',
          '**Problems** need a fix before the data replaces the simulation; **notes** are worth a look but do not block. '
          'Rows are never dropped: each output row carries `row_problems` / `row_notes` and its source row (`src`, `src_row`).', '',
          *summary, '', '## Details', '', *(details or ['No data entered yet.']), '',
          'Outputs: `client_<name>.csv`' + (' and `.parquet`' if have_parquet else ' (install pyarrow for parquet)') +
          '. Keys match 02_Cleaned: `wh_id`, `client_code`, `employee_id_norm`, `dist_cd` + `cust_cd`, hierarchy `code`.']
    with open(os.path.join(out, 'VALIDATION_REPORT.md'), 'w', encoding='utf-8') as f:
        f.write('\n'.join(md) + '\n')
    print('\n'.join(summary))
    print(f'\nwritten to {out}')


if __name__ == '__main__':
    main()