Add backend API module for FastAPI server

- backend/__init__.py
- backend/api.py - FastAPI routes for search, feedback, translation
- backend/services.py - Service layer for IR operations

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

backend/__init__.py

@@ -0,0 +1,6 @@
+"""
+UAE Knowledge System - Backend Package
+"""
+from .api import app
+
+__all__ = ["app"]

backend/api.py

@@ -0,0 +1,305 @@
+"""
+UAE Knowledge System - FastAPI Backend
+Serves the HTML frontend and provides search API
+"""
+import json
+import os
+import httpx
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from fastapi import FastAPI, Request
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import HTMLResponse, FileResponse
+from pydantic import BaseModel
+
+from .services import get_knowledge_base, get_retriever, search_knowledge_base, get_stats
+
+# DeepL API configuration
+DEEPL_API_KEY = os.environ.get("DEEPL_API_KEY", "")
+DEEPL_API_URL = "https://api-free.deepl.com/v2/translate"  # Use api.deepl.com for paid plans
+
+# ============================================================
+# Path Configuration
+# ============================================================
+PROJECT_ROOT = Path(__file__).parent.parent
+FRONTEND_DIR = PROJECT_ROOT / "frontend"
+DATA_DIR = PROJECT_ROOT / "data"
+
+# Feedback file location - use /data for HF Spaces persistence
+FEEDBACK_FILE = DATA_DIR / "feedback.json"
+
+# Translation cache file - persistent across restarts
+TRANSLATION_CACHE_FILE = DATA_DIR / "translations_cache.json"
+
+# ============================================================
+# Initialize FastAPI
+# ============================================================
+app = FastAPI(title="UAE Knowledge System", version="2.3.0")
+
+
+# ============================================================
+# Request/Response Models
+# ============================================================
+class SearchRequest(BaseModel):
+    query: str
+    category: str
+
+
+class FeedbackRequest(BaseModel):
+    query: str
+    category: str
+    entity_ratings: Dict[str, Dict[str, int]]
+    notes: str
+    results: List[str]
+
+
+class TranslateRequest(BaseModel):
+    texts: List[str]  # List of texts to translate
+    target_lang: str  # AR or ZH (DeepL uses ZH for Chinese)
+
+
+# ============================================================
+# Translation Cache (file-based, persistent across restarts)
+# ============================================================
+_translation_cache: Dict[str, str] = {}  # {text:lang: translated}
+
+
+def load_translation_cache() -> None:
+    """Load translation cache from file"""
+    global _translation_cache
+    if TRANSLATION_CACHE_FILE.exists():
+        try:
+            with open(TRANSLATION_CACHE_FILE, "r", encoding="utf-8") as f:
+                _translation_cache = json.load(f)
+            print(f"Loaded {len(_translation_cache)} cached translations")
+        except Exception as e:
+            print(f"Error loading translation cache: {e}")
+            _translation_cache = {}
+    else:
+        _translation_cache = {}
+
+
+def save_translation_cache() -> None:
+    """Save translation cache to file"""
+    try:
+        DATA_DIR.mkdir(parents=True, exist_ok=True)
+        with open(TRANSLATION_CACHE_FILE, "w", encoding="utf-8") as f:
+            json.dump(_translation_cache, f, ensure_ascii=False, indent=2)
+    except Exception as e:
+        print(f"Error saving translation cache: {e}")
+
+
+async def translate_with_deepl(texts: List[str], target_lang: str) -> List[str]:
+    """Translate texts using DeepL API"""
+    if not DEEPL_API_KEY:
+        return texts  # Return original if no API key
+
+    # Map our language codes to DeepL codes
+    lang_map = {"ar": "AR", "cn": "ZH"}
+    deepl_lang = lang_map.get(target_lang.lower(), target_lang.upper())
+
+    # Check cache first
+    results = []
+    texts_to_translate = []
+    text_indices = []
+
+    for i, text in enumerate(texts):
+        cache_key = f"{text}:{deepl_lang}"
+        if cache_key in _translation_cache:
+            results.append(_translation_cache[cache_key])
+        else:
+            results.append(None)  # Placeholder
+            texts_to_translate.append(text)
+            text_indices.append(i)
+
+    # Translate uncached texts
+    if texts_to_translate:
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    DEEPL_API_URL,
+                    data={
+                        "auth_key": DEEPL_API_KEY,
+                        "text": texts_to_translate,
+                        "target_lang": deepl_lang,
+                        "source_lang": "EN"
+                    },
+                    timeout=30.0
+                )
+
+                if response.status_code == 200:
+                    data = response.json()
+                    translations = data.get("translations", [])
+
+                    for j, trans in enumerate(translations):
+                        translated_text = trans.get("text", texts_to_translate[j])
+                        original_idx = text_indices[j]
+                        results[original_idx] = translated_text
+
+                        # Cache the translation
+                        cache_key = f"{texts_to_translate[j]}:{deepl_lang}"
+                        _translation_cache[cache_key] = translated_text
+
+                    # Save cache to file after new translations
+                    save_translation_cache()
+                else:
+                    # On error, use original texts
+                    for j, idx in enumerate(text_indices):
+                        results[idx] = texts_to_translate[j]
+
+        except Exception as e:
+            print(f"Translation error: {e}")
+            # On error, use original texts
+            for j, idx in enumerate(text_indices):
+                results[idx] = texts_to_translate[j]
+
+    return results
+
+
+# ============================================================
+# API Endpoints
+# ============================================================
+@app.get("/", response_class=HTMLResponse)
+async def root():
+    """Serve the main HTML page"""
+    html_path = FRONTEND_DIR / "index.html"
+    if html_path.exists():
+        return FileResponse(html_path)
+    return HTMLResponse("<h1>UAE Knowledge System</h1><p>index.html not found</p>")
+
+
+@app.get("/api/stats")
+async def api_stats():
+    """Get knowledge base statistics"""
+    return get_stats()
+
+
+@app.post("/api/search")
+async def api_search(request: SearchRequest):
+    """Search the knowledge base"""
+    try:
+        results = search_knowledge_base(request.query, top_k=5)
+        return {
+            "results": results,
+            "query": request.query,
+            "category": request.category,
+            "is_sensitive": False,
+            "sensitive_topic": None,
+            "sensitive_guidance": None
+        }
+    except Exception as e:
+        import traceback
+        return {"error": str(e), "traceback": traceback.format_exc()[:500]}
+
+
+@app.post("/api/feedback")
+async def api_feedback(request: FeedbackRequest, req: Request):
+    """Save user feedback"""
+    try:
+        # Ensure data directory exists
+        DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+        # Get client IP
+        client_ip = req.headers.get("x-forwarded-for", "").split(",")[0].strip()
+        if not client_ip:
+            client_ip = req.client.host if req.client else "unknown"
+
+        feedback = {
+            "timestamp": datetime.now().isoformat(),
+            "client_ip": client_ip,
+            "query": request.query,
+            "category": request.category,
+            "entity_ratings": request.entity_ratings,
+            "notes": request.notes,
+            "results": request.results
+        }
+
+        # Load existing feedback
+        if FEEDBACK_FILE.exists():
+            with open(FEEDBACK_FILE, "r", encoding="utf-8") as f:
+                all_feedback = json.load(f)
+        else:
+            all_feedback = []
+
+        all_feedback.append(feedback)
+
+        # Save feedback
+        with open(FEEDBACK_FILE, "w", encoding="utf-8") as f:
+            json.dump(all_feedback, f, ensure_ascii=False, indent=2)
+
+        return {"success": True, "total": len(all_feedback)}
+
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@app.post("/api/translate")
+async def api_translate(request: TranslateRequest):
+    """Translate texts using DeepL API"""
+    try:
+        if not DEEPL_API_KEY:
+            return {
+                "success": False,
+                "error": "Translation not configured (DEEPL_API_KEY not set)",
+                "translations": request.texts  # Return original texts
+            }
+
+        translations = await translate_with_deepl(request.texts, request.target_lang)
+        return {
+            "success": True,
+            "translations": translations,
+            "target_lang": request.target_lang
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "error": str(e),
+            "translations": request.texts  # Return original on error
+        }
+
+
+@app.get("/api/translate/status")
+async def api_translate_status():
+    """Check if translation is available"""
+    return {
+        "available": bool(DEEPL_API_KEY),
+        "provider": "DeepL" if DEEPL_API_KEY else None
+    }
+
+
+# ============================================================
+# Static Files - Serve frontend assets
+# ============================================================
+# Mount CSS
+app.mount("/css", StaticFiles(directory=str(FRONTEND_DIR / "css")), name="css")
+
+# Mount JavaScript
+app.mount("/js", StaticFiles(directory=str(FRONTEND_DIR / "js")), name="js")
+
+# Mount assets (images)
+app.mount("/assets", StaticFiles(directory=str(FRONTEND_DIR / "assets")), name="assets")
+
+
+# ============================================================
+# Startup Event
+# ============================================================
+@app.on_event("startup")
+async def startup_event():
+    """Pre-load retriever and cache on startup"""
+    print("Starting UAE Knowledge System API...")
+    # Load translation cache from file
+    load_translation_cache()
+    # Pre-load in background to speed up first request
+    get_knowledge_base()
+    get_retriever()
+    print("System ready!")
+
+
+# ============================================================
+# Run with Uvicorn (for direct execution)
+# ============================================================
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

