# ============================================== # ingest.py (v2025-UNIVERSAL-COMPAT) # ✅ 현재 db.py/init_db() (인자 없음)와 100% 호환 # ✅ 기존 questions 테이블 스키마( options_json / answer / pairs / sequence ) 기준으로 안전 동작 # ✅ 확장 모델( set_answer_keys / set_answer_steps / set_images / page_no 등 )이 있으면 자동 활용 # ✅ options는 항상 [{"key","text"}] 형태로 저장 → 웹 표기 안정화 # ✅ Steps/복수/중복 정답: # - 확장 모델이면 answer_json/answer_steps_json에 저장 # - 구버전이면 answer="A,B" / sequence=["E","B","C"] 로 fallback # ✅ rebuild_db=True: DB 파일 삭제 후 재생성 # ============================================== import json import os from typing import Any, Dict, List from db import SessionLocal, init_db, DB_PATH from models import Question # ----------------------------- # Helpers # ----------------------------- def _to_list_answer_keys(v: Any) -> List[str]: """ 정답 입력을 key 리스트로 정규화. - ["A","C"] -> ["A","C"] - "A" -> ["A"] - "A,C" -> ["A","C"] - "BE" -> ["B","E"] (전부 대문자 알파벳일 때만) - dict(steps 텍스트) -> [] (steps에서 처리) """ if v is None: return [] if isinstance(v, list): return [str(x).strip() for x in v if str(x).strip()] if isinstance(v, dict): return [] s = str(v).strip() if not s: return [] if "," in s: return [x.strip() for x in s.split(",") if x.strip()] if len(s) >= 2 and s.isalpha() and s.upper() == s: return list(s) return [s] def _normalize_options(opts: Any) -> List[Dict[str, str]]: """ options를 항상 표준 리스트 형태로: [{"key":"A","text":"..."}, ...] 지원: - list[str] - dict{key:text} - list[dict] (이미 key/text) """ if not opts: return [] # list[dict] if isinstance(opts, list) and opts and all(isinstance(x, dict) for x in opts): out = [] for o in opts: k = str(o.get("key", "")).strip() t = str(o.get("text", "")).strip() if k or t: out.append({"key": k, "text": t}) return out # list[str] if isinstance(opts, list): return [{"key": chr(65 + i), "text": str(opt).strip()} for i, opt in enumerate(opts)] # dict{key:text} if isinstance(opts, dict): return [{"key": str(k).strip(), "text": str(v).strip()} for k, v in opts.items()] return [] def _infer_steps_answer_keys(item: Dict[str, Any], options_std: List[Dict[str, str]]) -> List[str]: """ Steps 정답을 key 리스트로 뽑는다. 우선순위: 1) answer_steps(list) 2) sequence(list) 3) answer가 {"1":"텍스트", "2":"텍스트"} 형태면 options text 매칭으로 key 추정 """ if isinstance(item.get("answer_steps"), list): return [str(x).strip() for x in item["answer_steps"] if str(x).strip()] if isinstance(item.get("sequence"), list): return [str(x).strip() for x in item["sequence"] if str(x).strip()] ans = item.get("answer") if isinstance(ans, dict) and all(str(k).isdigit() for k in ans.keys()): text_to_key = {} for o in options_std: t = (o.get("text") or "").strip() if t and t not in text_to_key: text_to_key[t] = (o.get("key") or "").strip() keys = [] for i in sorted(int(x) for x in ans.keys()): t = str(ans.get(str(i), "")).strip() keys.append(text_to_key.get(t, "__UNKNOWN__")) return keys return [] def _load_json(json_path: str) -> List[Dict[str, Any]]: with open(json_path, "r", encoding="utf-8") as f: data = json.load(f) # Case Study: {"questions":[...]} if isinstance(data, dict) and "questions" in data: data = data["questions"] if not isinstance(data, list): raise ValueError("JSON root must be a list (or dict with 'questions').") # dict 아닌 것 제거 return [x for x in data if isinstance(x, dict)] def _normalize_item(item: Dict[str, Any]) -> Dict[str, Any]: # stem stem = (item.get("stem") or item.get("question") or item.get("q_text") or "") stem = str(stem).strip() explanation = str(item.get("explanation") or "").strip() qtype = item.get("question_type", "MCQ") code = item.get("code", "") category = item.get("category") or item.get("topic") or None subcategory = item.get("subcategory") or item.get("subtopic") or None # options 표준화 options_std = _normalize_options(item.get("options")) # 정렬키 (있으면) source_pages = item.get("source_pages") page_no = source_pages[0] if isinstance(source_pages, list) and source_pages else None page_legacy = item.get("page") q_no_on_page = item.get("q_no_on_page") global_no = item.get("global_no") or item.get("question_id") # 이미지 images = item.get("images") or item.get("image_urls") or [] if not isinstance(images, list): images = [] # steps 정답 answer_steps = _infer_steps_answer_keys(item, options_std) # 일반 정답 key들 answer_keys = [] if not answer_steps: if isinstance(item.get("answer_keys"), list): answer_keys = [str(x).strip() for x in item["answer_keys"] if str(x).strip()] else: answer_keys = _to_list_answer_keys(item.get("answer")) return { "stem": stem, "explanation": explanation, "question_type": qtype, "category": category, "subcategory": subcategory, "code": code, "options_std": options_std, "page": page_legacy, "page_no": page_no, "q_no_on_page": q_no_on_page, "global_no": global_no, "answer_keys": answer_keys, "answer_steps": answer_steps, # 레거시 유지 "pairs": item.get("pairs"), "sequence": item.get("sequence"), "images": images, "raw_answer": item.get("answer"), } # ----------------------------- # Ingest # ----------------------------- def ingest_questions(json_path: str, source_name: str = "imported", rebuild_db: bool = False) -> int: """ ✅ 현재 db.py/init_db()와 호환되는 통합 ingest - rebuild_db=True: DB 파일 삭제 후 init_db()로 새로 생성 """ json_path = str(json_path) if rebuild_db and DB_PATH.exists(): DB_PATH.unlink() print(f"[INFO] 🧹 Deleted DB: {DB_PATH}") # ✅ 현재 db.py는 인자 없는 init_db()만 지원 init_db() rows = _load_json(json_path) db = SessionLocal() try: count = 0 for raw in rows: qn = _normalize_item(raw) q = Question( page=qn["page"], stem=qn["stem"], explanation=qn["explanation"], question_type=qn["question_type"], category=qn["category"], subcategory=qn["subcategory"], source=source_name, code=qn["code"], ) # (확장 모델이면) 정렬키 저장 if hasattr(q, "page_no"): q.page_no = qn["page_no"] if hasattr(q, "q_no_on_page"): q.q_no_on_page = qn["q_no_on_page"] if hasattr(q, "global_no"): q.global_no = qn["global_no"] # ✅ options는 표준 list[dict]로 저장 (web 표시 안정화) q.set_options(qn["options_std"]) # ✅ Steps 정답 처리 if qn["answer_steps"]: if hasattr(q, "set_answer_steps"): q.set_answer_steps(qn["answer_steps"]) q.answer = "" # 확장 컬럼 쓰는 경우 레거시 비워도 OK else: # 구버전 fallback: sequence에 steps key 리스트 저장 q.sequence = json.dumps(qn["answer_steps"], ensure_ascii=False) q.answer = "" # steps는 answer 문자열 비교가 의미 없음 else: # ✅ 일반 정답(복수/중복 포함) if qn["answer_keys"]: if hasattr(q, "set_answer_keys"): q.set_answer_keys(qn["answer_keys"]) q.answer = "" else: # 구버전 fallback: answer="A,B,C" (중복도 그대로) q.answer = ",".join(qn["answer_keys"]) else: # 정답이 애매하면 원본 유지 q.answer = str(qn["raw_answer"] or "").strip() # 레거시 pairs/sequence 유지(있으면) if qn["sequence"] is not None: q.sequence = json.dumps(qn["sequence"], ensure_ascii=False) if isinstance(qn["sequence"], list) else qn["sequence"] if qn["pairs"] is not None: q.pairs = json.dumps(qn["pairs"], ensure_ascii=False) if isinstance(qn["pairs"], (dict, list)) else qn["pairs"] # ✅ images 저장(확장 모델이면) if hasattr(q, "set_images"): q.set_images(qn["images"]) db.add(q) count += 1 db.commit() print(f"[INFO] ✅ {count} 문항 DB 적재 완료 ({source_name})") return count except Exception as e: db.rollback() print(f"[ERROR] DB 적재 중 오류 발생 → {e}") raise finally: db.close() if __name__ == "__main__": # 너가 말한 실제 경로: data/questions.json path = os.getenv("QUESTIONS_JSON", "data/questions.json") source = os.getenv("SOURCE_NAME", "az104_dump") rebuild = os.getenv("REBUILD_DB", "0") == "1" ingest_questions(path, source_name=source, rebuild_db=rebuild)