policy:
  allow_implicit_invocation: true

# Spreadsheet Recipes

Use these as starting points. Keep amounts as strings, preserve source data, and validate the output with `peek.py`.

## Exact Decimal Transform

```python
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = []
# ///
import csv
from decimal import Decimal

with open("in.tsv", encoding="utf-8-sig", newline="") as f:
    reader = csv.DictReader(f, delimiter="\t")
    rows = list(reader)
    fields = list(reader.fieldnames or [])

if "value_usd" not in fields:
    fields.append("value_usd")

for row in rows:
    amount = Decimal(row["amount"])
    price = Decimal(row["price_usd"])
    row["value_usd"] = str(amount * price)

with open("out.tsv", "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=fields, delimiter="\t", lineterminator="\n")
    writer.writeheader()
    writer.writerows(rows)
```

## Idempotent Append

Use multiset difference, not set difference: identical rows can be legitimate.

```python
from collections import Counter

existing_counts = Counter(tuple(row.items()) for row in existing_rows)
to_append = []
for row in candidate_rows:
    key = tuple(row.items())
    if existing_counts[key] > 0:
        existing_counts[key] -= 1
    else:
        to_append.append(row)
```

## Schema Validation

Generate a draft schema from representative data, edit it, then validate future files. `qsv schema` creates stats
sidecars next to its input, so run it against a temp copy when the source directory must stay clean.

```sh
tmpdir=$(mktemp -d)
trap 'rm -rf "$tmpdir"' EXIT
cp txs.tsv "$tmpdir/txs.tsv"
(cd "$tmpdir" && qsv schema --stdout txs.tsv > txs.schema.json)
cp "$tmpdir/txs.schema.json" txs.schema.json
qsv validate schema txs.schema.json
qsv validate txs.tsv txs.schema.json
```

For house TSV validation after edits:

```sh
uv run "<skill-dir>/scripts/peek.py" txs.tsv --strict --house
```

## Keyed Diff

Use qsv when primary key values are unique; use daff for row/column-aware human review.

```sh
qsv extdedup --select tx_id txs.tsv --no-output
qsv diff --key tx_id --delimiter-output '\t' -o txs.diff.tsv before.tsv after.tsv
bunx daff before.tsv after.tsv
```

## External-Disclosure Profile

Use redaction when a report will be posted, published, or sent to a third party, or when the user asks. Internal agent
reports may include relevant unredacted samples.

```sh
uv run "<skill-dir>/scripts/profile.py" txs.tsv --markdown --redact-samples
```

The profile includes shape, issues, inferred types, null/cardinality signals, formula-prefix cells, and tool
availability without making transformation recommendations.

## Safe Workbook Creation

Use XlsxWriter for new workbooks and write formulas, not Python-computed constants.

```python
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["XlsxWriter"]
# ///
import xlsxwriter

wb = xlsxwriter.Workbook("report.xlsx", {"constant_memory": True})
ws = wb.add_worksheet("Report")
header = wb.add_format({"bold": True})
money = wb.add_format({"num_format": "$#,##0.00"})
ws.write_row(0, 0, ["asset", "amount", "price_usd", "value_usd"], header)
ws.write_row(1, 0, ["ETH", 1.5, 2400])
ws.write_formula(1, 3, "=B2*C2", money)
ws.freeze_panes(1, 0)
wb.close()
```

Then run:

```sh
uv run "<skill-dir>/scripts/recalc.py" report.xlsx
```

Deliver only when the command exits `0` with `"status": "success"`.

# Text Table Design

Design CSV and TSV files as durable interfaces for human maintenance and machine consumption. Existing project and
source contracts override house defaults.

## Vocabulary

Use these terms consistently:

- **Table interface** — everything a reader or writer must know: artifact class, serialization, header and column order,
  row grain, key, value grammar, ordering, provenance, and evolution expectations.
- **Artifact class** — whether the table is source evidence, a canonical authored table, or a generated output.
- **Row grain** — the single fact or entity represented by one row.
- **Key** — the stable column tuple that identifies a row, including whether duplicate keys or duplicate rows are
  legitimate.
- **Value grammar** — the allowed representation of each column: identifiers, enums, units, time zones, precision,
  dates, nulls, and other sentinel values.
- **Ordering contract** — whether row and column order is semantic, deterministic for review, or unconstrained.
- **Evolution contract** — how a schema change reaches producers, consumers, existing rows, generated artifacts, and
  validators.

## Design Workflow

### 1. Classify the artifact

- **Source evidence:** follow the owning project's preservation contract. Preserve source meaning, uncertainty, and any
  required schema or row ordering; never apply house defaults silently.
