125 lines
4.5 KiB
Python
125 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import requests
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class ZammadError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ZammadClient:
|
|
def __init__(self, base_url: str, token: str) -> None:
|
|
self.base = base_url.rstrip("/")
|
|
self.session = requests.Session()
|
|
self.session.headers.update(
|
|
{
|
|
"Authorization": f"Token token={token}",
|
|
"Accept": "application/json",
|
|
}
|
|
)
|
|
|
|
# Simple in-memory caches
|
|
self._user_cache: Dict[int, str] = {}
|
|
self._group_cache: Dict[int, str] = {}
|
|
self._state_cache: Dict[int, Dict[str, Any]] = {}
|
|
self._priority_cache: Dict[int, str] = {}
|
|
|
|
# HTTP helper
|
|
def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
|
|
url = f"{self.base}{path}"
|
|
try:
|
|
r = self.session.get(url, params=params, timeout=30)
|
|
except requests.RequestException as e:
|
|
raise ZammadError(str(e))
|
|
if r.status_code >= 400:
|
|
raise ZammadError(f"{r.status_code} {r.text}")
|
|
try:
|
|
return r.json()
|
|
except ValueError:
|
|
raise ZammadError("Invalid JSON from Zammad")
|
|
|
|
# Tickets search by created range
|
|
def search_tickets(self, date_from: str, date_to: str, per_page: int = 200) -> List[Dict[str, Any]]:
|
|
start = date_from[:10]
|
|
end = date_to[:10]
|
|
query = f"created_at:[{start} TO {end}]"
|
|
|
|
tickets: List[Dict[str, Any]] = []
|
|
page = 1
|
|
while True:
|
|
params = {"query": query, "per_page": per_page, "page": page}
|
|
data = self._get("/api/v1/tickets/search", params=params)
|
|
|
|
if isinstance(data, list):
|
|
batch = data
|
|
elif isinstance(data, dict):
|
|
if isinstance(data.get("tickets"), list):
|
|
batch = data["tickets"]
|
|
elif isinstance(data.get("rows"), list):
|
|
batch = data["rows"]
|
|
else:
|
|
batch = data.get("data", []) if isinstance(data.get("data"), list) else []
|
|
else:
|
|
batch = []
|
|
|
|
tickets.extend(batch)
|
|
if len(batch) < per_page:
|
|
break
|
|
page += 1
|
|
|
|
return tickets
|
|
|
|
# Lookups
|
|
def user_name(self, user_id: Optional[int]) -> str:
|
|
if not user_id:
|
|
return "Без владельца"
|
|
if user_id in self._user_cache:
|
|
return self._user_cache[user_id]
|
|
data = self._get(f"/api/v1/users/{user_id}")
|
|
name = (data.get("fullname") or data.get("firstname") or data.get("login") or str(user_id)).strip()
|
|
self._user_cache[user_id] = name
|
|
return name
|
|
|
|
def group_name(self, group_id: Optional[int]) -> str:
|
|
if not group_id:
|
|
return "Без группы"
|
|
if group_id in self._group_cache:
|
|
return self._group_cache[group_id]
|
|
data = self._get(f"/api/v1/groups/{group_id}")
|
|
name = (data.get("name") or str(group_id)).strip()
|
|
self._group_cache[group_id] = name
|
|
return name
|
|
|
|
def state(self, state_id: Optional[int]) -> Dict[str, Any]:
|
|
if not state_id:
|
|
return {"name": "unknown", "state_type": "unknown"}
|
|
if state_id in self._state_cache:
|
|
return self._state_cache[state_id]
|
|
data = self._get(f"/api/v1/ticket_states/{state_id}")
|
|
# Try to resolve state_type name if available
|
|
stype = data.get("state_type") or data.get("state_type_id")
|
|
state_type_name = None
|
|
if isinstance(stype, dict):
|
|
state_type_name = stype.get("name")
|
|
elif isinstance(stype, int):
|
|
try:
|
|
tdata = self._get(f"/api/v1/ticket_state_types/{stype}")
|
|
state_type_name = tdata.get("name")
|
|
except ZammadError:
|
|
state_type_name = None
|
|
result = {"name": data.get("name", str(state_id)), "state_type": state_type_name or "unknown"}
|
|
self._state_cache[state_id] = result
|
|
return result
|
|
|
|
def priority_name(self, priority_id: Optional[int]) -> str:
|
|
if not priority_id:
|
|
return "Без приоритета"
|
|
if priority_id in self._priority_cache:
|
|
return self._priority_cache[priority_id]
|
|
data = self._get(f"/api/v1/ticket_priorities/{priority_id}")
|
|
name = (data.get("name") or str(priority_id)).strip()
|
|
self._priority_cache[priority_id] = name
|
|
return name
|
|
|