Spreadsheets

policy:
  allow_implicit_invocation: true

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# ///
"""Inspect or validate a delimited text file (CSV/TSV) and print a JSON report.

Reports encoding, BOM, newline style, delimiter, header, shape, ragged and
empty rows, `-` null usage, issues, and sample rows. Read-only.

Usage:
  uv run scripts/peek.py <file> [--rows N] [--strict] [--expect-like REPORT]
  uv run scripts/peek.py <file> --house --redact-samples
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import re
import shutil
import subprocess
import sys
from collections import Counter
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Iterator, NoReturn

SNIFF_BYTES = 64 * 1024
CANDIDATE_DELIMITERS = ",\t;|"
BINARY_SUFFIXES = {".xlsx", ".xlsm", ".xls", ".ods", ".numbers", ".parquet"}
CELL_PREVIEW_LIMIT = 120
NEWLINE_CHUNK = 1024 * 1024
HOUSE_HEADER_RE = re.compile(r"^[a-z][a-z0-9_]*$")


def fail(message: str, hint: str | None = None) -> NoReturn:
    payload: dict[str, str] = {"error": message}
    if hint:
        payload["hint"] = hint
    print(json.dumps(payload, indent=2))
    sys.exit(2)


def decode_sample(raw: bytes) -> tuple[str, str, str | None, int]:
    """Return (text, encoding label, BOM label, bytes to skip while streaming)."""
    if raw.startswith(b"\xef\xbb\xbf"):
        return raw[3:].decode("utf-8", errors="replace"), "utf-8", "utf-8", 3
    if raw.startswith(b"\xff\xfe"):
        return raw[2:].decode("utf-16-le", errors="replace"), "utf-16-le", "utf-16-le", 2
    if raw.startswith(b"\xfe\xff"):
        return raw[2:].decode("utf-16-be", errors="replace"), "utf-16-be", "utf-16-be", 2
    try:
        return raw.decode("utf-8"), "utf-8", None, 0
    except UnicodeDecodeError:
        return raw.decode("latin-1"), "unknown-8bit (decoded as latin-1)", None, 0


def stream_codec(encoding_label: str) -> str:
    if encoding_label.startswith("unknown-8bit"):
        return "latin-1"
    return encoding_label


@contextmanager
def open_text(path: Path, encoding: str, skip_bytes: int = 0) -> Iterator[io.TextIOWrapper]:
    raw = path.open("rb")
    try:
        if skip_bytes:
            raw.seek(skip_bytes)
        with io.TextIOWrapper(raw, encoding=encoding, errors="replace", newline="") as text:
            yield text
    finally:
        if not raw.closed:
            raw.close()


def summarize_newlines(crlf: int, lf: int, cr: int, trailing_newline: bool) -> dict[str, Any]:
    present = [name for name, count in (("crlf", crlf), ("lf", lf), ("cr", cr)) if count]
    style = present[0] if len(present) == 1 else ("mixed" if present else "none")
    return {
        "style": style,
        "counts": {"crlf": crlf, "lf": lf, "cr": cr},
        "trailing_newline": trailing_newline,
    }


def newline_report(path: Path, encoding: str, skip_bytes: int) -> dict[str, Any]:
    crlf = 0
    lf = 0
    cr = 0
    pending_cr = False
    last_char: str | None = None

    with open_text(path, encoding, skip_bytes) as handle:
        while True:
            chunk = handle.read(NEWLINE_CHUNK)
            if chunk == "":
                break
            last_char = chunk[-1]
            if pending_cr:
                chunk = "\r" + chunk
                pending_cr = False
            if chunk.endswith("\r"):
                pending_cr = True
                chunk = chunk[:-1]

            crlf += chunk.count("\r\n")
            without_crlf = chunk.replace("\r\n", "")
            lf += without_crlf.count("\n")
            cr += without_crlf.count("\r")

    if pending_cr:
        cr += 1

    return summarize_newlines(crlf, lf, cr, last_char in ("\n", "\r"))


def pick_delimiter(path: Path, text: str) -> tuple[str, str]:
    if path.suffix.lower() in {".tsv", ".tab"}:
        return "\t", "extension"
    try:
        dialect = csv.Sniffer().sniff(text[:SNIFF_BYTES], delimiters=CANDIDATE_DELIMITERS)
        return dialect.delimiter, "sniffed"
    except csv.Error:
        first_line = text.splitlines()[0] if text else ""
        counts = {candidate: first_line.count(candidate) for candidate in CANDIDATE_DELIMITERS}
        best = max(counts, key=lambda candidate: counts[candidate])
        if counts[best] > 0:
            return best, "counted (most frequent in first line)"
        return ",", "fallback (no delimiter found)"


def preview(cell: str, redact: bool) -> str:
    if redact and cell not in ("", "-"):
        return "<redacted>"
    return cell if len(cell) <= CELL_PREVIEW_LIMIT else cell[:CELL_PREVIEW_LIMIT] + "..."


def issue(code: str, message: str, **details: Any) -> dict[str, Any]:
    payload: dict[str, Any] = {"code": code, "message": message}
    payload.update(details)
    return payload


def delimiter_label(delimiter: str) -> str:
    return "\\t" if delimiter == "\t" else delimiter


def structural_issues(report: dict[str, Any]) -> list[dict[str, Any]]:
    issues: list[dict[str, Any]] = []
    if report["encoding"] != "utf-8":
        issues.append(issue("non_utf8_encoding", "file encoding is not UTF-8", encoding=report["encoding"]))
    if report["bom"] is not None:
        issues.append(issue("bom_present", "file has a BOM", bom=report["bom"]))
    if report["newlines"]["style"] not in ("lf", "none"):
        issues.append(
            issue(
                "newline_style_not_lf",
                "newline style is not LF",
                style=report["newlines"]["style"],
                counts=report["newlines"]["counts"],
            )
        )
    if report["size_bytes"] > 0 and not report["newlines"]["trailing_newline"]:
        issues.append(issue("missing_trailing_newline", "file is missing a trailing newline"))
    if report["duplicate_headers"]:
        issues.append(
            issue(
                "duplicate_headers",
                "duplicate header names",
                headers=report["duplicate_headers"],
            )
        )
    if report["ragged_rows"]["count"]:
        issues.append(
            issue(
                "ragged_rows",
                "rows with column counts different from the header",
                count=report["ragged_rows"]["count"],
                first=report["ragged_rows"]["first"],
            )
        )
    if report["empty_rows"]:
        issues.append(issue("empty_rows", "empty data rows", count=report["empty_rows"]))
    validation = report.get("qsv_validation")
    if validation and validation.get("available") and "ok" in validation and not validation.get("ok"):
        issues.append(
            issue(
                "qsv_validation_failed",
                "qsv could not validate the file as RFC 4180-compatible CSV",
                detail=validation.get("stderr") or validation.get("stdout"),
            )
        )
    return issues


def house_issues(report: dict[str, Any]) -> list[dict[str, Any]]:
    issues: list[dict[str, Any]] = []
    if report["delimiter"] != "\t":
        issues.append(
            issue(
                "house_delimiter_not_tsv",
                "authored spreadsheet files should be TSV",
                delimiter=delimiter_label(report["delimiter"]),
            )
        )
    unsafe_headers = [header for header in report["header"] if not HOUSE_HEADER_RE.fullmatch(header)]
    if unsafe_headers:
        issues.append(
            issue(
                "house_headers_not_snake_case",
                "headers are not lowercase snake_case",
                headers=unsafe_headers,
            )
        )
    return issues


def load_expected_report(path: Path) -> dict[str, Any]:
    try:
        with path.open(encoding="utf-8") as handle:
            payload = json.load(handle)
    except OSError as exc:
        fail(f"could not read expected report {path}: {exc}")
    except json.JSONDecodeError as exc:
        fail(f"{path} is not valid JSON: {exc}")
    if not isinstance(payload, dict):
        fail(f"{path} is not a peek.py JSON object")
    return payload


def nested(payload: dict[str, Any], *keys: str) -> Any:
    value: Any = payload
    for key in keys:
        if not isinstance(value, dict):
            return None
        value = value.get(key)
    return value


def expected_count(expected: dict[str, Any], *keys: str) -> int:
    value = nested(expected, *keys)
    if value is None:
        return 0
    if type(value) is int:
        return value
    fail(f"expected report has non-integer {'.'.join(keys)}: {value!r}")


def compare_expected_like(report: dict[str, Any], expected: dict[str, Any]) -> list[dict[str, Any]]:
    issues: list[dict[str, Any]] = []

    comparisons = [
        (("columns",), "columns_changed", "column count changed"),
        (("header",), "header_changed", "header changed"),
        (("delimiter",), "delimiter_changed", "delimiter changed"),
        (("encoding",), "encoding_changed", "encoding changed"),
        (("newlines", "style"), "newline_style_changed", "newline style changed"),
        (("newlines", "trailing_newline"), "trailing_newline_changed", "trailing newline changed"),
        (("data_rows",), "data_rows_changed", "data row count changed"),
    ]
    for keys, code, message in comparisons:
        before = nested(expected, *keys)
        after = nested(report, *keys)
        if before != after:
            if keys == ("delimiter",):
                before = delimiter_label(str(before))
                after = delimiter_label(str(after))
            issues.append(issue(code, message, before=before, after=after))

    before_ragged = expected_count(expected, "ragged_rows", "count")
    after_ragged = int(nested(report, "ragged_rows", "count") or 0)
    if after_ragged > before_ragged:
        issues.append(issue("new_ragged_rows", "new ragged rows", before=before_ragged, after=after_ragged))

    before_empty = expected_count(expected, "empty_rows")
    after_empty = int(report.get("empty_rows") or 0)
    if after_empty > before_empty:
        issues.append(issue("new_empty_rows", "new empty rows", before=before_empty, after=after_empty))

    before_duplicates = set(expected.get("duplicate_headers") or [])
    after_duplicates = set(report.get("duplicate_headers") or [])
    new_duplicates = sorted(after_duplicates - before_duplicates)
    if new_duplicates:
        issues.append(issue("new_duplicate_headers", "new duplicate headers", headers=new_duplicates))

    if expected.get("bom") is None and report.get("bom") is not None:
        issues.append(issue("new_bom", "new BOM introduced", bom=report["bom"]))

    return issues


def parse_rows_python(
    path: Path,
    encoding: str,
    skip_bytes: int,
    delimiter: str,
    sample_rows: int,
    redact_samples: bool,
) -> dict[str, Any]:
    with open_text(path, encoding, skip_bytes) as handle:
        reader = csv.reader(handle, delimiter=delimiter)
        try:
            header = next(reader)
        except StopIteration:
            fail(f"{path.name} is empty")
        except csv.Error as exc:
            fail(f"{path.name} is not parseable as delimited text: {exc}")

        width = len(header)
        duplicate_headers = sorted(name for name, count in Counter(header).items() if count > 1)

        ragged_count = 0
        ragged_first: list[dict[str, int]] = []
        empty = 0
        dash_nulls = 0
        data_rows = 0
        sample: list[list[str]] = []

        try:
            for record_number, record in enumerate(reader, start=2):  # header is record 1
                data_rows += 1
                if len(sample) < sample_rows:
                    sample.append([preview(cell, redact_samples) for cell in record])
                if not record or all(cell.strip() == "" for cell in record):
                    empty += 1
                    continue
                if len(record) != width:
                    ragged_count += 1
                    if len(ragged_first) < 10:
                        ragged_first.append({"record": record_number, "columns": len(record)})
                dash_nulls += sum(1 for cell in record if cell == "-")
        except csv.Error as exc:
            fail(f"{path.name} is not parseable as delimited text: {exc}")

    return {
        "columns": width,
        "header": header,
        "duplicate_headers": duplicate_headers,
        "data_rows": data_rows,
        "empty_rows": empty,
        "ragged_rows": {"count": ragged_count, "first": ragged_first},
        "dash_null_cells": dash_nulls,
        "sample": sample,
    }


def strip_record_ending(raw_line: bytes) -> tuple[bytes, str | None]:
    if raw_line.endswith(b"\r\n"):
        return raw_line[:-2], "crlf"
    if raw_line.endswith(b"\n"):
        return raw_line[:-1], "lf"
    if raw_line.endswith(b"\r"):
        return raw_line[:-1], "cr"
    return raw_line, None


def decode_fields(raw_fields: list[bytes]) -> list[str] | None:
    try:
        return [field.decode("utf-8") for field in raw_fields]
    except UnicodeDecodeError:
        return None


def parse_rows_unquoted_fast(
    path: Path,
    skip_bytes: int,
    delimiter: str,
    sample_rows: int,
    redact_samples: bool,
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    """Fast parser for simple delimited files without quotes or embedded newlines."""
    delimiter_bytes = delimiter.encode("utf-8")
    if len(delimiter_bytes) != 1:
        return None

    crlf = 0
    lf = 0
    cr = 0
    trailing_newline = False
    header: list[str] | None = None
    width = 0
    duplicate_headers: list[str] = []
    ragged_count = 0
    ragged_first: list[dict[str, int]] = []
    empty = 0
    dash_nulls = 0
    data_rows = 0
    sample: list[list[str]] = []

    with path.open("rb") as handle:
        if skip_bytes:
            handle.seek(skip_bytes)
        for raw_record_number, raw_line in enumerate(handle, start=1):
            record_bytes, ending = strip_record_ending(raw_line)
            trailing_newline = ending is not None
            if ending == "crlf":
                crlf += 1
            elif ending == "lf":
                lf += 1
            elif ending == "cr":
                cr += 1
            if b'"' in record_bytes or b"\r" in record_bytes:
                return None

            if raw_record_number == 1:
                header = decode_fields(record_bytes.split(delimiter_bytes))
                if header is None:
                    return None
                width = len(header)
                duplicate_headers = sorted(name for name, count in Counter(header).items() if count > 1)
                continue

            data_rows += 1
            if record_bytes == b"":
                empty += 1
                if len(sample) < sample_rows:
                    sample.append([])
                continue

            record = decode_fields(record_bytes.split(delimiter_bytes))
            if record is None:
                return None
            if len(sample) < sample_rows:
                sample.append([preview(cell, redact_samples) for cell in record])
            if all(cell.strip() == "" for cell in record):
                empty += 1
                continue
            if len(record) != width:
                ragged_count += 1
                if len(ragged_first) < 10:
                    ragged_first.append({"record": raw_record_number, "columns": len(record)})
            dash_nulls += sum(1 for cell in record if cell == "-")

    if header is None:
        fail(f"{path.name} is empty")

    row_report = {
        "columns": width,
        "header": header,
        "duplicate_headers": duplicate_headers,
        "data_rows": data_rows,
        "empty_rows": empty,
        "ragged_rows": {"count": ragged_count, "first": ragged_first},
        "dash_null_cells": dash_nulls,
        "sample": sample,
    }
    return row_report, summarize_newlines(crlf, lf, cr, trailing_newline)


def qsv_validation(path: Path, delimiter: str, encoding: str) -> dict[str, Any]:
    qsv = shutil.which("qsv")
    if qsv is None:
        return {"available": False}
    if encoding != "utf-8":
        return {"available": True, "skipped": "non-utf8"}
    try:
        result = subprocess.run(
            [qsv, "validate", "-d", delimiter, str(path)],
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"available": True, "ok": False, "error": str(exc)}
    return {
        "available": True,
        "ok": result.returncode == 0,
        "returncode": result.returncode,
        "stdout": result.stdout.strip(),
        "stderr": result.stderr.strip(),
    }


def inspect_path(
    path: Path,
    *,
    rows: int = 5,
    strict: bool = False,
    expect_like: Path | None = None,
    expect_columns: int | None = None,
    engine: str = "auto",
    house: bool = False,
    redact_samples: bool = False,
) -> tuple[dict[str, Any], bool]:
    if rows < 0:
        fail("--rows must be >= 0")
    if expect_columns is not None and expect_columns < 1:
        fail("--expect-columns must be >= 1")
    if not path.is_file():
        fail(f"{path} is not a file")

    size = path.stat().st_size
    with path.open("rb") as handle:
        raw = handle.read(SNIFF_BYTES)

    if path.suffix.lower() in BINARY_SUFFIXES or raw[:4] == b"PK\x03\x04":
        fail(
            f"{path.name} is a binary spreadsheet, not delimited text",
            "follow references/xlsx.md (openpyxl, qsv excel, or DuckDB read_xlsx) instead of peek.py",
        )
    if b"\x00" in raw[:SNIFF_BYTES] and not raw.startswith((b"\xff\xfe", b"\xfe\xff")):
        fail(f"{path.name} looks binary (NUL bytes found)")

    text, encoding, bom, skip_bytes = decode_sample(raw)
    codec = stream_codec(encoding)
    delimiter, delimiter_source = pick_delimiter(path, text)

    validation = qsv_validation(path, delimiter, encoding) if engine == "auto" else {"available": False}
    engine_used = "python-csv"
    row_report: dict[str, Any]
    newlines: dict[str, Any]

    fast_report = None
    if engine == "auto" and encoding == "utf-8":
        fast_report = parse_rows_unquoted_fast(path, skip_bytes, delimiter, rows, redact_samples)
    if fast_report is not None:
        row_report, newlines = fast_report
        engine_used = "python-fast-unquoted"
    else:
        row_report = parse_rows_python(path, codec, skip_bytes, delimiter, rows, redact_samples)
        newlines = newline_report(path, codec, skip_bytes)

    report: dict[str, Any] = {
        "file": str(path),
        "size_bytes": size,
        "analysis_truncated": False,
        "engine_requested": engine,
        "engine": engine_used,
        "encoding": encoding,
        "bom": bom,
        "newlines": newlines,
        "delimiter": delimiter,
        "delimiter_source": delimiter_source,
        "qsv_validation": validation,
    }
    report.update(row_report)

    reported_issues = structural_issues(report)
    if house:
        reported_issues.extend(house_issues(report))
    should_fail = (strict or house) and bool(reported_issues)

    if expect_columns is not None and report["columns"] != expect_columns:
        reported_issues.append(
            issue(
                "expected_columns_mismatch",
                "column count does not match --expect-columns",
                expected=expect_columns,
                actual=report["columns"],
            )
        )
        should_fail = True

    if expect_like:
        expected = load_expected_report(expect_like)
        expected_issues = compare_expected_like(report, expected)
        reported_issues.extend(expected_issues)
        if expected_issues:
            should_fail = True

    report["status"] = "issues_found" if reported_issues else "ok"
    report["issues"] = reported_issues
    return report, should_fail


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", type=Path)
    parser.add_argument("--rows", type=int, default=5, help="sample rows to include (default: 5)")
    parser.add_argument("--strict", action="store_true", help="exit 1 on structural issues")
    parser.add_argument("--house", action="store_true", help="also enforce authored TSV house conventions")
    parser.add_argument("--redact-samples", action="store_true", help="redact non-null sample cell values")
    parser.add_argument("--expect-like", type=Path, help="exit 1 if shape/format drift from a previous peek report")
    parser.add_argument("--expect-columns", type=int, help="exit 1 unless the file has this many columns")
    parser.add_argument(
        "--engine",
        choices=("auto", "python"),
        default="auto",
        help="auto uses the fast unquoted parser and qsv validation when safe; python forces csv.reader",
    )
    args = parser.parse_args()

    report, should_fail = inspect_path(
        args.file,
        rows=args.rows,
        strict=args.strict,
        expect_like=args.expect_like,
        expect_columns=args.expect_columns,
        engine=args.engine,
        house=args.house,
        redact_samples=args.redact_samples,
    )
    print(json.dumps(report, indent=2, ensure_ascii=False))
    if should_fail:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# ///
"""Profile a delimited spreadsheet file and print JSON or Markdown.

