Sage100-vps/sage_client.py

import requests
from typing import Dict, List, Optional
from config import settings
import logging

logger = logging.getLogger(__name__)


class SageGatewayClient:
    """
    Client HTTP pour communiquer avec la gateway Sage Windows
    """

    def __init__(self):
        self.url = settings.sage_gateway_url.rstrip("/")
        self.headers = {
            "X-Sage-Token": settings.sage_gateway_token,
            "Content-Type": "application/json",
        }
        self.timeout = 30

    def _post(self, endpoint: str, data: dict = None, retries: int = 3) -> dict:
        """POST avec retry automatique"""
        import time

        for attempt in range(retries):
            try:
                r = requests.post(
                    f"{self.url}{endpoint}",
                    json=data or {},
                    headers=self.headers,
                    timeout=self.timeout,
                )
                r.raise_for_status()
                return r.json()
            except requests.exceptions.RequestException as e:
                if attempt == retries - 1:
                    logger.error(
                        f"❌ Échec après {retries} tentatives sur {endpoint}: {e}"
                    )
                    raise
                time.sleep(2**attempt)

    def _get(self, endpoint: str, params: dict = None, retries: int = 3) -> dict:
        """GET avec retry automatique"""
        import time

        for attempt in range(retries):
            try:
                r = requests.get(
                    f"{self.url}{endpoint}",
                    params=params or {},
                    headers=self.headers,
                    timeout=self.timeout,
                )
                r.raise_for_status()
                return r.json()
            except requests.exceptions.RequestException as e:
                if attempt == retries - 1:
                    logger.error(
                        f"❌ Échec GET après {retries} tentatives sur {endpoint}: {e}"
                    )
                    raise
                time.sleep(2**attempt)

    # === CLIENTS ===
    def lister_clients(self, filtre: str = "") -> List[Dict]:
        return self._post("/sage/clients/list", {"filtre": filtre}).get("data", [])

    def lire_client(self, code: str) -> Optional[Dict]:
        return self._post("/sage/clients/get", {"code": code}).get("data")

    # === ARTICLES ===
    def lister_articles(self, filtre: str = "") -> List[Dict]:
        return self._post("/sage/articles/list", {"filtre": filtre}).get("data", [])

    def lire_article(self, ref: str) -> Optional[Dict]:
        return self._post("/sage/articles/get", {"code": ref}).get("data")

    # === DEVIS ===
    def creer_devis(self, devis_data: Dict) -> Dict:
        return self._post("/sage/devis/create", devis_data).get("data", {})

    def lire_devis(self, numero: str) -> Optional[Dict]:
        return self._post("/sage/devis/get", {"code": numero}).get("data")

    # 🆕 US-A1: Lister devis
    def lister_devis(
        self, limit: int = 100, statut: Optional[int] = None
    ) -> List[Dict]:
        """Liste tous les devis avec filtres"""
        payload = {"limit": limit}
        if statut is not None:
            payload["statut"] = statut
        return self._post("/sage/devis/list", payload).get("data", [])

    # 🆕 US-A1: Changer statut devis
    def changer_statut_devis(self, numero: str, nouveau_statut: int) -> Dict:
        """Change le statut d'un devis"""
        return self._post(
            "/sage/devis/statut", {"numero": numero, "nouveau_statut": nouveau_statut}
        ).get("data", {})

    # === DOCUMENTS ===
    def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]:
        return self._post(
            "/sage/documents/get", {"numero": numero, "type_doc": type_doc}
        ).get("data")

    def transformer_document(
        self, numero_source: str, type_source: int, type_cible: int
    ) -> Dict:
        return self._post(
            "/sage/documents/transform",
            {
                "numero_source": numero_source,
                "type_source": type_source,
                "type_cible": type_cible,
            },
        ).get("data", {})

    def mettre_a_jour_champ_libre(
        self, doc_id: str, type_doc: int, nom_champ: str, valeur: str
    ) -> bool:
        resp = self._post(
            "/sage/documents/champ-libre",
            {
                "doc_id": doc_id,
                "type_doc": type_doc,
                "nom_champ": nom_champ,
                "valeur": valeur,
            },
        )
        return resp.get("success", False)

    # 🆕 US-A2: Lister commandes
    def lister_commandes(
        self, limit: int = 100, statut: Optional[int] = None
    ) -> List[Dict]:
        """Liste toutes les commandes"""
        payload = {"limit": limit}
        if statut is not None:
            payload["statut"] = statut
        return self._post("/sage/commandes/list", payload).get("data", [])

    # 🆕 US-A7: Lister factures
    def lister_factures(
        self, limit: int = 100, statut: Optional[int] = None
    ) -> List[Dict]:
        """Liste toutes les factures"""
        payload = {"limit": limit}
        if statut is not None:
            payload["statut"] = statut
        return self._post("/sage/factures/list", payload).get("data", [])

    # === CONTACTS ===
    def lire_contact_client(self, code_client: str) -> Optional[Dict]:
        return self._post("/sage/contact/read", {"code": code_client}).get("data")

    # 🆕 US-A5: Lire remise max client
    def lire_remise_max_client(self, code_client: str) -> float:
        """Récupère la remise max autorisée pour un client"""
        result = self._post("/sage/client/remise-max", {"code": code_client})
        return result.get("data", {}).get("remise_max", 10.0)

    # === CACHE ===
    def refresh_cache(self) -> Dict:
        return self._post("/sage/cache/refresh")

    def get_cache_info(self) -> Dict:
        """Récupère les infos du cache"""
        return self._get("/sage/cache/info").get("data", {})

    # === HEALTH ===
    def health(self) -> dict:
        try:
            r = requests.get(f"{self.url}/health", timeout=5)
            return r.json()
        except:
            return {"status": "down"}


# Instance globale
sage_client = SageGatewayClient()