PYTHON #4

.env

@@ -1,4 +1,4 @@
 MODEL_NAME=adibrino/LAPOR-AI
 ALLOWED_ORIGINS=https://lalim.vercel.app,http://localhost:8000,http://127.0.0.1:8000
-SERVICE_CODES_MAP="{'DPRKPCK': 'Perumahan Rakyat, Kawasan Permukiman dan Cipta Karya', 'DPUBM': 'Pekerjaan Umum Bina Marga', 'DPUSDA': 'Pekerjaan Umum Sumber Daya Air', 'DLH': 'Lingkungan Hidup', 'DINSOS': 'Sosial', 'BPBD': 'Penanggulangan Bencana Daerah', 'DISHUB': 'Perhubungan', 'DINKES': 'Kesehatan', 'SATPOLPP': 'Satuan Polisi Pamong Praja', 'DISKOMINFO': 'Komunikasi dan Informatika', 'DISNAKERTRANS': 'Tenaga Kerja dan Transmigrasi', 'DIPERTAKP': 'Pertanian dan Ketahanan Pangan', 'DISNAK': 'Peternakan', 'DKP': 'Kelautan dan Perikanan', 'DINDIK': 'Pendidikan', 'DISBUDPAR': 'Kebudayaan dan Pariwisata', 'DISPERINDAG': 'Perindustrian dan Perdagangan', 'DPMPTSP': 'Penanaman Modal dan Pelayanan Terpadu Satu Pintu', 'DISKOPUKM': 'Koperasi, Usaha Kecil dan Menengah', 'DISPORA': 'Kepemudaan dan Olahraga', 'DISPERPUSIP': 'Perpustakaan dan Kearsipan', 'BAPPEDA': 'Perencanaan Pembangunan Daerah', 'BAPENDA': 'Pajak dan Pendapatan Daerah', 'DP3AK': 'Pemberdayaan Perempuan, Perlindungan Anak dan Kependudukan'}"
+SERVICE_CODES_MAP={"DPRKPCK": "Perumahan Rakyat, Kawasan Permukiman dan Cipta Karya", "DPUBM": "Pekerjaan Umum Bina Marga", "DPUSDA": "Pekerjaan Umum Sumber Daya Air", "DLH": "Lingkungan Hidup", "DINSOS": "Sosial", "BPBD": "Penanggulangan Bencana Daerah", "DISHUB": "Perhubungan", "DINKES": "Kesehatan", "SATPOLPP": "Satuan Polisi Pamong Praja", "DISKOMINFO": "Komunikasi dan Informatika", "DISNAKERTRANS": "Tenaga Kerja dan Transmigrasi", "DIPERTAKP": "Pertanian dan Ketahanan Pangan", "DISNAK": "Peternakan", "DKP": "Kelautan dan Perikanan", "DINDIK": "Pendidikan", "DISBUDPAR": "Kebudayaan dan Pariwisata", "DISPERINDAG": "Perindustrian dan Perdagangan", "DPMPTSP": "Penanaman Modal dan Pelayanan Terpadu Satu Pintu", "DISKOPUKM": "Koperasi, Usaha Kecil dan Menengah", "DISPORA": "Kepemudaan dan Olahraga", "DISPERPUSIP": "Perpustakaan dan Kearsipan", "BAPPEDA": "Perencanaan Pembangunan Daerah", "BAPENDA": "Pajak dan Pendapatan Daerah", "DP3AK": "Pemberdayaan Perempuan, Perlindungan Anak dan Kependudukan"}
 IS_PRODUCTION=false

app.py

@@ -9,13 +9,13 @@ import shutil
 from typing import List, Any, Dict, Union
 
 from fastapi import FastAPI, UploadFile, File, Form
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, Response
 from fastapi.middleware.cors import CORSMiddleware
 import uvicorn
 from PIL import Image
 from dotenv import load_dotenv
 import ollama
-import spaces
+import spaces # type: ignore
 import gradio as gr
 
 load_dotenv()
@@ -25,15 +25,9 @@ MODEL_NAME: str = os.getenv("MODEL_NAME") or "adibrino/LAPOR-AI"
 IS_PRODUCTION: str = os.getenv("IS_PRODUCTION", "false")
 
 SERVICE_MAP_STR = os.getenv("SERVICE_CODES_MAP", "{}")
