From 5cab3229db7ac8ab8930cd2a59217199f8392cd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=8B=9C=EA=B3=A8=EC=95=BD=EC=82=AC?= Date: Fri, 23 Jan 2026 23:31:49 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20MSSQL=EC=97=90=EC=84=9C=20=EC=98=A4?= =?UTF-8?q?=EB=8A=98=20=ED=8C=90=EB=A7=A4=20=EC=A0=9C=ED=92=88=20=EC=9E=90?= =?UTF-8?q?=EB=8F=99=20import?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - import_products_from_mssql.py: MSSQL에서 오늘 판매된 제품 가져오기 - 바코드 + 제품명 자동 수집 (30개) - 제품명 기반 카테고리 자동 추론 - view_products.py: product_master 조회 스크립트 - 총 31개 제품 등록 완료 Co-Authored-By: Claude Sonnet 4.5 --- backend/import_products_from_mssql.py | 205 ++++++++++++++++++++++++++ backend/view_products.py | 63 ++++++++ 2 files changed, 268 insertions(+) create mode 100644 backend/import_products_from_mssql.py create mode 100644 backend/view_products.py diff --git a/backend/import_products_from_mssql.py b/backend/import_products_from_mssql.py new file mode 100644 index 0000000..f7b016a --- /dev/null +++ b/backend/import_products_from_mssql.py @@ -0,0 +1,205 @@ +""" +MSSQL에서 바코드 제품 데이터를 가져와서 product_master에 채우기 +""" + +import sys +import os +sys.path.insert(0, os.path.dirname(__file__)) + +from db.dbsetup import DatabaseManager +from sqlalchemy import text +import sqlite3 +import json + +def get_products_from_mssql(limit=30): + """MSSQL에서 바코드가 있는 제품 데이터 조회""" + db_manager = DatabaseManager() + + try: + session = db_manager.get_session('PM_PRES') + + # 오늘 판매된 제품 중 바코드가 있는 제품 조회 + query = text(f""" + SELECT TOP {limit} + S.BARCODE, + S.DrugCode, + ISNULL(G.GoodsName, '(약품명 없음)') AS GoodsName, + COUNT(*) as sales_count, + MAX(S.SL_NM_cost_a) as price, + MAX(S.SL_DT_appl) as last_sale_date + FROM SALE_SUB S + LEFT JOIN PM_DRUG.dbo.CD_GOODS G ON S.DrugCode = G.DrugCode + WHERE S.BARCODE IS NOT NULL + AND S.BARCODE != '' + AND G.GoodsName IS NOT NULL + AND S.SL_DT_appl = CONVERT(VARCHAR(8), GETDATE(), 112) + GROUP BY S.BARCODE, S.DrugCode, G.GoodsName + ORDER BY COUNT(*) DESC + """) + + results = session.execute(query).fetchall() + + products = [] + for r in results: + products.append({ + 'barcode': r.BARCODE, + 'drug_code': r.DrugCode, + 'product_name': r.GoodsName, + 'sales_count': r.sales_count, + 'price': float(r.price) if r.price else 0 + }) + + return products + + except Exception as e: + print(f"MSSQL 조회 오류: {e}") + return [] + finally: + db_manager.close_all() + + +def infer_category(product_name): + """제품명으로 카테고리 추론 (간단한 규칙 기반)""" + name = product_name.lower() + + categories = [] + + # 진통/소염 + if any(keyword in name for keyword in ['탁센', '이부', '아스피린', '게보린', '펜잘', '타이레놀']): + categories.append({'name': '진통제', 'score': 1.0}) + if any(keyword in name for keyword in ['소염', '진통']): + categories.append({'name': '진통소염제', 'score': 1.0}) + + # 소화기 + if any(keyword in name for keyword in ['베아제', '훼스탈', '소화', '위', '가스']): + categories.append({'name': '소화제', 'score': 1.0}) + if any(keyword in name for keyword in ['겔포스', '제산']): + categories.append({'name': '제산제', 'score': 1.0}) + + # 감기/호흡기 + if any(keyword in name for keyword in ['감기', '코', '기침', '목']): + categories.append({'name': '감기약', 'score': 1.0}) + if '판콜' in name or '판피린' in name: + categories.append({'name': '감기약', 'score': 1.0}) + + # 비타민 + if any(keyword in name for keyword in ['비타민', '마그', '칼슘', '철분']): + categories.append({'name': '복합비타민', 'score': 1.0}) + if any(keyword in name for keyword in ['간', '우루사', '밀크씨슬']): + categories.append({'name': '간영양제', 'score': 0.9}) + + # 외용 + if any(keyword in name for keyword in ['파스', '동전파스', '신신파스']): + categories.append({'name': '파스', 'score': 1.0}) + if any(keyword in name for keyword in ['안약', '인공눈물']): + categories.append({'name': '안약', 'score': 1.0}) + if any(keyword in name for keyword in ['연고', '크림']): + categories.append({'name': '연고', 'score': 1.0}) + + # 기타 + if any(keyword in name for keyword in ['항히스타민', '알레르기', '지르텍']): + categories.append({'name': '항히스타민제', 'score': 1.0}) + + # 카테고리가 없으면 기본값 + if not categories: + categories.append({'name': '기타', 'score': 0.5}) + + return categories + + +def insert_products_to_sqlite(products): + """SQLite product_master에 제품 삽입""" + db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db') + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + + inserted_count = 0 + skipped_count = 0 + + try: + for product in products: + barcode = product['barcode'] + product_name = product['product_name'] + + # 이미 존재하는지 확인 + cursor.execute("SELECT barcode FROM product_master WHERE barcode = ?", (barcode,)) + if cursor.fetchone(): + print(f"[SKIP] {barcode}: {product_name} (이미 존재)") + skipped_count += 1 + continue + + # 카테고리 추론 + categories = infer_category(product_name) + + # product_master 삽입 + cursor.execute(""" + INSERT INTO product_master + (barcode, product_name, drug_classification, is_verified) + VALUES (?, ?, ?, 0) + """, ( + barcode, + product_name, + '일반의약품' # 기본값 + )) + + # product_category_mapping 삽입 + for cat in categories: + # 카테고리가 존재하는지 확인 + cursor.execute( + "SELECT category_name FROM product_categories WHERE category_name = ?", + (cat['name'],) + ) + if cursor.fetchone(): + cursor.execute(""" + INSERT INTO product_category_mapping + (barcode, category_name, relevance_score) + VALUES (?, ?, ?) + """, (barcode, cat['name'], cat['score'])) + + print(f"[OK] {barcode}: {product_name} -> {[c['name'] for c in categories]}") + inserted_count += 1 + + conn.commit() + + print(f"\n{'='*80}") + print(f"총 {inserted_count}개 제품 추가됨") + print(f"총 {skipped_count}개 제품 스킵됨 (중복)") + print(f"{'='*80}") + + # 전체 제품 수 확인 + cursor.execute("SELECT COUNT(*) FROM product_master") + total_count = cursor.fetchone()[0] + print(f"\n전체 product_master 제품 수: {total_count}개") + + except Exception as e: + print(f"[ERROR] {e}") + conn.rollback() + finally: + conn.close() + + +def main(): + """메인 실행""" + print("="*80) + print("MSSQL에서 제품 데이터 가져오기") + print("="*80) + + # 1. MSSQL에서 제품 조회 + print("\n1단계: MSSQL에서 바코드 제품 30개 조회 중...") + products = get_products_from_mssql(limit=30) + + if not products: + print("[ERROR] MSSQL에서 제품을 가져오지 못했습니다.") + return + + print(f"[OK] {len(products)}개 제품 조회 완료\n") + + # 2. SQLite에 삽입 + print("2단계: SQLite product_master에 삽입 중...\n") + insert_products_to_sqlite(products) + + print("\n완료!") + + +if __name__ == '__main__': + main() diff --git a/backend/view_products.py b/backend/view_products.py new file mode 100644 index 0000000..2ef6ed3 --- /dev/null +++ b/backend/view_products.py @@ -0,0 +1,63 @@ +""" +product_master 제품 조회 +""" + +import sqlite3 +import os +import json + +def view_products(): + """product_master 제품 조회""" + db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db') + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + + try: + # 전체 제품 수 + cursor.execute("SELECT COUNT(*) FROM product_master") + total = cursor.fetchone()[0] + print(f"전체 제품 수: {total}개\n") + + # 카테고리별 제품 수 + print("="*80) + print("카테고리별 제품 수") + print("="*80) + cursor.execute(""" + SELECT + category_name, + COUNT(*) as count + FROM product_category_mapping + GROUP BY category_name + ORDER BY count DESC + """) + for cat, count in cursor.fetchall(): + print(f"{cat:20} : {count:3}개") + + # 제품 목록 (카테고리 포함) + print("\n" + "="*80) + print("제품 목록 (카테고리 포함)") + print("="*80) + cursor.execute(""" + SELECT + p.barcode, + p.product_name, + GROUP_CONCAT(m.category_name, ', ') as categories, + p.is_verified + FROM product_master p + LEFT JOIN product_category_mapping m ON p.barcode = m.barcode + GROUP BY p.barcode + ORDER BY p.product_name + """) + + for row in cursor.fetchall(): + barcode, name, cats, verified = row + verified_mark = "[V]" if verified else "[ ]" + cats_str = cats if cats else "(카테고리 없음)" + print(f"{verified_mark} {name:30} -> {cats_str}") + + finally: + conn.close() + + +if __name__ == '__main__': + view_products()