diff --git a/backend/scripts/fill_weight_from_dosage.py b/backend/scripts/fill_weight_from_dosage.py new file mode 100644 index 0000000..5f6e530 --- /dev/null +++ b/backend/scripts/fill_weight_from_dosage.py @@ -0,0 +1,463 @@ +# -*- coding: utf-8 -*- +""" +APDB weight_min_kg / weight_max_kg 일괄 채우기 +- dosage_instructions에서 (사이즈라벨, 체중구간) 쌍을 파싱 +- APC 레코드의 product_name에 포함된 사이즈 라벨로 매칭 + +매칭 전략: + 1. 제품명에 사이즈 라벨(소형견, 중형견 등)이 있으면 → 해당 체중구간 적용 + 2. 체중 구간이 1개뿐이면 → 전체 APC에 적용 + 3. 다중 구간인데 제품명에 라벨 없으면 → SKIP (안전) + +예외 처리: + - 사료/축산 관련(톤당 kg) → SKIP + - 축산용(max > 60kg) → SKIP + - 체중 구간 파싱 불가 → SKIP + +실행: python scripts/fill_weight_from_dosage.py [--commit] [--verbose] + 기본: dry-run (DB 변경 없음) + --commit: 실제 DB 업데이트 수행 + --verbose: 상세 로그 +""" +import sys +import io +import re +import argparse + +sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') +sys.path.insert(0, 'c:\\Users\\청춘약국\\source\\pharmacy-pos-qr-system\\backend') + +from sqlalchemy import text, create_engine + + +# ───────────────────────────────────────────── +# 1. 사이즈 라벨 정의 +# ───────────────────────────────────────────── + +SIZE_LABELS = { + '초소형견': 'XS', '초소형': 'XS', + '소형견': 'S', '소형': 'S', + '중형견': 'M', '중형': 'M', + '대형견': 'L', '대형': 'L', + '초대형견': 'XL', '초대형': 'XL', +} + +# 제품명에서 사이즈 감지용 (긴 것부터 먼저 매칭) +PRODUCT_NAME_SIZE_PATTERNS = [ + (r'초소형견', 'XS'), + (r'초소형', 'XS'), + (r'소형견', 'S'), + (r'소형', 'S'), + (r'중형견', 'M'), + (r'중형', 'M'), + (r'초대형견', 'XL'), + (r'초대형', 'XL'), + (r'대형견', 'L'), + (r'대형', 'L'), + # 영문/약어 + (r'\bSS\b', 'XS'), + (r'\bXS\b', 'XS'), + (r'[-\s]S\b', 'S'), + (r'\bS\(', 'S'), + (r'[-\s]M\b', 'M'), + (r'\bM\(', 'M'), + (r'[-\s]L\b', 'L'), + (r'\bL\(', 'L'), + (r'\bXL\b', 'XL'), +] + + +# ───────────────────────────────────────────── +# 2. 체중 구간 파싱 +# ───────────────────────────────────────────── + +def strip_html(html_text): + """HTML 태그 제거, 줄 단위 텍스트 반환""" + if not html_text: + return "" + t = html_text.replace('

', '\n').replace('