-try:
-    SERVICE_MAP = json.loads(SERVICE_MAP_STR)
-except json.JSONDecodeError:
-    SERVICE_MAP = {}
+SERVICE_MAP = json.loads(SERVICE_MAP_STR)
 
-if ALLOWED_ORIGINS_RAW == "*":
-    ALLOWED_ORIGINS = ["*"]
-else:
-    ALLOWED_ORIGINS = [origin.strip() for origin in ALLOWED_ORIGINS_RAW.split(",")]
+ALLOWED_ORIGINS = ["*"] if ALLOWED_ORIGINS_RAW == "*" else [origin.strip() for origin in ALLOWED_ORIGINS_RAW.split(",")]
 
 print(f"ALLOWED_ORIGINS: {ALLOWED_ORIGINS}")
 print(f"MODEL_NAME: {MODEL_NAME}")
@@ -85,7 +79,7 @@ def process_image_to_base64(image_bytes: bytes) -> Union[str, None]:
         return None
 
 @spaces.GPU(duration=60) 
-def run_inference(text_laporan: str, base64_images: List[str]) -> Dict[str, Any]:
+def run_inference(report_text: str, base64_images: List[str]) -> Dict[str, Any]:
     print("Starting GPU Inference...")
     
     try:
@@ -94,11 +88,11 @@ def run_inference(text_laporan: str, base64_images: List[str]) -> Dict[str, Any]
         print("Model not found in GPU context, pulling again...")
         subprocess.run(["ollama", "pull", MODEL_NAME], check=True)
 
