feat(scripts): APDB weight_min_kg/max_kg 일괄 채우기 스크립트
dosage_instructions에서 체중 구간을 파싱하여 weight 컬럼 업데이트. - 제품명 사이즈 라벨(소형견/중형견 등)로 체중구간 매칭 - 단일 체중구간 제품은 전체 APC에 적용 - 통합 제품(SS,S,M,L)은 안전하게 SKIP - 축산용(>60kg) 자동 제외 - dry-run 기본, --commit으로 실행 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
thug0bin
parent
7e7d06f32e
commit
90f88450be
463
backend/scripts/fill_weight_from_dosage.py
Normal file
463
backend/scripts/fill_weight_from_dosage.py
Normal file
@ -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('<p class="indent0">', '\n').replace('</p>', '')
|
||||
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()
|
||||
Loading…
Reference in New Issue
Block a user