import io import json import zipfile from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import StreamingResponse from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.models.document import Document from app.models.project import Project from app.routers.auth import get_current_agent from app.routers.documents import tiptap_to_markdown router = APIRouter(tags=["export"]) async def _get_doc_with_access(request: Request, document_id: str, db: AsyncSession) -> Document: """Get document and verify access.""" agent = await get_current_agent(request, db) result = await db.execute( select(Document).where( Document.id == document_id, Document.is_deleted == False, ) ) doc = result.scalar_one_or_none() if not doc: raise HTTPException(status_code=404, detail="Document not found") proj_result = await db.execute( select(Project).where( Project.id == doc.project_id, Project.agent_id == agent.id, Project.is_deleted == False, ) ) if not proj_result.scalar_one_or_none(): raise HTTPException(status_code=403, detail="Forbidden") return doc async def _get_project_with_access(request: Request, project_id: str, db: AsyncSession) -> tuple[Project, str]: """Get project and verify access. Returns (project, project_name).""" agent = await get_current_agent(request, db) result = await db.execute( select(Project).where( Project.id == project_id, Project.agent_id == agent.id, Project.is_deleted == False, ) ) project = result.scalar_one_or_none() if not project: raise HTTPException(status_code=404, detail="Project not found") return project, project.name @router.get("/api/v1/documents/{document_id}/export") async def export_document( request: Request, document_id: str, format: str = Query(..., regex="^(markdown|json)$"), db: AsyncSession = Depends(get_db), ): """ Export a single document as Markdown or JSON. """ doc = await _get_doc_with_access(request, document_id, db) if format == "markdown": # Convert tiptap to markdown if doc.tiptap_content: try: tiptap = json.loads(doc.tiptap_content) content = tiptap_to_markdown(tiptap) except json.JSONDecodeError: content = doc.content else: content = doc.content # Add frontmatter filename = f"{doc.title}.md" output = f"# {doc.title}\n\n{content}" return StreamingResponse( iter([output]), media_type="text/markdown", headers={ "Content-Disposition": f'attachment; filename="{filename}"' } ) else: # json # Parse tiptap_content tiptap_content = None if doc.tiptap_content: try: tiptap_content = json.loads(doc.tiptap_content) except json.JSONDecodeError: tiptap_content = None # Parse reasoning_steps reasoning_steps = [] if doc.reasoning_steps: try: reasoning_steps = json.loads(doc.reasoning_steps) except json.JSONDecodeError: reasoning_steps = [] # Parse confidence confidence = None if doc.confidence: try: confidence = float(doc.confidence) except (ValueError, TypeError): confidence = None # Parse outgoing_links outgoing_links = [] if doc.outgoing_links: try: outgoing_links = json.loads(doc.outgoing_links) except json.JSONDecodeError: outgoing_links = [] export_data = { "id": doc.id, "title": doc.title, "content": doc.content, "tiptap_content": tiptap_content, "created_at": doc.created_at.isoformat(), "updated_at": doc.updated_at.isoformat(), "metadata": { "reasoning_type": doc.reasoning_type, "confidence": confidence, "reasoning_steps": reasoning_steps, "model_source": doc.model_source, "outgoing_links": outgoing_links, } } filename = f"{doc.title}.json" output = json.dumps(export_data, indent=2, ensure_ascii=False) return StreamingResponse( iter([output]), media_type="application/json", headers={ "Content-Disposition": f'attachment; filename="{filename}"' } ) @router.get("/api/v1/projects/{project_id}/export") async def export_project( request: Request, project_id: str, format: str = Query(..., regex="^(zip|json)$"), include_metadata: bool = Query(True), db: AsyncSession = Depends(get_db), ): """ Export a complete project as ZIP (with .md files) or JSON. """ project, project_name = await _get_project_with_access(request, project_id, db) # Get all documents docs_result = await db.execute( select(Document).where( Document.project_id == project_id, Document.is_deleted == False, ).order_by(Document.created_at) ) all_docs = docs_result.scalars().all() # Check size (warn at 50MB, hard limit at 100MB) total_size = sum( len(d.content or "") + len(d.tiptap_content or "") + len(d.title) for d in all_docs ) if total_size > 100_000_000: raise HTTPException(status_code=507, detail="Project too large to export (max 100MB)") if format == "json": documents = [] for doc in all_docs: tiptap_content = None if doc.tiptap_content: try: tiptap_content = json.loads(doc.tiptap_content) except json.JSONDecodeError: pass outgoing_links = [] if doc.outgoing_links: try: outgoing_links = json.loads(doc.outgoing_links) except json.JSONDecodeError: pass metadata = {} if include_metadata: reasoning_steps = [] if doc.reasoning_steps: try: reasoning_steps = json.loads(doc.reasoning_steps) except json.JSONDecodeError: pass confidence = None if doc.confidence: try: confidence = float(doc.confidence) except: pass metadata = { "reasoning_type": doc.reasoning_type, "confidence": confidence, "reasoning_steps": reasoning_steps, "model_source": doc.model_source, } documents.append({ "id": doc.id, "title": doc.title, "content": doc.content, "tiptap_content": tiptap_content if include_metadata else None, "outgoing_links": outgoing_links, "metadata": metadata, }) export_data = { "project": { "id": project.id, "name": project.name, "description": project.description, "created_at": project.created_at.isoformat(), "updated_at": project.updated_at.isoformat(), }, "documents": documents, "exported_at": datetime.utcnow().isoformat(), "format_version": "3.0", } filename = f"{project_name}-export.json" output = json.dumps(export_data, indent=2, ensure_ascii=False) return StreamingResponse( iter([output]), media_type="application/json", headers={ "Content-Disposition": f'attachment; filename="{filename}"' } ) else: # zip buffer = io.BytesIO() with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zf: # Add project.json project_meta = { "id": project.id, "name": project.name, "description": project.description, "created_at": project.created_at.isoformat(), "updated_at": project.updated_at.isoformat(), } zf.writestr( "project.json", json.dumps(project_meta, indent=2, ensure_ascii=False) ) # Add documents for doc in all_docs: # Convert content to markdown if doc.tiptap_content: try: tiptap = json.loads(doc.tiptap_content) content = tiptap_to_markdown(tiptap) except json.JSONDecodeError: content = doc.content else: content = doc.content md_content = f"# {doc.title}\n\n{content}" safe_title = "".join(c if c.isalnum() or c in " -_" else "_" for c in doc.title) zf.writestr(f"documents/{safe_title}.md", md_content) buffer.seek(0) filename = f"{project_name}-export.zip" return StreamingResponse( iter([buffer.read()]), media_type="application/zip", headers={ "Content-Disposition": f'attachment; filename="{filename}"' } )