Deploy: exdev sync 2026-08-01 21:04:26

server-dev/oracle/tools/csv_to_insert.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""TROAD CSV 덤프를 Oracle INSERT 스크립트로 변환한다.
+
+사용법:
+  python3 csv_to_insert.py <csv> --table OPR_MAIN.T_INCIDENT_FTMS \
+      --numeric ROAD_NO,START_MILEPOST --out ../sql/20_v_incident_ftms_data.sql
+
+CSV 헤더가 대상 테이블의 컬럼명과 같아야 한다.
+공백만 있는 값은 NULL 로 넣는다.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from pathlib import Path
+
+BATCH_COMMIT = 50
+
+
+def sql_literal(value: str, numeric: bool, date: bool) -> str:
+    text = value.strip()
+    if not text:
+        return "NULL"
+    if date:
+        escaped = text.replace("'", "''")
+        return (
+            f"TO_TIMESTAMP('{escaped}', "
+            "'YYYY-MM-DD HH24:MI:SS.FF')"
+        )
+    if numeric:
+        try:
+            float(text)
+        except ValueError:
+            return "NULL"
+        return text
+    return "'" + text.replace("'", "''") + "'"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("csv_path")
+    parser.add_argument("--table", required=True)
+    parser.add_argument("--numeric", default="", help="숫자 컬럼 목록(콤마 구분)")
+    parser.add_argument(
+        "--date",
+        default="",
+        help="날짜/시간 컬럼 목록(콤마 구분, YYYY-MM-DD HH24:MI:SS.FF)",
+    )
+    parser.add_argument("--encoding", default="utf-8")
+    parser.add_argument("--out", required=True)
+    parser.add_argument(
+        "--no-truncate",
+        action="store_true",
+        help="기존 데이터 삭제 구문을 생략한다",
+    )
+    args = parser.parse_args()
+
+    numeric_cols = {c.strip().upper() for c in args.numeric.split(",") if c.strip()}
+    date_cols = {c.strip().upper() for c in args.date.split(",") if c.strip()}
+    overlap = numeric_cols & date_cols
+    if overlap:
+        parser.error(f"--numeric과 --date에 중복된 컬럼: {', '.join(sorted(overlap))}")
+    csv_path = Path(args.csv_path)
+
+    with csv_path.open(newline="", encoding=args.encoding) as fp:
+        rows = list(csv.reader(fp))
+
+    if not rows:
+        print(f"빈 CSV: {csv_path}", file=sys.stderr)
+        return 1
+
+    header = [h.strip().strip('"').upper() for h in rows[0]]
+    data = [r for r in rows[1:] if any(v.strip() for v in r)]
+    bad = [i for i, r in enumerate(data, start=2) if len(r) != len(header)]
+    if bad:
+        print(f"컬럼 수 불일치 라인: {bad[:10]}", file=sys.stderr)
+        return 1
+
+    out_path = Path(args.out)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with out_path.open("w", encoding="utf-8") as out:
+        out.write(f"-- 자동 생성: {csv_path.name} -> {args.table}\n")
+        out.write(f"-- 행 수: {len(data)}\n")
+        out.write("-- 생성 도구: server-dev/oracle/tools/csv_to_insert.py\n\n")
+        out.write("SET DEFINE OFF\n")
+        out.write("SET SQLBLANKLINES ON\n\n")
+        if not args.no_truncate:
+            out.write(f"DELETE FROM {args.table};\n\n")
+
+        cols = ", ".join(header)
+        for idx, row in enumerate(data, start=1):
+            values = ", ".join(
+                sql_literal(v, h in numeric_cols, h in date_cols)
+                for h, v in zip(header, row)
+            )
+            out.write(f"INSERT INTO {args.table} ({cols}) VALUES ({values});\n")
+            if idx % BATCH_COMMIT == 0:
+                out.write("COMMIT;\n")
+
+        out.write("\nCOMMIT;\n")
+
+    print(f"{len(data)} rows -> {out_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())