thug0bin/pharmacy-pos-qr-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: AI 기반 제품 카테고리 자동 태깅 및 UI 표시

Browse Source 
- OpenAI GPT-4o-mini로 31개 제품 자동 분류 (100% 커버리지)
- 관리자 페이지 사용자 상세 모달에 카테고리 뱃지 추가
- BARCODE 기반 제품-카테고리 매핑 (many-to-many)
- 카테고리별 색상 구분 (10가지 그라디언트 디자인)
- 제품 수동 분류 도구 추가 (update_product_category.py)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
시골약사 2026-01-23 23:56:28 +09:00
parent 6026f0aae8
commit 70d18a1954
4 changed files with 411 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

232
backend/ai_tag_products.py Normal file
View File

@ -0,0 +1,232 @@
"""
AI 기반 제품 자동 카테고리 태깅
"""
import sys
import os
# UTF-8 인코딩 강제 (Windows 한글 깨짐 방지)
if sys.platform == 'win32':
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
import sqlite3
import json
from openai import OpenAI
from dotenv import load_dotenv
# .env 파일 로드
load_dotenv()
# OpenAI 클라이언트 초기화
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
def get_uncategorized_products():
"""카테고리가 없는 제품 조회"""
db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
# 카테고리가 없는 제품 조회
cursor.execute("""
SELECT p.barcode, p.product_name
FROM product_master p
LEFT JOIN product_category_mapping m ON p.barcode = m.barcode
WHERE m.barcode IS NULL
ORDER BY p.product_name
""")
products = cursor.fetchall()
return [(barcode, name) for barcode, name in products]
finally:
conn.close()
def get_available_categories():
"""사용 가능한 카테고리 목록 조회"""
db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
cursor.execute("SELECT category_name, description FROM product_categories ORDER BY category_name")
categories = cursor.fetchall()
return categories
finally:
conn.close()
def ai_categorize_product(product_name, available_categories):
"""OpenAI API로 제품 카테고리 분류"""
# 카테고리 목록 포맷팅
category_list = "\n".join([f"- {cat[0]}: {cat[1] or '(설명 없음)'}" for cat in available_categories])
prompt = f"""당신은 약국 제품 분류 전문가입니다.
제품명: {product_name}
아래 카테고리 목록에서 이 제품에 가장 적합한 카테고리를 1~3개 선택하고, 각 카테고리의 관련도 점수(0.0~1.0)를 매겨주세요.
사용 가능한 카테고리:
{category_list}
응답은 반드시 아래 JSON 형식으로만 작성해주세요:
{{
"categories": [
{{"name": "카테고리명", "score": 1.0, "reason": "선택 이유"}},
{{"name": "카테고리명", "score": 0.8, "reason": "선택 이유"}}
]
}}
주의사항:
1. 가장 관련 있는 카테고리 1~3개만 선택
2. score는 0.0~1.0 범위 (주 카테고리는 1.0, 부 카테고리는 0.5~0.9)
3. 카테고리명은 위 목록에 있는 것만 사용
4. JSON 형식 외 다른 텍스트 추가하지 말 것"""
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "당신은 의약품 분류 전문가입니다. 항상 JSON 형식으로만 응답하세요."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=500
)
result_text = response.choices[0].message.content.strip()
# JSON 파싱
result = json.loads(result_text)
return result['categories']
except json.JSONDecodeError as e:
print(f"[ERROR] JSON 파싱 실패: {e}")
print(f"응답: {result_text}")
return []
except Exception as e:
print(f"[ERROR] AI 분류 실패: {e}")
return []
def update_product_categories(barcode, categories):
"""제품 카테고리 매핑 업데이트"""
db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
for cat in categories:
# 카테고리가 존재하는지 확인
cursor.execute(
"SELECT category_name FROM product_categories WHERE category_name = ?",
(cat['name'],)
)
if cursor.fetchone():
cursor.execute("""
INSERT OR REPLACE INTO product_category_mapping
(barcode, category_name, relevance_score)
VALUES (?, ?, ?)
""", (barcode, cat['name'], cat['score']))
conn.commit()
return True
except Exception as e:
print(f"[ERROR] DB 업데이트 실패: {e}")
conn.rollback()
return False
finally:
conn.close()
def main():
"""메인 실행"""
print("="*80)
print("AI 기반 제품 자동 카테고리 태깅")
print("="*80)
# 1. 카테고리가 없는 제품 조회
print("\n1단계: 카테고리 없는 제품 조회 중...")
uncategorized = get_uncategorized_products()
if not uncategorized:
print("[OK] 모든 제품이 이미 카테고리가 있습니다!")
return
print(f"[OK] {len(uncategorized)}개 제품 발견\n")
# 2. 사용 가능한 카테고리 조회
print("2단계: 사용 가능한 카테고리 조회 중...")
categories = get_available_categories()
print(f"[OK] {len(categories)}개 카테고리 사용 가능\n")
# 3. 각 제품 AI 태깅
print("3단계: AI 기반 제품 분류 시작...\n")
print("-"*80)
success_count = 0
fail_count = 0
for idx, (barcode, product_name) in enumerate(uncategorized, 1):
print(f"\n[{idx}/{len(uncategorized)}] {product_name}")
# AI로 카테고리 분류
suggested_categories = ai_categorize_product(product_name, categories)
if not suggested_categories:
print(f" [SKIP] AI 분류 실패")
fail_count += 1
continue
# 카테고리 출력
for cat in suggested_categories:
print(f" ├─ {cat['name']} (관련도: {cat['score']:.1f}) - {cat.get('reason', '')}")
# DB 업데이트
if update_product_categories(barcode, suggested_categories):
print(f" └─ [OK] 카테고리 매핑 완료")
success_count += 1
else:
print(f" └─ [ERROR] DB 업데이트 실패")
fail_count += 1
# 4. 결과 요약
print("\n" + "="*80)
print("AI 태깅 완료")
print("="*80)
print(f"성공: {success_count}개")
print(f"실패: {fail_count}개")
print("="*80)
# 5. 최종 통계
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]
cursor.execute("SELECT COUNT(DISTINCT barcode) FROM product_category_mapping")
categorized = cursor.fetchone()[0]
print(f"\n최종 통계:")
print(f" 전체 제품: {total}개")
print(f" 카테고리 있는 제품: {categorized}개")
print(f" 카테고리 없는 제품: {total - categorized}개")
finally:
conn.close()
if __name__ == '__main__':
main()

