tested ready for deployment

.gitignore

@@ -0,0 +1,62 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Virtual Environments
+venv/
+env/
+.venv/
+.env
+
+# Distribution / Packaging
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IDEs
+.vscode/
+.idea/
+
+# OS specific
+.DS_Store
+Thumbs.db

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PORT=7860
+
+WORKDIR /app
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

app/api/routes.py

@@ -0,0 +1,15 @@
+from fastapi import APIRouter, Depends
+
+from app.dependencies import get_category_service
+from app.schemas.categories import CategorizeRequest, CategoryPrediction
+from app.services.autocategorizer import AutoCategoryService
+
+router = APIRouter()
+
+
+@router.post("/categorize", response_model=CategoryPrediction, summary="Categorize a transaction note")
+async def categorize_transaction(
+    payload: CategorizeRequest,
+    service: AutoCategoryService = Depends(get_category_service),
+) -> CategoryPrediction:
+    return await service.categorize(payload.notes)

app/core/config.py

@@ -0,0 +1,25 @@
+from functools import lru_cache
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    mongo_uri: str = Field(..., alias="MONGO_URI")
+    mongo_db: str = Field("wallet_sync", alias="MONGO_DB")
+    mongo_collection: str = Field("autocategory", alias="MONGO_COLLECTION")
+
+    openai_api_key: str = Field(..., alias="OPENAI_API_KEY")
+    openai_model: str = Field("gpt-4o-mini", alias="OPENAI_MODEL")
+
+    category_cache_ttl_seconds: int = Field(300, alias="CATEGORY_CACHE_TTL")
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+settings = get_settings()

app/core/openai_client.py

@@ -0,0 +1,5 @@
+from openai import AsyncOpenAI
+
+from app.core.config import settings
+
+openai_client = AsyncOpenAI(api_key=settings.openai_api_key)

app/db/mongo.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection, AsyncIOMotorDatabase
+
+from app.core.config import settings
+
+_mongo_client: Optional[AsyncIOMotorClient] = None
+
+
+async def connect_to_mongo() -> None:
+    """Initialize the MongoDB client once during application startup."""
+    global _mongo_client
+    if _mongo_client is not None:
+        return
+
+    client = AsyncIOMotorClient(settings.mongo_uri, serverSelectionTimeoutMS=5000)
+    # Trigger a server selection to fail fast if the URI/config is invalid.
+    await client.server_info()
+    _mongo_client = client
+
+
+async def close_mongo_connection() -> None:
+    global _mongo_client
+    if _mongo_client is None:
+        return
+    _mongo_client.close()
+    _mongo_client = None
+
+
+def get_client() -> AsyncIOMotorClient:
+    if _mongo_client is None:
+        raise RuntimeError("MongoDB client is not initialized. Wait for startup to finish.")
+    return _mongo_client
+
+
+def get_database() -> AsyncIOMotorDatabase:
+    return get_client()[settings.mongo_db]
+
+
+def get_autocategory_collection() -> AsyncIOMotorCollection:
+    return get_database()[settings.mongo_collection]

app/dependencies.py

@@ -0,0 +1,20 @@
+from functools import lru_cache
+
+from app.core.config import settings
+from app.core.openai_client import openai_client
+from app.db.mongo import get_autocategory_collection
+from app.services.autocategorizer import AutoCategoryService
+
+
+@lru_cache
+def _get_service() -> AutoCategoryService:
+    return AutoCategoryService(
+        collection_getter=get_autocategory_collection,
+        openai_client=openai_client,
+        model=settings.openai_model,
+        cache_ttl_seconds=settings.category_cache_ttl_seconds,
+    )
+
+
+def get_category_service() -> AutoCategoryService:
+    return _get_service()

app/main.py

