오래된 파일에서 규칙에 맞게 데이터를 삽입하는 일회용 코드

lib/visitor_update.py

@@ -0,0 +1,87 @@
+# ./lib/visitor_update.py
+
+import os
+import sys
+import pandas as pd
+from datetime import datetime
+
+# 프로젝트 루트 경로 추가
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+from conf.db import get_session
+from conf.db_schema import pos
+from sqlalchemy import select
+
+# 상수 정의
+FILE_PATH = os.path.join(os.path.dirname(__file__), '..', 'data', 'visitor_raw.xlsx')
+CA01 = '매표소'
+CA02 = 'POS'
+CA03 = '입장료'
+BARCODE = 11111111
+DEFAULT_INT = 0
+
+
+def load_excel(filepath):
+    df = pd.read_excel(filepath)
+    df.columns = ['date', 'qty']
+    df['date'] = pd.to_datetime(df['date']).dt.date
+    return df
+
+
+def get_existing_dates(session, dates):
+    """DB에 이미 존재하는 날짜 목록 조회"""
+    stmt = select(pos.c.date).where(pos.c.date.in_(dates))
+    result = session.execute(stmt).scalars().all()
+    return set(result)
+
+
+def insert_data(df):
+    session = get_session()
+    try:
+        all_dates = set(df['date'].unique())
+        existing_dates = get_existing_dates(session, all_dates)
+
+        # 중복 날짜 제거
+        if existing_dates:
+            print(f"[INFO] 이미 존재하는 날짜는 건너뜁니다: {sorted(existing_dates)}")
+            df = df[~df['date'].isin(existing_dates)]
+
+        if df.empty:
+            print("[INFO] 삽입할 신규 데이터가 없습니다.")
+            return
+
+        for _, row in df.iterrows():
+            record = {
+                'date': row['date'],
+                'ca01': CA01,
+                'ca02': CA02,
+                'ca03': CA03,
+                'barcode': BARCODE,
+                'name': '입장객',
+                'qty': int(row['qty']),
+                'tot_amount': DEFAULT_INT,
+                'tot_discount': DEFAULT_INT,
+                'actual_amount': DEFAULT_INT
+            }
+            session.execute(pos.insert().values(**record))
+
+        session.commit()
+        print(f"[INFO] {len(df)}건의 데이터가 성공적으로 삽입되었습니다.")
+    except Exception as e:
+        session.rollback()
+        print(f"[ERROR] 데이터 저장 중 오류 발생: {e}")
+    finally:
+        session.close() 
+
+
+def main():
+    if not os.path.exists(FILE_PATH):
+        print(f"[ERROR] 파일을 찾을 수 없습니다: {FILE_PATH}")
+        return
+
+    df = load_excel(FILE_PATH)
+    insert_data(df)
+
+
+if __name__ == "__main__":
+    main()