"""CSV inference and visual review. Run with the training script's Python dependencies."""
from __future__ import annotations
import argparse
import csv
import html
import json
import random
import time
import unicodedata
from typing import cast
from dataclasses import asdict
from pathlib import Path
import xml.etree.ElementTree as ET
import torch
from train_tabfix import (BUSINESS_CATEGORIES as CATEGORIES, Consumer, Edit, Record, generate_repair, load_checkpoint, predict_edits, deterministic_findings, deterministic_repair, conditional_perplexity, replace_cell_xml, read_json, mapping, number)
def read_csv(path: Path) -> list[list[str]]:
with path.open(encoding="utf-8-sig", newline="") as stream:
rows = list(csv.reader(stream))
if not rows or not rows[0] or any(len(r) != len(rows[0]) for r in rows):
raise ValueError("Expected a rectangular, comma-delimited UTF-8 CSV with a header")
return rows
def write_csv(path: Path, rows: list[list[str]]) -> None:
with path.open("w", encoding="utf-8", newline="") as stream:
csv.writer(stream).writerows(rows)
def context_xml(rows: list[list[str]], selected: list[int], schema: str) -> str:
chunks = [f'', schema, '']
for i in selected:
chunks.append(f'')
for header, value in zip(rows[0], rows[i+1]):
chunks.append(f'| '+(html.escape(value) if value else '')+' | ')
chunks.append('
')
return '\n'.join(chunks+['', '
'])
def review(rows: list[list[str]], result: list[list[str]], reference: list[list[str]] | None,
audit: list[dict[str, object]], stats: dict[str, object]) -> str:
def show(value: str) -> str:
quoted = json.dumps(value, ensure_ascii=False)
visible = ''.join(f"\\u{ord(c):04x}" if unicodedata.category(c) == 'Cf' else c for c in quoted)
return html.escape(visible)
details: list[str] = []
for i, row in enumerate(rows[1:]):
for j, value in enumerate(row):
actual = result[i+1][j]
expected = reference[i+1][j] if reference else None
relevant = actual != value or (expected is not None and expected != value) or any(a['row']==i and a['column']==j for a in audit)
if not relevant:
continue
status = 'Changed' if actual != value else 'Unchanged'
if expected is not None:
status = ('Repaired' if actual == expected else 'Missed / incorrect') if value != expected else ('Unwanted change' if actual != value else 'Preserved')
color = 'good' if status in ('Repaired','Preserved') else 'bad'
categories = ', '.join(str(a['category']) for a in audit if a['row']==i and a['column']==j) or 'No detection'
details.append(f'| {i+1} | {html.escape(rows[0][j])} | {show(value)} | {show(actual)} | {show(expected) if expected is not None else "—"} | {status}{html.escape(categories)} |
')
full = '| Data row | '+''.join(''+html.escape(h)+' | ' for h in rows[0])+'
|---|
'
for i,row in enumerate(result[1:]):
full += f'| {i+1} | '+''.join(''+html.escape(v)+' | ' for j,v in enumerate(row))+'
'
cards = ''.join(''+html.escape(str(stats.get(key,'—')))+''+label+'
' for key,label in [('cells','Cells scanned'),('initial_faulty_cells','Errors in input'),('exactly_repaired_cells','Errors repaired'),('valid_cells_changed','Valid cells changed')])
return '''TabFix — end-to-end CSV review
Does TabFix fix this CSV?
Actual CSV → schema and context → predicted error locations → generated replacements → corrected CSV. The clean reference is used only after inference. No gold error locations or replacement text are supplied to the model. Row numbers below count data rows, excluding the header.
'''+''+cards+'
CSV inference review · Threshold '+str(stats['threshold'])+' · CPU inference '+str(stats['inference_seconds'])+' seconds (model loading excluded).
Errors and proposed changes
Strings are quoted to make spaces and empty values visible. Yellow cells below were changed. All other cells are included so unwanted edits remain visible.
Complete model output
Every detection and application decision
'+html.escape(json.dumps(audit,ensure_ascii=False,indent=2))+'