odoo-at-payroll/addons/l10n_at_payroll_agent/models/pv_agent_client.py

# Part of the odoo-at-payroll project. License: LGPL-3.
"""PV-Agent-Client: HTTP-Zugriff auf den eigenständigen Wissensdienst.

Odoo orchestriert und rechnet (System of Record, D25); der Agent antwortet
streng KB-gebunden. Der Client sendet ausschließlich Frage, Modus und den
schema-gebundenen Kontext — keine Personendaten, keine Rohobjekte. Der
Service-Key liegt am Unternehmen (`pv_agent_api_key`) und wird niemals
geloggt. HTTP-Zugriff über die stdlib (urllib) — keine Zusatzabhängigkeit.
"""
import json
import urllib.error
import urllib.request

from odoo import _, fields, models
from odoo.exceptions import UserError

DEFAULT_PV_AGENT_URL = "http://127.0.0.1:8080"
REQUEST_TIMEOUT_S = 240.0  # KB-Antworten beobachtet: ~40-160 s


class PvAgentHttpError(RuntimeError):
    """Interner Fehler-Wrapper für den Client-Test-Seam."""

    def __init__(self, status, detail=""):
        super().__init__(f"HTTP {status}: {detail}")
        self.status = status
        self.detail = detail


def _post_json(url, payload, api_key, request_id, timeout_s):
    """Test-Seam: einziger HTTP-Ausgang des Moduls (stdlib-only)."""
    headers = {
        "Content-Type": "application/json",
        "X-Request-ID": request_id,
    }
    if api_key:
        headers["Authorization"] = "Bearer " + api_key
    data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url, data=data, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=timeout_s) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        detail = ""
        try:
            detail = exc.read().decode("utf-8", errors="replace")
        except OSError:
            pass
        raise PvAgentHttpError(exc.code, detail) from exc
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        raise PvAgentHttpError(0, str(exc)) from exc


def _error_detail(status, detail):
    """Aus der Fehlerantwort des Agenten die neutrale Detailzeile ziehen —
    der Agent selbst liefert keine internen Hosts/Exceptions (v1-Vertrag)."""
    try:
        import json as _json
        body = _json.loads(detail)
        if isinstance(body, dict) and isinstance(body.get("detail"), str):
            return body["detail"]
    except (ValueError, TypeError):
        pass
    return ""


class PvAgentClient(models.AbstractModel):
    _name = "pv.agent.client"
    _description = "PV-Agent-Client (Wissensdienst / Review)"

    def _l10n_at_pv_agent_url(self):
        return self.env.company.pv_agent_url or DEFAULT_PV_AGENT_URL

    def _l10n_at_pv_agent_api_key(self):
        return self.env.company.pv_agent_api_key or ""

    def l10n_at_pv_ask_review(self, question, context, request_id):
        """POST /v1/ask (mode=review) — liefert die geprüfte Antwort des
        Agenten (dict). Fehler werden als UserError mit neutraler Meldung
        abgebildet; technische Details stehen nur im Server-Log."""
        url = self._l10n_at_pv_agent_url().rstrip("/") + "/v1/ask"
        payload = {
            "question": question,
            "mode": "review",
            "context": context,
        }
        try:
            status, body = _post_json(
                url, payload, self._l10n_at_pv_agent_api_key(), request_id,
                REQUEST_TIMEOUT_S)
        except PvAgentHttpError as exc:
            if exc.status == 401:
                raise UserError(_(
                    "PV-Agent: Service-Key fehlt oder ist ungültig. "
                    "PV-Agent-API-Key in den Unternehmensstammdaten prüfen."))
            if exc.status == 422:
                raise UserError(_(
                    "PV-Agent hat die Anfrage abgelehnt: %s",
                    _error_detail(exc.status, exc.detail) or "ungültige Anfrage"))
            if exc.status in (0, 502, 503, 504):
                raise UserError(_(
                    "PV-Agent nicht erreichbar (URL: %s). Dienststatus und "
                    "PV-Agent-URL in den Unternehmensstammdaten prüfen.",
                    self._l10n_at_pv_agent_url()))
            raise UserError(_("PV-Agent-Fehler (HTTP %s).", exc.status)) from exc
        return body


class ResCompany(models.Model):
    _inherit = "res.company"

    pv_agent_url = fields.Char(
        string="PV-Agent-URL",
        default=DEFAULT_PV_AGENT_URL,
        help="Basis-URL des eigenständigen Wissensdienstes, z. B. "
             "http://127.0.0.1:8080 wenn der Agent am selben Host läuft.",
        groups="hr_payroll.group_hr_payroll_user")
    pv_agent_api_key = fields.Char(
        string="PV-Agent Service-Key",
        help="Bearer-Key der Agent-API. Nur im Unternehmen hinterlegen — "
             "niemals in Logs oder Meldungen ausgeben.",
        groups="hr_payroll.group_hr_payroll_user")