# SHP 파일 여러 개를 하나의 PostGIS 테이블에 적재하기 > OSGeo4W와 ogr2ogr를 이용하여 Windows 환경에서 SHP 파일 여러 개를 하나의 PostGIS 테이블로 적재하는 방법 정리 > 모든 테이블구조와 좌표계가 일정해야함 --- ## 1. 환경 준비 ### 필수 설치 * **OSGeo4W** : [https://trac.osgeo.org/osgeo4w/](https://trac.osgeo.org/osgeo4w/) 설치 후 `OSGeo4W Shell` 사용 --- ## 2. 데이터 준비 * 적재할 SHP 파일을 한 폴더에 모음 예: ``` C:/upload/shp/ ``` --- ## 3. python code 작성 ```python import os import subprocess from glob import glob from datetime import datetime # ---------------------------- # 설정 # ---------------------------- PGHOST = "host" PGPORT = 5432 PGDB = "dbname" PGUSER = "username" PGPASS = "password" PGSCHEMA = "public" PGTABLE = "tablename" SRS = "EPSG:5186" SHP_FOLDER = r"C:\upload\shp\lsmd_cont_ldreg" os.makedirs(SHP_FOLDER, exist_ok=True) LOGFILE = os.path.join( SHP_FOLDER, f"import_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" ) ERROR_FILES = [] # ---------------------------- # 환경 변수에 PGPASSWORD 설정 # ---------------------------- env = os.environ.copy() env["PGPASSWORD"] = PGPASS # ---------------------------- # 로그 기록 함수 # ---------------------------- def logger(event, file_or_message, message=None): timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S') if message: log_line = f"{timestamp} | {event} | {file_or_message} | {message}" else: log_line = f"{timestamp} | {event} | {file_or_message}" with open(LOGFILE, "a", encoding="utf-8") as f: f.write(log_line + "\n") print(log_line) # ---------------------------- # ogr2ogr 실행 함수 # ---------------------------- def run_ogr2ogr(file_path, append=False): cmd = [ "ogr2ogr", "-f", "PostgreSQL", f"PG:host={PGHOST} port={PGPORT} dbname={PGDB} user={PGUSER} password={PGPASS}", file_path, "-nln", f"{PGSCHEMA}.{PGTABLE}", "-nlt", "PROMOTE_TO_MULTI", "-progress", "--config", "PG_USE_COPY", "YES", "-a_srs", SRS ] if not append: # create 옵션 cmd.extend([ "-lco", "GEOMETRY_NAME=geom", "-lco", "FID=gid" ]) else: # append 모드 cmd.append("-append") try: process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env) for line in process.stdout: line = line.strip() if line: print(line) # 진행률 출력 process.wait() if process.returncode != 0: raise subprocess.CalledProcessError(process.returncode, cmd) except subprocess.CalledProcessError as e: logger("ERROR", file_path, f"ogr2ogr failed with code {e.returncode}") ERROR_FILES.append(file_path) return False return True # ---------------------------- # SHP 파일 반복 처리 # ---------------------------- shp_files = glob(os.path.join(SHP_FOLDER, "*.shp")) first = True total_start = datetime.now() for i, shp in enumerate(shp_files, start=1): logger("START", shp) start_time = datetime.now() success = run_ogr2ogr(shp, append=not first) end_time = datetime.now() elapsed = (end_time - start_time).total_seconds() logger("END", shp, f"{elapsed:.2f}s (File {i}/{len(shp_files)})") if first and not success: logger("ERROR", "First file failed, aborting remaining imports.") break first = False # ---------------------------- # GIST 인덱스 생성 및 ANALYZE # ---------------------------- def create_index_analyze(): # GIST 인덱스 logger("START", "GIST index creation") try: subprocess.run( ["psql", "-h", PGHOST, "-p", str(PGPORT), "-U", PGUSER, "-d", PGDB, "-c", f"CREATE INDEX IF NOT EXISTS {PGTABLE}_geom_gist ON {PGSCHEMA}.{PGTABLE} USING GIST (geom);"], check=True, capture_output=True, text=True, env=env ) except subprocess.CalledProcessError as e: logger("ERROR", "GIST index creation", e.stderr) ERROR_FILES.append("GIST index creation") logger("END", "GIST index creation") # ANALYZE logger("START", "ANALYZE table") try: subprocess.run( ["psql", "-h", PGHOST, "-p", str(PGPORT), "-U", PGUSER, "-d", PGDB, "-c", f"ANALYZE {PGSCHEMA}.{PGTABLE};"], check=True, capture_output=True, text=True, env=env ) except subprocess.CalledProcessError as e: logger("ERROR", "ANALYZE table", e.stderr) ERROR_FILES.append("ANALYZE table") logger("END", "ANALYZE table") create_index_analyze() total_end = datetime.now() total_elapsed = (total_end - total_start).total_seconds() logger("INFO", "All SHP files processed!") logger("INFO", "Total elapsed time", f"{total_elapsed:.2f}s") # ---------------------------- # 실패 파일 기록 # ---------------------------- if ERROR_FILES: logger("INFO", "Failed tasks:") for f in ERROR_FILES: logger("INFO", f) logger("INFO", "Some tasks failed. See log for details.") else: logger("INFO", "All tasks completed successfully.") ``` --- ## 4. 데이터 확인 PostgreSQL 접속 후 데이터 확인: ```sql -- 총 데이터 수 확인 SELECT COUNT(*) FROM tablename; -- 좌표계 확인 SELECT ST_SRID(geom) FROM tablename LIMIT 1; ``` --- ## 5. 주의 사항 * SHP 파일 좌표계가 일치하지 않으면 위치가 엉뚱하게 표시될 수 있음 * 필드 구조가 다른 SHP 파일은 적재 실패 --- ### 6. 예시 화면 (OSGeo4W Shell) ``` 2025-09-06_20-44-10 | START | C:\upload\shp\lsmd_cont_ldreg\LSMD_CONT_LDREG_11_202508.shp WARNING: database "rsonegreet" has no actual collation version, but a version was recorded 0...10...20...30...40...50...60...70...80...90...100 - done in 00:00:13. 2025-09-06_20-44-23 | END | C:\upload\shp\lsmd_cont_ldreg\LSMD_CONT_LDREG_11_202508.shp | 13.20s (File 1/2) 2025-09-06_20-44-23 | START | C:\upload\shp\lsmd_cont_ldreg\LSMD_CONT_LDREG_41_202508.shp WARNING: database "rsonegreet" has no actual collation version, but a version was recorded 0...10...20...30...40...50...60...70...80...90...100 - done in 00:01:22. 2025-09-06_20-45-46 | END | C:\upload\shp\lsmd_cont_ldreg\LSMD_CONT_LDREG_41_202508.shp | 82.49s (File 2/2) 2025-09-06_20-45-46 | START | GIST index creation 2025-09-06_20-45-57 | END | GIST index creation 2025-09-06_20-45-57 | START | ANALYZE table 2025-09-06_20-45-57 | END | ANALYZE table 2025-09-06_20-45-57 | INFO | All SHP files processed! 2025-09-06_20-45-57 | INFO | Total elapsed time | 107.02s 2025-09-06_20-45-57 | INFO | All tasks completed successfully. ``` ---