381 lines
12 KiB
Python
381 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
동물약 마크다운 -> JSON 변환 파서
|
|
소스: new_anipharm 폴더의 마크다운 파일들
|
|
출력: data/drugs_from_md.json
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, List, Any
|
|
|
|
# 경로 설정
|
|
SOURCE_DIR = Path(r"C:\Users\청춘약국\source\new_anipharm")
|
|
OUTPUT_FILE = Path(r"C:\Users\청춘약국\source\animal-medication-api\data\drugs_from_md.json")
|
|
|
|
# 제외할 파일 (제품 정보가 아닌 파일들)
|
|
EXCLUDE_FILES = {
|
|
"anipharm_renewal_plan.md",
|
|
"drug_research_golden_pattern.md",
|
|
"esccap_gl_reference.md",
|
|
"GL1_5.md",
|
|
"site_fullrenewal.md",
|
|
"tosspayments_analysis.md",
|
|
"antibiotic_reference.md",
|
|
"heartworm_tick_combination_question.md",
|
|
"metronidazole_amx.md",
|
|
}
|
|
|
|
# 필드 매핑 (마크다운 표 항목 → JSON 필드)
|
|
TABLE_FIELD_MAP = {
|
|
"제품명": "name",
|
|
"영문명": "english_name",
|
|
"제조사": "manufacturer",
|
|
"분류": "category",
|
|
"약물 계열": "category",
|
|
"대상동물": "target_animal",
|
|
"대상 동물": "target_animal",
|
|
"투여 경로": "administration",
|
|
"투여경로": "administration",
|
|
"적응증": "efficacy",
|
|
"효능효과": "efficacy",
|
|
"효능·효과": "efficacy",
|
|
"성분": "ingredients",
|
|
"성분 (1정)": "ingredients",
|
|
"주성분": "ingredients",
|
|
"용법용량": "dosage",
|
|
"기본 용량": "dosage",
|
|
"보관": "storage",
|
|
"보관방법": "storage",
|
|
"유효기간": "shelf_life",
|
|
"금기": "precautions",
|
|
"금기사항": "precautions",
|
|
"주의사항": "precautions",
|
|
}
|
|
|
|
|
|
def parse_table(content: str, table_header: str = "제품 정보 요약") -> Dict[str, str]:
|
|
"""마크다운 표에서 제품 정보 추출"""
|
|
result = {}
|
|
|
|
# 표 찾기: "## 1. 제품 정보 요약" 또는 "## 제품 정보 요약" 등
|
|
pattern = rf"##\s*\d*\.?\s*{table_header}.*?\n([\s\S]*?)(?=\n##|\n---|\Z)"
|
|
match = re.search(pattern, content, re.IGNORECASE)
|
|
|
|
if not match:
|
|
return result
|
|
|
|
table_section = match.group(1)
|
|
|
|
# 표 행 파싱: | 항목 | 내용 |
|
|
row_pattern = r"\|\s*\*?\*?([^|*]+)\*?\*?\s*\|\s*([^|]+)\s*\|"
|
|
rows = re.findall(row_pattern, table_section)
|
|
|
|
for key, value in rows:
|
|
key = key.strip().replace("**", "")
|
|
value = value.strip().replace("**", "")
|
|
|
|
# 필드 매핑
|
|
for table_key, json_key in TABLE_FIELD_MAP.items():
|
|
if table_key in key:
|
|
result[json_key] = value
|
|
break
|
|
|
|
return result
|
|
|
|
|
|
def extract_json_block(content: str) -> Optional[Dict[str, Any]]:
|
|
"""DB 저장용 JSON 코드블록 추출"""
|
|
# "DB 저장용 JSON" 또는 "제품 메타데이터" 섹션의 첫 번째 JSON 블록
|
|
pattern = r"(?:DB 저장용 JSON|제품 메타데이터).*?```json\s*([\s\S]*?)```"
|
|
match = re.search(pattern, content, re.IGNORECASE)
|
|
|
|
if match:
|
|
json_str = match.group(1).strip()
|
|
try:
|
|
return json.loads(json_str)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def extract_apc_code(content: str, filename: str) -> str:
|
|
"""APC 코드 추출 (없으면 파일명 기반 생성)"""
|
|
# 패턴: APC-XXX 또는 apc_code 필드
|
|
patterns = [
|
|
r"APC[_-]?(\d+)",
|
|
r'"apc_code":\s*"([^"]+)"',
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, content, re.IGNORECASE)
|
|
if match:
|
|
code = match.group(1)
|
|
if not code.startswith("APC"):
|
|
code = f"APC-{code}"
|
|
return code
|
|
|
|
# 파일명 기반 생성
|
|
base = Path(filename).stem.upper().replace("_", "-")[:20]
|
|
return f"APC-{base}"
|
|
|
|
|
|
def parse_target_animal(value: Any) -> List[str]:
|
|
"""대상 동물 파싱"""
|
|
if isinstance(value, list):
|
|
return value
|
|
|
|
if isinstance(value, str):
|
|
# "개, 고양이" 또는 "개·고양이" 또는 "개/고양이"
|
|
animals = re.split(r"[,·/\s]+", value)
|
|
animals = [a.strip() for a in animals if a.strip()]
|
|
|
|
# 일반화
|
|
result = []
|
|
for a in animals:
|
|
a_lower = a.lower()
|
|
if "개" in a or "dog" in a_lower:
|
|
if "개" not in result:
|
|
result.append("개")
|
|
if "고양이" in a or "cat" in a_lower:
|
|
if "고양이" not in result:
|
|
result.append("고양이")
|
|
|
|
return result if result else ["개", "고양이"]
|
|
|
|
return ["개", "고양이"]
|
|
|
|
|
|
def parse_precautions(value: Any) -> List[str]:
|
|
"""주의사항/금기사항 파싱"""
|
|
if isinstance(value, list):
|
|
return value
|
|
|
|
if isinstance(value, str):
|
|
# 여러 항목 분리
|
|
items = re.split(r"[;,·]|\n", value)
|
|
return [item.strip() for item in items if item.strip()]
|
|
|
|
return []
|
|
|
|
|
|
def extract_title_name(content: str) -> tuple[str, str]:
|
|
"""마크다운 제목에서 제품명과 영문명 추출"""
|
|
# 첫 번째 # 제목 찾기
|
|
match = re.search(r"^#\s*(.+?)(?:\s*[-–—]\s*|\n)", content, re.MULTILINE)
|
|
if match:
|
|
title = match.group(1).strip()
|
|
# 괄호 안 영문명 추출
|
|
eng_match = re.search(r"\(([A-Za-z][A-Za-z\s\-®]+)\)", title)
|
|
eng_name = eng_match.group(1) if eng_match else ""
|
|
# 제품명 정리 (괄호 앞 부분)
|
|
name = re.sub(r"\s*\([^)]+\)\s*", "", title).strip()
|
|
return name, eng_name
|
|
return "", ""
|
|
|
|
|
|
def normalize_drug(data: Dict[str, Any], filename: str, content: str) -> Dict[str, Any]:
|
|
"""약품 데이터 정규화 및 기본값 적용"""
|
|
|
|
# 제목에서 제품명 추출 시도
|
|
title_name, title_eng = extract_title_name(content)
|
|
|
|
result = {
|
|
"apc_code": extract_apc_code(content, filename),
|
|
"name": title_name or "미정",
|
|
"english_name": title_eng or "",
|
|
"manufacturer": "미상",
|
|
"category": "",
|
|
"target_animal": ["개", "고양이"],
|
|
"administration": "경구",
|
|
"ingredients": "",
|
|
"efficacy": "",
|
|
"dosage": "",
|
|
"precautions": [],
|
|
"storage": "실온보관",
|
|
"shelf_life": "제조일로부터 24개월",
|
|
"source_file": filename,
|
|
}
|
|
|
|
# 데이터 병합
|
|
for key, value in data.items():
|
|
if key == "product_name":
|
|
result["name"] = value
|
|
elif key == "product_name_en":
|
|
result["english_name"] = value
|
|
elif key == "generic_name":
|
|
if not result["english_name"]:
|
|
result["english_name"] = value
|
|
elif key == "drug_class":
|
|
result["category"] = value
|
|
elif key == "target_animals":
|
|
result["target_animal"] = parse_target_animal(value)
|
|
elif key == "target_animal":
|
|
result["target_animal"] = parse_target_animal(value)
|
|
elif key == "route":
|
|
result["administration"] = value
|
|
elif key == "composition_per_tablet":
|
|
if isinstance(value, dict):
|
|
parts = []
|
|
for k, v in value.items():
|
|
if "mg" in k:
|
|
parts.append(f"{k.replace('_mg', '')}: {v}mg")
|
|
if parts:
|
|
result["ingredients"] = ", ".join(parts)
|
|
elif key == "dosing_single":
|
|
if isinstance(value, dict):
|
|
result["dosage"] = "; ".join(f"{k}: {v}" for k, v in value.items())
|
|
else:
|
|
result["dosage"] = str(value)
|
|
elif key == "contraindication":
|
|
result["precautions"] = parse_precautions(value)
|
|
elif key == "side_effects":
|
|
existing = result.get("precautions", [])
|
|
result["precautions"] = existing + parse_precautions(value)
|
|
elif key == "storage":
|
|
result["storage"] = value
|
|
elif key == "manufacturer":
|
|
result["manufacturer"] = value
|
|
elif key in result:
|
|
result[key] = value
|
|
|
|
# 이름에서 영문명 추출 시도
|
|
if result["name"] != "미정" and not result["english_name"]:
|
|
match = re.search(r"\(([A-Za-z][A-Za-z\s]+)\)", result["name"])
|
|
if match:
|
|
result["english_name"] = match.group(1)
|
|
|
|
return result
|
|
|
|
|
|
def extract_any_json_block(content: str) -> Optional[Dict[str, Any]]:
|
|
"""문서 내 첫 번째 product_name 포함 JSON 블록 추출"""
|
|
# 모든 ```json 블록 찾기
|
|
pattern = r"```json\s*([\s\S]*?)```"
|
|
matches = re.findall(pattern, content)
|
|
|
|
for json_str in matches:
|
|
try:
|
|
data = json.loads(json_str.strip())
|
|
# dict이고 product_name이 있으면 반환
|
|
if isinstance(data, dict) and "product_name" in data:
|
|
return data
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
return None
|
|
|
|
|
|
def parse_markdown_file(filepath: Path) -> Optional[Dict[str, Any]]:
|
|
"""단일 마크다운 파일 파싱"""
|
|
try:
|
|
with open(filepath, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
except Exception as e:
|
|
print(f" ❌ 파일 읽기 실패: {filepath.name} - {e}")
|
|
return None
|
|
|
|
# 제품 정보인지 확인 (제목에 제품명이 있거나, 제품 정보 요약 표가 있는 경우)
|
|
if "제품 정보 요약" not in content and "DB 저장용 JSON" not in content:
|
|
print(f" ⏭️ 제품 정보 없음: {filepath.name}")
|
|
return None
|
|
|
|
# 1. 표에서 추출
|
|
table_data = parse_table(content)
|
|
|
|
# 2. JSON 블록에서 추출 (두 가지 방법 시도)
|
|
json_data = extract_json_block(content)
|
|
if not json_data:
|
|
json_data = extract_any_json_block(content)
|
|
json_data = json_data or {}
|
|
|
|
# 3. 병합 (JSON 우선)
|
|
merged = {**table_data, **json_data}
|
|
|
|
# 4. 정규화 (데이터가 없어도 제목에서 추출 가능)
|
|
result = normalize_drug(merged, filepath.name, content)
|
|
|
|
# 최소 조건: 이름이 "미정"이 아니거나 JSON 데이터가 있는 경우
|
|
if result["name"] == "미정" and not json_data:
|
|
print(f" ⏭️ 파싱 데이터 없음: {filepath.name}")
|
|
return None
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
"""메인 실행"""
|
|
print("=" * 60)
|
|
print("동물약 마크다운 → JSON 변환 파서")
|
|
print("=" * 60)
|
|
print(f"소스: {SOURCE_DIR}")
|
|
print(f"출력: {OUTPUT_FILE}")
|
|
print()
|
|
|
|
# 마크다운 파일 목록
|
|
md_files = list(SOURCE_DIR.glob("*.md"))
|
|
print(f"총 {len(md_files)}개 마크다운 파일 발견")
|
|
print()
|
|
|
|
drugs = []
|
|
success = 0
|
|
skipped = 0
|
|
failed = 0
|
|
|
|
for filepath in sorted(md_files):
|
|
# 제외 파일 확인
|
|
if filepath.name in EXCLUDE_FILES:
|
|
print(f" ⏭️ 제외: {filepath.name}")
|
|
skipped += 1
|
|
continue
|
|
|
|
print(f"📄 처리 중: {filepath.name}")
|
|
|
|
result = parse_markdown_file(filepath)
|
|
|
|
if result:
|
|
drugs.append(result)
|
|
print(f" ✅ 성공: {result['name']}")
|
|
success += 1
|
|
else:
|
|
failed += 1
|
|
|
|
# 중복 제거 (이름 기준)
|
|
seen_names = set()
|
|
unique_drugs = []
|
|
for drug in drugs:
|
|
name = drug["name"]
|
|
if name not in seen_names:
|
|
seen_names.add(name)
|
|
unique_drugs.append(drug)
|
|
else:
|
|
print(f" ⚠️ 중복 제거: {name}")
|
|
|
|
# 결과 저장
|
|
OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(unique_drugs, f, ensure_ascii=False, indent=2)
|
|
|
|
print()
|
|
print("=" * 60)
|
|
print("변환 완료")
|
|
print("=" * 60)
|
|
print(f"✅ 성공: {success}개")
|
|
print(f"⏭️ 스킵: {skipped}개")
|
|
print(f"❌ 실패: {failed}개")
|
|
print(f"📦 최종 약품 수: {len(unique_drugs)}개")
|
|
print(f"💾 저장: {OUTPUT_FILE}")
|
|
|
|
# 변환된 약품 목록 출력
|
|
print()
|
|
print("변환된 약품 목록:")
|
|
for i, drug in enumerate(unique_drugs, 1):
|
|
print(f" {i}. {drug['name']} ({drug['english_name'] or 'N/A'})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|