pos_update_gui.py > pos_update_daily_product.py 로 입력할 파일에 맞춰 파일명 변경

lib/pos_update_daily_product.py

@@ -0,0 +1,158 @@
+# POS Update
+'''
+포스 데이터를 추출한 엑셀파일을 업데이트
+OK포스 > 매출관리 > 일자별 > 상품별 > 날짜 지정 > 조회줄수 5000으로 변경 > 엑셀
+추출파일을 ./data에 복사
+본 파일 실행하면 자동으로 mariadb의 DB에 삽입함.
+
+'''
+
+
+import sys, os
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+import tkinter as tk
+import pandas as pd
+from tkinter import filedialog, messagebox
+from sqlalchemy.dialects.mysql import insert as mysql_insert
+from sqlalchemy.exc import IntegrityError
+
+from conf import db, db_schema
+from lib.common import load_config
+
+CONFIG = load_config()
+DATA_DIR = os.path.join(os.path.dirname(__file__), '../data')
+
+def update_pos_table(engine, table, df):
+    with engine.begin() as conn:
+        for idx, row in df.iterrows():
+            data = row.to_dict()
+            stmt = mysql_insert(table).values(**data)
+
+            # insert ... on duplicate key update (복합 unique key 기준)
+            update_data = {
+                'qty': data['qty'],
+                'tot_amount': data['tot_amount'],
+                'tot_discount': data['tot_discount'],
+                'actual_amount': data['actual_amount']
+            }
+
+            stmt = stmt.on_duplicate_key_update(**update_data)
+
+            try:
+                conn.execute(stmt)
+            except IntegrityError as e:
+                print(f"[ERROR] {idx + 1}행 삽입 실패: {e}")
+
+    print("[DONE] 모든 데이터 삽입 완료")
+
+def process_file(filepath, table, engine):
+    print(f"[INFO] 처리 시작: {filepath}")
+    try:
+        ext = os.path.splitext(filepath)[-1].lower()
+
+        if ext == ".xls":
+            df = pd.read_excel(filepath, header=5, engine="xlrd")
+        elif ext == ".xlsx":
+            df = pd.read_excel(filepath, header=5, engine="openpyxl")
+        else:
+            raise ValueError("지원하지 않는 파일 형식입니다.")
+
+        df = df[df.iloc[:, 0] != '합계']
+
+        df.rename(columns={
+            '일자': 'date',
+            '대분류': 'ca01',
+            '중분류': 'ca02',
+            '소분류': 'ca03',
+            '상품코드': 'barcode',
+            '상품명': 'name',
+            '수량': 'qty',
+            '총매출액': 'tot_amount',
+            '총할인액': 'tot_discount',
+            '실매출액': 'actual_amount'
+        }, inplace=True)
+
+        if 'idx' in df.columns:
+            df = df.drop(columns=['idx'])
+
+        df['date'] = pd.to_datetime(df['date']).dt.date
+        df['barcode'] = df['barcode'].astype(int)
+        df['qty'] = df['qty'].astype(int)
+        df['tot_amount'] = df['tot_amount'].astype(int)
+        df['tot_discount'] = df['tot_discount'].astype(int)
+        df['actual_amount'] = df['actual_amount'].astype(int)
+
+        if df.empty:
+            print("[WARN] 데이터가 없습니다.")
+            return False, 0
+
+    except Exception as e:
+        print(f"[ERROR] 로드 실패: {e}")
+        return False, 0
+
+    print(f"[INFO] 데이터 건수: {len(df)}")
+    update_pos_table(engine, table, df)
+    print(f"[INFO] 처리 완료: {filepath}")
+    return True, len(df)
+
+def batch_process_files(table, engine):
+    files = [f for f in os.listdir(DATA_DIR) if f.startswith("일자별 (상품별)") and f.endswith(('.xlsx', '.xls'))]
+
+    if not files:
+        print("[INFO] 처리할 파일이 없습니다.")
+        return False
+
+    print(f"[INFO] {len(files)}개의 파일을 찾았습니다.")
+    total_rows = 0
+
+    print(f"[INFO] 처리된 전체 데이터 건수: {total_rows}")
+    return True
+
+def run_pos_update():
+    filepath = filedialog.askopenfilename(
+        filetypes=[("Excel Files", "*.xlsx *.xls")],
+        title="파일을 선택하세요"
+    )
+    if not filepath:
+        return
+
+    engine = db.engine
+    try:
+        table = db_schema.pos
+    except AttributeError:
+        messagebox.showerror("DB 오류", "'pos' 테이블이 db_schema에 정의되어 있지 않습니다.")
+        return
+
+    if messagebox.askyesno("확인", f"'{os.path.basename(filepath)}' 파일을 'pos' 테이블에 업로드 하시겠습니까?"):
+        success, count = process_file(filepath, table, engine)
+        if success:
+            print(f"[INFO] 수동 선택된 파일 처리 완료: {count}건")
+            messagebox.showinfo("완료", f"DB 업데이트가 완료되었습니다.\n총 {count}건 처리됨.")
+
+def main():
+    engine = db.engine
+    try:
+        table = db_schema.pos
+    except AttributeError:
+        print("[ERROR] 'pos' 테이블이 db_schema에 정의되어 있지 않습니다.")
+        return
+
+    batch_done = batch_process_files(table, engine)
+    if not batch_done:
+        print("[ERROR] batch_process_files 실행 실패")
+#        # GUI 시작
+#        root = tk.Tk()
+#        root.title("POS 데이터 업데이트")
+#        root.geometry("300x150")
+#
+#        lbl = tk.Label(root, text="POS 데이터 업데이트")
+#        lbl.pack(pady=20)
+#
+#        btn = tk.Button(root, text="데이터 선택 및 업데이트", command=run_pos_update)
+#        btn.pack()
+#
+#        root.mainloop()
+
+if __name__ == "__main__":
+    main()