Add basis workspace filters, stale-on-reimport, and variant usage metrics

app/services/ai_generation.py

@@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.config import get_settings
 from app.models.item import Item
+from app.models.ai_generation_run import AIGenerationRun
 from app.models.tryout import Tryout
 from app.models.user_answer import UserAnswer
 from app.schemas.ai import GeneratedQuestion
@@ -493,6 +494,9 @@ async def save_ai_question(
     level: Literal["mudah", "sedang", "sulit"],
     ai_model: str,
     db: AsyncSession,
+    generation_run_id: int | None = None,
+    source_snapshot_question_id: int | None = None,
+    variant_status: str = "draft",
 ) -> Optional[int]:
     """
     Save AI-generated question to database.
@@ -523,6 +527,9 @@ async def save_ai_question(
             generated_by="ai",
             ai_model=ai_model,
             basis_item_id=basis_item_id,
+            generation_run_id=generation_run_id,
+            source_snapshot_question_id=source_snapshot_question_id,
+            variant_status=variant_status,
             calibrated=False,
             ctt_p=None,
             ctt_bobot=None,
@@ -547,6 +554,50 @@ async def save_ai_question(
         return None
 
 
+async def create_generation_run(
+    basis_item_id: int,
+    target_level: Literal["mudah", "sulit"],
+    requested_count: int,
+    model: str,
+    created_by: str,
+    db: AsyncSession,
+    source_snapshot_question_id: int | None = None,
+    operator_notes: str | None = None,
+    prompt_version: str = "v1",
+) -> int:
+    run = AIGenerationRun(
+        basis_item_id=basis_item_id,
+        source_snapshot_question_id=source_snapshot_question_id,
+        target_level=target_level,
+        requested_count=requested_count,
+        model=model,
+        prompt_version=prompt_version,
+        operator_notes=operator_notes,
+        created_by=created_by,
+    )
+    db.add(run)
+    await db.flush()
+    return int(run.id)
+
+
+async def generate_questions_batch(
+    basis_item: Item,
+    target_level: Literal["mudah", "sulit"],
+    ai_model: str,
+    count: int,
+) -> list[GeneratedQuestion]:
+    generated_items: list[GeneratedQuestion] = []
+    for _ in range(count):
+        generated = await generate_question(
+            basis_item=basis_item,
+            target_level=target_level,
+            ai_model=ai_model,
+        )
+        if generated is not None:
+            generated_items.append(generated)
+    return generated_items
+
+
 async def get_ai_stats(db: AsyncSession) -> Dict[str, Any]:
     """
     Get AI generation statistics.