109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""Minimal EspoCRM REST client (X-Api-Key auth, in-run retries)."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
|
|
import requests
|
|
|
|
|
|
class EspoError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class EspoClient:
|
|
def __init__(self, base: str, api_key: str, retries: int = 2,
|
|
backoff: float = 2.0):
|
|
self.base = base.rstrip("/")
|
|
self.session = requests.Session()
|
|
self.session.headers["X-Api-Key"] = api_key
|
|
self.retries = retries
|
|
self.backoff = backoff
|
|
|
|
def _req(self, method: str, path: str, **kw):
|
|
last = None
|
|
for attempt in range(self.retries + 1):
|
|
try:
|
|
r = self.session.request(method, f"{self.base}/{path}",
|
|
timeout=30, **kw)
|
|
if r.status_code < 500:
|
|
if r.status_code >= 400:
|
|
raise EspoError(
|
|
f"{method} {path} -> {r.status_code} "
|
|
f"{r.headers.get('X-Status-Reason', '')}")
|
|
return r.json() if r.text else {}
|
|
last = EspoError(f"{method} {path} -> {r.status_code}")
|
|
except (requests.ConnectionError, requests.Timeout) as exc:
|
|
last = exc
|
|
time.sleep(self.backoff * (attempt + 1))
|
|
raise EspoError(str(last))
|
|
|
|
def search(self, entity, wtype, attr, value, select="name", max_size=50):
|
|
params = {"where[0][type]": wtype, "where[0][attribute]": attr,
|
|
"where[0][value]": value, "select": select,
|
|
"maxSize": max_size}
|
|
return self._req("GET", entity, params=params).get("list", [])
|
|
|
|
def find_opportunity_by_link(self, url):
|
|
hits = self.search("Opportunity", "equals", "cProjektlink", url)
|
|
return hits[0] if hits else None
|
|
|
|
def team_id(self, name):
|
|
hits = self.search("Team", "equals", "name", name)
|
|
if not hits:
|
|
raise EspoError(
|
|
f"Team '{name}' nicht gefunden — cowork-api braucht Lese-"
|
|
f"Zugriff auf Team, und das Team muss existieren")
|
|
return hits[0]["id"]
|
|
|
|
def ensure_account(self, name, acc_type):
|
|
for hit in self.search("Account", "contains", "name",
|
|
core_token(name), select="name,type"):
|
|
if hit["name"].strip().lower() == name.strip().lower():
|
|
return hit["id"]
|
|
return self._req("POST", "Account",
|
|
json={"name": name, "type": acc_type})["id"]
|
|
|
|
def ensure_contact(self, first, last, account_id):
|
|
full = f"{first} {last}".strip().lower()
|
|
for hit in self.search("Contact", "contains", "name", last,
|
|
select="name,accountName"):
|
|
if hit["name"].strip().lower() == full:
|
|
return hit["id"]
|
|
return self._req("POST", "Contact",
|
|
json={"firstName": first, "lastName": last,
|
|
"accountId": account_id})["id"]
|
|
|
|
def unique_opportunity_name(self, name):
|
|
existing = {h["name"] for h in self.search(
|
|
"Opportunity", "startsWith", "name", name, max_size=100)}
|
|
if name not in existing:
|
|
return name
|
|
n = 2
|
|
while f"{name} ({n})" in existing:
|
|
n += 1
|
|
return f"{name} ({n})"
|
|
|
|
def create_opportunity(self, payload):
|
|
return self._req("POST", "Opportunity", json=payload)
|
|
|
|
def get_opportunity(self, oid):
|
|
return self._req("GET", f"Opportunity/{oid}")
|
|
|
|
def delete(self, entity, oid):
|
|
return self._req("DELETE", f"{entity}/{oid}")
|
|
|
|
|
|
def core_token(name: str) -> str:
|
|
for tok in re.split(r"[^\wÄÖÜäöüß]+", name or ""):
|
|
if len(tok) > 2:
|
|
return tok
|
|
return name
|
|
|
|
|
|
def split_person(full: str):
|
|
parts = (full or "").split()
|
|
if not parts:
|
|
return ("", "")
|
|
return (" ".join(parts[:-1]), parts[-1])
|