backend/services.py

@@ -0,0 +1,133 @@
+"""
+UAE Knowledge System - Backend Services
+Handles knowledge base and retriever initialization
+"""
+import sys
+from pathlib import Path
+
+# Add parent directory to path for imports
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from ir.retriever import EntityRetriever, RetrievalOutput
+from ir.knowledge_base import KnowledgeBase
+
+# ============================================================
+# Global State
+# ============================================================
+_retriever = None
+_knowledge_base = None
+
+# Paths relative to project root
+PROJECT_ROOT = Path(__file__).parent.parent
+INDEX_CACHE_PATH = PROJECT_ROOT / "ir" / "cache" / "dense_index"
+
+
+def get_knowledge_base() -> KnowledgeBase:
+    """Lazy load knowledge base"""
+    global _knowledge_base
+    if _knowledge_base is None:
+        print("Loading knowledge base...")
+        _knowledge_base = KnowledgeBase(debug=False)
+    return _knowledge_base
+
+
+def get_retriever():
+    """Get the dense retriever (cached)"""
+    global _retriever
+    if _retriever is not None:
+        return _retriever
+
+    from ir.retrievers.dense import DenseRetriever
+
+    print("Loading dense retriever...")
+    retriever = DenseRetriever(model_name="bge-m3", debug=False)
+    kb = get_knowledge_base()
+
+    # Try to load cached index
+    if INDEX_CACHE_PATH.exists():
+        print(f"Loading cached index from {INDEX_CACHE_PATH}...")
+        if retriever.load_index(str(INDEX_CACHE_PATH)):
+            print("Cached index loaded!")
+        else:
+            print("Cache load failed, building index...")
+            retriever.build_index_from_knowledge_base(kb)
+            retriever.save_index(str(INDEX_CACHE_PATH))
+    else:
+        print("Building dense index (this may take a while)...")
+        retriever.build_index_from_knowledge_base(kb)
+        INDEX_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
+        retriever.save_index(str(INDEX_CACHE_PATH))
+        print("Index built and cached!")
+
+    _retriever = retriever
+    return retriever
+
+
+def search_knowledge_base(query: str, top_k: int = 5):
+    """
+    Search the knowledge base and return formatted results
+    """
+    retriever = get_retriever()
+    kb = get_knowledge_base()
+
+    # Perform search
+    results = retriever.search(query, top_k=top_k)
+
+    # Format results
+    formatted_results = []
+    for metadata, score in results:
+        entity_id = metadata.get("entity_id", "")
+        entity_name = metadata.get("entity_name", "Unknown")
+
+        # Get full entity data from KB
+        raw_data = kb.get_raw_entity(entity_id) if entity_id else None
+
+        result = {
+            "entity_id": entity_id,
+            "entity_name": entity_name,
+            "score": score,
+            "chunk_type": metadata.get("chunk_type", ""),
+            "subcategory": "",
+            "emirate": "",
+            "is_royal": False,
+            "summary": "",
+            "must_answer": []
+        }
+
+        if raw_data:
+            facts_data = raw_data.get('facts', {})
+            metadata_kb = raw_data.get('metadata', {})
+
+            result["subcategory"] = raw_data.get('subcategory', '')
+            result["emirate"] = metadata_kb.get('emirate', '')
+            result["is_royal"] = metadata_kb.get('is_royal', False)
+            result["summary"] = facts_data.get('summary_paragraph', '')
+
+            # Extract must-answer facts
+            must_answer = facts_data.get('must_answer', [])
+            result["must_answer"] = [
+                fact.get('fact', fact) if isinstance(fact, dict) else str(fact)
+                for fact in must_answer[:5]
+            ]
+
+            # Include full entity data for detailed view
+            result["full_entity"] = raw_data
+
+        formatted_results.append(result)
+
+    return formatted_results
+
+
+def get_stats():
+    """Get knowledge base statistics"""
+    try:
+        kb = get_knowledge_base()
+        entities = len(kb.entities)
+        # Fixed: 8 knowledge categories as defined in the system
+        return {
+            "entities": entities,
+            "categories": 8,
+            "version": "2.3.0"
+        }
+    except Exception as e:
+        return {"entities": 0, "categories": 8, "error": str(e)}