feat: did optimization and cleanup

app/api/deps.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from fastapi import Depends
+from typing import Any, Dict, Optional, Tuple
+
+from fastapi import Depends, Request
 
 from app.core.security import require_api_key
 from app.services.auth_service import AuthService
@@ -54,3 +56,16 @@ def get_vector_store_service() -> VectorStoreService:
 
 def require_auth(token: str = Depends(require_api_key)) -> str:
     return token
+
+
+def get_redis_scripts(request: Request) -> Tuple[Optional[Any], Dict[str, str]]:
+    """Extract Redis client and Lua scripts from app state.
+
+    Consolidates the repeated pattern:
+        redis = getattr(request.app.state, "redis", None)
+        scripts = getattr(request.app.state, "scripts", None)
+    """
+    return (
+        getattr(request.app.state, "redis", None),
+        getattr(request.app.state, "scripts", {}),
+    )

app/api/v1/chat.py

@@ -8,7 +8,7 @@ from typing import Any, AsyncGenerator, Dict, List, Optional
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
-from app.api.deps import require_auth
+from app.api.deps import get_redis_scripts, require_auth
 from app.services.chat_service import _stream_chat_completion, chat_completion
 from app.utils.schema_utils import validate_response_format
 
@@ -31,8 +31,7 @@ async def _stream_events(
     temperature = body.get("temperature", 0.7)
     top_p = body.get("top_p", 0.9)
     response_format = body.get("response_format")
-    redis = getattr(request.app.state, "redis", None)
-    scripts = getattr(request.app.state, "scripts", None)
+    redis, scripts = get_redis_scripts(request)
 
     try:
         async for chunk in _stream_chat_completion(
@@ -119,8 +118,7 @@ async def create_chat_completion(
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
-    redis = getattr(request.app.state, "redis", None)
-    scripts = getattr(request.app.state, "scripts", None)
+    redis, scripts = get_redis_scripts(request)
 
     if stream:
         return StreamingResponse(

app/api/v1/csv_analysis.py

@@ -6,7 +6,7 @@ from typing import Annotated, Any, Dict, List, Optional
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
 from pydantic import BaseModel, ValidationError
 
-from app.api.deps import require_auth
+from app.api.deps import get_redis_scripts, require_auth
 from app.config import get_settings
 from app.services.chat_service import chat_completion
 from app.services.csv_analysis_service import (
@@ -221,8 +221,7 @@ async def csv_chat(
         {"role": "user", "content": query},
     ]
 
-    redis = getattr(request.app.state, "redis", None)
-    scripts = getattr(request.app.state, "scripts", None)
+    redis, scripts = get_redis_scripts(request)
 
     try:
         ai_response = await chat_completion(

app/api/v1/json_extract.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field

app/api/v1/qr_decoder.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List, Optional
+from typing import List, Optional
 
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
 from pydantic import BaseModel, Field

app/services/chat_service.py

@@ -22,7 +22,8 @@ from app.config import (
     OPENROUTER_MIMIKA_MODEL,
     get_settings,
 )
-from app.utils.json_utils import extract_json_blocks, extract_single_json
+from app.utils.http_utils import create_session_timeout
+from app.utils.json_utils import extract_single_json
 from app.utils.schema_utils import generate_schema_prompt, validate_against_schema
 
 logger = logging.getLogger(__name__)
@@ -51,6 +52,48 @@ def _refresh_keys() -> None:
 _refresh_keys()
 
 
+async def _call_llm_api(
+    base_url: str,
+    path: str,
+    api_key: str,
+    model: str,
+    prepared_messages: List[Dict[str, str]],
+    *,
+    max_tokens: int = DEFAULT_MAX_TOKENS,
+    temperature: float = DEFAULT_TEMPERATURE,
+    top_p: float = DEFAULT_TOP_P,
+    stream: bool = False,
+) -> Optional[Dict[str, Any]]:
+    """Common LLM API call pattern: build payload → session → POST → parse JSON.
+
+    Consolidates the duplicated HTTP call logic across call_openrouter_mimika,
+    call_meganova, call_aion_labs, and their _no_redis/stream variants.
+    """
+    payload: Dict[str, Any] = {
+        "model": model,
+        "messages": prepared_messages,
+        "max_tokens": max_tokens,
+        "temperature": temperature,
+        "top_p": top_p,
+        "stream": stream,
+    }
+
+    timeout = create_session_timeout()
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        async with session.post(
+            f"{base_url}{path}",
+            json=payload,
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+            },
+        ) as resp:
+            if resp.status != 200:
+                logger.warning("LLM API HTTP %d from %s", resp.status, base_url)
+                return None
+            return await resp.json()
+
+
 def build_default_system_prompt(model_name: Optional[str]) -> str:
     name = model_name or "agentdeck-1.0"
     return (
@@ -185,33 +228,22 @@ async def call_openrouter_mimika(
         return None
 
     prepared = prepare_messages(messages, response_format)
-    payload = {
-        "model": OPENROUTER_MIMIKA_MODEL,
-        "messages": prepared,
-        "max_tokens": max_tokens,
-        "temperature": temperature,
-        "top_p": top_p,
-        "stream": False,
-    }
-
     logger.info("Calling OpenRouter Mimika...")
     try:
-        timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.post(
-                f"{OPENROUTER_MIMIKA_BASE_URL}{OPENROUTER_MIMIKA_CHAT_PATH}",
-                json=payload,
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "Content-Type": "application/json",
-                },
-            ) as resp:
-                if resp.status != 200:
-                    logger.warning("OpenRouter Mimika HTTP %d", resp.status)
-                    return None
-                data = await resp.json()
-                attach_json_content(data, response_format)
-                return data
+        data = await _call_llm_api(
+            OPENROUTER_MIMIKA_BASE_URL,
+            OPENROUTER_MIMIKA_CHAT_PATH,
+            api_key,
+            OPENROUTER_MIMIKA_MODEL,
+            prepared,
+            max_tokens=max_tokens,
+            temperature=temperature,
+            top_p=top_p,
+        )
+        if data is None:
+            return None
+        attach_json_content(data, response_format)
+        return data
     except Exception as exc:
         logger.warning("OpenRouter Mimika failed: %s", exc)
         return None
@@ -360,7 +392,7 @@ async def call_meganova(
         }
 
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+            timeout = create_session_timeout()
             async with aiohttp.ClientSession(timeout=timeout) as session:
                 async with session.post(
                     f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
@@ -442,7 +474,7 @@ async def call_aion_labs(
         logger.info("[aion] Attempt %s/%s", attempt + 1, total_tries)
 
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+            timeout = create_session_timeout()
             async with aiohttp.ClientSession(timeout=timeout) as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
@@ -569,33 +601,22 @@ async def call_meganova_no_redis(
         return None
 
     prepared = prepare_messages(messages, response_format)
-    payload = {
-        "model": target_model or MODELS[1],
-        "messages": prepared,
-        "max_tokens": max_tokens,
-        "temperature": temperature,
-        "top_p": top_p,
-        "stream": False,
-    }
-
     logger.info("Calling MegaNova (no Redis)...")
     try:
-        timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.post(
-                f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
-                json=payload,
-                headers={
-                    "Authorization": f"Bearer {api_key}",
-                    "Content-Type": "application/json",
-                },
-            ) as resp:
-                if resp.status != 200:
-                    logger.warning("MegaNova HTTP %d", resp.status)
-                    return None
-                data = await resp.json()
-                attach_json_content(data, response_format)
-                return data
+        data = await _call_llm_api(
+            MEGANOVA_BASE_URL,
+            MEGANOVA_CHAT_PATH,
+            api_key,
+            target_model or MODELS[1],
+            prepared,
+            max_tokens=max_tokens,
+            temperature=temperature,
+            top_p=top_p,
+        )
+        if data is None:
+            return None
+        attach_json_content(data, response_format)
+        return data
     except Exception as exc:
         logger.warning("MegaNova failed: %s", exc)
         return None
@@ -619,32 +640,22 @@ async def call_aion_labs_no_redis(
 
     for attempt, key in enumerate(keys[:3]):
         prepared = prepare_messages(messages, response_format)
-        payload = {
-            "model": model,
-            "messages": prepared,
-            "max_tokens": max_tokens,
-            "temperature": temperature,
-            "top_p": top_p,
-            "stream": False,
-        }
-
         logger.info("[aion] (no redis) Attempt %s/%s", attempt + 1, min(len(keys), 3))
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.post(
-                    f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
-                    json=payload,
-                    headers={
-                        "Authorization": f"Bearer {key}",
-                        "Content-Type": "application/json",
-                    },
-                ) as resp:
-                    if resp.status != 200:
-                        continue
-                    data = await resp.json()
-                    attach_json_content(data, response_format)
-                    return data
+            data = await _call_llm_api(
+                AION_LABS_BASE_URL,
+                AION_LABS_CHAT_PATH,
+                key,
+                model,
+                prepared,
+                max_tokens=max_tokens,
+                temperature=temperature,
+                top_p=top_p,
+            )
+            if data is None:
+                continue
+            attach_json_content(data, response_format)
+            return data
         except Exception:
             continue
 
@@ -717,7 +728,7 @@ async def _stream_meganova(
         }
 
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+            timeout = create_session_timeout()
             async with aiohttp.ClientSession(timeout=timeout) as session:
                 async with session.post(
                     f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
@@ -766,7 +777,7 @@ async def _stream_meganova_no_redis(
     }
 
     try:
-        timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+        timeout = create_session_timeout()
         async with aiohttp.ClientSession(timeout=timeout) as session:
             async with session.post(
                 f"{MEGANOVA_BASE_URL}{MEGANOVA_CHAT_PATH}",
@@ -819,7 +830,7 @@ async def _stream_aion_labs(
         }
 
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+            timeout = create_session_timeout()
             async with aiohttp.ClientSession(timeout=timeout) as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
@@ -865,7 +876,7 @@ async def _stream_aion_labs_no_redis(
         }
 
         try:
-            timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+            timeout = create_session_timeout()
             async with aiohttp.ClientSession(timeout=timeout) as session:
                 async with session.post(
                     f"{AION_LABS_BASE_URL}{AION_LABS_CHAT_PATH}",
@@ -904,7 +915,7 @@ async def _stream_openrouter_mimika(
     }
 
     try:
-        timeout = aiohttp.ClientTimeout(total=_settings.request_timeout_ms / 1000)
+        timeout = create_session_timeout()
         async with aiohttp.ClientSession(timeout=timeout) as session:
             async with session.post(
                 f"{OPENROUTER_MIMIKA_BASE_URL}{OPENROUTER_MIMIKA_CHAT_PATH}",

app/services/code_executor_service.py

@@ -2,17 +2,15 @@ from __future__ import annotations
 
 import asyncio
 import logging
-import os
 import re
 import shutil
-import signal
-import subprocess
 import tempfile
 import time
 from pathlib import Path
 from typing import Optional
 
 from app.config import get_settings
+from app.utils.subprocess_utils import run_subprocess
 
 _logger = logging.getLogger(__name__)
 
@@ -168,33 +166,7 @@ class CodeExecutorService:
         cmd: list[str],
         timeout: float,
     ) -> dict:
-        proc = subprocess.Popen(
-            cmd,
-            stdin=subprocess.DEVNULL,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        try:
-            stdout_bytes, stderr_bytes = proc.communicate(timeout=timeout)
-            timed_out = False
-        except subprocess.TimeoutExpired:
-            try:
-                if os.name == "nt":
-                    proc.kill()
-                else:
-                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-            except Exception:
-                proc.kill()
-            stdout_bytes, stderr_bytes = proc.communicate()
-            timed_out = True
-
-        max_out = self._max_output_bytes
-        return {
-            "stdout": (stdout_bytes.decode("utf-8", errors="replace")[:max_out] if stdout_bytes else ""),
-            "stderr": (stderr_bytes.decode("utf-8", errors="replace")[:max_out] if stderr_bytes else ""),
-            "exit_code": proc.returncode,
-            "timed_out": timed_out,
-        }
+        return run_subprocess(cmd, timeout, self._max_output_bytes)
 
     async def check_runtimes(self) -> dict[str, str]:
         status = {}

app/services/csv_analysis_service.py

@@ -3,10 +3,7 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-import os
 import shutil
-import signal
-import subprocess
 import sys
 import tempfile
 import time
@@ -14,17 +11,16 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
 from urllib.parse import unquote, urlparse
 
-import aiohttp
-
 from app.services.code_executor_service import CodeSanitizer
 from app.services.dataset_metadata_service import extract_metadata
+from app.utils.http_utils import download_url
+from app.utils.subprocess_utils import run_subprocess
 
 logger = logging.getLogger(__name__)
 
 _PYTHON = getattr(sys, "executable", None) or "python3"
 _MAX_OUTPUT_BYTES = 65536
 _MAX_CONCURRENT = 8
-_DOWNLOAD_CHUNK_SIZE = 256 * 1024
 _DOWNLOAD_TIMEOUT = 120
 
 _semaphore = asyncio.Semaphore(_MAX_CONCURRENT)
@@ -35,18 +31,8 @@ class CSVAnalysisError(Exception):
 
 
 async def _download_file(url: str) -> bytes:
-    timeout = aiohttp.ClientTimeout(total=_DOWNLOAD_TIMEOUT)
-    try:
-        async with aiohttp.ClientSession(timeout=timeout) as session:
-            async with session.get(url) as resp:
-                if resp.status != 200:
-                    raise CSVAnalysisError(f"HTTP {resp.status} when fetching {url}")
-                chunks: List[bytes] = []
-                async for chunk in resp.content.iter_chunked(_DOWNLOAD_CHUNK_SIZE):
-                    chunks.append(chunk)
-                return b"".join(chunks)
-    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
-        raise CSVAnalysisError(f"Download failed for {url}: {exc}") from exc
+    data, _ = await download_url(url, timeout_seconds=_DOWNLOAD_TIMEOUT)
+    return data
 
 
 async def _resolve_source(source: Union[str, bytes]) -> Tuple[bytes, Optional[str]]:
@@ -62,32 +48,7 @@ async def _resolve_source(source: Union[str, bytes]) -> Tuple[bytes, Optional[st
 
 
 def _run_subprocess(cmd: List[str], timeout: float, max_output: int) -> Dict[str, Any]:
-    proc = subprocess.Popen(
-        cmd,
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
-    try:
-        stdout_bytes, stderr_bytes = proc.communicate(timeout=timeout)
-        timed_out = False
-    except subprocess.TimeoutExpired:
-        try:
-            if os.name == "nt":
-                proc.kill()
-            else:
-                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-        except Exception:
-            proc.kill()
-        stdout_bytes, stderr_bytes = proc.communicate()
-        timed_out = True
-
-    return {
-        "stdout": (stdout_bytes.decode("utf-8", errors="replace")[:max_output] if stdout_bytes else ""),
-        "stderr": (stderr_bytes.decode("utf-8", errors="replace")[:max_output] if stderr_bytes else ""),
-        "exit_code": proc.returncode,
-        "timed_out": timed_out,
-    }
+    return run_subprocess(cmd, timeout, max_output)
 
 
 _CHAT_SCRIPT = """\

app/utils/http_utils.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+from typing import List, Optional, Tuple
+from urllib.parse import unquote, urlparse
+
+import aiohttp
+
+from app.config import get_settings
+
+_logger = logging.getLogger(__name__)
+_settings = get_settings()
+
+
+def create_session_timeout(total_seconds: Optional[float] = None) -> aiohttp.ClientTimeout:
+    """Create an aiohttp timeout using the configured request timeout."""
+    if total_seconds is None:
+        total_seconds = _settings.request_timeout_ms / 1000
+    return aiohttp.ClientTimeout(total=total_seconds)
+
+
+async def download_url(
+    url: str,
+    *,
+    timeout_seconds: float = 120,
+    max_size_bytes: Optional[int] = None,
+    chunk_size: int = 256 * 1024,
+) -> Tuple[bytes, Optional[str]]:
+    """Download a file from a URL with streaming and size guard.
+
+    Returns (data, filename) where filename may be None.
+    Consolidates _download_file from csv_analysis_service and
+    _download_from_url from dataset_metadata_service.
+    """
+    timeout = aiohttp.ClientTimeout(total=timeout_seconds)
+    try:
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.get(url) as resp:
+                if resp.status != 200:
+                    raise RuntimeError(f"HTTP {resp.status} when fetching {url}")
+
+                if max_size_bytes and resp.content_length and resp.content_length > max_size_bytes:
+                    raise RuntimeError(
+                        f"Remote file advertises {resp.content_length} bytes, "
+                        f"limit is {max_size_bytes}"
+                    )
+
+                chunks: List[bytes] = []
+                total = 0
+                async for chunk in resp.content.iter_chunked(chunk_size):
+                    total += len(chunk)
+                    if max_size_bytes and total > max_size_bytes:
+                        raise RuntimeError(f"Download exceeded {max_size_bytes} bytes")
+                    chunks.append(chunk)
+
+                data = b"".join(chunks)
+    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+        raise RuntimeError(f"Download failed for {url}: {exc}") from exc
+
+    parsed = urlparse(url)
+    filename = unquote(Path(parsed.path).name) if parsed.path else None
+    return data, filename

app/utils/subprocess_utils.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import subprocess
+from typing import Dict, Any, List
+
+_logger = logging.getLogger(__name__)
+
+
+def run_subprocess(
+    cmd: List[str],
+    timeout: float,
+    max_output: int = 65536,
+) -> Dict[str, Any]:
+    """Run a subprocess with timeout handling and output capture.
+
+    Consolidates the duplicate subprocess logic from code_executor_service
+    and csv_analysis_service.
+    """
+    proc = subprocess.Popen(
+        cmd,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    try:
+        stdout_bytes, stderr_bytes = proc.communicate(timeout=timeout)
+        timed_out = False
+    except subprocess.TimeoutExpired:
+        try:
+            if os.name == "nt":
+                proc.kill()
+            else:
+                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+        except Exception:
+            proc.kill()
+        stdout_bytes, stderr_bytes = proc.communicate()
+        timed_out = True
+
+    return {
+        "stdout": (stdout_bytes.decode("utf-8", errors="replace")[:max_output] if stdout_bytes else ""),
+        "stderr": (stderr_bytes.decode("utf-8", errors="replace")[:max_output] if stderr_bytes else ""),
+        "exit_code": proc.returncode,
+        "timed_out": timed_out,
+    }