Deploy: exdev sync 2026-08-13 13:38:36
This commit is contained in:
bkyungee
Regular → Executable
+110
-110
@@ -1,110 +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())
|
||||
#!/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())
|
||||
|
||||
Reference in New Issue
Block a user