Fix ingest.py: use correct service signatures (chunk_document, summarize_book, file_path args)

app/api/ingest.py

@@ -4,23 +4,21 @@ Routes (per implementation plan):
   POST /api/admin/{author_slug}/books/upload          — Upload a book file
   GET  /api/admin/{author_slug}/books/upload/progress — SSE ingestion progress
 
-RULE: Requires Author JWT (Bearer token).
-RULE: Magic byte validation before any processing.
+Saves uploaded file to disk first, then passes the path to services
+that expect file_path (not raw bytes).
 """
 
 import asyncio
 import json
+import os
+import tempfile
+from pathlib import Path
 
 from fastapi import APIRouter, Depends, File, UploadFile, HTTPException
 from fastapi.responses import StreamingResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.dependencies import get_db, get_current_user, get_redis
-from app.services.file_validator import validate_file
-from app.services.parser import parse_document
-from app.services.chunker import chunk_text
-from app.services.embeddings import embed_and_store
-from app.services.summarizer import generate_summary
 from app.repositories.book_repo import BookRepository
 from app.repositories.document_repo import DocumentRepository
 from app.config import get_settings
@@ -41,42 +39,42 @@ async def upload_book(
     db: AsyncSession = Depends(get_db),
     redis=Depends(get_redis),
 ):
-    """Upload a book file (PDF/EPUB/DOCX/TXT) and start async ingestion.
-
-    Responds immediately with book_id. Use /progress SSE to track ingestion.
-    """
-    # Validate file (magic bytes, not just extension)
+    """Upload a book file (PDF/EPUB/DOCX/TXT) and start async ingestion."""
     contents = await file.read()
-    ext = validate_file(contents, file.filename or "")
+    filename = file.filename or "upload"
 
     book_repo = BookRepository(db)
     doc_repo = DocumentRepository(db)
 
     book = await book_repo.create(
         author_id=current_user.id,
-        title=title or (file.filename or "Untitled").rsplit(".", 1)[0],
+        title=title or filename.rsplit(".", 1)[0],
         genre=genre,
         price=price,
         buy_url=buy_url,
         status="processing",
     )
 
+    # Save to temp dir so file-path-based services can access it
+    upload_dir = Path(cfg.UPLOAD_STORAGE_PATH)
+    upload_dir.mkdir(parents=True, exist_ok=True)
+    suffix = Path(filename).suffix or ".bin"
+    tmp_path = upload_dir / f"{book.id}{suffix}"
+    tmp_path.write_bytes(contents)
+
     doc = await doc_repo.create(
         book_id=book.id,
         author_id=current_user.id,
-        filename=file.filename or "upload",
-        file_extension=ext,
+        filename=filename,
+        file_extension=suffix.lstrip("."),
         file_size_bytes=len(contents),
     )
     await db.commit()
 
-    # Fire async ingestion in background
     asyncio.create_task(
         _run_ingestion(
-            contents=contents,
-            ext=ext,
+            file_path=str(tmp_path),
             book_id=book.id,
-            doc_id=doc.id,
             author_id=current_user.id,
             title=book.title,
             redis=redis,
@@ -94,7 +92,7 @@ async def ingestion_progress(
     current_user=Depends(get_current_user),
     redis=Depends(get_redis),
 ):
-    """SSE stream for real-time ingestion progress of a specific book."""
+    """SSE stream for real-time ingestion progress."""
     async def event_stream():
         channel = f"ingestion:{current_user.id}:{book_id}"
         pubsub = redis.pubsub()
@@ -117,36 +115,46 @@ async def ingestion_progress(
 
 
 async def _publish(redis, channel: str, stage: str, progress: int, detail: str = "") -> None:
-    """Publish an SSE progress event to Redis pub/sub."""
-    await redis.publish(channel, json.dumps({"stage": stage, "progress": progress, "detail": detail}))
+    if redis:
+        try:
+            await redis.publish(channel, json.dumps({"stage": stage, "progress": progress, "detail": detail}))
+        except Exception:
+            pass
 
 
 async def _run_ingestion(
-    contents: bytes,
-    ext: str,
+    file_path: str,
     book_id: str,
-    doc_id: str,
     author_id: str,
     title: str,
     redis,
     db: AsyncSession,
 ) -> None:
-    """Full async ingestion: parse → chunk → embed → summarize."""
+    """Full async ingestion: validate → parse → chunk → embed → summarize."""
     channel = f"ingestion:{author_id}:{book_id}"
     book_repo = BookRepository(db)
 
     try:
-        await _publish(redis, channel, "parsing", 10)
-        text = parse_document(contents, ext)
+        await _publish(redis, channel, "validating", 5)
+        from app.services.file_validator import validate_file
+        ext = validate_file(file_path)
+
+        await _publish(redis, channel, "parsing", 20)
+        from app.services.parser import parse_document
+        parse_result = parse_document(file_path, ext)
+        text = parse_result.text if hasattr(parse_result, "text") else str(parse_result)
 
-        await _publish(redis, channel, "chunking", 30)
-        chunks = chunk_text(text)
+        await _publish(redis, channel, "chunking", 40)
+        from app.services.chunker import chunk_document
+        chunks = chunk_document(text)
 
-        await _publish(redis, channel, "embedding", 50)
+        await _publish(redis, channel, "embedding", 60)
+        from app.services.embeddings import embed_and_store
         collection_name = await embed_and_store(chunks, author_id, book_id, title)
 
-        await _publish(redis, channel, "summarizing", 80)
-        summary = await generate_summary(text[:3000], title)
+        await _publish(redis, channel, "summarizing", 85)
+        from app.services.summarizer import summarize_book
+        summary = await summarize_book(text[:3000])
 
         await book_repo.update(book_id, {
             "status": "active",
@@ -155,10 +163,17 @@ async def _run_ingestion(
             "chunk_count": len(chunks),
         })
         await db.commit()
-
         await _publish(redis, channel, "complete", 100)
 
     except Exception as e:
         await book_repo.update(book_id, {"status": "failed"})
         await db.commit()
         await _publish(redis, channel, "error", 0, str(e))
+
+    finally:
+        # Clean up temp file
+        try:
+            if os.path.exists(file_path):
+                os.remove(file_path)
+        except Exception:
+            pass