Radix-Skills/skills/data-science/erp-data-reconciliation/scripts/excel-analyzer.py

"""
Generic Excel file analyzer for ERP reconciliation projects.

Usage:
    python analyze_excel.py <directory> [--output analysis.json]

This script:
1. Scans a directory for .xlsx files
2. Identifies sheets, headers, data ranges
3. Detects Danish number formats
4. Prints null counts and sample values
5. Saves structured JSON for downstream processing

To customize for a project, adjust:
- EXPECTED_COLS: columns that must exist for a row to be considered a header
- AMOUNT_COL_PATTERNS: regex patterns for amount columns
- DATE_COL_PATTERNS: regex patterns for date columns
"""
import pandas as pd
import openpyxl
import os
import json
import re
from collections import OrderedDict

# --- CUSTOMIZE PER PROJECT ---
EXPECTED_COLS = {'Dato', 'Faktura', 'Bilag', 'Tekst', 'Beløb'}
AMOUNT_COL_PATTERNS = [
    r'(?i)bel[oø]b',
    r'(?i)amount',
    r'(?i)sum',
    r'(?i)rest',
    r'(?i)forfalden',
]
DATE_COL_PATTERNS = [
    r'(?i)dato',
    r'(?i)date',
    r'(?i)posting',
    r'(?i)entry',
]
REFERENCE_COL_PATTERNS = [
    r'(?i)reference',
    r'(?i)faktura',
    r'(?i)bilag',
    r'(?i)document number',
    r'(?i)invoice',
]
# --- END CUSTOMIZE ---


def classify_column(col_name):
    """Guess the semantic type of a column from its name."""
    cn = str(col_name)
    for p in AMOUNT_COL_PATTERNS:
        if re.search(p, cn):
            return 'amount'
    for p in DATE_COL_PATTERNS:
        if re.search(p, cn):
            return 'date'
    for p in REFERENCE_COL_PATTERNS:
        if re.search(p, cn):
            return 'reference'
    return 'other'


def find_header_row(filepath, sheet=0, required_cols=None):
    """Find the first row that looks like a header."""
    if required_cols is None:
        required_cols = EXPECTED_COLS
    wb = openpyxl.load_workbook(filepath, data_only=True)
    ws = wb[wb.sheetnames[sheet]]
    for i, row in enumerate(ws.iter_rows(values_only=True), start=0):
        vals = set(str(v) for v in row if v is not None)
        if required_cols.issubset(vals):
            return i
    return None


def parse_danish_number(val):
    """Normalize Danish text numbers like '-2.543.803,93' to float."""
    if pd.isna(val):
        return float('nan')
    if isinstance(val, (int, float)):
        return float(val)
    s = str(val).strip()
    if not s:
        return float('nan')
    # Danish format: thousands sep = '.', decimal = ','
    if '.' in s and ',' in s:
        s = s.replace('.', '').replace(',', '.')
    elif ',' in s:
        s = s.replace(',', '.')
    try:
        return float(s)
    except ValueError:
        return float('nan')


def analyze_file(filepath):
    info = {
        "filepath": filepath,
        "filename": os.path.basename(filepath),
        "size_kb": round(os.path.getsize(filepath) / 1024, 1),
        "sheets": OrderedDict(),
    }
    xl = pd.ExcelFile(filepath)
    for sheet in xl.sheet_names:
        info["sheets"][sheet] = analyze_sheet(filepath, sheet)
    return info


def analyze_sheet(filepath, sheet):
    # Try default header=0 first; if columns look wrong, try header detection
    df = pd.read_excel(filepath, sheet_name=sheet, engine='openpyxl')
    header_row = None
    # Simple heuristic: if ALL columns are unnamed, try detecting header
    unnamed = [c for c in df.columns if 'Unnamed' in str(c)]
    if len(unnamed) > len(df.columns) * 0.5:
        hr = find_header_row(filepath, sheet)
        if hr is not None:
            df = pd.read_excel(filepath, sheet_name=sheet, engine='openpyxl', header=hr)
            header_row = hr

    cols = OrderedDict()
    for col in df.columns:
        col_type = classify_column(str(col))
        is_date = pd.api.types.is_datetime64_any_dtype(df[col])
        is_num = pd.api.types.is_numeric_dtype(df[col])
        null_count = int(df[col].isna().sum())
        unique_count = int(df[col].nunique())
        samples = []
        for v in df[col].dropna().head(5):
            if len(samples) >= 3:
                break
            try:
                if is_date:
                    samples.append(str(pd.Timestamp(v).date()))
                else:
                    samples.append(str(v)[:60])
            except Exception:
                samples.append(str(v)[:60])

        # Detect danish number format
        danish_flag = False
        if is_num:
            for v in df[col].dropna().head(10):
                if isinstance(v, str) and (',' in v or ('.' in v and ',' in v.replace('.', ''))):
                    danish_flag = True
                    break

        cols[str(col)] = {
            "semantic_type": col_type,
            "pandas_dtype": str(df[col].dtype),
            "null_count": null_count,
            "null_pct": round(null_count / len(df) * 100, 1),
            "unique_count": unique_count,
            "is_date": bool(is_date),
            "is_numeric": bool(is_num),
            "danish_number_format": danish_flag,
            "sample_values": samples,
        }

    empty_rows = int((df.isna().all(axis=1)).sum())
    dupes = int(df.duplicated().sum())

    # Try to find date range
    date_range = None
    for col, meta in cols.items():
        if meta["is_date"]:
            non_null = df[col].dropna()
            if len(non_null) > 0:
                date_range = {
                    "min": str(non_null.min().date()),
                    "max": str(non_null.max().date()),
                    "column": col,
                }
                break

    return {
        "rows": len(df),
        "columns": len(df.columns),
        "detected_header_row": header_row,
        "empty_rows": empty_rows,
        "duplicate_rows": dupes,
        "date_range": date_range,
        "columns": cols,
    }


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Analyze Excel files for data reconciliation")
    parser.add_argument("directory", help="Directory containing .xlsx files")
    parser.add_argument("--output", default="excel_analysis.json", help="Output JSON file")
    args = parser.parse_args()

    results = OrderedDict()
    for f in sorted(os.listdir(args.directory)):
        if f.endswith('.xlsx') and not f.startswith('~'):
            fp = os.path.join(args.directory, f)
            key = os.path.splitext(f)[0]
            results[key] = analyze_file(fp)
            print(f"Analyzed: {f}")

    with open(args.output, 'w', encoding='utf-8') as fh:
        json.dump(results, fh, indent=2, ensure_ascii=False)
    print(f"\nSaved: {args.output}")


if __name__ == "__main__":
    main()