- **Canonical authored table:** optimize for clear maintenance and deterministic machine use. Apply the entrypoint's
  authored text-table invariants unless the project defines a stronger convention.
- **Generated output:** find and edit the owning source or generator. Treat the emitted table as a review surface, not
  an independent schema.

### 2. Define row identity

State the row grain in one sentence before choosing columns. Give each row one coherent meaning; do not hide repeated
records, nested structures, or unrelated facts in a delimited cell unless the source contract requires it.

Define the key explicitly. Use a natural key when its components are stable and unambiguous; otherwise use an owned
stable identifier. Record whether duplicate keys or identical rows are valid. Never infer a key solely from uniqueness
in a sample.

### 3. Define columns and value grammar

- Give each column one meaning. Split overloaded fields when their parts are independently queried, validated, or
  changed.
- Make units and time zones explicit in the header or owning schema, and use one convention consistently.
- Distinguish null, unknown, not applicable, zero, and empty text. Do not alternate between blank and `-` for the same
  meaning.
- Define enum spelling and case, identifier canonicalization, decimal precision, and date/time representation.
- Order columns for the dominant reading and editing flow. Keep related fields adjacent and prefer stable semantic order
  over alphabetical order.
- Keep provenance sufficient to trace authored or reconstructed facts without embedding an entire narrative in one cell.

Normalize entities that change independently or whose repetition creates inconsistent copies. Denormalize when
self-contained rows materially improve auditability or maintenance and the duplicated facts are stable or validated. Do
not maintain two columns that compete as the source of truth for the same fact.

### 4. Choose serialization

Follow an existing table's established contract. For a newly authored text table, use the entrypoint's house defaults;
choose CSV instead of TSV only for source fidelity or consumer interoperability. Use a real CSV/TSV parser and writer
whenever delimiters, quotes, or newlines can occur in cells.

### 5. Design validation before writing

Identify the owning schema or validator and the task-specific semantic invariants. Structural validation must cover
encoding, delimiter, header, width, and row count expectations. Semantic validation must cover keys, required values,
enums, ordering, cross-column relationships, and any intentional duplicates.

## Evolving a Table Interface

For every added, removed, renamed, reordered, split, or merged column, or any change to row grain, key, value grammar,
or ordering:

1. Inventory the schema source, producers, consumers, existing tables, generated artifacts, and validators.
2. Update the owning schema or producer rather than patching generated outputs.
3. Migrate every in-scope existing row atomically, preserving values not changed by the migration.
4. Regenerate derived artifacts through their owner.
5. Validate the exact header and order, row counts, keys, and task-specific semantic invariants.
6. Decide compatibility explicitly when an active consumer requires it; do not add speculative aliases or shims.

A schema that parses while existing rows, producers, or consumers remain stale is not a completed migration.

## Review Checklist

- The artifact class and row grain are explicit.
- The key and duplicate policy are explicit.
- Every column has one meaning and a defined value grammar.
- Units, time zones, nulls, enums, and identifiers are unambiguous.
- Row and column ordering are intentional.
- Provenance and generated/source ownership are clear.
- Validation proves both structure and domain semantics.
- Schema evolution covers existing data and every active producer and consumer.

# Excel (.xlsx) Workflows

