bkyungee
111 lines
3.6 KiB
Python
Executable File
111 lines
3.6 KiB
Python
Executable File
#!/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())
|