hyungi
0a7402b327
study_memo_cards 추출 파이프라인 + 버전키 폴러 + needs_review 컬럼. 운영 SR 코드(session_finalize/quiz_selection) 무수정.
- migrations 287~298: study_memo_cards/_evidence/_jobs/_progress(P1 휴면)·study_reminders·study_topics.focused_at·study_questions needs_review 3컬럼. dedup PARTIAL UNIQUE(deleted_at IS NULL).
- 워커: in-process RAG gather → MLX {cards} → 카드 가드(정량=evidence 원문 등장·cue/cloze 누출·dedup) → supersede 구버전 retire → append. 별 consumer 로 기존 study_queue 격리.
- 폴러 study_card_enqueue: 버전키 NOT EXISTS(source_version) 멱등 + ai_explanation_generated_at NOT NULL 가드 + per-poll LIMIT(thundering-herd).
- 검증: 실 prod 스키마 덤프 위 12 마이그 적용 OK + dedup/supersede/active-unique 기능 7/7 PASS + 정규화 util 15/15.
plan: PKM plans/2026-06-05-study-memo-card-p1-plan.html
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
"""공부 암기노트 카드 — 정량 토큰 정규화 + dedup 키 + 누출/근거 1차 primitives.
|
|
|
|
정규화 정책(보수적 = restrictive):
|
|
- NFC 유니코드 정규화
|
|
- 수치와 단위 사이 공백 제거 ('0.5 MPa' -> '0.5MPa')
|
|
- 천단위 구분자(콤마) 제거 ('1,000kg' -> '1000kg'), 숫자 3자리 그룹 한정
|
|
- 단위 환산 절대 금지 (원문 표기 보존 — LLM 오변환을 정규화로 흡수하지 않음)
|
|
|
|
대소문자는 보존한다 (MPa vs mpa 는 다른 단위라 lowercase 안 함).
|
|
dedup_hash = sha256(source_question_id | format | normalize_token(정답토큰)).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
import unicodedata
|
|
|
|
# 수치 다음의 공백 + (단위로 시작하는) 토큰 사이 공백 제거.
|
|
_NUM_UNIT_SPACE = re.compile(r"(\d)\s+(?=[A-Za-z℃°%‰Ωµμ/])")
|
|
# 천단위 콤마: 숫자 뒤 콤마 + 정확히 3자리 숫자 그룹이 이어질 때만 (소수점/일반 콤마 보호).
|
|
_THOUSANDS = re.compile(r"(?<=\d),(?=\d{3}(?:\D|$))")
|
|
_WS = re.compile(r"\s+")
|
|
# cloze 빈칸 마커: [____] / [___] / {{...}} / ____ 등.
|
|
_BLANK = re.compile(r"\[_+\]|\{\{[^}]*\}\}|_{2,}")
|
|
_DIGIT = re.compile(r"\d")
|
|
|
|
|
|
def normalize_token(s: str | None) -> str:
|
|
"""단일 정답 토큰 정규화 (대소문자 보존). dedup 키·근거 매칭의 단위."""
|
|
if not s:
|
|
return ""
|
|
s = unicodedata.normalize("NFC", s)
|
|
s = _NUM_UNIT_SPACE.sub(r"\1", s)
|
|
s = _THOUSANDS.sub("", s)
|
|
return s.strip()
|
|
|
|
|
|
def normalize_for_match(s: str | None) -> str:
|
|
"""근거 텍스트/문장 비교용 — 토큰 정규화 + 공백 축약 (대소문자 보존)."""
|
|
if not s:
|
|
return ""
|
|
s = normalize_token(s)
|
|
return _WS.sub(" ", s).strip()
|
|
|
|
|
|
def compute_dedup_hash(source_question_id: int | None, fmt: str, answer_token: str | None) -> str:
|
|
"""정본 키: sha256(source_question_id | format | normalize_token(정답토큰))."""
|
|
key = f"{source_question_id}|{fmt}|{normalize_token(answer_token)}"
|
|
return hashlib.sha256(key.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def is_quantitative(token: str | None) -> bool:
|
|
"""숫자를 포함하면 정량 토큰 (정량 cloze 는 evidence 원문 등장 필수)."""
|
|
return bool(_DIGIT.search(normalize_token(token)))
|
|
|
|
|
|
def text_contains(haystack: str | None, needle: str | None) -> bool:
|
|
"""needle(정답토큰)이 haystack 안에 정규화 후 부분문자열로 등장하면 True."""
|
|
n = normalize_for_match(needle)
|
|
if not n:
|
|
return False
|
|
return n in normalize_for_match(haystack)
|
|
|
|
|
|
def is_cue_leak(cue: str | None, answer_token: str | None) -> bool:
|
|
"""cue(앞면)에 정답토큰이 노출되면 True (drop 대상)."""
|
|
return text_contains(cue, answer_token)
|
|
|
|
|
|
def is_cloze_self_leak(cloze_text: str | None, answer_token: str | None) -> bool:
|
|
"""cloze_text 의 빈칸 마커를 제거한 평문에 정답토큰이 노출되면 True (drop 대상)."""
|
|
if not cloze_text:
|
|
return False
|
|
stripped = _BLANK.sub(" ", cloze_text)
|
|
return text_contains(stripped, answer_token)
|
|
|
|
|
|
def matching_evidence(answer_token: str | None, evidence_refs: list[dict]) -> list[dict]:
|
|
"""정답토큰이 snippet 에 등장하는 evidence_refs 만 반환 (citation 적재용)."""
|
|
out: list[dict] = []
|
|
for ref in evidence_refs or []:
|
|
if text_contains(ref.get("snippet"), answer_token):
|
|
out.append(ref)
|
|
return out
|