diff --git a/conf/db.py b/conf/db.py index df6b738..71cc931 100644 --- a/conf/db.py +++ b/conf/db.py @@ -18,8 +18,16 @@ db_cfg = config['database'] db_url = f"mysql+pymysql://{db_cfg['user']}:{db_cfg['password']}@{db_cfg['host']}/{db_cfg['name']}?charset=utf8mb4" # MySQL 연결이 끊겼을 때 자동 재시도 옵션 포함 -engine = create_engine(db_url, pool_pre_ping=True) +engine = create_engine( + db_url, + pool_pre_ping=True, + pool_recycle=3600, # 3600초 = 1시간 +) + Session = sessionmaker(bind=engine) +def get_engine(): + return engine + def get_session(): return Session() diff --git a/conf/db_schema.py b/conf/db_schema.py index 95e345e..73af925 100644 --- a/conf/db_schema.py +++ b/conf/db_schema.py @@ -1,7 +1,7 @@ # db_schema.py import os import yaml -from sqlalchemy import Table, Column, Date, Integer, String, Float, Text, MetaData, UniqueConstraint, DateTime +from sqlalchemy import Table, Column, Date, Integer, String, Float, Text, MetaData, UniqueConstraint, DateTime, Time, PrimaryKeyConstraint from sqlalchemy.sql import func BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) @@ -164,6 +164,15 @@ ga4 = Table( mysql_charset='utf8mb4' ) +holiday = Table( + get_full_table_name('holiday'), metadata, + Column('date', String(8), primary_key=True, comment='날짜 (YYYYMMDD)'), + Column('name', String(50), nullable=False, comment='휴일명'), + Column('created_at', DateTime, server_default=func.now(), comment='등록일시'), + Column('updated_at', DateTime, server_default=func.now(), onupdate=func.now(), comment='수정일시'), + comment='한국천문연구원 특일정보' +) + pos = Table( get_full_table_name('pos'), metadata, Column('idx', Integer, primary_key=True, autoincrement=True), @@ -180,11 +189,38 @@ pos = Table( UniqueConstraint('date', 'ca01', 'ca02', 'ca03', 'name', 'barcode', name='uniq_pos_composite') ) -holiday = Table( - get_full_table_name('holiday'), metadata, - Column('date', String(8), primary_key=True, comment='날짜 (YYYYMMDD)'), - Column('name', String(50), nullable=False, comment='휴일명'), - Column('created_at', DateTime, server_default=func.now(), comment='등록일시'), - Column('updated_at', DateTime, server_default=func.now(), onupdate=func.now(), comment='수정일시'), - comment='한국천문연구원 특일정보' +pos_billdata = Table( + get_full_table_name('pos_billdata'), metadata, + Column('sale_date', Date, nullable=False), + Column('shop_cd', String(20), nullable=False), + Column('pos_no', Integer, nullable=False), + Column('bill_no', Integer, nullable=False), + Column('product_cd', String(20), nullable=False), + Column('division', String(10)), + Column('table_no', String(20)), + Column('order_time', Time), + Column('pay_time', Time), + Column('barcode', String(20)), + Column('product_name', String(100)), + Column('qty', Integer), + Column('tot_sale_amt', Integer), + Column('erp_cd', String(50)), + Column('remark', Text), + Column('dc_amt', Integer), + Column('dc_type', String(50)), + Column('dcm_sale_amt', Integer), + Column('net_amt', Integer), + Column('vat_amt', Integer), + PrimaryKeyConstraint('sale_date', 'shop_cd', 'pos_no', 'bill_no', 'product_cd') +) + +pos_shop_name = Table( + get_full_table_name('pos_shop_name'), metadata, + Column('shop_cd', String(20), primary_key=True, nullable=False), + Column('shop_name', String(100), nullable=False), + Column('used', Integer, nullable=False, default=1, comment='사용여부 (1=사용, 0=미사용)'), + Column('created_at', DateTime, server_default=func.current_timestamp(), comment='등록일시'), + Column('updated_at', DateTime, server_default=func.current_timestamp(), onupdate=func.current_timestamp(), comment='수정일시'), + mysql_engine='InnoDB', + mysql_charset='utf8mb4', ) diff --git a/lib/common.py b/lib/common.py index 42dbb57..2ef9416 100644 --- a/lib/common.py +++ b/lib/common.py @@ -1,5 +1,6 @@ # common.py import os, yaml +import logging def load_config(): """ @@ -8,3 +9,13 @@ def load_config(): path = os.path.join(os.path.dirname(__file__), '..', 'conf', 'config.yaml') with open(path, encoding='utf-8') as f: return yaml.safe_load(f) + +def get_logger(name): + logger = logging.getLogger(name) + if not logger.handlers: + handler = logging.StreamHandler() + formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s') + handler.setFormatter(formatter) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + return logger diff --git a/lib/file_watch.py b/lib/file_watch.py new file mode 100644 index 0000000..4fc7ba6 --- /dev/null +++ b/lib/file_watch.py @@ -0,0 +1,81 @@ +import time +import os +from watchdog.observers import Observer +from watchdog.events import FileSystemEventHandler +import threading + +import pos_update_bill +import pos_update_daily_product + +DATA_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '../data')) + +FILE_EXTENSIONS = ('.xls', '.xlsx') +BILL_PREFIX = "영수증별매출상세현황" +DAILY_PRODUCT_PREFIX = "일자별 (상품별)" + +class NewFileHandler(FileSystemEventHandler): + def __init__(self): + super().__init__() + self._lock = threading.Lock() + self._processing_files = set() + + def on_created(self, event): + if event.is_directory: + return + filepath = event.src_path + filename = os.path.basename(filepath) + if not filename.endswith(FILE_EXTENSIONS): + return + + # 처리 대상 여부 확인 + if filename.startswith(BILL_PREFIX) or filename.startswith(DAILY_PRODUCT_PREFIX): + print(f"[WATCHER] 신규 파일 감지: {filename}") + threading.Thread(target=self.process_file, args=(filepath, filename), daemon=True).start() + + def process_file(self, filepath, filename): + with self._lock: + if filename in self._processing_files: + print(f"[WATCHER] {filename} 이미 처리 중") + return + self._processing_files.add(filename) + + try: + time.sleep(3) # 파일 쓰기 완료 대기 + + print(f"[WATCHER] 파일 처리 시작: {filename}") + if filename.startswith(BILL_PREFIX): + pos_update_bill.main() + elif filename.startswith(DAILY_PRODUCT_PREFIX): + pos_update_daily_product.main() + else: + print(f"[WATCHER] 처리 대상이 아님: {filename}") + return + + except Exception as e: + print(f"[WATCHER] 처리 중 오류 발생: {filename} / {e}") + else: + try: + os.remove(filepath) + print(f"[WATCHER] 파일 처리 완료 및 삭제: {filename}") + except Exception as e: + print(f"[WATCHER] 파일 삭제 실패: {filename} / {e}") + finally: + with self._lock: + self._processing_files.discard(filename) + +def start_watching(): + print(f"[WATCHER] '{DATA_DIR}' 폴더 감시 시작") + event_handler = NewFileHandler() + observer = Observer() + observer.schedule(event_handler, DATA_DIR, recursive=False) + observer.start() + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("[WATCHER] 감시 종료 요청 수신, 종료 중...") + observer.stop() + observer.join() + +if __name__ == "__main__": + start_watching() diff --git a/lib/holiday.py b/lib/holiday.py index 642c049..340b6eb 100644 --- a/lib/holiday.py +++ b/lib/holiday.py @@ -4,7 +4,7 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) import yaml import requests import xml.etree.ElementTree as ET -from datetime import datetime, date +from datetime import date, datetime, timedelta from sqlalchemy import select, insert, delete # config.yaml 경로 및 로딩 @@ -134,8 +134,40 @@ def is_korean_holiday(dt: date) -> bool: finally: session.close() +def get_holiday_dates(start_date: date, end_date: date) -> set[date]: + """특정 기간 내의 휴일 목록 반환""" + session = db.get_session() + try: + stmt = select(holiday_table.c.date).where( + holiday_table.c.date.between(start_date.strftime("%Y%m%d"), end_date.strftime("%Y%m%d")) + ) + results = session.execute(stmt).scalars().all() + return set(datetime.strptime(d, "%Y%m%d").date() for d in results) + finally: + session.close() + + +def get_weekday_dates(start_date: date, end_date: date) -> set[date]: + """특정 기간 중 평일(월~금 & 비휴일) 목록 반환""" + holiday_dates = get_holiday_dates(start_date, end_date) + result = set() + curr = start_date + while curr <= end_date: + if curr.weekday() < 5 and curr not in holiday_dates: # 월(0)~금(4) + result.add(curr) + curr += timedelta(days=1) + return result + if __name__ == "__main__": - print("📌 특일정보 초기화 시작") + print("📌 휴일 테스트 시작") init_holidays() - print("✅ 특일정보 초기화 완료") + + from datetime import date + start = date(2025, 1, 1) + end = date(2025, 12, 31) + + holidays = get_holiday_dates(start, end) + print(f"🔍 {start} ~ {end} 사이 휴일 {len(holidays)}건") + for d in sorted(holidays): + print(" -", d) \ No newline at end of file diff --git a/lib/pos_update_bill.py b/lib/pos_update_bill.py new file mode 100644 index 0000000..7089281 --- /dev/null +++ b/lib/pos_update_bill.py @@ -0,0 +1,263 @@ +""" +영수증별매출상세현황 엑셀파일을 기반으로 MariaDB에 데이터 업데이트 + +1. 파일은 ./data 폴더에 위치 (파일명: '영수증별매출상세현황*.xls[x]') +2. 중복된 데이터는 update 처리됨 (on duplicate key update) +3. 처리 후 파일 자동 삭제 (파일 삭제 로직은 필요시 추가 가능) +""" + +import os +import sys +import re +import pandas as pd +from datetime import datetime +from sqlalchemy.dialects.mysql import insert +from sqlalchemy import select + +# 상위 경로를 sys.path에 추가해 프로젝트 내 모듈 임포트 가능하게 설정 +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +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') + +# 처리 대상 파일명 패턴: '영수증별매출상세현황'으로 시작하고 .xls 또는 .xlsx 확장자 +FILE_PATTERN = re.compile(r"^영수증별매출상세현황.*\.xls[x]?$") + +# 엑셀 상단 A3셀 형식 예: "조회일자 : 2025-07-27 매장선택 : [V83728] 퍼스트(삐아또" +HEADER_PATTERN = re.compile(r"조회일자\s*:\s*(\d{4}-\d{2}-\d{2})\s+매장선택\s*:\s*\[(\w+)]\s*(.+)") + +def extract_file_info(filepath: str): + """ + 엑셀 파일 상단에서 조회일자, 매장코드, 매장명을 추출한다. + A3 셀 (2행 0열, 0부터 시작 기준) 데이터를 정규식으로 파싱. + + Args: + filepath (str): 엑셀파일 경로 + + Returns: + tuple: (sale_date: date, shop_cd: str, shop_name: str) + + Raises: + ValueError: 정규식 매칭 실패 시 + """ + print(f"[INFO] {filepath} 상단 조회일자 및 매장 정보 추출 시작") + df_head = pd.read_excel(filepath, header=None, nrows=5) + first_row = df_head.iloc[2, 0] # 3행 A열 (0-based index) + + match = HEADER_PATTERN.search(str(first_row)) + if not match: + raise ValueError(f"[ERROR] 조회일자 및 매장 정보 추출 실패: {filepath}") + + sale_date = datetime.strptime(match.group(1), "%Y-%m-%d").date() + shop_cd = match.group(2) + shop_name = match.group(3).strip() + print(f"[INFO] 추출된 조회일자: {sale_date}, 매장코드: {shop_cd}, 매장명: {shop_name}") + return sale_date, shop_cd, shop_name + +def load_excel_data(filepath: str): + """ + 지정한 컬럼만 읽고, 헤더는 6번째 행(0-based index 5)으로 지정. + '합계'라는 단어가 '포스번호' 컬럼에 있으면 그 행부터 제거한다. + + Args: + filepath (str): 엑셀파일 경로 + + Returns: + pd.DataFrame: 전처리된 데이터프레임 + + Raises: + ValueError: 필수 컬럼 누락 시 + """ + print(f"[INFO] {filepath} 데이터 영역 로드 시작") + usecols = [ + "포스번호", "영수증번호", "구분", "테이블명", "최초주문", "결제시각", + "상품코드", "바코드", "상품명", "수량", "총매출액", "ERP 매핑코드", + "비고", "할인액", "할인구분", "실매출액", "가액", "부가세" + ] + # header=5 => 6번째 행이 컬럼명 + df = pd.read_excel(filepath, header=5, dtype=str) + # 컬럼명 좌우 공백 제거 + df.columns = df.columns.str.strip() + + # '합계'인 행의 인덱스 찾기 및 제거 + if '합계' in df['포스번호'].values: + idx = df[df['포스번호'] == '합계'].index[0] + df = df.loc[:idx-1] + print(f"[INFO] '합계' 행 이후 데이터 제거: {idx}번째 행부터 제외") + + # 필수 컬럼 존재 여부 체크 + if not set(usecols).issubset(df.columns): + raise ValueError(f"[ERROR] 필수 컬럼 누락: 현재 컬럼 {df.columns.tolist()}") + + df = df[usecols] + print(f"[INFO] {filepath} 데이터 영역 로드 완료, 데이터 건수: {len(df)}") + return df + +def normalize_data(df: pd.DataFrame, sale_date, shop_cd): + """ + 컬럼명을 내부 규칙에 맞게 변경하고, 숫자 필드를 정수형으로 변환한다. + 조회일자와 매장코드를 데이터프레임에 추가. + + Args: + df (pd.DataFrame): 원본 데이터프레임 + sale_date (date): 조회일자 + shop_cd (str): 매장코드 + + Returns: + pd.DataFrame: 정규화된 데이터프레임 + """ + print(f"[INFO] 데이터 정규화 시작") + def to_int(x): + try: + return int(str(x).replace(",", "").strip()) + except: + return 0 + + df.rename(columns={ + "포스번호": "pos_no", + "영수증번호": "bill_no", + "구분": "division", + "테이블명": "table_no", + "최초주문": "order_time", + "결제시각": "pay_time", + "상품코드": "product_cd", + "바코드": "barcode", + "상품명": "product_name", + "수량": "qty", + "총매출액": "tot_sale_amt", + "ERP 매핑코드": "erp_cd", + "비고": "remark", + "할인액": "dc_amt", + "할인구분": "dc_type", + "실매출액": "dcm_sale_amt", + "가액": "net_amt", + "부가세": "vat_amt" + }, inplace=True) + + df["sale_date"] = sale_date + df["shop_cd"] = shop_cd + + # 숫자형 컬럼 정수 변환 + int_fields = ["qty", "tot_sale_amt", "dc_amt", "dcm_sale_amt", "net_amt", "vat_amt"] + for field in int_fields: + df[field] = df[field].apply(to_int) + + # pos_no, bill_no는 반드시 int로 변환 + df["pos_no"] = df["pos_no"].astype(int) + df["bill_no"] = df["bill_no"].astype(int) + + print(f"[INFO] 데이터 정규화 완료") + return df + +def upsert_data(df: pd.DataFrame, batch_size: int = 500) -> int: + """ + SQLAlchemy insert 구문을 사용하여 + 중복 PK 발생 시 update 처리 (on duplicate key update) + 대량 데이터는 batch_size 단위로 나누어 처리 + + Args: + df (pd.DataFrame): DB에 삽입할 데이터 + batch_size (int): 한번에 처리할 데이터 건수 (기본 500) + + Returns: + int: 영향 받은 총 행 수 + """ + print(f"[INFO] DB 저장 시작") + df = df.where(pd.notnull(df), None) # NaN → None 변환 + + engine = db.get_engine() + metadata = db_schema.metadata + table = db_schema.pos_billdata + total_affected = 0 + + with engine.connect() as conn: + for start in range(0, len(df), batch_size): + batch_df = df.iloc[start:start+batch_size] + records = batch_df.to_dict(orient="records") + insert_stmt = insert(table).values(records) + + update_fields = { + col.name: insert_stmt.inserted[col.name] + for col in table.columns + if col.name not in table.primary_key.columns + } + upsert_stmt = insert_stmt.on_duplicate_key_update(update_fields) + + try: + result = conn.execute(upsert_stmt) + conn.commit() + total_affected += result.rowcount + print(f"[INFO] 배치 처리 완료: {start} ~ {start+len(records)-1} / 영향 행 수: {result.rowcount}") + except Exception as e: + print(f"[ERROR] 배치 처리 실패: {start} ~ {start+len(records)-1} / 오류: {e}") + # 필요 시 raise 하거나 continue로 다음 배치 진행 가능 + raise + + print(f"[INFO] DB 저장 전체 완료, 총 영향 행 수: {total_affected}") + return total_affected + + +def ensure_shop_exists(shop_cd, shop_name): + """ + 매장 정보 테이블에 매장코드가 없으면 신규 등록한다. + + Args: + shop_cd (str): 매장 코드 + shop_name (str): 매장 명 + """ + print(f"[INFO] 매장 존재 여부 확인: {shop_cd}") + engine = db.get_engine() + conn = engine.connect() + shop_table = db_schema.pos_shop_name + + try: + query = shop_table.select().where(shop_table.c.shop_cd == shop_cd) + result = conn.execute(query).fetchone() + if result is None: + print(f"[INFO] 신규 매장 등록: {shop_cd} / {shop_name}") + ins = shop_table.insert().values(shop_cd=shop_cd, shop_name=shop_name) + conn.execute(ins) + conn.commit() + else: + print(f"[INFO] 기존 매장 존재: {shop_cd}") + except Exception as e: + print(f"[ERROR] 매장 확인/등록 실패: {e}") + raise + finally: + conn.close() + +def main(): + """ + 대상 데이터 파일 목록을 찾고, 파일별로 처리 진행한다. + 처리 성공 시 저장 건수를 출력하고, 실패 시 오류 메시지 출력. + """ + files = [f for f in os.listdir(DATA_DIR) if FILE_PATTERN.match(f)] + print(f"[INFO] 발견된 파일 {len(files)}개") + + for file in files: + filepath = os.path.join(DATA_DIR, file) + print(f"[INFO] 파일: {file} 처리 시작") + + try: + sale_date, shop_cd, shop_name = extract_file_info(filepath) + ensure_shop_exists(shop_cd, shop_name) + + raw_df = load_excel_data(filepath) + df = normalize_data(raw_df, sale_date, shop_cd) + + affected = upsert_data(df) + print(f"[DONE] 처리 완료: {file} / 저장 건수: {affected}") + + # 처리 완료 후 파일 삭제 (필요 시 활성화) + # os.remove(filepath) + # print(f"[INFO] 처리 완료 후 파일 삭제: {file}") + + except Exception as e: + print(f"[ERROR] {file} 처리 실패: {e}") + +if __name__ == "__main__": + main() diff --git a/lib/pos_update_gui.py b/lib/pos_update_daily_product.py similarity index 70% rename from lib/pos_update_gui.py rename to lib/pos_update_daily_product.py index b60dae6..9ddd326 100644 --- a/lib/pos_update_gui.py +++ b/lib/pos_update_daily_product.py @@ -1,22 +1,16 @@ # 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 +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from conf import db, db_schema from lib.common import load_config @@ -29,14 +23,12 @@ def update_pos_table(engine, table, df): 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: @@ -50,7 +42,6 @@ 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": @@ -73,8 +64,7 @@ def process_file(filepath, table, engine): '실매출액': 'actual_amount' }, inplace=True) - if 'idx' in df.columns: - df = df.drop(columns=['idx']) + df.drop(columns=[col for col in ['idx'] if col in df.columns], inplace=True) df['date'] = pd.to_datetime(df['date']).dt.date df['barcode'] = df['barcode'].astype(int) @@ -98,7 +88,6 @@ def process_file(filepath, table, engine): 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 @@ -114,36 +103,15 @@ def batch_process_files(table, engine): total_rows += count try: os.remove(full_path) - print(f"[INFO] 처리 완료 후 파일 삭제: {fname}") + print(f"[INFO] 파일 삭제 완료: {fname}") deleted_files += 1 except Exception as e: - print(f"[WARN] 파일 삭제 실패: {fname}, {e}") + print(f"[WARN] 파일 삭제 실패: {fname} / {e}") - print(f"[INFO] 처리된 전체 데이터 건수: {total_rows}") + print(f"[INFO] 총 처리 데이터 건수: {total_rows}") print(f"[INFO] 삭제된 파일 수: {deleted_files}") 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: @@ -154,18 +122,7 @@ def main(): batch_done = batch_process_files(table, engine) if not batch_done: - # 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() + print("[INFO] 처리할 데이터가 없습니다.") if __name__ == "__main__": main() diff --git a/lib/pos_view_gui.py b/lib/pos_view_gui.py index 0897885..4c06faf 100644 --- a/lib/pos_view_gui.py +++ b/lib/pos_view_gui.py @@ -9,12 +9,13 @@ from tkcalendar import DateEntry from datetime import datetime, timedelta from sqlalchemy import select, func, between from conf import db_schema, db +from lib import holiday # 휴일 기능 -# Windows DPI Awareness 설정 (윈도우 전용) +# Windows DPI Awareness 설정 if sys.platform == "win32": import ctypes try: - ctypes.windll.shcore.SetProcessDpiAwareness(1) # SYSTEM_AWARE = 1 + ctypes.windll.shcore.SetProcessDpiAwareness(1) except Exception: pass @@ -26,30 +27,23 @@ class PosViewGUI(ctk.CTk): super().__init__() self.title("POS 데이터 조회") - self.geometry("900x500") - self.configure(fg_color="#f0f0f0") # 배경색 맞춤 + self.geometry("1100x700") + self.configure(fg_color="#f0f0f0") ctk.set_appearance_mode("light") ctk.set_default_color_theme("blue") - # 폰트 세팅 - NanumGothic이 없으면 Arial 대체 try: self.label_font = ("NanumGothic", 13) except Exception: self.label_font = ("Arial", 13) - # Treeview 스타일 설정 (ttk 스타일) style = ttk.Style(self) style.theme_use('default') - style.configure("Treeview", - font=("NanumGothic", 12), - rowheight=30) # 높이 조절로 글씨 깨짐 방지 - style.configure("Treeview.Heading", - font=("NanumGothic", 13, "bold")) + style.configure("Treeview", font=("NanumGothic", 12), rowheight=30) + style.configure("Treeview.Heading", font=("NanumGothic", 13, "bold")) - # --- 위젯 배치 --- - - # 날짜 범위 + # 날짜 필터 ctk.CTkLabel(self, text="시작일:", anchor="w", font=self.label_font, fg_color="#f0f0f0")\ .grid(row=0, column=0, padx=10, pady=5, sticky="e") self.start_date_entry = DateEntry(self, width=12, background='darkblue', foreground='white') @@ -60,6 +54,18 @@ class PosViewGUI(ctk.CTk): self.end_date_entry = DateEntry(self, width=12, background='darkblue', foreground='white') self.end_date_entry.grid(row=0, column=3, padx=10, pady=5, sticky="w") + # 날짜유형 라디오버튼 + self.date_filter_var = ctk.StringVar(value="전체") + ctk.CTkLabel(self, text="날짜유형:", font=self.label_font, fg_color="#f0f0f0")\ + .grid(row=0, column=4, padx=(10, 0), pady=5, sticky="e") + + ctk.CTkRadioButton(self, text="전체", variable=self.date_filter_var, value="전체")\ + .grid(row=0, column=5, padx=2, pady=5, sticky="w") + ctk.CTkRadioButton(self, text="휴일", variable=self.date_filter_var, value="휴일")\ + .grid(row=0, column=6, padx=2, pady=5, sticky="w") + ctk.CTkRadioButton(self, text="평일", variable=self.date_filter_var, value="평일")\ + .grid(row=0, column=7, padx=2, pady=5, sticky="w") + # 대분류 ctk.CTkLabel(self, text="대분류 :", anchor="w", font=self.label_font, fg_color="#f0f0f0")\ .grid(row=1, column=0, padx=10, pady=5, sticky="e") @@ -82,9 +88,9 @@ class PosViewGUI(ctk.CTk): # 조회 버튼 self.search_btn = ctk.CTkButton(self, text="조회", command=self.search, fg_color="#0d6efd", hover_color="#0b5ed7", text_color="white") - self.search_btn.grid(row=3, column=0, columnspan=4, pady=10) + self.search_btn.grid(row=3, column=0, columnspan=8, pady=10) - # 결과 Treeview + # 상품별 트리뷰 self.DISPLAY_COLUMNS = ['ca01', 'ca02', 'ca03', 'name', 'qty', 'tot_amount', 'tot_discount', 'actual_amount'] self.COLUMN_LABELS = { 'ca01': '대분류', @@ -97,30 +103,40 @@ class PosViewGUI(ctk.CTk): 'actual_amount': '실매출액' } - self.tree = ttk.Treeview(self, columns=self.DISPLAY_COLUMNS, show='headings', height=15) + self.tree = ttk.Treeview(self, columns=self.DISPLAY_COLUMNS, show='headings', height=12) for col in self.DISPLAY_COLUMNS: self.tree.heading(col, text=self.COLUMN_LABELS[col]) self.tree.column(col, width=120, anchor='center') - self.tree.grid(row=4, column=0, columnspan=4, padx=10, pady=10, sticky="nsew") + self.tree.grid(row=4, column=0, columnspan=8, padx=10, pady=10, sticky="nsew") + + # 날짜 요약 트리뷰 + self.date_tree = ttk.Treeview(self, columns=['date', 'qty', 'tot_amount', 'actual_amount'], show='headings', height=6) + self.date_tree.heading('date', text='일자') + self.date_tree.heading('qty', text='수량합') + self.date_tree.heading('tot_amount', text='총매출합') + self.date_tree.heading('actual_amount', text='실매출합') + + for col in ['date', 'qty', 'tot_amount', 'actual_amount']: + self.date_tree.column(col, width=150, anchor='center') + + self.date_tree.grid(row=5, column=0, columnspan=8, padx=10, pady=(0, 10), sticky="nsew") - # 그리드 가중치 설정 (창 크기에 따라 트리뷰 확장) self.grid_rowconfigure(4, weight=1) - for col_index in range(4): + self.grid_rowconfigure(5, weight=1) + for col_index in range(8): self.grid_columnconfigure(col_index, weight=1) - # 날짜 기본값 설정 (전날부터 7일 전까지) + # 날짜 기본값 end_date = datetime.today().date() - timedelta(days=1) start_date = end_date - timedelta(days=6) self.start_date_entry.set_date(start_date) self.end_date_entry.set_date(end_date) - # 초기 대분류, 소분류 콤보박스 값 불러오기 self.load_ca01_options() def on_ca01_selected(self, value): - # print("대분류 선택됨:", value) 디버깅용 self.load_ca03_options() - + def load_ca01_options(self): start_date = self.start_date_entry.get_date() end_date = self.end_date_entry.get_date() @@ -148,7 +164,7 @@ class PosViewGUI(ctk.CTk): result = conn.execute(stmt) ca03_list = [row[0] for row in result.fetchall()] self.ca03_combo.configure(values=['전체'] + ca03_list) - self.ca03_combo.set('전체') # 항상 기본값으로 초기화 + self.ca03_combo.set('전체') def search(self): start_date = self.start_date_entry.get_date() @@ -156,8 +172,34 @@ class PosViewGUI(ctk.CTk): ca01_val = self.ca01_combo.get() ca03_val = self.ca03_combo.get() name_val = self.name_entry.get().strip() + date_filter = self.date_filter_var.get() + + print("🔍 date_filter:", date_filter, + "| start:", start_date, "end:", end_date) + if date_filter == "휴일": + valid_dates = holiday.get_holiday_dates(start_date, end_date) + print("🚩 반환된 휴일 날짜 리스트:", valid_dates) + + conditions = [] + + if date_filter == "전체": + conditions.append(between(pos_table.c.date, start_date, end_date)) + else: + if date_filter == "휴일": + valid_dates = holiday.get_holiday_dates(start_date, end_date) + elif date_filter == "평일": + valid_dates = holiday.get_weekday_dates(start_date, end_date) + else: + valid_dates = set() + + if not valid_dates: + messagebox.showinfo("알림", f"{date_filter}에 해당하는 데이터가 없습니다.") + self.tree.delete(*self.tree.get_children()) + self.date_tree.delete(*self.date_tree.get_children()) + return + + conditions.append(pos_table.c.date.in_(valid_dates)) - conditions = [between(pos_table.c.date, start_date, end_date)] if ca01_val != '전체': conditions.append(pos_table.c.ca01 == ca01_val) if ca03_val != '전체': @@ -166,6 +208,7 @@ class PosViewGUI(ctk.CTk): conditions.append(pos_table.c.name.like(f"%{name_val}%")) with engine.connect() as conn: + # 상품별 stmt = select( pos_table.c.ca01, pos_table.c.ca02, @@ -179,11 +222,42 @@ class PosViewGUI(ctk.CTk): result = conn.execute(stmt).mappings().all() + # 날짜별 요약 + date_stmt = select( + pos_table.c.date, + func.sum(pos_table.c.qty).label("qty"), + func.sum(pos_table.c.tot_amount).label("tot_amount"), + func.sum(pos_table.c.actual_amount).label("actual_amount") + ).where(*conditions).group_by(pos_table.c.date).order_by(pos_table.c.date) + + date_summary = conn.execute(date_stmt).mappings().all() + + # 트리뷰 초기화 self.tree.delete(*self.tree.get_children()) + self.date_tree.delete(*self.date_tree.get_children()) + + # 상품별 출력 for row in result: values = tuple(row[col] for col in self.DISPLAY_COLUMNS) self.tree.insert('', 'end', values=values) + # 날짜별 출력 + total_qty = total_amount = total_actual = 0 + for row in date_summary: + self.date_tree.insert('', 'end', values=( + row['date'].strftime("%Y-%m-%d"), + row['qty'], + row['tot_amount'], + row['actual_amount'] + )) + total_qty += row['qty'] + total_amount += row['tot_amount'] + total_actual += row['actual_amount'] + + # 총합계 추가 + self.date_tree.insert('', 'end', values=("총합계", total_qty, total_amount, total_actual)) + + if __name__ == "__main__": try: import tkcalendar diff --git a/requirements.txt b/requirements.txt index 6b0461a..aed05ad 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,4 +12,5 @@ statsmodels scikit-learn customtkinter tkcalendar -tabulate \ No newline at end of file +tabulate +watchdog \ No newline at end of file