', '') + t = re.sub(r'<[^>]+>', '', t) + return t + + +def is_livestock_context(text_content): + """축산/사료 관련인지 판단""" + # 톤당 kg은 사료 관련 + if '톤당' in text_content and 'kg' in text_content: + # 체중 구간이 별도로 있는 경우는 반려동물일 수 있음 + if '체중' not in text_content and '형견' not in text_content: + return True + return False + + +def parse_weight_ranges(dosage_instructions): + """ + dosage_instructions에서 (사이즈라벨, 체중min, 체중max) 리스트를 추출. + 체중 구간만 추출하며, 성분 용량은 무시. + + Returns: + list of dict: [{'min': 0, 'max': 11, 'size': 'S', 'label': '소형견'}, ...] + """ + if not dosage_instructions: + return [] + + txt = strip_html(dosage_instructions) + + # 축산/사료 관련 제외 + if is_livestock_context(txt): + return [] + + ranges = [] + seen = set() # (size, min, max) 중복 방지 + + # ── 전처리: 줄 분리된 사이즈+체중 합치기 ── + # HTML 변환 후 빈 줄이 끼어있을 수 있음: + # "소형견 1chewable 68㎍ 57mg\n\n(체중0-11kg)" + # → "소형견 1chewable 68㎍ 57mg (체중0-11kg)" + lines = txt.split('\n') + # 빈 줄 제거한 리스트 (인덱스 보존) + non_empty = [(i, line.strip()) for i, line in enumerate(lines) if line.strip()] + + merged_set = set() # 합쳐진 줄 인덱스 (원본 기준) + merged_lines = [] + + for idx, (orig_i, stripped) in enumerate(non_empty): + if orig_i in merged_set: + continue + # 현재 줄에 사이즈 라벨이 있고, 다음 비어있지 않은 줄이 (체중...) 패턴이면 합치기 + if idx + 1 < len(non_empty): + next_orig_i, next_stripped = non_empty[idx + 1] + if (re.match(r'\(체중', next_stripped) + and re.search(r'(초소형|소형|중형|대형|초대형)견?', stripped)): + merged_lines.append(stripped + ' ' + next_stripped) + merged_set.add(next_orig_i) + continue + merged_lines.append(stripped) + + txt = '\n'.join(merged_lines) + + def add_range(size, wmin, wmax, label): + """중복 방지하며 범위 추가""" + if wmax > 60: # 반려동물 체중 범위 초과 → 축산용 + return + if wmax <= wmin: + return + key = (size, wmin, wmax) + if key not in seen: + seen.add(key) + ranges.append({'min': wmin, 'max': wmax, 'size': size, 'label': label}) + + def get_size(label_text): + """라벨 텍스트 → 사이즈 코드""" + return SIZE_LABELS.get(label_text + '견', SIZE_LABELS.get(label_text)) + + # ── 패턴1: "X형견(체중A-Bkg)" / "X형견 ... (체중A-Bkg)" ── + # 예: "소형견(체중0-11kg)", "중형견(체중12-22kg)" + # 예(줄 합침): "소형견 1chewable 68㎍ 57mg (체중0-11kg)" + for m in re.finditer( + r'(초소형|소형|중형|대형|초대형)견?\s*(?:용\s*)?\(?(?:체중\s*)?' + r'(\d+\.?\d*)\s*[-~~]\s*(\d+\.?\d*)\s*kg', + txt + ): + label = m.group(1) + add_range(get_size(label), float(m.group(2)), float(m.group(3)), label + '견') + + # ── 패턴1b: 같은 줄에 라벨과 (체중...)이 먼 경우 ── + # 예: "소형견 1chewable 68㎍ 57mg 1개월 (체중0-11kg)" + for m in re.finditer( + r'(초소형|소형|중형|대형|초대형)견?\b[^\n]*?\(체중\s*' + r'(\d+\.?\d*)\s*[-~~]\s*(\d+\.?\d*)\s*kg\)', + txt + ): + label = m.group(1) + add_range(get_size(label), float(m.group(2)), float(m.group(3)), label + '견') + + # ── 패턴2: "체중A~Bkg X형견용" ── + # 예: "체중12~22kg 중형견용(M)" + for m in re.finditer( + r'(?:체중\s*)?(\d+\.?\d*)\s*[-~~]\s*(\d+\.?\d*)\s*kg\s*(?:의\s*)?' + r'(초소형|소형|중형|대형|초대형)견?', + txt + ): + label = m.group(3) + add_range(get_size(label), float(m.group(1)), float(m.group(2)), label + '견') + + # ── 패턴3: "Akg이하 X형견" / "~Akg X형견" ── + # 예: "11kg이하 소형견용" + for m in re.finditer( + r'(?:체중\s*)?[~~]?\s*(\d+\.?\d*)\s*kg\s*(?:이하|까지)?\s*(?:의\s*)?' + r'(초소형|소형|중형|대형|초대형)견?', + txt + ): + label = m.group(2) + add_range(get_size(label), 0, float(m.group(1)), label + '견') + + # ── 패턴4: "(Akg~Bkg의 X형견에게)" ── + # 예: "(5.7kg ~11kg의 소형견에게 본제 1정 투여)" + for m in re.finditer( + r'\(\s*(\d+\.?\d*)\s*kg?\s*[-~~]\s*(\d+\.?\d*)\s*kg\s*(?:의\s*)?' + r'(초소형|소형|중형|대형|초대형)견', + txt + ): + label = m.group(3) + add_range(get_size(label), float(m.group(1)), float(m.group(2)), label + '견') + + # ── 패턴4b: "(Akg.Bkg의 X형견에게)" - 마침표 구분자 ── + # 예: "(12kg.22kg의 중형견에게)" + for m in re.finditer( + r'\(\s*(\d+\.?\d*)\s*kg\s*\.\s*(\d+\.?\d*)\s*kg\s*(?:의\s*)?' + r'(초소형|소형|중형|대형|초대형)견', + txt + ): + label = m.group(3) + add_range(get_size(label), float(m.group(1)), float(m.group(2)), label + '견') + + # ── 패턴5: "Akg이하의 X형견에게" ── + # 예: "(5.6kg이하의 초소형견에게)" + for m in re.finditer( + r'(\d+\.?\d*)\s*kg\s*이하\s*(?:의\s*)?(초소형|소형|중형|대형|초대형)견', + txt + ): + label = m.group(2) + add_range(get_size(label), 0, float(m.group(1)), label + '견') + + # ── 패턴6: 테이블 "A~B | 제품명 X형견용" ── + # 예: "2~3.5 넥스가드 스펙트라 츄어블 초소형견용 1" + # 예: "2-5 | 프론트라인 트리액트 초소형견용 | 0.5ml" + for m in re.finditer( + r'(\d+\.?\d*)\s*[-~~]\s*(\d+\.?\d*)\s*[\s|]*[^\n]*?' + r'(초소형|소형|중형|대형|초대형)견?용?', + txt + ): + label = m.group(3) + wmin, wmax = float(m.group(1)), float(m.group(2)) + if wmax <= 60: + add_range(get_size(label), wmin, wmax, label + '견') + + # ── 패턴7: "체중 Akg 미만 X형견용" ── + # 예: "체중 15kg 미만 소, 중형견용" + for m in re.finditer( + r'체중\s*(\d+\.?\d*)\s*kg\s*미만\s*[^\n]*?' + r'(초소형|소형|중형|대형|초대형)견', + txt + ): + label = m.group(2) + add_range(get_size(label), 0, float(m.group(1)), label + '견') + + # ── 패턴8: 라벨 없이 체중 구간만 (반려동물 키워드 있을 때) ── + if not ranges and ('개' in txt or '고양이' in txt or '반려' in txt or '애완' in txt): + for m in re.finditer( + r'(?:체중\s*)?(\d+\.?\d*)\s*[-~~]\s*(\d+\.?\d*)\s*kg', + txt + ): + wmin, wmax = float(m.group(1)), float(m.group(2)) + if wmax <= 60 and wmax > wmin: + add_range(None, wmin, wmax, None) + + for m in re.finditer( + r'(?:체중\s*)?[~~]\s*(\d+\.?\d*)\s*kg|(\d+\.?\d*)\s*kg\s*(?:이하|까지)', + txt + ): + val = m.group(1) or m.group(2) + wmax = float(val) + if wmax <= 60: + add_range(None, 0, wmax, None) + + # 정렬 (min 기준) + ranges.sort(key=lambda x: x['min']) + return ranges + + +def is_multi_size_product_name(product_name): + """ + 제품명에 여러 사이즈가 함께 들어있는 통합 제품인지 판단. + 예: "하트커버(SS,S,M,L)정" → True + """ + if not product_name: + return False + # 여러 사이즈 약어가 한 제품명에 있는 경우 + if re.search(r'[(\(].*(?:SS|XS).*[,/].*(?:S|M|L).*[)\)]', product_name): + return True + # 소형/중형/대형 등이 2개 이상 포함된 경우 + size_count = sum(1 for kw in ['초소형', '소형', '중형', '대형', '초대형'] + if kw in product_name) + if size_count >= 2: + return True + return False + + +def detect_size_from_product_name(product_name): + """ + 제품명에서 사이즈 라벨을 감지. + Returns: 'XS', 'S', 'M', 'L', 'XL' 또는 None + 통합 제품(SS,S,M,L 등 여러 사이즈)은 None 반환. + """ + if not product_name: + return None + # 통합 제품 제외 + if is_multi_size_product_name(product_name): + return None + for pattern, size in PRODUCT_NAME_SIZE_PATTERNS: + if re.search(pattern, product_name): + return size + return None + + +# ───────────────────────────────────────────── +# 3. 메인 로직 +# ───────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser(description='APDB weight_min_kg/weight_max_kg 일괄 채우기') + parser.add_argument('--commit', action='store_true', help='실제 DB 업데이트 수행 (기본: dry-run)') + parser.add_argument('--verbose', '-v', action='store_true', help='상세 로그') + args = parser.parse_args() + + dry_run = not args.commit + + if dry_run: + print("=" * 60) + print(" DRY-RUN 모드 (DB 변경 없음)") + print(" 실제 업데이트: python scripts/fill_weight_from_dosage.py --commit") + print("=" * 60) + else: + print("=" * 60) + print(" COMMIT 모드 - DB에 실제 업데이트합니다") + print("=" * 60) + + pg = create_engine('postgresql://admin:trajet6640@192.168.0.87:5432/apdb_master') + conn = pg.connect() + + # ── 동물용의약품 중 dosage_instructions에 kg 있고 weight 미입력인 APC 전체 조회 ── + apcs = conn.execute(text(''' + SELECT apc, item_seq, product_name, dosage, dosage_instructions, + product_type + FROM apc + WHERE product_type = '동물용의약품' + AND dosage_instructions ILIKE '%%kg%%' + AND weight_min_kg IS NULL + ORDER BY item_seq, apc + ''')).fetchall() + + print(f'\n대상 APC 레코드: {len(apcs)}건') + + # item_seq별로 그룹핑 + from collections import defaultdict + items = defaultdict(list) + di_cache = {} + for row in apcs: + items[row.item_seq].append(row) + if row.item_seq not in di_cache: + di_cache[row.item_seq] = row.dosage_instructions + + print(f'대상 item_seq: {len(items)}건\n') + + stats = { + 'total_items': len(items), + 'updated': 0, + 'matched_by_name': 0, + 'matched_single': 0, + 'skipped_no_parse': 0, + 'skipped_livestock': 0, + 'skipped_multi_no_label': 0, + } + + updates = [] # (apc, weight_min, weight_max, product_name, reason) + + for item_seq, apc_rows in items.items(): + di = di_cache[item_seq] + first_name = apc_rows[0].product_name + + # 체중 구간 파싱 + weight_ranges = parse_weight_ranges(di) + + if not weight_ranges: + stats['skipped_no_parse'] += 1 + if args.verbose: + print(f' SKIP (파싱불가): {first_name} ({item_seq})') + continue + + # 축산용 필터 (max > 60kg인 구간이 있으면 전체 SKIP) + if any(r['max'] > 60 for r in weight_ranges): + stats['skipped_livestock'] += 1 + if args.verbose: + large = [r for r in weight_ranges if r['max'] > 60] + print(f' SKIP (축산용): {first_name} ({item_seq}) max={large[0]["max"]}kg') + continue + + if len(weight_ranges) == 1: + # ── 체중 구간 1개 → 전체 APC에 적용 ── + wr = weight_ranges[0] + for row in apc_rows: + updates.append((row.apc, wr['min'], wr['max'], row.product_name, '단일구간')) + stats['matched_single'] += len(apc_rows) + stats['updated'] += len(apc_rows) + if args.verbose: + print(f' 적용 (단일구간): {first_name} → {wr["min"]}~{wr["max"]}kg ({len(apc_rows)}건)') + + else: + # ── 체중 구간 여러 개 → 제품명의 사이즈 라벨로 매칭 ── + size_to_weight = {} + for wr in weight_ranges: + if wr['size']: + size_to_weight[wr['size']] = (wr['min'], wr['max']) + + for row in apc_rows: + size = detect_size_from_product_name(row.product_name) + if size and size in size_to_weight: + wmin, wmax = size_to_weight[size] + updates.append((row.apc, wmin, wmax, row.product_name, f'제품명→{size}')) + stats['matched_by_name'] += 1 + stats['updated'] += 1 + if args.verbose: + print(f' 적용 (제품명 {size}): {row.product_name} → {wmin}~{wmax}kg') + else: + stats['skipped_multi_no_label'] += 1 + if args.verbose: + print(f' SKIP (다중구간+라벨없음): {row.product_name} ' + f'(감지={size}, 가용={list(size_to_weight.keys())})') + + # ── 결과 출력 ── + print('\n' + '=' * 60) + print(' 결과 요약') + print('=' * 60) + print(f' 대상 item_seq: {stats["total_items"]}건') + print(f' 업데이트할 APC: {stats["updated"]}건') + print(f' - 단일구간 적용: {stats["matched_single"]}건') + print(f' - 제품명 라벨 매칭: {stats["matched_by_name"]}건') + print(f' SKIP - 파싱 불가: {stats["skipped_no_parse"]}건') + print(f' SKIP - 축산용 (>60kg): {stats["skipped_livestock"]}건') + print(f' SKIP - 다중구간+라벨없음: {stats["skipped_multi_no_label"]}건') + + if updates: + print(f'\n === 업데이트 미리보기 (처음 30건) ===') + for apc, wmin, wmax, pname, reason in updates[:30]: + print(f' {apc} | {pname[:35]:35s} → {wmin}~{wmax}kg [{reason}]') + if len(updates) > 30: + print(f' ... 외 {len(updates) - 30}건') + + # ── DB 업데이트 ── + if not dry_run and updates: + print(f'\n DB 업데이트 시작...') + tx = conn.begin() + try: + for apc_code, wmin, wmax, _, _ in updates: + conn.execute(text(''' + UPDATE apc + SET weight_min_kg = :wmin, weight_max_kg = :wmax + WHERE apc = :apc + '''), {'wmin': wmin, 'wmax': wmax, 'apc': apc_code}) + tx.commit() + print(f' 완료: {len(updates)}건 업데이트') + except Exception as e: + tx.rollback() + print(f' 오류 발생, 롤백: {e}') + elif not dry_run and not updates: + print('\n 업데이트할 항목이 없습니다.') + + conn.close() + print('\n완료.') + + +if __name__ == '__main__': + main()