"""Repuso Monitoring API - official Python client. Zero dependencies (standard library only). from repuso import Repuso repuso = Repuso("YOUR_API_KEY") data = repuso.reviews.list(profiles="123", min_rating=4) Auth is HTTP Basic with an empty username and your API key as the password. """ import base64 import json import urllib.error import urllib.parse import urllib.request class RepusoError(Exception): def __init__(self, message, status=None, body=None): super().__init__(message) self.status = status self.body = body class _Namespace: def __init__(self, client): self._c = client class _Reviews(_Namespace): def list(self, **params): return self._c.request("GET", "/reviews/get", params=params) def get(self, review_id): return self._c.request("GET", f"/reviews/get/{review_id}") def create(self, data, idempotency_key=None): return self._c.request("POST", "/reviews/add", body=data, idempotency_key=idempotency_key) def create_bulk(self, reviews, idempotency_key=None): return self._c.request("POST", "/reviews/add_bulk", body={"reviews": reviews}, idempotency_key=idempotency_key) def update(self, review_id, data): return self._c.request("PUT", f"/reviews/update/{review_id}", body=data) def remove(self, review_id): return self._c.request("DELETE", f"/reviews/delete/{review_id}") class _Profiles(_Namespace): def list(self, **params): return self._c.request("GET", "/profiles/get", params=params) def get(self, profile_id): return self._c.request("GET", f"/profiles/get/{profile_id}") def types(self): return self._c.request("GET", "/profiles/types") def add(self, data, idempotency_key=None): return self._c.request("POST", "/profiles/add", body=data, idempotency_key=idempotency_key) def update(self, profile_id, data): return self._c.request("PUT", f"/profiles/update/{profile_id}", body=data) def remove(self, profile_id): return self._c.request("DELETE", f"/profiles/delete/{profile_id}") def refresh(self, profile_id): return self._c.request("POST", f"/profiles/refresh/{profile_id}") def summary(self, profile_id, **params): return self._c.request("GET", f"/profiles/ai/summary/{profile_id}", params=params) def generate_summary(self, profile_id, **data): return self._c.request("POST", f"/profiles/ai/summary/{profile_id}", body=data or None) def insights(self, profile_id, **params): return self._c.request("GET", f"/profiles/ai/insights/{profile_id}", params=params) def rating_history(self, profile_id, **params): return self._c.request("GET", f"/profiles/rating_history/{profile_id}", params=params) def discover(self, name, platforms, **extra): return self._c.request("POST", "/profiles/discover", body={"name": name, "platforms": platforms, **extra}) class _Reports(_Namespace): def ratings(self, **params): return self._c.request("GET", "/reports/ratings", params=params) def average_rating(self, **params): return self._c.request("GET", "/reports/average_rating", params=params) def reviews_over_time(self, **params): return self._c.request("GET", "/reports/reviews_over_time", params=params) def rating_distribution(self, **params): return self._c.request("GET", "/reports/rating_distribution", params=params) class _Webhooks(_Namespace): def list(self): return self._c.request("GET", "/webhooks/get") def add(self, data): return self._c.request("POST", "/webhooks/add", body=data) def update(self, hook_id, data): return self._c.request("PUT", f"/webhooks/update/{hook_id}", body=data) def remove(self, hook_id): return self._c.request("DELETE", f"/webhooks/delete/{hook_id}") def test(self, hook_id): return self._c.request("POST", f"/webhooks/test/{hook_id}") def deliveries(self, hook_id): return self._c.request("GET", f"/webhooks/deliveries/{hook_id}") class _Groups(_Namespace): def list(self): return self._c.request("GET", "/groups") def create(self, data): return self._c.request("POST", "/groups", body=data) def update(self, group_id, data): return self._c.request("PUT", f"/groups/{group_id}", body=data) def remove(self, group_id): return self._c.request("DELETE", f"/groups/{group_id}") class _Usage(_Namespace): def get(self): return self._c.request("GET", "/usage") def log(self): return self._c.request("GET", "/usage/log") class Repuso: def __init__(self, api_key, base_url="https://api.repuso.com/public/v1", timeout=30): if not api_key: raise ValueError("Repuso: an API key is required") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self._auth = "Basic " + base64.b64encode((":" + api_key).encode("utf-8")).decode("ascii") self.reviews = _Reviews(self) self.profiles = _Profiles(self) self.reports = _Reports(self) self.webhooks = _Webhooks(self) self.groups = _Groups(self) self.usage = _Usage(self) def request(self, method, path, params=None, body=None, idempotency_key=None): url = self.base_url + path if params: clean = {k: v for k, v in params.items() if v is not None and v != ""} if clean: url += "?" + urllib.parse.urlencode(clean) headers = {"Authorization": self._auth, "Accept": "application/json"} data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode("utf-8") if idempotency_key: headers["Idempotency-Key"] = idempotency_key req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw else None except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") try: parsed = json.loads(raw) if raw else None except ValueError: parsed = raw msg = parsed.get("msg") if isinstance(parsed, dict) else (raw or e.reason) raise RepusoError(msg, status=e.code, body=parsed) from None