36
backend/app.py
View File

@ -818,9 +818,10 @@ def admin_user_detail(user_id):
else:
transaction_date = '-'
# SALE_SUB + CD_GOODS JOIN
# SALE_SUB + CD_GOODS JOIN (BARCODE 추가)
sale_items_query = text("""
SELECT
S.BARCODE,
S.DrugCode,
ISNULL(G.GoodsName, '(약품명 없음)') AS goods_name,
S.SL_NM_item AS quantity,
@ -837,17 +838,36 @@ def admin_user_detail(user_id):
{'transaction_id': transaction_id}
).fetchall()
# 상품 리스트 변환
items = [
{
# 상품 리스트 변환 (카테고리 포함)
items = []
for item in items_raw:
barcode = item.BARCODE
# SQLite에서 제품 카테고리 조회
categories = []
if barcode:
cursor.execute("""
SELECT category_name, relevance_score
FROM product_category_mapping
WHERE barcode = ?
ORDER BY relevance_score DESC
""", (barcode,))
for cat_row in cursor.fetchall():
categories.append({
'name': cat_row[0],
'score': cat_row[1]
})
items.append({
'code': item.DrugCode,
'barcode': barcode,
'name': item.goods_name,
'qty': int(item.quantity or 0),
'price': int(item.price or 0),
'total': int(item.total or 0)
}
for item in items_raw
]
'total': int(item.total or 0),
'categories': categories
})
# 상품 요약 생성 ("첫번째상품명 외 N개")
if items:

45
backend/templates/admin.html
View File

@ -352,6 +352,38 @@
display: none;
}
}
/* 카테고리 뱃지 스타일 */
.category-badge {
display: inline-block;
padding: 4px 10px;
margin: 2px 4px 2px 0;
border-radius: 12px;
font-size: 11px;
font-weight: 600;
white-space: nowrap;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: #ffffff;
box-shadow: 0 2px 4px rgba(102, 126, 234, 0.3);
transition: all 0.2s ease;
}
.category-badge:hover {
transform: translateY(-1px);
box-shadow: 0 4px 8px rgba(102, 126, 234, 0.4);
}
/* 제품 카테고리별 색상 */
.category-badge.cat-진통제 { background: linear-gradient(135deg, #f06292 0%, #e91e63 100%); }
.category-badge.cat-소화제 { background: linear-gradient(135deg, #64b5f6 0%, #1976d2 100%); }
.category-badge.cat-감기약 { background: linear-gradient(135deg, #4db6ac 0%, #00796b 100%); }
.category-badge.cat-복합비타민 { background: linear-gradient(135deg, #ffb74d 0%, #f57c00 100%); }
.category-badge.cat-피로회복제 { background: linear-gradient(135deg, #a1887f 0%, #6d4c41 100%); }
.category-badge.cat-소염제 { background: linear-gradient(135deg, #ff8a65 0%, #d84315 100%); }
.category-badge.cat-연고 { background: linear-gradient(135deg, #90a4ae 0%, #546e7a 100%); }
.category-badge.cat-파스 { background: linear-gradient(135deg, #81c784 0%, #388e3c 100%); }
.category-badge.cat-간영양제 { background: linear-gradient(135deg, #ba68c8 0%, #8e24aa 100%); }
.category-badge.cat-위장약 { background: linear-gradient(135deg, #4fc3f7 0%, #0288d1 100%); }
</style>
</head>
<body>
@ -960,10 +992,21 @@
`;
purchase.items.forEach(item => {
// 카테고리 뱃지 생성
let categoriesBadges = '';
if (item.categories && item.categories.length > 0) {
item.categories.forEach(cat => {
categoriesBadges += `<span class="category-badge cat-${cat.name}">${cat.name}</span>`;
});
}
html += `
<tr style="border-bottom: 1px solid #f1f3f5;">
<td style="padding: 10px; font-size: 13px; color: #495057;">${item.code}</td>
<td style="padding: 10px; font-size: 13px; color: #212529; font-weight: 500;">${item.name}</td>
<td style="padding: 10px; font-size: 13px; color: #212529; font-weight: 500;">
<div style="margin-bottom: 4px;">${item.name}</div>
${categoriesBadges ? `<div style="margin-top: 6px;">${categoriesBadges}</div>` : ''}
</td>
<td style="padding: 10px; text-align: right; font-size: 13px; color: #495057;">${item.qty}</td>
<td style="padding: 10px; text-align: right; font-size: 13px; color: #495057;">${item.price.toLocaleString()}원</td>
<td style="padding: 10px; text-align: right; font-size: 13px; color: #6366f1; font-weight: 600;">${item.total.toLocaleString()}원</td>

107
backend/update_product_category.py Normal file
View File

@ -0,0 +1,107 @@
"""
제품 카테고리 수동 업데이트
"""
import sys
import sqlite3
import os
# UTF-8 인코딩 강제 (Windows 한글 깨짐 방지)
if sys.platform == 'win32':
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
def update_category(product_name, category_name, relevance_score=1.0):
"""제품 카테고리 업데이트"""
db_path = os.path.join(os.path.dirname(__file__), 'db', 'mileage.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
try:
# 제품 바코드 조회
cursor.execute(
"SELECT barcode FROM product_master WHERE product_name = ?",
(product_name,)
)
result = cursor.fetchone()
if not result:
print(f"[ERROR] '{product_name}' 제품을 찾을 수 없습니다.")
return False
barcode = result[0]
# 카테고리 매핑 추가 (이미 있으면 업데이트)
cursor.execute("""
INSERT OR REPLACE INTO product_category_mapping
(barcode, category_name, relevance_score)
VALUES (?, ?, ?)
""", (barcode, category_name, relevance_score))
conn.commit()
print(f"[OK] '{product_name}' → '{category_name}' (관련도: {relevance_score})")
return True
except Exception as e:
print(f"[ERROR] {e}")
conn.rollback()
return False
finally:
conn.close()
def show_stats():
"""제품 통계 조회"""
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]
# 카테고리가 있는 제품 수
cursor.execute("""
SELECT COUNT(DISTINCT barcode)
FROM product_category_mapping
""")
categorized = cursor.fetchone()[0]
# 카테고리가 없는 제품 수
uncategorized = total - categorized
print("\n" + "="*80)
print("제품 통계")
print("="*80)
print(f"전체 제품 수: {total}개")
print(f"카테고리 있는 제품: {categorized}개")
print(f"카테고리 없는 제품: {uncategorized}개")
print("="*80)
# 카테고리별 제품 수
cursor.execute("""
SELECT category_name, COUNT(*) as count
FROM product_category_mapping
GROUP BY category_name
ORDER BY count DESC
""")
print("\n카테고리별 제품 수:")
for cat, count in cursor.fetchall():
print(f" {cat:20} : {count:3}개")
finally:
conn.close()
if __name__ == '__main__':
print("제품 카테고리 업데이트")
print("="*80)
# 소하자임플러스정 → 소화제
update_category("소하자임플러스정", "소화제", 1.0)
# 통계 출력
show_stats()