Contents: [Choose the Path](#choose-the-path) · [Read](#read) · [Create](#create) · [Edit](#edit) ·
[Formulas](#formulas) · [Recalculate](#recalculate) · [Convert](#convert) · [Formatting Defaults](#formatting-defaults)
· [Pitfalls](#pitfalls)

## Choose the Path

- **Values only** (analyze, extract, convert): use `qsv excel` for fast export/metadata, DuckDB for SQL over `.xlsx`, or
  `fastexcel` when Python/Polars/Arrow is truly needed. Prefer converting to TSV early and doing the real work there.
- **New workbook deliverable** (formulas, styling, tables, charts): build with XlsxWriter, then run the mandatory
  recalculation loop.
- **Existing workbook edit** (preserve sheets, formulas, macros where possible): use openpyxl surgically, then run the
  recalculation loop.

## Read

Fast metadata/export with qsv:

```sh
qsv excel --metadata J book.xlsx                 # sheet/table/header metadata as JSON
qsv excel --sheet Trades -d '\t' -q -o trades.tsv book.xlsx
qsv excel --table Table1 -d '\t' -q -o table1.tsv book.xlsx
```

DuckDB — values-only SQL, precision-safe:

```sql
FROM read_xlsx('book.xlsx', all_varchar = true);                    -- first sheet
FROM read_xlsx('book.xlsx', sheet = 'Trades', all_varchar = true);  -- named sheet
```

fastexcel — fast Python read when a dataframe/Arrow path is needed:

```sh
uv run --with fastexcel --with polars python -c "
import fastexcel, polars as pl
sheet = fastexcel.read_excel('book.xlsx').load_sheet_by_name('Trades')
df = pl.DataFrame(sheet)
print(df.shape)
"
```

openpyxl — existing workbook structure and formulas:

```python
from openpyxl import load_workbook

wb = load_workbook("book.xlsx")                        # formulas as strings
wb_values = load_workbook("book.xlsx", data_only=True) # cached values from last save
for ws in wb.worksheets:
    print(ws.title, ws.max_row, ws.max_column)
```

- `data_only=True` returns the values cached by the last application that saved the file. A workbook freshly written by
  openpyxl has no cache — formula cells read as `None` until recalculated.
- Never save a workbook loaded with `data_only=True`: formulas are silently replaced by values, permanently.
- Large values-only reads should not default to openpyxl; prefer `qsv excel`, DuckDB, or fastexcel.
- Bulk multi-sheet dump when pandas is genuinely convenient: `uv run --with pandas python -c "..."` with
  `pd.read_excel(path, sheet_name=None, dtype=str)` — `dtype=str` is non-negotiable for amount columns.

## Create

```python
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["XlsxWriter"]
# ///
import xlsxwriter

wb = xlsxwriter.Workbook("portfolio.xlsx", {"constant_memory": True})
ws = wb.add_worksheet("Portfolio")
header = wb.add_format({"bold": True})
money = wb.add_format({"num_format": "$#,##0.00"})
ws.write_row(0, 0, ["asset", "amount", "price_usd", "value_usd"], header)
ws.write_row(1, 0, ["ETH", 1.5, 2400])
ws.write_formula(1, 3, "=B2*C2", money)
ws.write_row(2, 0, ["BTC", 0.25, 64000])
ws.write_formula(2, 3, "=B3*C3", money)
ws.write(3, 0, "total")
ws.write_formula(3, 3, "=SUM(D2:D3)", money)
ws.freeze_panes(1, 0)
ws.set_column(0, 0, 10)
ws.set_column(1, 3, 14)
wb.close()
```

Then recalculate — see [Recalculate](#recalculate).

## Edit

```python
from openpyxl import load_workbook

wb = load_workbook("book.xlsx")
ws = wb["Sheet1"]
ws["B2"] = "new value"
ws.insert_rows(3)
ws.delete_cols(5)
extra = wb.create_sheet("Extra")
wb.save("book.xlsx")
```

- Match the existing workbook's conventions — fonts, number formats, layout — exactly; never restyle while editing.
- Cell coordinates are 1-based: `ws.cell(row=1, column=1)` is `A1`.
- Keep the original file (or a copy) until the edited output is verified.

## Formulas

A spreadsheet must stay recalculable: when source cells change, derived cells must follow. Write formulas, not constants
computed in Python.

```python
ws["D10"] = "=SUM(D2:D9)"                        # not the Python-side sum
ws["E2"] = "=D2/$D$10"                           # absolute ref for a shared denominator
ws["B2"] = "=Inputs!B2*(1+Inputs!B3)"            # assumptions live in cells, not literals
```

- Put assumptions (rates, fees, multipliers) in dedicated cells and reference them; no magic numbers inside formulas.
- Guard divisions: `=IF(C2=0, 0, B2/C2)`.
- Mind the offset: with one header row, list/DataFrame row `N` lands on worksheet row `N + 2`. Verify two or three
  references against the actual data before filling a whole column.
- Cross-sheet references quote names containing spaces: `='FX Rates'!B2`.

## Recalculate

Python libraries write formula strings without computing them; errors only become visible after a real engine
recalculates. Always finish with:

```sh
uv run "<skill-dir>/scripts/recalc.py" book.xlsx [timeout-seconds]   # exits nonzero unless status is success
```

The script resolves LibreOffice (`soffice` on `PATH`, else the macOS app bundle under `/Applications` or
`~/Applications`), uses an isolated temporary LibreOffice profile, recalculates and saves the workbook in place, then
audits every cell and prints JSON:

```json
{
  "status": "errors_found",
  "total_formulas": 41,
  "uncached_formulas": 0,
  "total_errors": 2,
  "errors": { "#DIV/0!": { "count": 2, "cells": ["Portfolio!D7", "Portfolio!E7"] } }
}
```

Loop until `status` is `success`: fix the listed cells, rerun. Statuses:

- `success` — deliverable.
- `errors_found` — exits `1`; fix and rerun. Typical causes: `#REF!` broken references after inserting/deleting rows or
  columns; `#DIV/0!` unguarded division; `#VALUE!` text where a number is expected; `#NAME?` misspelled function or
  unquoted sheet name.
- `recalc_incomplete` — exits `1`; LibreOffice did not write cached values.
- `error` — exits `1`; the JSON `hint` says what to do (e.g. `brew install --cask libreoffice` when LibreOffice is
  missing).

Use `--soft` only when automation must capture a non-success report without failing the shell command.

## Convert

```sh
# xlsx/xls/xlsb/ods -> tsv (one sheet, fast values-only export)
qsv excel --sheet Trades -d '\t' -q -o trades.tsv book.xlsx

# xlsx -> tsv via DuckDB when you need SQL/range/filter control
duckdb -c "COPY (FROM read_xlsx('book.xlsx', sheet = 'Trades', all_varchar = true)) TO 'trades.tsv' (FORMAT csv, DELIMITER '\t', HEADER true)"

# tsv -> xlsx (values only, no styling)
duckdb -c "INSTALL excel; LOAD excel; COPY (FROM read_csv('data.tsv', delim = '\t', header = true, all_varchar = true)) TO 'data.xlsx' (FORMAT xlsx, HEADER true)"
```

`read_xlsx` autoloads DuckDB's excel extension; `COPY ... (FORMAT xlsx)` does not — keep the
`INSTALL excel; LOAD excel;` prefix.

`all_varchar` keeps amounts as text on both sides — the precision rule survives conversion. Type the columns only when
explicitly asked.

After any `.xlsx` to TSV export, validate the TSV before using or delivering it:

```sh
uv run "<skill-dir>/scripts/peek.py" book.tsv --strict
```

When replacing an existing TSV export and the sheet shape should not change, save a before-report first and finish with
`--expect-like`.

## Formatting Defaults

For new workbooks; conventions in an existing template always win.

- One font family for the whole workbook (Calibri or Arial); bold header row; freeze it (`ws.freeze_panes = "A2"`).
- Number formats, not data mangling: set `cell.number_format` (`"yyyy-mm-dd"`, `"0.0%"`, `"#,##0.00;(#,##0.00)"`)
  instead of writing formatted strings into cells.
- Money: a currency `number_format` like `"$#,##0.00"` or a unit-suffixed header (`value_usd`); never currency symbols
  inside cell values.
- Set column widths so nothing displays truncated.
- Zero formula errors at delivery — enforced by the recalc loop.

## Pitfalls

- Homebrew-cask LibreOffice stays Gatekeeper-quarantined, and `soffice` writes `.pyc` files into the app bundle on first
  run, breaking its signature seal — a later GUI launch then claims "LibreOffice.app is damaged". The app is fine; do
  not trash it: `xattr -dr com.apple.quarantine /Applications/LibreOffice.app` (or install with `--no-quarantine`).
  Headless recalculation is unaffected either way.
- openpyxl round-trips drop charts and images, and can degrade pivot tables and other advanced features. If a workbook
  contains them, confine edits to what was asked, save to a new file, and tell the user what may be lost.
- `.xlsm`: pass `keep_vba=True` to `load_workbook`, or the macros are stripped.
- Write `datetime`/`date` objects for date cells (with a date `number_format`), not strings — strings stay text and
  break date arithmetic.
- Column letters: use `openpyxl.utils.get_column_letter` / `column_index_from_string`; never hand-compute (column 64 is
  `BL`, not `BK`).
- `.numbers` files are out of scope: ask the user to export CSV/xlsx from Numbers first (`open -a Numbers <file>`).

#!/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 cells with spreadsheet formula prefixes.

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_prefix_cells(
    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 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] = {
        "schema_version": 2,
        "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_prefix_cells": scan_formula_prefix_cells(
            args.file,
            delimiter,
            peek_report["encoding"],
            peek_report["header"],
            args.redact_samples,
        ),
    }
    formula_issue = args.external_data and report["formula_prefix_cells"]["count"] > 0
    report["status"] = "issues_found" if peek_report["issues"] or formula_issue 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 {
        "schema_version": 2,
        "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,
    }


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_prefix_cells"]["count"]:
            lines.append(f"- `formula_prefix_cells`: {report['formula_prefix_cells']['count']} 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'}")
    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(
        "--external-data",
        action="store_true",
        help="treat formula-prefix cells as external-data safety issues",
    )
    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()