"""study_memo_cards / study_memo_card_evidence ORM (공부 암기노트 Phase 1). study_questions(MCQ) 와 별개로, 풀이/근거에서 추출한 암기 플래시카드 본체. - source_kind: question(P1) / subject_note / document(P3 예약) - format: qa(cue->fact) / cloze(빈칸). 강한 enum 미사용 (read-time 매핑). - source_generated_at: 추출 당시 ai_explanation_generated_at — 버전 핀/stale 판정. - needs_review DEFAULT true: 생성물이라 검토 대기로 입고. dedup_hash PARTIAL UNIQUE(migration 288, WHERE deleted_at IS NULL) 가 중복 최종 방어선. 정정/삭제 후 supersede(구버전 카드 deleted_at 마킹)로 stale 잔류 0 — append 전에 호출해 살아있는 구카드가 새 추출을 ON CONFLICT 로 막지 않게 한다. """ from __future__ import annotations from datetime import datetime from typing import Any, Sequence from sqlalchemy import ( BigInteger, Boolean, DateTime, ForeignKey, Integer, String, Text, func, select, text, update, ) from sqlalchemy.dialects.postgresql import JSONB, insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, mapped_column from core.database import Base class StudyMemoCard(Base): __tablename__ = "study_memo_cards" id: Mapped[int] = mapped_column(BigInteger, primary_key=True) user_id: Mapped[int] = mapped_column( BigInteger, ForeignKey("users.id", ondelete="CASCADE"), nullable=False ) study_topic_id: Mapped[int] = mapped_column( BigInteger, ForeignKey("study_topics.id", ondelete="CASCADE"), nullable=False ) source_kind: Mapped[str] = mapped_column(String(40), nullable=False) source_question_id: Mapped[int | None] = mapped_column( BigInteger, ForeignKey("study_questions.id", ondelete="CASCADE") ) source_subject_note_id: Mapped[int | None] = mapped_column(BigInteger) format: Mapped[str] = mapped_column(String(20), nullable=False) cue: Mapped[str] = mapped_column(Text, nullable=False) fact: Mapped[str] = mapped_column(Text, nullable=False) cloze_text: Mapped[str | None] = mapped_column(Text) extra: Mapped[dict | None] = mapped_column(JSONB) source_generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) dedup_hash: Mapped[str] = mapped_column(String(64), nullable=False) needs_review: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) flagged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) flagged_by: Mapped[str | None] = mapped_column(String(40)) model: Mapped[str | None] = mapped_column(String(120)) generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # '그냥 공부'(cram) 봤다 기록 (SR 무관, migration 300) view_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) last_viewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=datetime.now, nullable=False ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) class StudyMemoCardEvidence(Base): """append-only citation. UPDATE/DELETE 없음.""" __tablename__ = "study_memo_card_evidence" id: Mapped[int] = mapped_column(BigInteger, primary_key=True) card_id: Mapped[int] = mapped_column( BigInteger, ForeignKey("study_memo_cards.id", ondelete="CASCADE"), nullable=False ) source_type: Mapped[str] = mapped_column(String(40), nullable=False) source_id: Mapped[int | None] = mapped_column(BigInteger) chunk_index: Mapped[int | None] = mapped_column(Integer) snippet: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=datetime.now, nullable=False ) async def supersede_old_cards( session: AsyncSession, *, source_question_id: int, keep_generated_at: datetime | None, ) -> list[int]: """같은 문제의 '다른 버전' 카드를 deleted_at 마킹(retire). 새 source_generated_at 카드 적재 '전에' 호출 — 살아있는 구버전 카드가 dedup PARTIAL UNIQUE 로 새 추출을 막는 것을 방지(정정-후 stale 잔류 0). 같은 버전은 보존. Returns: retire 되며 '발행 중이던'(needs_review=False) 카드 id 목록 — 발행 tombstone 대상(호출측이 enqueue). 검수 안 됐던(미발행) retire 카드는 tombstone 불요라 제외. """ # 발행 중이던 retire 대상 선캡처(update 전) — 미발행 카드 스푸리어스 tombstone 회피. published_retired = ( await session.execute( select(StudyMemoCard.id).where( StudyMemoCard.source_question_id == source_question_id, StudyMemoCard.deleted_at.is_(None), StudyMemoCard.source_generated_at.is_distinct_from(keep_generated_at), StudyMemoCard.needs_review.is_(False), ) ) ).scalars().all() stmt = ( update(StudyMemoCard) .where( StudyMemoCard.source_question_id == source_question_id, StudyMemoCard.deleted_at.is_(None), StudyMemoCard.source_generated_at.is_distinct_from(keep_generated_at), ) .values(deleted_at=func.now()) ) await session.execute(stmt) return list(published_retired) async def append_card( session: AsyncSession, *, user_id: int, study_topic_id: int, source_kind: str, source_question_id: int | None, format: str, cue: str, fact: str, cloze_text: str | None, dedup_hash: str, source_generated_at: datetime | None, model: str | None, generated_at: datetime | None, needs_review: bool = True, ) -> int | None: """카드 1장 INSERT. dedup_hash PARTIAL UNIQUE 충돌 시 None (DO NOTHING). Returns: 새 card.id, 또는 중복으로 건너뛰면 None. """ stmt = ( pg_insert(StudyMemoCard) .values( user_id=user_id, study_topic_id=study_topic_id, source_kind=source_kind, source_question_id=source_question_id, format=format, cue=cue, fact=fact, cloze_text=cloze_text, dedup_hash=dedup_hash, source_generated_at=source_generated_at, needs_review=needs_review, model=model, generated_at=generated_at, ) .on_conflict_do_nothing( index_elements=["dedup_hash"], index_where=text("deleted_at IS NULL"), ) .returning(StudyMemoCard.id) ) result = await session.execute(stmt) return result.scalar_one_or_none() async def append_card_evidence( session: AsyncSession, *, card_id: int, refs: Sequence[dict[str, Any]], ) -> int: """카드 인용 append-only INSERT. refs: [{source_type, source_id?, chunk_index?, snippet?}].""" rows = [ { "card_id": card_id, "source_type": r.get("source_type") or "unknown", "source_id": r.get("source_id"), "chunk_index": r.get("chunk_index"), "snippet": r.get("snippet"), } for r in refs ] if not rows: return 0 await session.execute(pg_insert(StudyMemoCardEvidence).values(rows)) return len(rows) async def record_card_view( session: AsyncSession, *, user_id: int, card_id: int ) -> bool: """'그냥 공부'(cram) 봤다 기록 — view_count++ + last_viewed_at. SR(progress) 무관. needs_review 무관(검수 안 된 카드도 가볍게 둘러볼 수 있음), 본인·미삭제 카드만. Returns: 기록됨 여부. """ stmt = ( update(StudyMemoCard) .where( StudyMemoCard.id == card_id, StudyMemoCard.user_id == user_id, StudyMemoCard.deleted_at.is_(None), ) .values(view_count=StudyMemoCard.view_count + 1, last_viewed_at=func.now()) ) result = await session.execute(stmt) return (result.rowcount or 0) > 0 async def flag_cards_for_source( session: AsyncSession, *, source_question_id: int, reason: str, ) -> list[int]: """소스 문제 정정/삭제 시 파생 카드를 needs_review=auto 마킹(임시 플래그). 최종 stale 정리는 워커 supersede 가 책임 — 이건 사용자 가시화용 즉시 플래그. reason: 'source_changed' | 'source_deleted'. Returns: 플래그로 '발행 자격을 잃은'(직전 needs_review=False) 카드 id 목록 — 발행 tombstone 대상(호출측 enqueue). 이미 검수대기였던(미발행) 카드는 제외. """ # 발행 중이던 카드 선캡처(update 전) — 플래그로 needs_review=True 가 되면 발행 자격 상실. published_ids = ( await session.execute( select(StudyMemoCard.id).where( StudyMemoCard.source_question_id == source_question_id, StudyMemoCard.deleted_at.is_(None), StudyMemoCard.needs_review.is_(False), ) ) ).scalars().all() stmt = ( update(StudyMemoCard) .where( StudyMemoCard.source_question_id == source_question_id, StudyMemoCard.deleted_at.is_(None), ) .values(needs_review=True, flagged_by=reason, flagged_at=func.now()) ) await session.execute(stmt) return list(published_ids)