예전 날씨파일을 업데이트하기위한 일회용 코드

lib/weatherFileUpdate.py

@@ -0,0 +1,96 @@
+# weatherFileUpdate.py
+
+import os
+import csv
+import sys
+from datetime import datetime
+from sqlalchemy import select, and_
+
+# 경로 설정
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from conf import db, db_schema
+
+CSV_FILENAME = 'weather.csv'  # 데이터 파일명
+CSV_PATH = os.path.join(os.path.dirname(__file__), '../data', CSV_FILENAME)
+
+weather_table = db_schema.fg_manager_static_weather
+
+STN_ID = 99  # 고정된 stnId
+
+def parse_float(value):
+    try:
+        f = float(value)
+        return f if f == f else 0.0  # NaN 체크, NaN일 경우 0.0
+    except (ValueError, TypeError):
+        return 0.0
+
+def load_csv(filepath):
+    rows = []
+    try:
+        with open(filepath, newline='', encoding='utf-8') as csvfile:
+            reader = csv.DictReader(csvfile)
+            for row in reader:
+                try:
+                    date = datetime.strptime(row['날짜'], '%Y-%m-%d').date()
+
+                    data = {
+                        'date': date,
+                        'stnId': STN_ID,
+                        'minTa': parse_float(row.get('최저기온', 0)),
+                        'maxTa': parse_float(row.get('최고기온', 0)),
+                        'sumRn': parse_float(row.get('일강수량\n(mm)', 0)),
+                        'avgWs': parse_float(row.get('평균풍속\n(m/s)', 0)),
+                        'avgRhm': parse_float(row.get('습도', 0)),
+                    }
+
+                    rows.append(data)
+                except Exception as e:
+                    print(f"[WARN] 잘못된 행 건너뜀: {row} / 오류: {e}")
+    except FileNotFoundError:
+        print(f"[ERROR] 파일이 존재하지 않음: {filepath}")
+        sys.exit(1)
+
+    return rows
+
+def row_exists(session, date, stnId):
+    stmt = select(weather_table.c.date).where(
+        and_(
+            weather_table.c.date == date,
+            weather_table.c.stnId == stnId
+        )
+    )
+    return session.execute(stmt).scalar() is not None
+
+def insert_rows(rows):
+    inserted = 0
+    skipped = 0
+    session = db.get_session()
+    try:
+        for row in rows:
+            if row_exists(session, row['date'], row['stnId']):
+                skipped += 1
+                continue
+            session.execute(weather_table.insert().values(**row))
+            inserted += 1
+        session.commit()
+    except Exception as e:
+        session.rollback()
+        print(f"[ERROR] DB 삽입 실패: {e}")
+    finally:
+        session.close()
+    return inserted, skipped
+
+def main():
+    print(f"[INFO] CSV 파일 로드: {CSV_PATH}")
+    rows = load_csv(CSV_PATH)
+    print(f"[INFO] 총 행 수: {len(rows)}")
+
+    if not rows:
+        print("[WARN] 삽입할 데이터가 없습니다.")
+        return
+
+    inserted, skipped = insert_rows(rows)
+    print(f"[DONE] 삽입 완료: {inserted}건, 건너뜀: {skipped}건")
+
+if __name__ == "__main__":
+    main()