Read-only. Uses peek.py for structure, qsv for fast column statistics when
available, and a local scan for spreadsheet formula-injection risks.

Usage:
  uv run scripts/profile.py data.tsv
  uv run scripts/profile.py data.tsv --markdown --redact-samples
"""

from __future__ import annotations

import argparse
import csv
import json
import shutil
import subprocess
import sys
from collections import Counter
from pathlib import Path
from typing import Any

import peek

FORMULA_PREFIXES = ("=", "+", "@")
MAX_FINDINGS = 20
XLSX_SUFFIXES = {".xlsx", ".xlsm", ".xls", ".xlsb", ".ods"}


def run_command(args: list[str], timeout: int = 60) -> dict[str, Any]:
    try:
        result = subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"ok": False, "error": str(exc), "args": args}
    return {
        "ok": result.returncode == 0,
        "returncode": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "args": args,
    }


def tool_version(binary: str) -> dict[str, Any]:
    path = shutil.which(binary)
    if path is None:
        return {"available": False}
    result = run_command([path, "--version"], timeout=10)
    version = (result.get("stdout") or result.get("stderr") or "").strip().splitlines()
    return {"available": True, "path": path, "version": version[0] if version else None}


def qsv_args(command: str, delimiter: str, file: Path, extra: list[str] | None = None) -> list[str] | None:
    qsv = shutil.which("qsv")
    if qsv is None:
        return None
    args = [qsv, command]
    if extra:
        args.extend(extra)
    args.extend(["-d", delimiter, str(file)])
    return args


def parse_csv_stdout(stdout: str) -> list[dict[str, str]]:
    return list(csv.DictReader(stdout.splitlines()))


def collect_stats(file: Path, delimiter: str) -> dict[str, Any]:
    args = qsv_args("stats", delimiter, file, ["--cache-threshold", "0", "--cardinality"])
    if args is None:
        return {"available": False}
    result = run_command(args)
    if not result["ok"]:
        return {
            "available": True,
            "ok": False,
            "stderr": result.get("stderr", "").strip(),
            "stdout": result.get("stdout", "").strip(),
        }
    rows = parse_csv_stdout(result["stdout"])
    columns = []
    for row in rows:
        columns.append(
            {
                "field": row.get("field", ""),
                "type": row.get("type", ""),
                "nullcount": parse_number(row.get("nullcount", "")),
                "sparsity": parse_number(row.get("sparsity", "")),
                "cardinality": parse_number(row.get("cardinality", "")),
                "uniqueness_ratio": parse_number(row.get("uniqueness_ratio", "")),
                "min": row.get("min", ""),
                "max": row.get("max", ""),
                "max_precision": row.get("max_precision", ""),
            }
        )
    return {"available": True, "ok": True, "columns": columns}


def collect_frequency(file: Path, delimiter: str, top: int, redact: bool) -> dict[str, Any]:
    args = qsv_args("frequency", delimiter, file, ["--limit", str(top), "--json", "--no-stats"])
    if args is None:
        return {"available": False}
    result = run_command(args)
    if not result["ok"]:
        return {
            "available": True,
            "ok": False,
            "stderr": result.get("stderr", "").strip(),
            "stdout": result.get("stdout", "").strip(),
        }
    try:
        payload = json.loads(result["stdout"])
    except json.JSONDecodeError as exc:
        return {"available": True, "ok": False, "error": str(exc)}
    if redact:
        for field in payload.get("fields", []):
            for frequency in field.get("frequencies", []):
                value = frequency.get("value")
                if value not in ("", "-", "<ALL_UNIQUE>", "HIGH_CARDINALITY"):
                    frequency["value"] = "<redacted>"
    return {"available": True, "ok": True, "report": payload}


def parse_number(value: str) -> int | float | str | None:
    if value == "":
        return None
    try:
        if "." in value:
            return float(value)
        return int(value)
    except ValueError:
        return value


def scan_formula_injection(
    file: Path,
    delimiter: str,
    encoding: str,
    header: list[str],
    redact: bool,
) -> dict[str, Any]:
    codec = peek.stream_codec(encoding)
    findings: list[dict[str, Any]] = []
    count = 0
    try:
        with peek.open_text(file, codec) as handle:
            reader = csv.reader(handle, delimiter=delimiter)
            next(reader, None)
            for row_number, row in enumerate(reader, start=2):
                for index, cell in enumerate(row):
                    if cell.startswith(FORMULA_PREFIXES):
                        count += 1
                        if len(findings) < MAX_FINDINGS:
                            findings.append(
                                {
                                    "record": row_number,
                                    "column": header[index] if index < len(header) else str(index + 1),
                                    "value": "<redacted>" if redact else peek.preview(cell, False),
                                }
                            )
    except csv.Error as exc:
        return {"count": count, "first": findings, "error": str(exc)}
    return {"count": count, "first": findings}


def header_quality(header: list[str]) -> dict[str, Any]:
    duplicate_headers = sorted(name for name, count in Counter(header).items() if count > 1)
    unsafe = [name for name in header if not peek.HOUSE_HEADER_RE.fullmatch(name)]
    empty = [index + 1 for index, name in enumerate(header) if name == ""]
    return {"duplicates": duplicate_headers, "unsafe_house_headers": unsafe, "empty_header_positions": empty}


def recommendations(report: dict[str, Any]) -> list[str]:
    recs: list[str] = []
    peek_report = report.get("peek", {})
    if peek_report.get("issues"):
        recs.append("Fix structural issues before transforming or aggregating.")
    if not report["tools"]["qsv"]["available"]:
        recs.append("Install qsv for faster validation, profiling, frequency, and Excel export workflows.")
    if not report["tools"]["duckdb"]["available"]:
        recs.append("Install DuckDB for SQL joins, pivots, and exact DECIMAL aggregations.")
    formula_count = report.get("formula_injection", {}).get("count", 0)
    if formula_count:
        recs.append("Escape external cells starting with =, +, or @ before writing to CSV/XLSX.")
    high_cardinality = [
        col["field"]
        for col in report.get("stats", {}).get("columns", [])
        if isinstance(col.get("uniqueness_ratio"), (float, int)) and col["uniqueness_ratio"] >= 0.95
    ]
    if high_cardinality:
        recs.append("Treat high-uniqueness columns as identifiers; avoid full frequency tables on them.")
    if not recs:
        recs.append("File is structurally clean; proceed with qsv for simple operations or DuckDB for SQL.")
    return recs


def profile_delimited(args: argparse.Namespace) -> dict[str, Any]:
    peek_report, _should_fail = peek.inspect_path(
        args.file,
        rows=args.rows,
        strict=False,
        engine=args.engine,
        house=args.house,
        redact_samples=args.redact_samples,
    )
    delimiter = peek_report["delimiter"]
    report: dict[str, Any] = {
        "file": str(args.file),
        "kind": "delimited",
        "tools": {"qsv": tool_version("qsv"), "duckdb": tool_version("duckdb")},
        "peek": peek_report,
        "header_quality": header_quality(peek_report["header"]),
        "stats": collect_stats(args.file, delimiter),
        "frequency": collect_frequency(args.file, delimiter, args.top, args.redact_samples),
        "formula_injection": scan_formula_injection(
            args.file,
            delimiter,
            peek_report["encoding"],
            peek_report["header"],
            args.redact_samples,
        ),
    }
    report["recommendations"] = recommendations(report)
    report["status"] = "issues_found" if peek_report["issues"] or report["formula_injection"]["count"] else "ok"
    return report


def profile_workbook(args: argparse.Namespace) -> dict[str, Any]:
    qsv = shutil.which("qsv")
    metadata: dict[str, Any]
    if qsv is None:
        metadata = {"available": False}
    else:
        result = run_command([qsv, "excel", "--metadata", "J", str(args.file)])
        if result["ok"]:
            try:
                metadata = {"available": True, "ok": True, "report": json.loads(result["stdout"])}
            except json.JSONDecodeError as exc:
                metadata = {"available": True, "ok": False, "error": str(exc)}
        else:
            metadata = {
                "available": True,
                "ok": False,
                "stdout": result.get("stdout", "").strip(),
                "stderr": result.get("stderr", "").strip(),
            }
    return {
        "file": str(args.file),
        "kind": "workbook",
        "status": "ok" if metadata.get("ok") else "needs_manual_workflow",
        "tools": {"qsv": tool_version("qsv"), "duckdb": tool_version("duckdb")},
        "metadata": metadata,
        "recommendations": [
            "Use qsv excel for fast values-only export or sheet metadata.",
            "Use DuckDB read_xlsx(all_varchar=true) for SQL over .xlsx values.",
            "Use openpyxl only for editing existing workbook structure or formulas.",
            "Run recalc.py after writing formulas and treat non-success as a failed delivery.",
        ],
    }


def markdown_table_row(values: list[Any]) -> str:
    escaped = [str(value if value is not None else "").replace("|", "\\|") for value in values]
    return "| " + " | ".join(escaped) + " |"


def render_markdown(report: dict[str, Any]) -> str:
    lines = ["# Spreadsheet Profile", ""]
    lines.append(f"- File: `{report['file']}`")
    lines.append(f"- Kind: `{report['kind']}`")
    lines.append(f"- Status: `{report['status']}`")
    if report["kind"] == "delimited":
        peek_report = report["peek"]
        lines.append(f"- Shape: {peek_report['data_rows']} rows x {peek_report['columns']} columns")
        lines.append(f"- Delimiter: `{peek.delimiter_label(peek_report['delimiter'])}`")
        lines.append("")
        lines.append("## Issues")
        if peek_report["issues"]:
            lines.extend(f"- `{item['code']}`: {item['message']}" for item in peek_report["issues"])
        else:
            lines.append("- None")
        if report["formula_injection"]["count"]:
            lines.append(f"- `formula_injection`: {report['formula_injection']['count']} risky cells")
        lines.append("")
        lines.append("## Columns")
        lines.append(markdown_table_row(["field", "type", "nulls", "cardinality", "unique"]))
        lines.append(markdown_table_row(["---", "---", "---", "---", "---"]))
        for column in report.get("stats", {}).get("columns", [])[:30]:
            lines.append(
                markdown_table_row(
                    [
                        column.get("field"),
                        column.get("type"),
                        column.get("nullcount"),
                        column.get("cardinality"),
                        column.get("uniqueness_ratio"),
                    ]
                )
            )
    else:
        lines.append("")
        lines.append("## Workbook Metadata")
        metadata = report["metadata"]
        if metadata.get("ok"):
            payload = metadata.get("report", {})
            lines.append(f"- Sheets: {payload.get('sheet_count', payload.get('number_of_sheets', 'unknown'))}")
        else:
            lines.append(f"- Metadata unavailable: {metadata.get('stderr') or metadata.get('error') or 'qsv missing'}")
    lines.append("")
    lines.append("## Recommendations")
    lines.extend(f"- {item}" for item in report["recommendations"])
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", type=Path)
    parser.add_argument("--rows", type=int, default=5, help="sample rows to include from peek.py")
    parser.add_argument("--top", type=int, default=5, help="top values per column for qsv frequency")
    parser.add_argument("--house", action="store_true", help="also enforce authored TSV house conventions")
    parser.add_argument("--redact-samples", action="store_true", help="redact sample and frequency values")
    parser.add_argument("--markdown", action="store_true", help="print Markdown instead of JSON")
    parser.add_argument("--engine", choices=("auto", "python"), default="auto", help="peek.py engine")
    args = parser.parse_args()

    if args.top < 1:
        peek.fail("--top must be >= 1")
    if not args.file.is_file():
        peek.fail(f"{args.file} is not a file")

    suffix = args.file.suffix.lower()
    report = profile_workbook(args) if suffix in XLSX_SUFFIXES else profile_delimited(args)
    if args.markdown:
        print(render_markdown(report), end="")
    else:
        print(json.dumps(report, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["openpyxl>=3.1"]
# ///
"""Recalculate formulas in an .xlsx/.xlsm with headless LibreOffice, then audit it.