-    response: Any = ollama.chat(
+    response: Any = ollama.chat( # type: ignore
         model=MODEL_NAME,
         messages=[{
             'role': 'user',
-            'content': text_laporan,
+            'content': report_text,
             'images': base64_images if base64_images else None # type: ignore
         }],
         format='json',
@@ -106,105 +100,111 @@ def run_inference(text_laporan: str, base64_images: List[str]) -> Dict[str, Any]
     )
     
     if isinstance(response, dict):
-        return response
+        return response # type: ignore
     return dict(response)
 
 @app.get("/")
 def health_check():
-    return {"status": "Python Backend with ZeroGPU is running."}
+    return Response("Python Backend is running.")
 
 @app.post("/api/analyze")
-async def analyze(
-    laporan: str = Form(...),
+async def analyze( # type: ignore
+    report: str = Form(...),
     images: List[UploadFile] = File(...)
 ):
-    if not laporan or len(laporan) < 10:
-        return JSONResponse(
-            status_code=400, 
-            content={"status": "error", "message": "Deskripsi laporan wajib diisi minimal 10 karakter."}
-        )
+    try:
+        if not report or len(report) < 10:
+            return JSONResponse(
+                status_code=400, 
+                content={"status": "error", "message": "Deskripsi laporan wajib diisi minimal 10 karakter."}
+            )
+
+        if not images:
+            return JSONResponse(
+                status_code=400,
+                content={"status": "error", "message": "Wajib melampirkan minimal 1 foto bukti."}
+            )
+
+        base64_images: List[str] = []
+        for img_file in images:
+            content = await img_file.read()
+            if len(content) > 0:
+                b64 = process_image_to_base64(content)
+                if b64:
+                    base64_images.append(b64)
+        
+        if not base64_images:
+            return JSONResponse(
+                status_code=400,
+                content={"status": "error", "message": "File gambar tidak valid/corrupt."}
+            )
+
+        max_retries = 3
+        last_exception = None
+            
+        print("Report Text:", report)
+
+        for attempt in range(max_retries):
+            try:
+                print(f"Attempting AI Analysis... ({attempt + 1}/{max_retries})")
+                
+                response_raw = run_inference(report, base64_images)
+
+                if 'message' not in response_raw or 'content' not in response_raw['message']:
+                    raise ValueError("Empty response structure from AI")
+
+                content_str = response_raw['message']['content']
+                
+                ai_content = json.loads(content_str)
+
+                required_keys = ["title", "category", "priority", "service_code"]
+                missing_keys = [key for key in required_keys if key not in ai_content]
+                if missing_keys:
+                    raise ValueError(f"Missing keys in JSON: {missing_keys}")
+
+                service_code = ai_content["service_code"]
+                if service_code not in SERVICE_MAP:
+                    print(f"Warning: Service code {service_code} unknown.")
+                
+                priority = str(ai_content["priority"]).lower()
+                if priority not in ['high', 'medium', 'low']:
+                    priority = 'medium'
+                ai_content["priority"] = priority
+
+                data = { # type: ignore
+                    "status": "success",
+                    "data": ai_content,
+                    "meta": {
+                        "model": MODEL_NAME,
+                        'processing_time_sec': (response_raw.get("total_duration", 0)) / 1e9,
+                        "images_analyzed": len(base64_images),
+                        "attempts": attempt + 1
+                    }
+                }
+                
+                print("AI Success")
+                print(json.dumps(data, indent=2, ensure_ascii=True))
+                
+                return data # type: ignore
+            except Exception as e:
+                print(f"Attempt {attempt + 1} failed: {str(e)}")
+                last_exception = e
+                time.sleep(1)
+                continue
 
-    if not images:
         return JSONResponse(
-            status_code=400,
-            content={"status": "error", "message": "Wajib melampirkan minimal 1 foto bukti."}
+            status_code=500,
+            content={"status": "error", "message": f"AI Failed: {str(last_exception)}"}
         )
-
-    base64_images: List[str] = []
-    for img_file in images:
-        content = await img_file.read()
-        if len(content) > 0:
-            b64 = process_image_to_base64(content)
-            if b64:
-                base64_images.append(b64)
+    except Exception as e:
+        raise e
     
-    if not base64_images:
-         return JSONResponse(
-            status_code=400,
-            content={"status": "error", "message": "File gambar tidak valid/corrupt."}
-        )
-
-    max_retries = 3
-    last_exception = None
-
-    for attempt in range(max_retries):
-        try:
-            print(f"Attempting AI Analysis... ({attempt + 1}/{max_retries})")
-            
-            response_raw = run_inference(laporan, base64_images)
-
-            if 'message' not in response_raw or 'content' not in response_raw['message']:
-                raise ValueError("Empty response structure from AI")
-
-            content_str = response_raw['message']['content']
-            
-            ai_content = json.loads(content_str)
-
-            required_keys = ["title", "category", "priority", "service_code"]
-            missing_keys = [key for key in required_keys if key not in ai_content]
-            if missing_keys:
-                raise ValueError(f"Missing keys in JSON: {missing_keys}")
-
-            service_code = ai_content["service_code"]
-            if service_code not in SERVICE_MAP:
-                print(f"Warning: Service code {service_code} unknown.")
-            
-            priority = str(ai_content["priority"]).lower()
-            if priority not in ['high', 'medium', 'low']:
-                priority = 'medium'
-            ai_content["priority"] = priority
-
-            data = {
-                "status": "success",
-                "data": ai_content,
-                "meta": {
-                    "model": MODEL_NAME,
-                    'processing_time_sec': (response_raw.get("total_duration", 0)) / 1e9,
-                    "images_analyzed": len(base64_images),
-                    "attempts": attempt + 1
-                }
-            }
-            
-            print("AI Success")
-            return data
-
-        except Exception as e:
-            print(f"Attempt {attempt + 1} failed: {str(e)}")
-            last_exception = e
-            time.sleep(1)
-            continue
-
-    return JSONResponse(
-        status_code=500,
-        content={"status": "error", "message": f"AI Failed: {str(last_exception)}"}
-    )
-
 if __name__ == "__main__":
     with gr.Blocks() as demo:
         gr.Markdown("# LAPOR AI API Backend")
         gr.Markdown("This space hosts the API at `/api/analyze`.")
         gr.Markdown(f"**Model:** {MODEL_NAME}")
     
-    app = gr.mount_gradio_app(app, demo, path="/")
+    app = gr.mount_gradio_app(app, demo, path="/") # type: ignore
     
     uvicorn.run(app, host="0.0.0.0", port=7860)