@@ -0,0 +1,29 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from app.api.routes import router as api_router
+from app.db.mongo import close_mongo_connection, connect_to_mongo
+
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    await connect_to_mongo()
+    try:
+        yield
+    finally:
+        await close_mongo_connection()
+
+
+app = FastAPI(
+    title="Auto Categorizer API",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+app.include_router(api_router, prefix="/api/v1")
+
+
+@app.get("/health", tags=["Health"])
+async def health_check() -> dict[str, str]:
+    return {"status": "ok"}

app/schemas/categories.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel, Field
+
+
+class CategorizeRequest(BaseModel):
+    notes: str = Field(..., min_length=1, description="Full transaction note.")
+
+
+class CategoryPrediction(BaseModel):
+    title: str = Field(..., description="High-level category title.")
+    subcategory: str = Field(..., description="Specific subcategory chosen by the model.")

app/services/autocategorizer.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import ast
+import asyncio
+import json
+import re
+import time
+from typing import Callable, Dict, List, Optional
+
+from fastapi import HTTPException
+from motor.motor_asyncio import AsyncIOMotorCollection
+from openai import AsyncOpenAI
+
+from app.core.config import settings
+from app.schemas.categories import CategoryPrediction
+
+
+class AutoCategoryService:
+    """Classifies transaction notes into the closest Mongo-backed category."""
+
+    def __init__(
+        self,
+        collection_getter: Callable[[], AsyncIOMotorCollection],
+        openai_client: AsyncOpenAI,
+        model: str,
+        cache_ttl_seconds: int,
+    ) -> None:
+        self._collection_getter = collection_getter
+        self._openai_client = openai_client
+        self._model = model
+        self._cache_ttl_seconds = cache_ttl_seconds
+        self._cached_categories: List[Dict[str, object]] | None = None
+        self._last_loaded: float = 0.0
+        self._lock = asyncio.Lock()
+
+    def _collection(self) -> AsyncIOMotorCollection:
+        return self._collection_getter()
+
+    async def categorize(self, notes: str) -> CategoryPrediction:
+        categories = await self._get_categories()
+        if not categories:
+            raise HTTPException(status_code=500, detail="No categories configured.")
+
+        formatted_categories = self._format_categories(categories)
+        user_prompt = (
+            "Transaction note:\n"
+            f"{notes}\n\n"
+            "Available categories and subcategories:\n"
+            f"{formatted_categories}\n\n"
+            "Respond with the exact title and subcategory from the list above."
+        )
+
+        request_payload = dict(
+            model=self._model,
+            input=[
+                {
+                    "role": "system",
+                    "content": (
+                        "You classify financial transactions into the closest category. "
+                        "Only use the provided title and subcategory options. "
+                        "Output valid JSON with keys 'title' and 'subcategory'."
+                    ),
+                },
+                {"role": "user", "content": [{"type": "input_text", "text": user_prompt}]},
+            ],
+        )
+
+        try:
+            response = await self._openai_client.responses.create(
+                response_format={"type": "json_object"},
+                **request_payload,
+            )
+        except TypeError as exc:
+            # Older openai-python clients (pre 1.3x) do not yet support response_format.
+            if "response_format" not in str(exc):
+                raise
+            response = await self._openai_client.responses.create(**request_payload)
+
+        try:
+            payload = self._parse_response_payload(response)
+        except ValueError as exc:
+            raise HTTPException(status_code=502, detail="Failed to parse model output.") from exc
+
+        title = payload.get("title")
+        subcategory = payload.get("subcategory")
+        if not isinstance(title, str) or not isinstance(subcategory, str):
+            raise HTTPException(status_code=502, detail="Model response missing category fields.")
+
+        return CategoryPrediction(title=title.strip(), subcategory=subcategory.strip())
+
+    def _parse_response_payload(self, response) -> Dict[str, object]:
+        raw_text = self._extract_response_text(response)
+        if not raw_text:
+            raise ValueError("Model response did not contain text content.")
+
+        cleaned = self._strip_code_fence(raw_text)
+        candidates = [cleaned]
+
+        json_snippet = self._extract_json_snippet(cleaned)
+        if json_snippet and json_snippet not in candidates:
+            candidates.append(json_snippet)
+
+        for candidate in candidates:
+            for parser in (self._try_parse_json, self._try_parse_literal_dict, self._try_parse_key_values):
+                payload = parser(candidate)
+                if payload:
+                    return payload
+
+        raise ValueError("Unable to coerce model response into a payload.")
+
+    @staticmethod
+    def _extract_response_text(response) -> str:
+        text = getattr(response, "output_text", "") or ""
+        if isinstance(text, str) and text.strip():
+            return text.strip()
+
+        outputs = getattr(response, "output", []) or []
+        for output in outputs:
+            contents = getattr(output, "content", []) or []
+            for content in contents:
+                value = getattr(content, "text", None)
+                if isinstance(value, str) and value.strip():
+                    return value.strip()
+
+        return ""
+
+    @staticmethod
+    def _strip_code_fence(raw_text: str) -> str:
+        text = raw_text.strip()
+        if text.startswith("```") and text.endswith("```"):
+            lines = text.split("\n")
+            # Drop first and last fence line
+            if len(lines) >= 2:
+                text = "\n".join(lines[1:-1]).strip()
+        return text
+
+    @staticmethod
+    def _extract_json_snippet(raw_text: str) -> Optional[str]:
+        start = raw_text.find("{")
+        end = raw_text.rfind("}")
+        if start == -1 or end == -1 or end <= start:
+            return None
+        return raw_text[start : end + 1]
+
+    @staticmethod
+    def _try_parse_json(raw_text: str) -> Optional[Dict[str, object]]:
+        text = raw_text.strip()
+        if not text:
+            return None
+        try:
+            payload = json.loads(text)
+        except json.JSONDecodeError:
+            return None
+        return payload if isinstance(payload, dict) else None
+
+    @staticmethod
+    def _try_parse_literal_dict(raw_text: str) -> Optional[Dict[str, object]]:
+        try:
+            payload = ast.literal_eval(raw_text)
+        except (SyntaxError, ValueError):
+            return None
+        return payload if isinstance(payload, dict) else None
+
+    @staticmethod
+    def _try_parse_key_values(raw_text: str) -> Optional[Dict[str, object]]:
+        title: Optional[str] = None
+        subcategory: Optional[str] = None
+        for chunk in re.split(r"[\n;,]+", raw_text):
+            if ":" in chunk:
+                key, value = chunk.split(":", 1)
+            elif "=" in chunk:
+                key, value = chunk.split("=", 1)
+            else:
+                continue
+            key_normalized = key.strip().lower()
+            value_clean = value.strip().strip('"\'')
+            if not value_clean:
+                continue
+            if key_normalized in {"title", "category"}:
+                title = value_clean
+            elif key_normalized in {"subcategory", "sub_category", "sub"}:
+                subcategory = value_clean
+
+        if title and subcategory:
+            return {"title": title, "subcategory": subcategory}
+
+        return None
+
+    async def _get_categories(self) -> List[Dict[str, object]]:
+        async with self._lock:
+            now = time.monotonic()
+            if self._cached_categories and (now - self._last_loaded) < self._cache_ttl_seconds:
+                return self._cached_categories
+
+            categories: List[Dict[str, object]] = []
+            async for doc in self._collection().find({}, {"_id": 0}):
+                categories.append(
+                    {
+                        "title": doc.get("title", ""),
+                        "subcategories": doc.get("subcategories", []),
+                    }
+                )
+
+            self._cached_categories = categories
+            self._last_loaded = now
+            return categories
+
+    @staticmethod
+    def _format_categories(categories: List[Dict[str, object]]) -> str:
+        lines = []
+        for category in categories:
+            subs = category.get("subcategories") or []
+            subs_text = ", ".join(subs) if subs else "Unspecified"
+            lines.append(f"- {category.get('title', 'Unknown')}: {subs_text}")
+        return "\n".join(lines)

requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.111.0
+uvicorn[standard]==0.30.1
+motor==3.4.0
+pymongo>=4.6.0,<4.7.0
+pydantic-settings==2.2.1
+python-dotenv==1.0.1
+openai>=1.30.1,<2.0.0