Saves the workbook in place and prints a JSON report whose `status` is one of:
success | errors_found | recalc_incomplete | error.

By default, every non-success status exits nonzero. Use --soft to preserve the
old report-only behavior for formula errors and incomplete recalculation.

Usage: uv run scripts/recalc.py <workbook> [timeout-seconds] [--soft]
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
import tempfile
from collections import defaultdict
from pathlib import Path
from typing import Any, NoReturn

EXCEL_ERRORS = {"#VALUE!", "#DIV/0!", "#REF!", "#NAME?", "#NULL!", "#NUM!", "#N/A", "#SPILL!", "#CALC!"}
LIBREOFFICE_ERROR = re.compile(r"Err:\d{3}")
MAX_CELLS_PER_ERROR = 20
DEFAULT_TIMEOUT = 60

MACRO_SUB = "RecalcAndSaveClose"
MACRO_XML = f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE script:module PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "module.dtd">
<script:module xmlns:script="http://openoffice.org/2000/script" script:name="Module1" script:language="StarBasic">
Sub {MACRO_SUB}()
    ThisComponent.calculateAll()
    ThisComponent.store()
    ThisComponent.close(True)
End Sub
</script:module>
"""
SCRIPT_XLC = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE library:libraries PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "libraries.dtd">
<library:libraries xmlns:library="http://openoffice.org/2000/library" xmlns:xlink="http://www.w3.org/1999/xlink">
 <library:library library:name="Standard" library:link="false"/>
</library:libraries>
"""
DIALOG_XLC = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE library:libraries PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "libraries.dtd">
<library:libraries xmlns:library="http://openoffice.org/2000/library" xmlns:xlink="http://www.w3.org/1999/xlink"/>
"""
SCRIPT_XLB = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE library:library PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "library.dtd">
<library:library xmlns:library="http://openoffice.org/2000/library" library:name="Standard" library:readonly="false" library:passwordprotected="false">
 <library:element library:name="Module1"/>
</library:library>
"""
DIALOG_XLB = """<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE library:library PUBLIC "-//OpenOffice.org//DTD OfficeDocument 1.0//EN" "library.dtd">
<library:library xmlns:library="http://openoffice.org/2000/library" library:name="Standard" library:readonly="false" library:passwordprotected="false"/>
"""


