Upload 14 files

RAG_SYSTEM_PLAN.md

@@ -139,7 +139,7 @@ Request body:
 {
   "question": "What is the penalty for late tax filing?",
   "top_k": 5,
-  "model": "gemini-2.0-flash"
+  "model": "gemini-2.5-flash"
 }
 ```
 
@@ -201,7 +201,7 @@ Key settings in `ingest.py`:
 ## Models Used
 
 - **Embeddings**: `text-embedding-004` (768 dimensions)
-- **Generation**: `gemini-2.0-flash` (default, fast)
+- **Generation**: `gemini-2.5-flash` (default, fast)
   - Can also use `gemini-2.0-pro` for complex reasoning
 
 ## Security Considerations

app.py

@@ -7,7 +7,7 @@ from contextlib import asynccontextmanager
 from collections import defaultdict
 
 from fastapi import FastAPI, HTTPException, UploadFile, File, Request, Depends, Form
-from typing import Optional
+from typing import Optional, List
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.security import APIKeyHeader
 from pydantic import BaseModel, Field
@@ -147,6 +147,18 @@ class HealthResponse(BaseModel):
     vectors_indexed: int
 
 
+class YearlyWrapRequest(BaseModel):
+    year: int = Field(default=2024, ge=2000, le=2030)
+
+
+class YearlyWrapResponse(BaseModel):
+    analysis: dict
+    video_script: Optional[dict]
+    video_url: Optional[str]
+    status: str
+    message: str
+
+
 @app.get("/", response_model=dict)
 async def root():
     return {
@@ -209,7 +221,8 @@ async def ask_question(
     
     image_data = None
     image_mime_type = None
-    
+    document_text = ""
+
     if image and image.filename:
         allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
         if image.content_type not in allowed_types:
@@ -222,7 +235,38 @@ async def ask_question(
         
         image_data = await image.read()
         image_mime_type = image.content_type
-    
+
+    if document and document.filename:
+        allowed_exts = [".pdf", ".doc", ".docx", ".txt"]
+        if not any(document.filename.lower().endswith(ext) for ext in allowed_exts):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid document type. Allowed: {', '.join(allowed_exts)}"
+            )
+        if document.size and document.size > 50 * 1024 * 1024:
+            raise HTTPException(status_code=400, detail="Document too large. Max 50MB.")
+
+        doc_content = await document.read()
+
+        try:
+            from rag.ingest import extract_text_from_file
+            import tempfile
+            import os
+
+            with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(document.filename)[1]) as tmp_file:
+                tmp_file.write(doc_content)
+                tmp_file_path = tmp_file.name
+
+            try:
+                document_text = extract_text_from_file(Path(tmp_file_path))
+                if document_text:
+                    document_text = f"[User Uploaded Document: {document.filename}]\n{document_text}"
+            finally:
+                os.unlink(tmp_file_path)
+
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Error processing document: {str(e)}")
+
     try:
         query_embedding = generate_query_embedding(gemini_client, question)
     except Exception as e:
@@ -265,7 +309,21 @@ async def ask_question(
         })
     
     context = "\n\n---\n\n".join(context_parts)
-    
+
+    if document_text:
+        context = f"""[Tax Document Analysis - User Uploaded File]
+{document_text}
+
+--- Tax Law Reference Context ---
+{context}
+
+[TAX ANALYSIS INSTRUCTIONS]
+- Analyze the uploaded document for tax-relevant information, forms, and declarations
+- Identify tax amounts, deadlines, compliance requirements, and filing obligations
+- Cross-reference with Nigerian tax laws from the retrieved context
+- Provide specific guidance on tax declarations, calculations, and compliance
+- Highlight any missing information or additional documents needed"""
+
     conversation_history = conversation_sessions.get(session_id, [])
     
     try:
@@ -354,7 +412,7 @@ async def ingest_document(
 async def get_stats(api_key: str = Depends(verify_api_key)):
     if pinecone_index is None:
         raise HTTPException(status_code=503, detail="Pinecone not initialized.")
-    
+
     try:
         stats = pinecone_index.describe_index_stats()
         return StatsResponse(
@@ -366,6 +424,139 @@ async def get_stats(api_key: str = Depends(verify_api_key)):
         raise HTTPException(status_code=500, detail=f"Error getting stats: {str(e)}")
 
 
+@app.post("/yearly-wrap", response_model=YearlyWrapResponse)
+async def create_yearly_wrap(
+    request: Request,
+    year: int = Form(..., ge=2000, le=2030),
+    api_key: str = Depends(verify_api_key)
+):
+    """
+    Create a yearly financial wrap video from account statements and financial documents.
+
+    Upload your bank statements, investment reports, tax documents, and financial images
+    to generate a personalized yearly financial summary video.
+    """
+    # Parse multipart form data manually to handle optional files properly
+    form = await request.form()
+
+    # Get documents and images from form data
+    documents = []
+    images = []
+
+    for field_name, field_value in form.items():
+        if field_name.startswith("documents"):
+            if hasattr(field_value, 'filename') and field_value.filename:
+                documents.append(field_value)
+        elif field_name.startswith("images"):
+            if hasattr(field_value, 'filename') and field_value.filename:
+                images.append(field_value)
+
+    if not documents and not images:
+        raise HTTPException(
+            status_code=400,
+            detail="Please upload at least one financial document or image"
+        )
+
+    # Process documents
+    document_texts = []
+    image_data_list = []
+
+    # Handle documents
+    if documents:
+        for doc in documents:
+            if doc.filename:
+                allowed_exts = [".pdf", ".doc", ".docx", ".txt"]
+                if not any(doc.filename.lower().endswith(ext) for ext in allowed_exts):
+                    raise HTTPException(
+                        status_code=400,
+                        detail=f"Document type not supported: {doc.filename}"
+                    )
+                if doc.size and doc.size > 50 * 1024 * 1024:
+                    raise HTTPException(status_code=400, detail="Document too large. Max 50MB.")
+
+                doc_content = await doc.read()
+
+                try:
+                    from rag.ingest import extract_text_from_file
+                    import tempfile
+
+                    with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(doc.filename)[1]) as tmp_file:
+                        tmp_file.write(doc_content)
+                        tmp_file_path = tmp_file.name
+
+                    try:
+                        text = extract_text_from_file(Path(tmp_file_path))
+                        if text:
+                            document_texts.append(f"[Document: {doc.filename}]\n{text}")
+                    finally:
+                        os.unlink(tmp_file_path)
+
+                except Exception as e:
+                    raise HTTPException(status_code=500, detail=f"Error processing document {doc.filename}: {str(e)}")
+
+    # Handle images
+    if images:
+        for img in images:
+            if img.filename:
+                allowed_types = ["image/jpeg", "image/png", "image/gif", "image/webp"]
+                if img.content_type not in allowed_types:
+                    raise HTTPException(
+                        status_code=400,
+                        detail=f"Image type not supported: {img.filename}"
+                    )
+                if img.size and img.size > 10 * 1024 * 1024:
+                    raise HTTPException(status_code=400, detail="Image too large. Max 10MB.")
+
+                img_data = await img.read()
+                image_data_list.append(img_data)
+
+    # Analyze financial data
+    try:
+        from rag.utils import analyze_financial_documents
+        analysis = analyze_financial_documents(document_texts, image_data_list)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error analyzing financial data: {str(e)}")
+
+    # Generate video script and video
+    video_script = None
+    video_url = None
+
+    try:
+        from rag.utils import create_video_script, generate_yearly_wrap_video
+
+        # Step 1: Create professional video script
+        video_script = create_video_script(analysis)
+
+        # Step 2: Generate video from script
+        video_url = generate_yearly_wrap_video(analysis)
+
+        if video_url:
+            status = "completed"
+            message = f"Yearly financial wrap for {year} created successfully with professional video script and animation!"
+        else:
+            status = "script_only"
+            message = f"Yearly financial wrap script created successfully! Video generation is temporarily unavailable due to service limitations, but you have a complete professional script ready."
+
+    except Exception as e:
+        error_msg = str(e)
+        if "Video generation" in error_msg:
+            # Script succeeded, video failed
+            status = "script_only"
+            message = f"Financial analysis and video script created, but video generation failed: {error_msg}"
+        else:
+            # Both failed
+            status = "analysis_only"
+            message = f"Financial analysis completed, but script/video generation failed: {error_msg}"
+
+    return YearlyWrapResponse(
+        analysis=analysis,
+        video_script=video_script,
+        video_url=video_url,
+        status=status,
+        message=message
+    )
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)

rag/.env

@@ -1,6 +1,7 @@
-GEMINI_API_KEY=AIzaSyBEEUyeYjDewcGffhHSbtsHjuhngyi3Coo
+GEMINI_API_KEY=AIzaSyCxyrpiDh8ancmuUd55Ak2hAEPyC9dfzqM
 PINECONE_API_KEY=pcsk_2BGUcf_CzBnGUF9jP7UTgL6Ned77DVj6zV75RBGyKfFMxVqzw36bAQAc6HiH1nwdMLBoYA
 PINECONE_INDEX=sabitax
+LUMA_API_KEY=
 
 # Security
 API_KEY=11e10c46685090a8a464f7c8a8f09cd519b69836935a2c8897b71472e2b74138

rag/utils.py

@@ -7,6 +7,10 @@ from dotenv import load_dotenv
 from google import genai
 from google.genai import types
 from PIL import Image
+try:
+    from lumaai import LumaAI
+except ImportError:
+    LumaAI = None
 
 load_dotenv()
 
@@ -128,41 +132,73 @@ Respond naturally and conversationally. Introduce yourself as SabiTax in a frien
                     role = "User" if msg["role"] == "user" else "You (SabiTax)"
                     history_text += f"{role}: {msg['content']}\n"
                 history_text += "\n"
-            
-            prompt = f"""You are SabiTax, a friendly and conversational legal and tax expert assistant specializing in Nigerian law and taxation. You talk to users like you're having a natural conversation with a friend - warm, approachable, and easy to understand.
-
-Your style:
-- Talk naturally, like you're chatting over coffee
-- Use "you" and "I" - make it personal and engaging
-- Be warm and friendly, not robotic or formal
-- Use everyday language and simple explanations
-- Reference previous parts of the conversation when relevant: "As I mentioned earlier..." or "Building on what we discussed..."
-- Ask follow-up questions if helpful: "Does that make sense?" or "Want me to explain that differently?"
-- Show enthusiasm about helping: "Great question!" or "I'm happy to help with that!"
-
-Your approach:
-1. **Reason through the information**: Think about what the user really needs to know
-2. **Break it down simply**: Translate complex legal stuff into everyday language
-3. **Make it practical**: Focus on "what this means for you" and "what you need to do"
-4. **Prioritize current info**: Always mention the most recent laws first (2025 over 2020, etc.) and note if something's been updated
-5. **Continue the conversation**: If this is part of an ongoing discussion, naturally reference what was said before
+
+            prompt = f"""You are SabiTax, Nigeria's comprehensive tax assistant. You help Nigerians with all aspects of tax compliance, from understanding laws to filing returns and analyzing documents.
+
+Your expertise covers:
+- **General Tax Questions**: Rates, deadlines, deductions, tax planning
+- **Tax Calculations**: Personal income tax, company tax, VAT, capital gains tax
+- **Form Guidance**: How to complete and file tax forms (Form A, Form B, etc.)
+- **Compliance Requirements**: What to declare, when to file, penalties
+- **Document Analysis**: Reviewing tax returns, financial statements, receipts
+- **Tax Optimization**: Legal ways to minimize tax liability
+- **Business Taxes**: Company registration, payroll taxes, VAT compliance
+
+Your communication style:
+- Professional yet approachable, like a trusted tax consultant
+- Explain complex concepts in simple, everyday Nigerian English
+- Use clear examples: "If you earn N3 million yearly, your tax is calculated as..."
+- Be encouraging and patient with all tax-related questions
+- Always emphasize compliance and accuracy
+
+How you handle different types of questions:
+
+**For General Tax Questions:**
+- Provide accurate information from current Nigerian tax laws
+- Break down calculations step-by-step
+- Reference specific sections of tax acts
+- Give practical examples relevant to Nigerian taxpayers
+
+**For Document Analysis:**
+- Identify the type of document and its tax purpose
+- Extract key tax information (amounts, dates, taxpayer details)
+- Check for compliance with Nigerian tax requirements
+- Point out missing information or potential issues
+
+**For Tax Calculations:**
+- Use current tax rates and brackets
+- Show step-by-step calculations
+- Explain deductions and allowances
+- Calculate final tax payable
+
+**For Filing Guidance:**
+- Explain which forms to use and when
+- Guide through form completion
+- Highlight common mistakes to avoid
+- Provide filing deadlines and methods
+
+Tax-specific guidelines:
+- Always reference current legislation (2025 acts take precedence)
+- Use Nigerian Naira (₦) for amounts
+- Reference FIRS (Federal Inland Revenue Service) procedures
+- Explain tax terms clearly: "Assessable profit means your business income after expenses"
+- Highlight tax incentives and reliefs available to Nigerians
+- Emphasize voluntary compliance over penalties
 
 Important rules:
-- Answer based ONLY on the provided context from the documents
-- Always prioritize the most recent/current legislation (e.g., 2025 acts over 2020 acts)
-- If there's old info, mention it's been updated: "The old 2020 law has been replaced by the 2025 act..."
-- Explain everything in simple terms - no legal jargon without explanation
-- Use examples and analogies to make things clearer
-- If you don't have enough info, say so honestly: "I don't have enough details on that, but here's what I know..."
-- Keep it conversational - use short paragraphs, bullet points when helpful, but write like you're talking
-- If the user is continuing a topic from earlier, acknowledge it and build on the previous conversation
+- Base answers on the provided context from indexed tax documents
+- If context doesn't have enough information, clearly state this
+- When analyzing documents, be thorough but practical
+- Suggest consulting a professional tax advisor for complex situations
+- Always promote ethical tax practices and full compliance
+- If something is unclear, ask for clarification rather than assuming
 
 {history_text}Context from documents:
 {context}
 
 Question: {question}
 
-Respond naturally and conversationally. Explain things like you're helping a friend understand their taxes. Be clear, friendly, and focus on what they actually need to know. If this continues a previous topic, reference it naturally."""
+Provide comprehensive, accurate tax guidance. Whether it's a general question, document analysis, or calculation help, explain everything clearly and help the user understand their Nigerian tax obligations."""
 
     if image_data:
         img = Image.open(io.BytesIO(image_data))
@@ -195,6 +231,228 @@ Respond naturally and conversationally. Explain things like you're helping a fri
     raise Exception("Failed to generate answer after multiple attempts")
 
 
+def analyze_financial_documents(document_texts: list, image_data_list: list) -> dict:
+    """Analyze financial documents and extract key financial metrics."""
+
+    # Combine all document texts
+    all_text = "\n\n".join(document_texts) if document_texts else ""
+
+    # Add image analysis if images are provided
+    if image_data_list:
+        all_text += "\n\n[Image Analysis]: Please analyze any financial data visible in the images."
+
+    if not all_text.strip():
+        return {
+            "income": "No income data found",
+            "expenses": "No expense data found",
+            "savings": "No savings data found",
+            "achievements": "No financial achievements identified",
+            "summary": "Unable to analyze financial documents"
+        }
+
+    analysis_prompt = f"""Analyze this financial document and extract key financial information. Focus on:
+
+1. **Income**: Total income, sources, trends
+2. **Expenses**: Major expense categories and amounts
+3. **Savings**: Savings rate, emergency fund, investments
+4. **Achievements**: Financial milestones, debt reduction, investment growth
+5. **Summary**: Overall financial health and key insights
+
+Document content:
+{all_text[:10000]}
+
+Provide a structured analysis with specific amounts where available. If amounts aren't specified, use descriptive terms like "significant" or "moderate"."""
+
+    try:
+        client = get_gemini_client()
+        response = client.models.generate_content(
+            model="gemini-2.5-flash",
+            contents=analysis_prompt
+        )
+
+        # Parse the response to extract structured data
+        analysis_text = response.text
+
+        # Simple parsing - in production you might want more sophisticated parsing
+        analysis = {
+            "income": "Analysis completed - see detailed summary",
+            "expenses": "Analysis completed - see detailed summary",
+            "savings": "Analysis completed - see detailed summary",
+            "achievements": "Analysis completed - see detailed summary",
+            "summary": analysis_text[:1000]  # Truncate for response size
+        }
+
+        return analysis
+
+    except Exception as e:
+        return {
+            "income": "Error analyzing documents",
+            "expenses": "Error analyzing documents",
+            "savings": "Error analyzing documents",
+            "achievements": "Error analyzing documents",
+            "summary": f"Analysis failed: {str(e)}"
+        }
+
+
+def create_video_script(financial_analysis: dict) -> dict:
+    """Create a professional video script using Gemini 2.5 Flash."""
+
+    summary = financial_analysis.get("summary", "A year of financial growth and achievements")
+    income = financial_analysis.get("income", "Steady income growth")
+    expenses = financial_analysis.get("expenses", "Managed expenses effectively")
+    savings = financial_analysis.get("savings", "Built savings successfully")
+    achievements = financial_analysis.get("achievements", "Achieved financial goals")
+
+    script_prompt = f"""Create a professional year-in-review financial video script based on this user data:
+
+FINANCIAL DATA:
+- Summary: {summary}
+- Income: {income}
+- Expenses: {expenses}
+- Savings: {savings}
+- Achievements: {achievements}
+
+Create a cinematic 15-20 second video with 4-6 scenes. Output JSON with:
+
+{{
+  "scenes": [
+    {{
+      "scene_number": 1,
+      "duration": "3-4 seconds",
+      "description": "Brief scene description",
+      "video_prompt": "Detailed prompt for video generation AI",
+      "voiceover": "Voiceover text for this scene"
+    }}
+  ],
+  "music_mood": "uplifting, motivational, professional",
+  "overall_theme": "Financial success and growth",
+  "total_duration": "15-20 seconds"
+}}
+
+Make it professional, celebratory, and focused on financial achievements. Use Nigerian context where appropriate."""
+
+    try:
+        client = get_gemini_client()
+
+        response = client.models.generate_content(
+            model="gemini-2.5-flash",
+            contents=script_prompt
+        )
+
+        # Try to parse as JSON, fallback to text processing
+        try:
+            import json
+            script_data = json.loads(response.text.strip())
+            return script_data
+        except json.JSONDecodeError:
+            # Fallback: create structured script from text
+            return {
+                "scenes": [
+                    {
+                        "scene_number": 1,
+                        "duration": "5 seconds",
+                        "description": "Financial overview and achievements",
+                        "video_prompt": f"Create a professional financial recap video showing: {summary}. Use animated charts, money visualizations, and success indicators.",
+                        "voiceover": f"This year brought remarkable financial growth: {summary}"
+                    }
+                ],
+                "music_mood": "uplifting, professional",
+                "overall_theme": "Financial success story",
+                "total_duration": "15 seconds"
+            }
+
+    except Exception as e:
+        # Ultimate fallback
+        return {
+            "scenes": [
+                {
+                    "scene_number": 1,
+                    "duration": "5 seconds",
+                    "description": "Financial success visualization",
+                    "video_prompt": f"Professional financial recap: {summary}. Show growing charts, money animations, success celebrations.",
+                    "voiceover": f"A year of financial achievements: {summary}"
+                }
+            ],
+            "music_mood": "motivational",
+            "overall_theme": "Financial growth",
+            "total_duration": "10 seconds"
+        }
+
+
+def generate_yearly_wrap_video(financial_analysis: dict) -> str:
+    """Generate yearly financial wrap video using Gemini + LumaAI pipeline."""
+
+    # Step 1: Gemini creates the video script
+    script_data = create_video_script(financial_analysis)
+
+    # Step 2: Use the script to create video prompt
+    scenes = script_data.get("scenes", [])
+    if not scenes:
+        # Fallback scene
+        scenes = [{
+            "scene_number": 1,
+            "duration": "5 seconds",
+            "description": "Financial overview",
+            "video_prompt": f"Professional financial recap video showing: {financial_analysis.get('summary', 'Financial achievements')}. Animated charts, money visualizations, success indicators.",
+            "voiceover": f"Financial highlights: {financial_analysis.get('summary', 'Great year')}"
+        }]
+
+    # Combine all scene prompts into one comprehensive prompt
+    combined_prompt = f"""Create a cinematic financial year-in-review video:
+
+{scenes[0]['video_prompt']}
+
+Style: Professional, celebratory, modern financial visualization with animated charts, money effects, and success celebrations."""
+
+    try:
+        if LumaAI is None:
+            raise Exception("Video generation not available - lumaai not installed")
+
+        luma_api_key = os.environ.get("LUMAAI_API_KEY")
+        if not luma_api_key:
+            raise Exception("LUMAAI_API_KEY environment variable is not set")
+
+        client = LumaAI(auth_token=luma_api_key)
+
+        generation = client.generations.create(
+            prompt=combined_prompt,
+            model="ray-2",
+            aspect_ratio="16:9",
+            resolution="720p",
+            duration="5s",
+            loop=False
+        )
+
+        # Poll for completion
+        max_attempts = 30  # 2.5 minutes max
+        for attempt in range(max_attempts):
+            status = client.generations.get(generation.id)
+
+            if status.state == "completed":
+                return status.assets.video
+            elif status.state == "failed":
+                failure_reason = getattr(status, 'failure_reason', 'Unknown failure')
+                raise Exception(f"Video generation failed: {failure_reason}")
+            elif status.state in ["dreaming", "in_progress", "pending"]:
+                # Still processing, continue polling
+                pass
+            else:
+                # Unknown state, log and continue
+                print(f"Unknown generation state: {status.state}")
+
+            time.sleep(5)
+
+        raise Exception("Video generation timed out after 2.5 minutes")
+
+    except Exception as e:
+        error_msg = str(e).lower()
+        if "unavailable" in error_msg or "rate limit" in error_msg or "quota" in error_msg:
+            # Service temporarily unavailable - return None instead of failing
+            return None
+        else:
+            raise Exception(f"Video generation failed: {str(e)}")
+
+
 def clean_text(text: str) -> str:
     text = text.encode('utf-8', errors='ignore').decode('utf-8')
     text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', text)

requirements.txt

@@ -8,3 +8,6 @@ google-genai
 pydantic
 python-dotenv
 Pillow
+requests
+docx2txt
+lumaai

test_statement.txt

@@ -0,0 +1,10 @@
+Bank Statement 2024
+
+Income: ₦5,000,000
+Expenses: ₦3,200,000
+Savings: ₦1,800,000
+
+Financial Summary:
+- Good year with increased savings
+- Reduced expenses by 10%
+- Investment portfolio grew by 15%