from __future__ import annotations
import argparse
import csv
import json
import os
import shutil
import sys
ROW_CAP = 10_000
def find_csv_assets(script_path: str) -> str | None:
script_dir = os.path.dirname(os.path.abspath(script_path))
root = os.path.dirname(os.path.dirname(script_dir))
runtime = os.path.join(root, "ext", "csv")
if os.path.isdir(runtime):
return runtime
source = os.path.join(root, "share", "wyvern", "ext", "csv")
if os.path.isdir(source):
return source
return None
def parse_csv(path: str, row_cap: int = ROW_CAP) -> tuple[list[str], list[list[str]], bool]:
try:
with open(path, newline="", encoding="utf-8-sig") as handle:
reader = csv.reader(handle)
try:
header = next(reader)
except StopIteration as exc:
raise ValueError("CSV file is empty") from exc
columns = [cell.strip() if isinstance(cell, str) else str(cell) for cell in header]
if not columns or all(col == "" for col in columns):
raise ValueError("CSV file has no header columns")
rows: list[list[str]] = []
truncated = False
for raw in reader:
if len(rows) >= row_cap:
truncated = True
break
normalized = [raw[i] if i < len(raw) else "" for i in range(len(columns))]
rows.append(normalized)
return columns, rows, truncated
except OSError as exc:
raise ValueError(f"could not read CSV '{path}': {exc}") from exc
except csv.Error as exc:
raise ValueError(f"malformed CSV '{path}': {exc}") from exc
def write_rows_json(out_dir: str, columns: list[str], rows: list[list[str]], truncated: bool) -> None:
data_dir = os.path.join(out_dir, "data")
os.makedirs(data_dir, exist_ok=True)
payload = {
"columns": columns,
"rows": rows,
"meta": {"truncated": truncated, "row_count": len(rows)},
}
dest = os.path.join(data_dir, "rows.json")
with open(dest, "w", encoding="utf-8") as handle:
json.dump(payload, handle, ensure_ascii=False)
def copy_static_assets(out_dir: str, assets_root: str) -> None:
mapping = (
("pages/view.html", os.path.join("pages", "view.html")),
("shared/table.js", os.path.join("shared", "table.js")),
("shared/table.css", os.path.join("shared", "table.css")),
)
for rel, dest_rel in mapping:
src = os.path.join(assets_root, rel)
if not os.path.isfile(src):
raise ValueError(f"missing packaged CSV asset: {src}")
dest = os.path.join(out_dir, dest_rel)
os.makedirs(os.path.dirname(dest), exist_ok=True)
shutil.copy2(src, dest)
def _md_cell(value: str) -> str:
return str(value).replace("|", "\\|").replace("\n", " ").replace("\r", "")
def to_markdown(columns: list[str], rows: list[list[str]]) -> str:
header = "| " + " | ".join(_md_cell(col) for col in columns) + " |"
sep = "| " + " | ".join("---" for _ in columns) + " |"
body = ["| " + " | ".join(_md_cell(cell) for cell in row) + " |" for row in rows]
return "\n".join([header, sep, *body]) + "\n"
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Stage a CSV as a wyvern table view or markdown.")
parser.add_argument("csv_file", help="Path to the CSV file")
parser.add_argument("--out", required=True, help="Output tmpdir")
parser.add_argument(
"--format",
required=True,
choices=("html", "markdown"),
help="html stages files; markdown writes a pipe table to stdout",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
columns, rows, truncated = parse_csv(args.csv_file)
os.makedirs(args.out, exist_ok=True)
if args.format == "html":
assets = find_csv_assets(__file__)
if assets is None:
raise ValueError(
"could not locate share/wyvern/ext/csv assets relative to this script"
)
write_rows_json(args.out, columns, rows, truncated)
copy_static_assets(args.out, assets)
else:
sys.stdout.write(to_markdown(columns, rows))
return 0
except ValueError as exc:
print(f"csv_to_view: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())