def report(payload: dict[str, Any], code: int = 0) -> NoReturn:
    print(json.dumps(payload, indent=2))
    sys.exit(code)


def find_soffice() -> str | None:
    on_path = shutil.which("soffice")
    if on_path:
        return on_path
    for candidate in (
        Path("/Applications/LibreOffice.app/Contents/MacOS/soffice"),
        Path.home() / "Applications/LibreOffice.app/Contents/MacOS/soffice",
    ):
        if candidate.is_file():
            return str(candidate)
    return None


def run_soffice(args: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)


def profile_arg(profile_root: Path) -> str:
    return f"-env:UserInstallation={profile_root.resolve().as_uri()}"


def write_profile_macro(profile_root: Path) -> None:
    basic = profile_root / "user/basic"
    standard = basic / "Standard"
    standard.mkdir(parents=True, exist_ok=True)
    (basic / "script.xlc").write_text(SCRIPT_XLC, encoding="utf-8")
    (basic / "dialog.xlc").write_text(DIALOG_XLC, encoding="utf-8")
    (standard / "script.xlb").write_text(SCRIPT_XLB, encoding="utf-8")
    (standard / "dialog.xlb").write_text(DIALOG_XLB, encoding="utf-8")
    (standard / "Module1.xba").write_text(MACRO_XML, encoding="utf-8")


