from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, EmailStr, Field from typing import List, Dict, Any from fastapi.responses import FileResponse import httpx import os import json from dotenv import load_dotenv import logging from datetime import datetime, timezone # Setup logging with timestamp logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Load environment variables load_dotenv() # System information CURRENT_UTC = "2025-05-27 17:27:01" CURRENT_USER = "farhanwew" # Get API key from environment OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY") if not OPENROUTER_API_KEY: raise ValueError("OPENROUTER_API_KEY not found in environment variables") app = FastAPI( title="Cuan.in API", description="API untuk rekomendasi bisnis UMKM yang personalized", version="1.0.0", ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], # Adjust in production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Models class BusinessProfile(BaseModel): business_interest: str = Field(..., description="Minat bisnis pengguna") target_market: str = Field(..., description="Target pasar yang diinginkan") description: str = Field(..., description="Deskripsi usaha yang diinginkan") email: EmailStr = Field(..., description="Email pengguna") class RecommendationResponse(BaseModel): primary_recommendation: Dict[str, Any] alternative_recommendations: List[Dict[str, Any]] success_factors: List[str] challenges: List[str] next_steps: List[str] def extract_json(text: str) -> str: """ Try to extract JSON substring from text, if there's text before/after JSON. """ try: # Find from first curly brace to last curly brace start = text.index("{") end = text.rindex("}") + 1 return text[start:end] except ValueError: # If not found, return original text return text async def call_llm(prompt: str) -> dict: for attempt in range(2): # Try maximum 2 times async with httpx.AsyncClient(timeout=60.0) as client: try: response = await client.post( "https://openrouter.ai/api/v1/chat/completions", headers={ "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "https://cuan.in", # Replace with your actual domain "X-Title": "Cuan.in - UMKM Business Recommendations" }, json={ "model": "google/gemini-2.0-flash-001", "messages": [{"role": "user", "content": prompt}], "response_format": {"type": "json_object"}, "max_tokens": 8048, }, ) if response.status_code != 200: logging.error(f"Error from LLM API: {response.status_code} - {response.text}") raise HTTPException(status_code=response.status_code, detail=f"Error from LLM API: {response.text}") result = response.json() llm_response_raw = result["choices"][0]["message"]["content"] # Log the raw response for debugging logging.info(f"Raw LLM response text: {llm_response_raw}") # Clean and validate the response cleaned_response = extract_json(llm_response_raw) # Add additional validation before parsing if not cleaned_response.strip().startswith("{"): logging.error(f"Invalid JSON structure: {cleaned_response}") raise json.JSONDecodeError("Invalid JSON structure", cleaned_response, 0) parsed_response = json.loads(cleaned_response) # Validate the response structure required_keys = { "primary_recommendation", "alternative_recommendations", "success_factors", "challenges", "next_steps" } if not all(key in parsed_response for key in required_keys): logging.error(f"Missing required keys in response: {parsed_response}") raise ValueError("Response missing required keys") return parsed_response except json.JSONDecodeError as e: logging.error(f"JSON decoding error on attempt {attempt+1}: {e}") logging.error(f"Problematic JSON content: {cleaned_response}") if attempt == 1: # Last attempt raise HTTPException( status_code=502, detail=f"Invalid JSON from LLM. Original error: {str(e)}" ) logging.info("Retrying due to JSON decode error...") except Exception as e: logging.error(f"Unexpected error on attempt {attempt+1}: {str(e)}") if attempt == 1: # Last attempt raise HTTPException( status_code=500, detail=f"Error processing LLM response: {str(e)}" ) logging.info("Retrying due to unexpected error...") # API endpoints @app.get("/") async def root(): return { "message": "Welcome to Cuan.in API - Rekomendasi Bisnis UMKM", "version": "1.0.0", "current_utc": CURRENT_UTC, "status": "operational" } @app.get("/favicon.ico") async def favicon(): return FileResponse("static/favicon.ico") @app.get("/status") async def status(): return { "status": "operational", "timestamp": CURRENT_UTC, "user": CURRENT_USER, "api_version": "1.0.0" } @app.post("/recommendations", response_model=RecommendationResponse) async def get_recommendations(profile: BusinessProfile): prompt = f""" Berikan respons dalam format JSON yang valid dengan struktur berikut tanpa teks tambahan apapun: {{ "primary_recommendation": {{ "nama_usaha": "string", "deskripsi": "string", "modal_yang_dibutuhkan": "string", "perkiraan_keuntungan": "string", "skala_usaha": "string", "target_pasar": "string" }}, "alternative_recommendations": [ {{ "nama_usaha": "string", "deskripsi_singkat": "string", "modal_yang_dibutuhkan": "string" }} ], "success_factors": ["string"], "challenges": ["string"], "next_steps": ["string"] }} Berikan rekomendasi bisnis UMKM berdasarkan profil berikut: - Minat Bisnis: {profile.business_interest} - Target Pasar: {profile.target_market} - Deskripsi: {profile.description} PENTING: 1. Pastikan output adalah JSON yang valid dengan tanda kutip ganda (\") dan tanpa teks tambahan di luar struktur JSON. 2. Berikan minimal 2 alternative_recommendations. 3. Berikan minimal 3 item untuk success_factors, challenges, dan next_steps. 4. Semua nilai harus dalam Bahasa Indonesia. 5. Modal dan keuntungan harus dalam format angka dengan mata uang Rupiah (contoh: "Rp 5.000.000"). """ # Log the request logging.info(f"Received recommendation request for email: {profile.email}") try: recommendations = await call_llm(prompt) # Log successful response logging.info(f"Successfully generated recommendations for email: {profile.email}") return recommendations except Exception as e: # Log the error logging.error(f"Error generating recommendations for email {profile.email}: {str(e)}") raise # Startup and shutdown events @app.on_event("startup") async def startup_event(): logging.info(f"API Started - Current UTC: {CURRENT_UTC}") logging.info(f"Current User: {CURRENT_USER}") @app.on_event("shutdown") async def shutdown_event(): logging.info("API Shutting down") # Run with: uvicorn main:app --reload if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)