arrr-erp/app/modules/accounting/ai_provider.py

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any


class AccountingAIProviderError(RuntimeError):
    pass


@dataclass
class ProviderResult:
    provider: str
    model: str
    data: dict[str, Any]
    raw: dict[str, Any]
    input_tokens: int = 0
    output_tokens: int = 0
    latency_ms: int = 0


class AccountingSemanticProvider:
    provider_name = "base"

    def is_configured(self) -> bool:
        raise NotImplementedError

    def model_name(self) -> str:
        raise NotImplementedError

    def classify(self, *, system_text: str, user_payload: dict[str, Any], schema: dict[str, Any]) -> ProviderResult:
        raise NotImplementedError


class OpenAIResponsesProvider(AccountingSemanticProvider):
    provider_name = "openai"

    def __init__(self):
        self.api_key = (os.getenv("OPENAI_API_KEY") or "").strip()
        self.model = (os.getenv("ACCOUNTING_AI_MODEL") or "gpt-5-mini").strip()
        self.base_url = (os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1").rstrip("/")
        self.timeout = max(10, min(120, int(os.getenv("ACCOUNTING_AI_TIMEOUT_SECONDS", "45") or 45)))

    def is_configured(self) -> bool:
        return bool(self.api_key and self.model)

    def model_name(self) -> str:
        return self.model

    @staticmethod
    def _extract_output_text(response: dict[str, Any]) -> str:
        value = response.get("output_text")
        if isinstance(value, str) and value.strip():
            return value.strip()

        for item in response.get("output") or []:
            if not isinstance(item, dict) or item.get("type") != "message":
                continue
            for content in item.get("content") or []:
                if not isinstance(content, dict):
                    continue
                if content.get("type") in {"output_text", "text"}:
                    text = content.get("text")
                    if isinstance(text, str) and text.strip():
                        return text.strip()
        raise AccountingAIProviderError("AI response did not contain structured output text.")

    def classify(self, *, system_text: str, user_payload: dict[str, Any], schema: dict[str, Any]) -> ProviderResult:
        if not self.is_configured():
            raise AccountingAIProviderError(
                "Server AI is not configured. Set OPENAI_API_KEY and ACCOUNTING_AI_MODEL on the ERP server."
            )

        body = {
            "model": self.model,
            "store": False,
            "input": [
                {
                    "role": "system",
                    "content": [{"type": "input_text", "text": system_text}],
                },
                {
                    "role": "user",
                    "content": [{"type": "input_text", "text": json.dumps(user_payload, ensure_ascii=False)}],
                },
            ],
            "text": {
                "format": {
                    "type": "json_schema",
                    "name": "accounting_semantic_classification",
                    "strict": True,
                    "schema": schema,
                }
            },
        }

        request = urllib.request.Request(
            self.base_url + "/responses",
            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
            method="POST",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
                "User-Agent": "ARRR-ERP-Accounting-AI/12",
            },
        )

        started = time.monotonic()
        last_error = None
        for attempt in range(2):
            try:
                with urllib.request.urlopen(request, timeout=self.timeout) as response:
                    raw = json.loads(response.read().decode("utf-8"))
                text = self._extract_output_text(raw)
                data = json.loads(text)
                usage = raw.get("usage") or {}
                return ProviderResult(
                    provider=self.provider_name,
                    model=str(raw.get("model") or self.model),
                    data=data,
                    raw=raw,
                    input_tokens=int(usage.get("input_tokens") or 0),
                    output_tokens=int(usage.get("output_tokens") or 0),
                    latency_ms=int((time.monotonic() - started) * 1000),
                )
            except urllib.error.HTTPError as exc:
                detail = exc.read().decode("utf-8", errors="ignore")[:2000]
                last_error = AccountingAIProviderError(
                    f"AI provider HTTP {exc.code}: {detail or exc.reason}"
                )
                if exc.code not in {429, 500, 502, 503, 504} or attempt:
                    raise last_error
                time.sleep(1.0)
            except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
                last_error = AccountingAIProviderError(f"AI provider request failed: {exc}")
                if attempt:
                    raise last_error
                time.sleep(1.0)

        raise last_error or AccountingAIProviderError("AI provider request failed.")


def get_accounting_semantic_provider() -> AccountingSemanticProvider:
    provider = (os.getenv("ACCOUNTING_AI_PROVIDER") or "openai").strip().lower()
    if provider == "openai":
        return OpenAIResponsesProvider()
    raise AccountingAIProviderError(
        f"Unsupported ACCOUNTING_AI_PROVIDER '{provider}'. Current Phase 12 provider: openai."
    )