def ensure_macro(soffice: str, profile_root: Path) -> None:
    profile_root.mkdir(parents=True, exist_ok=True)
    run_soffice(
        [soffice, profile_arg(profile_root), "--headless", "--norestore", "--terminate_after_init"],
        timeout=60,
    )
    write_profile_macro(profile_root)


def recalculate(soffice: str, profile_root: Path, workbook: Path, timeout: int) -> subprocess.CompletedProcess[str]:
    uri = f"vnd.sun.star.script:Standard.Module1.{MACRO_SUB}?language=Basic&location=application"
    return run_soffice(
        [
            soffice,
            profile_arg(profile_root),
            "--headless",
            "--norestore",
            "--nolockcheck",
            uri,
            str(workbook),
        ],
        timeout=timeout,
    )


def audit(workbook: Path) -> dict[str, Any]:
    from openpyxl import load_workbook

    with_values = load_workbook(workbook, data_only=True)
    try:
        with_formulas = load_workbook(workbook, data_only=False)
    except Exception:
        with_values.close()
        raise

    try:
        errors: dict[str, list[str]] = defaultdict(list)
        total_errors = 0
        total_formulas = 0
        uncached = 0

        for name in with_formulas.sheetnames:
            formula_ws = with_formulas[name]
            value_ws = with_values[name]
            for row in formula_ws.iter_rows():
                for cell in row:
                    if isinstance(cell.value, str) and cell.value.startswith("="):
                        total_formulas += 1
                        if value_ws[cell.coordinate].value is None:
                            uncached += 1
            for row in value_ws.iter_rows():
                for cell in row:
                    if isinstance(cell.value, str):
                        token = cell.value.strip()
                        if token in EXCEL_ERRORS or LIBREOFFICE_ERROR.fullmatch(token):
                            errors[token].append(f"{name}!{cell.coordinate}")
                            total_errors += 1
    finally:
        with_values.close()
        with_formulas.close()

    if total_errors:
        status = "errors_found"
    elif total_formulas and uncached == total_formulas:
        status = "recalc_incomplete"
    else:
        status = "success"

    return {
        "file": str(workbook),
        "status": status,
        "total_formulas": total_formulas,
        "uncached_formulas": uncached,
        "total_errors": total_errors,
        "errors": {
            kind: {"count": len(cells), "cells": cells[:MAX_CELLS_PER_ERROR]}
            for kind, cells in sorted(errors.items())
        },
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("workbook", type=Path)
    parser.add_argument("timeout_seconds", nargs="?", type=int, default=DEFAULT_TIMEOUT)
    parser.add_argument("--soft", action="store_true", help="exit 0 after formula errors; still fails operational errors")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    workbook = args.workbook.resolve()
    timeout = args.timeout_seconds
    if timeout < 1:
        report({"status": "error", "error": "timeout_seconds must be >= 1"}, code=1)
    if not workbook.is_file():
        report({"status": "error", "error": f"{workbook} does not exist"}, code=1)

    soffice = find_soffice()
    if soffice is None:
        report(
            {
                "status": "error",
                "error": "LibreOffice not found: no soffice on PATH and no LibreOffice.app in /Applications or ~/Applications",
                "hint": "brew install --cask libreoffice",
            },
            code=1,
        )

    try:
        with tempfile.TemporaryDirectory(prefix="spreadsheet-recalc-lo-") as tmp:
            profile_root = Path(tmp) / "profile"
            ensure_macro(soffice, profile_root)
            result = recalculate(soffice, profile_root, workbook, timeout)
    except subprocess.TimeoutExpired:
        report(
            {
                "status": "error",
                "error": f"LibreOffice timed out after {timeout}s",
                "hint": f"rerun with a larger timeout: uv run scripts/recalc.py {workbook.name} {timeout * 3}",
            },
            code=1,
        )

    if result.returncode != 0:
        report(
            {
                "status": "error",
                "error": f"LibreOffice exited with status {result.returncode}",
                "stdout": result.stdout.strip(),
                "stderr": result.stderr.strip(),
            },
            code=1,
        )

    try:
        audit_result = audit(workbook)
    except Exception as exc:
        report({"status": "error", "error": f"could not audit workbook: {exc}"}, code=1)

    if audit_result["status"] == "recalc_incomplete":
        audit_result["hint"] = "no cached values were written; close other LibreOffice instances and rerun"

    exit_code = 0 if audit_result["status"] == "success" or args.soft else 1
    report(audit_result, code=exit_code)


if __name__ == "__main__":
    main()

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["openpyxl>=3.1"]
# ///
"""Smoke tests for the spreadsheet skill helper scripts."""

from __future__ import annotations

import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

from openpyxl import Workbook

ROOT = Path(__file__).resolve().parent
PEEK = ROOT / "peek.py"
PROFILE = ROOT / "profile.py"
RECALC = ROOT / "recalc.py"


def run(args: list[str], check: bool = True) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(args, capture_output=True, text=True, check=False)
    if check and result.returncode != 0:
        raise AssertionError(
            f"command failed: {' '.join(args)}\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}"
        )
    return result


def run_json(args: list[str], check: bool = True) -> tuple[subprocess.CompletedProcess[str], dict]:
    result = run(args, check=check)
    return result, json.loads(result.stdout)


def test_peek(tmp: Path) -> None:
    good = tmp / "good.tsv"
    good.write_text("asset\tamount\tnote\nETH\t1.0\tok\nBTC\t-\tsafe\n", encoding="utf-8")

    _, auto = run_json([sys.executable, str(PEEK), str(good), "--strict", "--house"])
    _, python = run_json([sys.executable, str(PEEK), str(good), "--engine", "python", "--strict"])
    assert auto["status"] == "ok"
    assert auto["engine"] == "python-fast-unquoted"
    assert auto["data_rows"] == python["data_rows"] == 2
    assert auto["dash_null_cells"] == python["dash_null_cells"] == 1
    assert auto["header"] == python["header"]

    _, redacted = run_json([sys.executable, str(PEEK), str(good), "--redact-samples"])
    assert redacted["sample"][0] == ["<redacted>", "<redacted>", "<redacted>"]
    assert redacted["sample"][1][1] == "-"

    ragged = tmp / "ragged.tsv"
    ragged.write_text("asset\tamount\nETH\t1\nBTC\n", encoding="utf-8")
    result, payload = run_json([sys.executable, str(PEEK), str(ragged), "--strict"], check=False)
    assert result.returncode == 1
    assert payload["status"] == "issues_found"
    assert any(item["code"] == "ragged_rows" for item in payload["issues"])


def test_profile(tmp: Path) -> None:
    risky = tmp / "risky.tsv"
    risky.write_text("asset\tamount\tnote\nETH\t1.0\t=cmd\nBTC\t-\t+cmd\n", encoding="utf-8")
    _, payload = run_json(
        [sys.executable, str(PROFILE), str(risky), "--redact-samples", "--top", "2"]
    )
    assert payload["status"] == "issues_found"
    assert payload["formula_injection"]["count"] == 2
    assert payload["formula_injection"]["first"][0]["value"] == "<redacted>"
    assert payload["stats"]["available"] in (True, False)


def test_recalc(tmp: Path) -> None:
    if shutil.which("soffice") is None and not Path("/Applications/LibreOffice.app/Contents/MacOS/soffice").is_file():
        print("skip recalc: LibreOffice not found")
        return

    clean = tmp / "clean.xlsx"
    wb = Workbook()
    ws = wb.active
    ws["A1"] = 1
    ws["A2"] = 2
    ws["A3"] = "=SUM(A1:A2)"
    wb.save(clean)
    result, payload = run_json([sys.executable, str(RECALC), str(clean), "30"])
    assert result.returncode == 0
    assert payload["status"] == "success"

    bad = tmp / "bad.xlsx"
    wb = Workbook()
    ws = wb.active
    ws["A1"] = 0
    ws["A2"] = "=1/A1"
    wb.save(bad)
    result, payload = run_json([sys.executable, str(RECALC), str(bad), "30"], check=False)
    assert result.returncode == 1
    assert payload["status"] == "errors_found"
    assert payload["total_errors"] == 1


def main() -> None:
    with tempfile.TemporaryDirectory(prefix="spreadsheet-skill-tests-") as tmp_dir:
        tmp = Path(tmp_dir)
        test_peek(tmp)
        test_profile(tmp)
        test_recalc(tmp)
    print("spreadsheet helper tests passed")


if __name__ == "__main__":
    main()