React Architectural Patterns

#!/usr/bin/env python3
"""Read and inspect CSV/TSV files with encoding detection and structure analysis.

Usage:
    python3 csv_read.py <file_path> [--head N] [--encoding ENCODING] [--delimiter DELIM] [--info]

Options:
    --head N         Show first N rows (default: 20)
    --encoding ENC   Force encoding (default: auto-detect)
    --delimiter D    Force delimiter (',' or '\\t' or ';' etc. default: auto-detect)
    --info           Show file structure info only (column names, types, row count)
    --search TERM    Search for rows containing TERM in any column
    --column COL     Filter output to specific column name(s), comma-separated
"""

import argparse
import csv
import io
import json
import os
import sys


def detect_encoding(file_path: str) -> str:
    """Detect file encoding by reading BOM or trying common encodings."""
    with open(file_path, "rb") as f:
        raw = f.read(min(os.path.getsize(file_path), 65536))

    # Check BOM
    if raw.startswith(b"\xef\xbb\xbf"):
        return "utf-8-sig"
    if raw.startswith(b"\xff\xfe"):
        return "utf-16-le"
    if raw.startswith(b"\xfe\xff"):
        return "utf-16-be"

    # Try encodings in order
    for enc in ["utf-8", "shift_jis", "euc-jp", "iso-2022-jp", "cp932", "latin-1"]:
        try:
            raw.decode(enc)
            return enc
        except (UnicodeDecodeError, UnicodeError):
            continue

    return "utf-8"


def detect_delimiter(sample: str) -> str:
    """Detect delimiter from file content sample."""
    sniffer = csv.Sniffer()
    try:
        dialect = sniffer.sniff(sample, delimiters=",\t;|")
        return dialect.delimiter
    except csv.Error:
        # Fallback: count occurrences
        for delim in [",", "\t", ";", "|"]:
            if delim in sample:
                return delim
        return ","


def infer_column_type(values: list) -> str:
    """Infer column type from sample values."""
    non_empty = [v for v in values if v.strip()]
    if not non_empty:
        return "empty"

    int_count = 0
    float_count = 0
    for v in non_empty:
        try:
            int(v)
            int_count += 1
            continue
        except ValueError:
            pass
        try:
            float(v)
            float_count += 1
        except ValueError:
            pass

    total = len(non_empty)
    if int_count == total:
        return "integer"
    if (int_count + float_count) == total:
        return "float"
    return "string"


def main():
    parser = argparse.ArgumentParser(description="Read and inspect CSV/TSV files")
    parser.add_argument("file_path", help="Path to the CSV/TSV file")
    parser.add_argument("--head", type=int, default=20, help="Number of rows to show")
    parser.add_argument("--encoding", default=None, help="Force encoding")
    parser.add_argument("--delimiter", default=None, help="Force delimiter")
    parser.add_argument("--info", action="store_true", help="Show structure info only")
    parser.add_argument("--search", default=None, help="Search for rows containing term")
    parser.add_argument("--column", default=None, help="Filter to specific columns (comma-separated)")
    args = parser.parse_args()

    if not os.path.exists(args.file_path):
        print(f"Error: File not found: {args.file_path}", file=sys.stderr)
        sys.exit(1)

    # Detect encoding
    encoding = args.encoding or detect_encoding(args.file_path)

    # Read file
    with open(args.file_path, "r", encoding=encoding, errors="replace") as f:
        content = f.read()

    # Detect delimiter
    delimiter = args.delimiter or detect_delimiter(content[:4096])
    if delimiter == "\\t":
        delimiter = "\t"

    # Parse CSV
    reader = csv.reader(io.StringIO(content), delimiter=delimiter)
    rows = list(reader)

    if not rows:
        print("Error: File is empty", file=sys.stderr)
        sys.exit(1)

    headers = rows[0]
    data_rows = rows[1:]

    # Column filter
    col_indices = None
    if args.column:
        col_names = [c.strip() for c in args.column.split(",")]
        col_indices = []
        for cn in col_names:
            try:
                idx = headers.index(cn)
                col_indices.append(idx)
            except ValueError:
                print(f"Warning: Column '{cn}' not found", file=sys.stderr)

    if args.info:
        # Structure info mode
        info = {
            "file": args.file_path,
            "encoding": encoding,
            "delimiter": repr(delimiter),
            "total_rows": len(data_rows),
            "total_columns": len(headers),
            "columns": [],
        }
        sample_size = min(100, len(data_rows))
        for i, header in enumerate(headers):
            sample_values = [r[i] for r in data_rows[:sample_size] if i < len(r)]
            non_empty = sum(1 for v in sample_values if v.strip())
            col_info = {
                "index": i,
                "name": header,
                "type": infer_column_type(sample_values),
                "non_empty_rate": f"{non_empty}/{sample_size}" if sample_size > 0 else "N/A",
                "sample_values": sample_values[:3],
            }
            info["columns"].append(col_info)

        print(json.dumps(info, indent=2, ensure_ascii=False))
        return

    # Search mode
    if args.search:
        term = args.search.lower()
        matched = [r for r in data_rows if any(term in cell.lower() for cell in r)]
        print(f"Search: '{args.search}' - {len(matched)} matches found")
        print(delimiter.join(headers))
        for row in matched[: args.head]:
            if col_indices:
                row = [row[i] for i in col_indices if i < len(row)]
            print(delimiter.join(row))
        return

    # Display mode
    display_headers = headers
    if col_indices:
        display_headers = [headers[i] for i in col_indices]

    print(f"[{args.file_path}] encoding={encoding} delimiter={repr(delimiter)} rows={len(data_rows)} cols={len(headers)}")
    print(delimiter.join(display_headers))
    print("-" * 80)

    for row in data_rows[: args.head]:
        if col_indices:
            row = [row[i] for i in col_indices if i < len(row)]
        print(delimiter.join(row))

    if len(data_rows) > args.head:
        print(f"... ({len(data_rows) - args.head} more rows)")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Write and transform CSV/TSV files with encoding and format control.

