"""Author RAG — Q&A Service. Business logic for custom Q&A training data: CRUD, CSV import/export, and question embedding for semantic matching. """ import csv import io import re import json import structlog from fastapi import HTTPException, UploadFile from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.models.custom_qa import CustomQA from app.repositories.qa_repo import QARepository from app.schemas.admin import QACreateRequest, QAUpdateRequest from app.services.pipeline.helpers import _get_openai_client logger = structlog.get_logger(__name__) cfg = get_settings() MAX_QA_PAIRS = 500 _SCRIPT_TAG_RE = re.compile(r"]*>.*?", re.IGNORECASE | re.DOTALL) _HTML_TAG_RE = re.compile(r"<[^>]+>") def _sanitize_qa_text(text: str) -> str: """Strip script tags and HTML from Q&A CSV imports (R-146).""" cleaned = _SCRIPT_TAG_RE.sub("", text) cleaned = _HTML_TAG_RE.sub("", cleaned) return cleaned.strip() async def _embed_question_safe(question: str) -> str | None: """Embed a Q&A question for semantic matching. Returns JSON string or None.""" try: client = _get_openai_client() resp = await client.embeddings.create( model=cfg.OPENAI_EMBEDDING_MODEL, input=question, ) return json.dumps(resp.data[0].embedding) except Exception as e: logger.warning( "Q&A embedding failed — saving without embedding (Jaccard fallback active)", error=str(e), ) return None def _qa_to_dict(qa: CustomQA) -> dict: """Serialize a CustomQA row for API response.""" return { "id": qa.id, "book_id": qa.book_id, "question": qa.question, "answer": qa.answer, "priority": qa.priority, "is_active": qa.is_active, "match_count": qa.match_count, "match_threshold": qa.match_threshold, "category": qa.category, "created_at": qa.created_at.isoformat() if qa.created_at else None, } class QAService: """Orchestrates custom Q&A management for admin panel.""" def __init__(self, db: AsyncSession) -> None: self._db = db self._qa = QARepository(db) async def list_qa(self, author_id: str, book_id: str | None = None) -> dict: """List all custom Q&A pairs for the author.""" items = await self._qa.list_for_author(author_id, book_id) return { "qa_pairs": [_qa_to_dict(qa) for qa in items], "total": len(items), } async def create_qa(self, author_id: str, body: QACreateRequest) -> dict: """Create a new custom Q&A pair.""" count = await self._qa.count_for_author(author_id) if count >= MAX_QA_PAIRS: raise HTTPException(400, "Maximum 500 Q&A pairs allowed") qa = CustomQA( author_id=author_id, book_id=body.book_id, question=body.question, answer=body.answer, priority=body.priority, category=body.category, match_threshold=body.match_threshold, embedding_json=await _embed_question_safe(body.question), ) self._db.add(qa) await self._db.commit() return { "id": qa.id, "message": "Q&A pair created", "has_embedding": qa.embedding_json is not None, } async def import_qa_csv(self, author_id: str, file: UploadFile) -> dict: """Bulk import Q&A pairs from CSV (columns: question,answer,category,priority).""" contents = await file.read() text = contents.decode("utf-8-sig") reader = csv.DictReader(io.StringIO(text)) if not reader.fieldnames or "question" not in reader.fieldnames or "answer" not in reader.fieldnames: raise HTTPException(400, "CSV must have 'question' and 'answer' columns") current_count = await self._qa.count_for_author(author_id) imported = 0 skipped = 0 errors: list[str] = [] for i, row in enumerate(reader, 1): if current_count + imported >= MAX_QA_PAIRS: errors.append(f"Row {i}: Limit of 500 Q&A pairs reached") break q = _sanitize_qa_text((row.get("question") or "").strip()) a = _sanitize_qa_text((row.get("answer") or "").strip()) if not q or not a: skipped += 1 continue if len(q) > 500 or len(a) > 2000: errors.append(f"Row {i}: Question or answer too long") skipped += 1 continue qa = CustomQA( author_id=author_id, question=q[:500], answer=a[:2000], category=(row.get("category") or "").strip()[:50] or None, priority=int(row.get("priority") or 0), ) self._db.add(qa) imported += 1 await self._db.commit() return { "imported": imported, "skipped": skipped, "errors": errors[:10], "message": f"Imported {imported} Q&A pairs", } async def export_qa_csv(self, author_id: str) -> StreamingResponse: """Export all Q&A pairs as a CSV download.""" items = await self._qa.list_for_author(author_id) output = io.StringIO() writer = csv.writer(output) writer.writerow([ "question", "answer", "category", "priority", "is_active", "match_count", ]) for qa in items: writer.writerow([ qa.question, qa.answer, qa.category or "", qa.priority, qa.is_active, qa.match_count, ]) output.seek(0) return StreamingResponse( iter([output.getvalue()]), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=qa_pairs.csv"}, ) async def update_qa( self, author_id: str, qa_id: str, body: QAUpdateRequest, ) -> dict: """Update an existing Q&A pair.""" qa = await self._qa.get_for_author(author_id, qa_id) if not qa: raise HTTPException(404, "Q&A pair not found") question_changed = body.question is not None if body.question is not None: qa.question = body.question if body.answer is not None: qa.answer = body.answer if body.priority is not None: qa.priority = body.priority if body.category is not None: qa.category = body.category if body.match_threshold is not None: qa.match_threshold = body.match_threshold if hasattr(body, "is_active") and body.is_active is not None: qa.is_active = body.is_active if hasattr(body, "book_id") and body.book_id is not None: qa.book_id = body.book_id if question_changed or qa.embedding_json is None: qa.embedding_json = await _embed_question_safe(qa.question) await self._db.commit() return { "message": "Q&A pair updated", "has_embedding": qa.embedding_json is not None, } async def delete_qa(self, author_id: str, qa_id: str) -> dict: """Delete a Q&A pair.""" qa = await self._qa.get_for_author(author_id, qa_id) if not qa: raise HTTPException(404, "Q&A pair not found") await self._db.delete(qa) await self._db.commit() return {"message": "Q&A pair deleted"}