Convert SQL query output into csv style output

sql_output_to_csv.py

#!/usr/bin/env python3
"""
Convert MySQL-style boxed query output (ASCII table with +---+ and | cells) to CSV.

Usage:
  python sql_output_to_csv.py < query.txt > out.csv
  python sql_output_to_csv.py query.txt -o out.csv
  cat query.txt | python sql_output_to_csv.py
"""

from __future__ import annotations

import argparse
import csv
import re
import sys

# Lines like +------+---+ that delimit the table (only + and -).
SEPARATOR_LINE_PATTERN = re.compile(r"^\+[-+]+$")

_HELP_EPILOG = """Examples:
  python sql_output_to_csv.py < query.txt > out.csv
  python sql_output_to_csv.py query.txt -o out.csv
  cat query.txt | python sql_output_to_csv.py"""


def is_separator_line(line: str) -> bool:
    stripped = line.strip()
    if not stripped.startswith("+"):
        return False
    return bool(SEPARATOR_LINE_PATTERN.match(stripped))


def parse_pipe_row(line: str) -> list[str] | None:
    if "|" not in line:
        return None
    cells = [c.strip() for c in line.split("|")]
    while cells and cells[0] == "":
        cells.pop(0)
    while cells and cells[-1] == "":
        cells.pop()
    return cells if cells else None


def boxed_sql_to_header_and_rows(lines: list[str]) -> tuple[list[str], list[list[str]]]:
    rows: list[list[str]] = []
    for line in lines:
        if is_separator_line(line):
            continue
        parsed = parse_pipe_row(line)
        if parsed is not None:
            rows.append(parsed)
    if not rows:
        return [], []
    return rows[0], rows[1:]


def main() -> int:
    parser = argparse.ArgumentParser(
        prog="sql_output_to_csv.py",
        description=(
            "Convert MySQL-style boxed query output "
            "(ASCII +---+ borders and | column cells) to CSV."
        ),
        epilog=_HELP_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="Input file (default: stdin)",
    )
    parser.add_argument(
        "-o",
        "--output",
        help="Output CSV file (default: stdout)",
    )
    args = parser.parse_args()

    if args.input:
        with open(args.input, encoding="utf-8") as fin:
            content = fin.read()
    else:
        content = sys.stdin.read()

    header, data = boxed_sql_to_header_and_rows(content.splitlines())

    if args.output:
        with open(args.output, "w", encoding="utf-8", newline="") as fout:
            writer = csv.writer(fout)
            if header:
                writer.writerow(header)
            writer.writerows(data)
    else:
        writer = csv.writer(sys.stdout)
        if header:
            writer.writerow(header)
        writer.writerows(data)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())