"""Shared paths and helpers for the Binzagr cleaning pipeline (02_Cleaned).

Rules for this pipeline:
  * Deterministic only - no estimation, imputation or simulation.
  * Raw files are never modified; every cleaned row keeps lineage (src_file + src_line/src_row).
  * Anything suspicious is FLAGGED, not silently changed. Values are only set to null when the
    source value is a provable placeholder/corruption (e.g. 01/01/1900 dates), and a flag says so.
"""
import os, json, hashlib, gzip, shutil
import numpy as np
import pandas as pd

_HERE = os.path.dirname(os.path.abspath(__file__))
# Set BZ_RAW / BZ_LOOSE to where the raw files are on your machine (defaults are the paths used for the delivered run).
RAW = os.environ.get('BZ_RAW', '/home/claude/binzagr/raw')                       # unzipped client zips, one sub-folder per zip
LOOSE = os.environ.get('BZ_LOOSE', '/mnt/user-data/uploads/Downloads/OneDrive_1_10-21-2026')  # loose files (Payment Term.xlsx)
OUT = os.environ.get('BZ_OUT', os.path.normpath(os.path.join(_HERE, '..')))      # default: the folder above scripts/
DATA = os.path.join(OUT, 'data')        # parquet (ML-ready, typed)
CSV = os.path.join(OUT, 'csv')          # csv / csv.gz (same content)
AUDIT = os.path.join(OUT, 'audit')      # removed rows, stats
STATS_DIR = os.path.join(OUT, 'audit')

for d in (DATA, CSV, AUDIT):
    os.makedirs(d, exist_ok=True)

CSV_GZIP_THRESHOLD = 20 * 1024 * 1024   # csv bigger than this is written gzipped
MAX_FILE = 29 * 1024 * 1024             # keep every file under the 30 MB transfer limit


def save_stats(name, obj):
    with open(os.path.join(STATS_DIR, f'stats_{name}.json'), 'w', encoding='utf-8') as f:
        json.dump(obj, f, indent=2, default=str, ensure_ascii=False)


def load_stats(name):
    with open(os.path.join(STATS_DIR, f'stats_{name}.json'), encoding='utf-8') as f:
        return json.load(f)


def _csv_bytes(df):
    return df.to_csv(index=False, date_format='%Y-%m-%d %H:%M:%S').encode('utf-8-sig')


def _write_one(df, pdir, cdir, name):
    ppath = os.path.join(pdir, f'{name}.parquet')
    df.to_parquet(ppath, index=False, compression='zstd')
    b = _csv_bytes(df)
    if len(b) > CSV_GZIP_THRESHOLD:
        cpath = os.path.join(cdir, f'{name}.csv.gz')
        with gzip.open(cpath, 'wb', compresslevel=6) as f:
            f.write(b)
    else:
        cpath = os.path.join(cdir, f'{name}.csv')
        with open(cpath, 'wb') as f:
            f.write(b)
    return ppath, cpath


def write_table(df, name, subdir=None):
    """Write one table as parquet + csv (csv gzipped when large).
    If any file would exceed MAX_FILE the table is split into row-ordered parts: name_part1, name_part2 ...
    Returns list of written paths."""
    pdir = os.path.join(DATA, subdir) if subdir else DATA
    cdir = os.path.join(CSV, subdir) if subdir else CSV
    os.makedirs(pdir, exist_ok=True); os.makedirs(cdir, exist_ok=True)
    nparts = 1
    while True:
        written = []
        bounds = np.linspace(0, len(df), nparts + 1).astype(int)
        too_big = False
        for i in range(nparts):
            part = df.iloc[bounds[i]:bounds[i + 1]]
            nm = name if nparts == 1 else f'{name}_part{i + 1}'
            pp, cp = _write_one(part, pdir, cdir, nm)
            written += [pp, cp]
            if max(os.path.getsize(pp), os.path.getsize(cp)) > MAX_FILE:
                too_big = True
        if not too_big:
            return written
        for p in written:
            os.remove(p)
        nparts += 1
        if nparts > 12:
            raise RuntimeError(f'{name}: cannot split under the transfer limit')


def sha256(path):
    h = hashlib.sha256()
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(1 << 20), b''):
            h.update(chunk)
    return h.hexdigest()


def nullify(s, tokens=('NULL', '')):
    """Map textual null tokens to real nulls (no other change)."""
    return s.mask(s.isin(tokens))