feat: MSSQL에서 오늘 판매 제품 자동 import
- import_products_from_mssql.py: MSSQL에서 오늘 판매된 제품 가져오기 - 바코드 + 제품명 자동 수집 (30개) - 제품명 기반 카테고리 자동 추론 - view_products.py: product_master 조회 스크립트 - 총 31개 제품 등록 완료 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
39539639b7
commit
5cab3229db
205
backend/import_products_from_mssql.py
Normal file
205
backend/import_products_from_mssql.py
Normal file
@ -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()
|
||||
63
backend/view_products.py
Normal file
63
backend/view_products.py
Normal file
@ -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()
|
||||
Loading…
Reference in New Issue
Block a user