228 lines
8.3 KiB
Python
228 lines
8.3 KiB
Python
import uuid
|
|
from dataclasses import dataclass
|
|
|
|
from ..models.pack import WordBankPack, WordBankPackEntry, WordBankPackFlashcardTemplate
|
|
from ...outbound.postgres.repositories.pack_repository import PackRepository
|
|
from ...outbound.postgres.repositories.vocab_repository import VocabRepository
|
|
from ...outbound.postgres.repositories.flashcard_repository import FlashcardRepository
|
|
from ...outbound.postgres.repositories.dictionary_repository import DictionaryRepository
|
|
|
|
|
|
class DuplicateEntryError(Exception):
|
|
"""Raised when a pack would add plain cards that are identical to ones already in the bank."""
|
|
|
|
def __init__(self, duplicate_surface_texts: list[str]) -> None:
|
|
self.duplicate_surface_texts = duplicate_surface_texts
|
|
joined = ", ".join(f'"{t}"' for t in duplicate_surface_texts)
|
|
super().__init__(
|
|
f"You already have the following word(s) in your bank: {joined}. "
|
|
"Remove them first, or add the pack once they have been cleared."
|
|
)
|
|
|
|
|
|
class PackNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class PackApplicationResult:
|
|
added_surface_texts: list[str]
|
|
|
|
|
|
class PackService:
|
|
def __init__(
|
|
self,
|
|
pack_repo: PackRepository,
|
|
vocab_repo: VocabRepository,
|
|
flashcard_repo: FlashcardRepository,
|
|
dict_repo: DictionaryRepository,
|
|
) -> None:
|
|
self.pack_repo = pack_repo
|
|
self.vocab_repo = vocab_repo
|
|
self.flashcard_repo = flashcard_repo
|
|
self.dict_repo = dict_repo
|
|
|
|
async def create_pack(
|
|
self,
|
|
name: str,
|
|
name_target: str,
|
|
description: str,
|
|
description_target: str,
|
|
source_lang: str,
|
|
target_lang: str,
|
|
proficiencies: list[str],
|
|
) -> WordBankPack:
|
|
return await self.pack_repo.create_pack(
|
|
name=name,
|
|
name_target=name_target,
|
|
description=description,
|
|
description_target=description_target,
|
|
source_lang=source_lang,
|
|
target_lang=target_lang,
|
|
proficiencies=proficiencies,
|
|
)
|
|
|
|
async def update_pack(
|
|
self,
|
|
pack_id: uuid.UUID,
|
|
name: str | None = None,
|
|
name_target: str | None = None,
|
|
description: str | None = None,
|
|
description_target: str | None = None,
|
|
proficiencies: list[str] | None = None,
|
|
) -> WordBankPack:
|
|
pack = await self.pack_repo.get_pack(pack_id)
|
|
if pack is None:
|
|
raise PackNotFoundError(f"Pack {pack_id} not found")
|
|
return await self.pack_repo.update_pack(
|
|
pack_id=pack_id,
|
|
name=name,
|
|
name_target=name_target,
|
|
description=description,
|
|
description_target=description_target,
|
|
proficiencies=proficiencies,
|
|
)
|
|
|
|
async def publish_pack(self, pack_id: uuid.UUID) -> WordBankPack:
|
|
pack = await self.pack_repo.get_pack(pack_id)
|
|
if pack is None:
|
|
raise PackNotFoundError(f"Pack {pack_id} not found")
|
|
return await self.pack_repo.publish_pack(pack_id)
|
|
|
|
async def add_entry_to_pack(
|
|
self,
|
|
pack_id: uuid.UUID,
|
|
sense_id: uuid.UUID | None,
|
|
surface_text: str,
|
|
) -> WordBankPackEntry:
|
|
pack = await self.pack_repo.get_pack(pack_id)
|
|
if pack is None:
|
|
raise PackNotFoundError(f"Pack {pack_id} not found")
|
|
return await self.pack_repo.add_entry(
|
|
pack_id=pack_id,
|
|
sense_id=sense_id,
|
|
surface_text=surface_text,
|
|
)
|
|
|
|
async def add_flashcard_template_to_entry(
|
|
self,
|
|
pack_entry_id: uuid.UUID,
|
|
prompt_text: str,
|
|
answer_text: str,
|
|
prompt_context_text: str | None = None,
|
|
answer_context_text: str | None = None,
|
|
) -> WordBankPackFlashcardTemplate:
|
|
return await self.pack_repo.add_flashcard_template(
|
|
pack_entry_id=pack_entry_id,
|
|
prompt_text=prompt_text,
|
|
answer_text=answer_text,
|
|
prompt_context_text=prompt_context_text,
|
|
answer_context_text=answer_context_text,
|
|
)
|
|
|
|
async def add_pack_to_user_bank(
|
|
self,
|
|
pack_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
source_lang: str,
|
|
target_lang: str,
|
|
) -> PackApplicationResult:
|
|
pack = await self.pack_repo.get_pack(pack_id)
|
|
if pack is None or not pack.is_published:
|
|
raise PackNotFoundError(f"Pack {pack_id} not found")
|
|
|
|
entries = await self.pack_repo.get_entries_for_pack(pack_id)
|
|
if not entries:
|
|
return PackApplicationResult(added_surface_texts=[])
|
|
|
|
pair = await self.vocab_repo.get_or_create_language_pair(user_id, source_lang, target_lang)
|
|
language_pair_id = uuid.UUID(pair.id)
|
|
|
|
entry_ids = [uuid.UUID(e.id) for e in entries]
|
|
templates_by_entry = await self.pack_repo.get_templates_for_entries(entry_ids)
|
|
existing_sense_ids = await self.vocab_repo.get_sense_ids_for_user_in_pair(
|
|
user_id, language_pair_id
|
|
)
|
|
|
|
# Detect plain-card duplicates: entries whose sense is already in the user's bank
|
|
# and whose templates carry no context text (would produce identical plain cards).
|
|
duplicates = []
|
|
for entry in entries:
|
|
if entry.sense_id is None or entry.sense_id not in existing_sense_ids:
|
|
continue
|
|
entry_templates = templates_by_entry.get(entry.id, [])
|
|
has_context = any(
|
|
t.prompt_context_text or t.answer_context_text for t in entry_templates
|
|
)
|
|
if not has_context:
|
|
duplicates.append(entry.surface_text)
|
|
|
|
if duplicates:
|
|
raise DuplicateEntryError(duplicates)
|
|
|
|
added: list[str] = []
|
|
for entry in entries:
|
|
bank_entry = await self.vocab_repo.add_entry(
|
|
user_id=user_id,
|
|
language_pair_id=language_pair_id,
|
|
surface_text=entry.surface_text,
|
|
entry_pathway="pack",
|
|
sense_id=uuid.UUID(entry.sense_id) if entry.sense_id else None,
|
|
disambiguation_status="auto_resolved" if entry.sense_id else "pending",
|
|
pack_entry_id=uuid.UUID(entry.id),
|
|
)
|
|
|
|
entry_templates = templates_by_entry.get(entry.id, [])
|
|
if entry_templates:
|
|
for template in entry_templates:
|
|
await self.flashcard_repo.create_flashcard(
|
|
user_id=user_id,
|
|
bank_entry_id=uuid.UUID(bank_entry.id),
|
|
source_lang=pair.source_lang,
|
|
target_lang=pair.target_lang,
|
|
prompt_text=template.prompt_text,
|
|
answer_text=template.answer_text,
|
|
prompt_context_text=template.prompt_context_text,
|
|
answer_context_text=template.answer_context_text,
|
|
source_pack_flashcard_template_id=uuid.UUID(template.id),
|
|
)
|
|
elif entry.sense_id:
|
|
# Fallback: no templates — generate plain direction cards from the dictionary
|
|
await self._generate_plain_cards(
|
|
bank_entry_id=uuid.UUID(bank_entry.id),
|
|
user_id=user_id,
|
|
sense_id=uuid.UUID(entry.sense_id),
|
|
source_lang=pair.source_lang,
|
|
target_lang=pair.target_lang,
|
|
)
|
|
|
|
added.append(entry.surface_text)
|
|
|
|
return PackApplicationResult(added_surface_texts=added)
|
|
|
|
async def _generate_plain_cards(
|
|
self,
|
|
bank_entry_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
sense_id: uuid.UUID,
|
|
source_lang: str,
|
|
target_lang: str,
|
|
) -> None:
|
|
sense = await self.dict_repo.get_sense(sense_id)
|
|
if sense is None:
|
|
return
|
|
lemma = await self.dict_repo.get_lemma(uuid.UUID(sense.lemma_id))
|
|
if lemma is None:
|
|
return
|
|
for prompt, answer in [
|
|
(lemma.headword, sense.gloss),
|
|
(sense.gloss, lemma.headword),
|
|
]:
|
|
await self.flashcard_repo.create_flashcard(
|
|
user_id=user_id,
|
|
bank_entry_id=bank_entry_id,
|
|
source_lang=source_lang,
|
|
target_lang=target_lang,
|
|
prompt_text=prompt,
|
|
answer_text=answer,
|
|
)
|