Usage:
    python3 csv_write.py <output_path> --json <json_data> [options]
    python3 csv_write.py <output_path> --from <input_path> [options]

Modes:
    --json DATA      Write CSV from JSON array of objects or array of arrays
    --from PATH      Convert/transform an existing CSV file

Options:
    --encoding ENC   Output encoding (default: utf-8)
    --delimiter D    Output delimiter (default: ',')
    --bom            Add BOM for Excel compatibility (utf-8-sig)
    --tsv            Shorthand for --delimiter '\\t' with .tsv extension
    --sort COL       Sort by column name
    --sort-desc COL  Sort by column name descending
    --filter EXPR    Filter rows: 'column==value' or 'column!=value' or 'column~=regex'
    --select COLS    Select specific columns (comma-separated names)
    --rename OLD=NEW Rename a column (can be repeated)
    --dedupe COL     Remove duplicate rows based on column
    --no-header      Omit header row in output
"""

import argparse
import csv
import io
import json
import os
import re
import sys

# Reuse detect_encoding from csv_read
sys.path.insert(0, os.path.dirname(__file__))
from csv_read import detect_delimiter, detect_encoding


def parse_filter(expr: str):
    """Parse a filter expression like 'col==val', 'col!=val', 'col~=regex'."""
    for op in ["~=", "!=", "=="]:
        if op in expr:
            col, val = expr.split(op, 1)
            return col.strip(), op, val.strip()
    raise ValueError(f"Invalid filter expression: {expr}. Use ==, !=, or ~=")


def apply_filter(rows: list, headers: list, expr: str) -> list:
    """Filter rows based on expression."""
    col, op, val = parse_filter(expr)
    try:
        col_idx = headers.index(col)
    except ValueError:
        print(f"Warning: Filter column '{col}' not found", file=sys.stderr)
        return rows

    result = []
    for row in rows:
        cell = row[col_idx] if col_idx < len(row) else ""
        if op == "==" and cell == val:
            result.append(row)
        elif op == "!=" and cell != val:
            result.append(row)
        elif op == "~=" and re.search(val, cell):
            result.append(row)
    return result


def main():
    parser = argparse.ArgumentParser(description="Write and transform CSV/TSV files")
    parser.add_argument("output_path", help="Output file path")
    parser.add_argument("--json", default=None, help="JSON data (array of objects or arrays)")
    parser.add_argument("--json-file", default=None, help="Path to JSON file")
    parser.add_argument("--from", dest="from_path", default=None, help="Input CSV to transform")
    parser.add_argument("--encoding", default="utf-8", help="Output encoding")
    parser.add_argument("--delimiter", default=",", help="Output delimiter")
    parser.add_argument("--bom", action="store_true", help="Add UTF-8 BOM")
    parser.add_argument("--tsv", action="store_true", help="Output as TSV")
    parser.add_argument("--sort", default=None, help="Sort by column")
    parser.add_argument("--sort-desc", default=None, help="Sort descending by column")
    parser.add_argument("--filter", action="append", default=[], help="Filter expression")
    parser.add_argument("--select", default=None, help="Select columns (comma-separated)")
    parser.add_argument("--rename", action="append", default=[], help="Rename column OLD=NEW")
    parser.add_argument("--dedupe", default=None, help="Deduplicate by column")
    parser.add_argument("--no-header", action="store_true", help="Omit header")
    args = parser.parse_args()

    if args.tsv:
        args.delimiter = "\t"
    if args.delimiter == "\\t":
        args.delimiter = "\t"

    encoding = "utf-8-sig" if args.bom else args.encoding

    # Load data
    headers = []
    data_rows = []

    if args.json or args.json_file:
        # JSON input
        if args.json_file:
            with open(args.json_file, "r", encoding="utf-8") as f:
                data = json.load(f)
        else:
            data = json.loads(args.json)

        if not data:
            print("Error: Empty JSON data", file=sys.stderr)
            sys.exit(1)

        if isinstance(data[0], dict):
            # Array of objects
            headers = list(data[0].keys())
            for obj in data:
                for k in obj.keys():
                    if k not in headers:
                        headers.append(k)
            data_rows = [[str(obj.get(h, "")) for h in headers] for obj in data]
        else:
            # Array of arrays - first row is headers
            headers = [str(h) for h in data[0]]
            data_rows = [[str(c) for c in row] for row in data[1:]]

    elif args.from_path:
        # Transform existing CSV
        src_encoding = detect_encoding(args.from_path)
        with open(args.from_path, "r", encoding=src_encoding, errors="replace") as f:
            content = f.read()
        src_delimiter = detect_delimiter(content[:4096])
        reader = csv.reader(io.StringIO(content), delimiter=src_delimiter)
        rows = list(reader)
        if rows:
            headers = rows[0]
            data_rows = rows[1:]
    else:
        print("Error: Provide --json, --json-file, or --from", file=sys.stderr)
        sys.exit(1)

    # Apply transformations
    # Filter
    for flt in args.filter:
        data_rows = apply_filter(data_rows, headers, flt)

    # Sort
    sort_col = args.sort or args.sort_desc
    if sort_col:
        try:
            sort_idx = headers.index(sort_col)
            reverse = args.sort_desc is not None
            data_rows.sort(key=lambda r: r[sort_idx] if sort_idx < len(r) else "", reverse=reverse)
        except ValueError:
            print(f"Warning: Sort column '{sort_col}' not found", file=sys.stderr)

    # Deduplicate
    if args.dedupe:
        try:
            dedupe_idx = headers.index(args.dedupe)
            seen = set()
            unique_rows = []
            for row in data_rows:
                key = row[dedupe_idx] if dedupe_idx < len(row) else ""
                if key not in seen:
                    seen.add(key)
                    unique_rows.append(row)
            data_rows = unique_rows
        except ValueError:
            print(f"Warning: Dedupe column '{args.dedupe}' not found", file=sys.stderr)

    # Rename columns
    for rename_expr in args.rename:
        old, new = rename_expr.split("=", 1)
        try:
            idx = headers.index(old.strip())
            headers[idx] = new.strip()
        except ValueError:
            print(f"Warning: Rename column '{old}' not found", file=sys.stderr)

    # Select columns
    if args.select:
        col_names = [c.strip() for c in args.select.split(",")]
        col_indices = []
        for cn in col_names:
            try:
                col_indices.append(headers.index(cn))
            except ValueError:
                print(f"Warning: Select column '{cn}' not found", file=sys.stderr)
        headers = [headers[i] for i in col_indices]
        data_rows = [[r[i] for i in col_indices if i < len(r)] for r in data_rows]

    # Write output
    os.makedirs(os.path.dirname(os.path.abspath(args.output_path)), exist_ok=True)

    with open(args.output_path, "w", encoding=encoding, newline="") as f:
        writer = csv.writer(f, delimiter=args.delimiter)
        if not args.no_header:
            writer.writerow(headers)
        writer.writerows(data_rows)

    print(f"Written: {args.output_path} ({len(data_rows)} rows, {len(headers)} columns, encoding={encoding})")


if __name__ == "__main__":
    main()