Agent Zero
df50ea8b4b
- Map real Rentman fields: price -> rental_price, folder -> category/subcategory (resolved via folder hierarchy to top-level categories), code -> number, shop_description_long (fallback short/external_remark) -> description (HTML stripped) - Sync only in_shop items; exclude archived/temporary from public catalog - Extract known brands from equipment names (d&b, Shure, Pioneer, ...) - Build specifications from physical fields (power, weight, dimensions) - Persist equipment images in docker volume (survive redeploys) - Router: filter available items, add price_asc/price_desc sorting - Frontend: show prices in catalog card, detail page and cart with per-day subtotal; price sorting options - Tests: use real Rentman field names in fixtures, mock folder_map
305 lines
12 KiB
Python
305 lines
12 KiB
Python
"""Rentman API client wrapper for equipment import and request submission."""
|
|
import httpx
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
RENTMAN_BASE_URL = "https://api.rentman.net"
|
|
|
|
|
|
class RentmanService:
|
|
"""Wrapper around the Rentman REST API using httpx."""
|
|
|
|
def __init__(self, token: str | None = None) -> None:
|
|
self._token = token or settings.rentman_api_token
|
|
self._base_url = RENTMAN_BASE_URL
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {
|
|
"Authorization": f"Bearer {self._token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def get_equipment_page(self, limit: int = 100, offset: int = 0) -> dict[str, Any]:
|
|
"""Fetch a single page of equipment from Rentman.
|
|
|
|
Returns the raw JSON response dict with 'data' and optional 'itemCount'.
|
|
"""
|
|
url = f"{self._base_url}/equipment"
|
|
params = {"limit": limit, "offset": offset}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.get(url, headers=self._headers(), params=params)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def get_all_equipment(self, limit: int = 100) -> list[dict[str, Any]]:
|
|
"""Paginate through all equipment pages until data is empty."""
|
|
all_items: list[dict[str, Any]] = []
|
|
offset = 0
|
|
while True:
|
|
page = await self.get_equipment_page(limit=limit, offset=offset)
|
|
data = page.get("data", [])
|
|
if not data:
|
|
break
|
|
all_items.extend(data)
|
|
offset += limit
|
|
return all_items
|
|
|
|
|
|
async def get_all_folders(self) -> list[dict[str, Any]]:
|
|
"""Paginate through all folders until data is empty."""
|
|
all_folders: list[dict[str, Any]] = []
|
|
offset = 0
|
|
while True:
|
|
url = f"{self._base_url}/folders"
|
|
params = {"limit": 100, "offset": offset}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.get(url, headers=self._headers(), params=params)
|
|
resp.raise_for_status()
|
|
page = resp.json()
|
|
data = page.get("data", [])
|
|
if not data:
|
|
break
|
|
all_folders.extend(data)
|
|
offset += 100
|
|
return all_folders
|
|
|
|
async def get_folder_map(self) -> dict[str, dict[str, str]]:
|
|
"""Build a map: folder path -> {category: top-level name, subcategory: folder name}.
|
|
|
|
Walks the parent chain so every folder resolves to its top-level ancestor.
|
|
"""
|
|
folders = await self.get_all_folders()
|
|
by_path = {f"/folders/{f.get('id')}": f for f in folders}
|
|
|
|
def top_ancestor(folder: dict) -> dict:
|
|
current = folder
|
|
seen = set()
|
|
while current.get("parent") and current["parent"] not in seen:
|
|
seen.add(current["parent"])
|
|
parent = by_path.get(current["parent"])
|
|
if not parent:
|
|
break
|
|
current = parent
|
|
return current
|
|
|
|
folder_map: dict[str, dict[str, str]] = {}
|
|
for path, folder in by_path.items():
|
|
ancestor = top_ancestor(folder)
|
|
folder_map[path] = {
|
|
"category": ancestor.get("name", ""),
|
|
"subcategory": folder.get("name", ""),
|
|
}
|
|
return folder_map
|
|
|
|
async def get_file_url(self, file_id: str | int) -> str | None:
|
|
"""Fetch the S3 URL for a file from Rentman.
|
|
|
|
GET /files/{file_id} → response contains 'url' field with S3 link.
|
|
"""
|
|
url = f"{self._base_url}/files/{file_id}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.get(url, headers=self._headers())
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
file_data = data.get("data", data)
|
|
file_url = file_data.get("url")
|
|
if file_url:
|
|
return file_url
|
|
logger.warning("No url field in file response for file_id=%s", file_id)
|
|
return None
|
|
except Exception as exc:
|
|
logger.warning("Failed to fetch file URL for file_id=%s: %s", file_id, exc)
|
|
return None
|
|
|
|
# Well-known rental brands for extraction from equipment names
|
|
KNOWN_BRANDS = [
|
|
"d&b audiotechnik", "d&b", "L-Acoustics", "Shure", "Sennheiser", "Pioneer",
|
|
"RCF", "Soundcraft", "Cameo", "Eurolite", "Globaltruss", "Klotz",
|
|
"Martin", "JBL", "QSC", "Yamaha", "Behringer", "ROBE", "Chauvet",
|
|
"PL-Audio", "Showtec", "American DJ", "ADJ", "GLP", "Wireless Solution",
|
|
"Sennheiser", "AKG", "Rode", "Audix", "Palmer", "Neutrik", "Layher",
|
|
]
|
|
|
|
@staticmethod
|
|
def _strip_html(text: str) -> str:
|
|
"""Remove HTML tags and normalize whitespace in description fields."""
|
|
if not text:
|
|
return ""
|
|
text = re.sub(r"<br\s*/?>", "\n", text, flags=re.IGNORECASE)
|
|
text = re.sub(r"</p>", "\n", text, flags=re.IGNORECASE)
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
text = text.replace(" ", " ").replace("&", "&")
|
|
text = text.replace("<", "<").replace(">", ">").replace(""", '"')
|
|
text = re.sub(r"[ \t]+", " ", text)
|
|
text = re.sub(r"\n\s*\n+", "\n\n", text)
|
|
return text.strip()
|
|
|
|
def _extract_brand(self, name: str) -> str:
|
|
"""Extract a known brand from the equipment name, if present."""
|
|
lowered = (name or "").lower()
|
|
for brand in self.KNOWN_BRANDS:
|
|
if lowered.startswith(brand.lower() + " ") or lowered.startswith(brand.lower() + "-"):
|
|
return brand
|
|
return ""
|
|
|
|
def _resolve_category(self, folder_path: str | None, folder_map: dict[str, dict[str, str]] | None) -> tuple[str, str]:
|
|
"""Resolve (category, subcategory) from the Rentman folder path.
|
|
|
|
category = top-level folder name (e.g. "Tontechnik")
|
|
subcategory = direct folder name (e.g. "Mikrofone")
|
|
"""
|
|
if not folder_path or not folder_map:
|
|
return "", ""
|
|
info = folder_map.get(folder_path)
|
|
if not info:
|
|
return "", ""
|
|
return info.get("category", ""), info.get("subcategory", "")
|
|
|
|
def _build_specifications(self, raw: dict[str, Any]) -> dict[str, Any]:
|
|
"""Collect physical specs from flat Rentman fields when set."""
|
|
specs: dict[str, Any] = {}
|
|
field_map = [
|
|
("Leistung (W)", "power"),
|
|
("Gewicht (kg)", "empty_weight"),
|
|
("Hoehe (mm)", "height"),
|
|
("Breite (mm)", "width"),
|
|
("Laenge (mm)", "length"),
|
|
("Volumen (m3)", "volume"),
|
|
]
|
|
for label, key in field_map:
|
|
value = raw.get(key)
|
|
if isinstance(value, (int, float)) and value > 0:
|
|
specs[label] = value
|
|
return specs
|
|
|
|
async def transform_equipment(
|
|
self,
|
|
raw: dict[str, Any],
|
|
folder_map: dict[str, dict[str, str]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Map a raw Rentman equipment object to equipment_cache schema.
|
|
|
|
Uses the real Rentman field names:
|
|
- price -> rental_price (per day, shop price)
|
|
- folder ("/folders/N") -> category/subcategory via folder_map
|
|
- shop_description_long (fallbacks: shop_description_short,
|
|
external_remark) -> description (HTML stripped)
|
|
- code -> number (article number)
|
|
- in_shop/in_archive -> availability for the public catalog
|
|
"""
|
|
image_path = raw.get("image")
|
|
image_urls: list[str] = []
|
|
|
|
if image_path and isinstance(image_path, str):
|
|
match = re.search(r"/files/(\d+)", image_path)
|
|
if match:
|
|
file_id = match.group(1)
|
|
s3_url = await self.get_file_url(file_id)
|
|
if s3_url:
|
|
image_urls = [s3_url]
|
|
|
|
category, subcategory = self._resolve_category(raw.get("folder"), folder_map)
|
|
|
|
description = self._strip_html(
|
|
raw.get("shop_description_long")
|
|
or raw.get("shop_description_short")
|
|
or raw.get("external_remark")
|
|
or ""
|
|
)
|
|
|
|
price = raw.get("price")
|
|
try:
|
|
price = float(price) if price is not None else None
|
|
except (TypeError, ValueError):
|
|
price = None
|
|
|
|
available = bool(raw.get("in_shop")) and not bool(raw.get("in_archive")) and not bool(raw.get("temporary"))
|
|
|
|
return {
|
|
"rentman_id": str(raw.get("id", "")),
|
|
"name": raw.get("name", "") or raw.get("displayname", ""),
|
|
"number": raw.get("code", "") or raw.get("number", ""),
|
|
"category": category,
|
|
"subcategory": subcategory,
|
|
"description": description,
|
|
"specifications": self._build_specifications(raw) or None,
|
|
"images": image_urls,
|
|
"rental_price": price,
|
|
"brand": self._extract_brand(raw.get("name", "")),
|
|
"available": available,
|
|
"update_hash": raw.get("updateHash", ""),
|
|
}
|
|
|
|
async def create_project_request(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""POST /projectrequests to create a new project request in Rentman."""
|
|
url = f"{self._base_url}/projectrequests"
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.post(url, headers=self._headers(), json=payload)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def add_equipment_to_request(
|
|
self, request_id: str, equipment_payload: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""POST /projectrequests/{id}/projectrequestequipment for a single item."""
|
|
url = f"{self._base_url}/projectrequests/{request_id}/projectrequestequipment"
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.post(url, headers=self._headers(), json=equipment_payload)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
@staticmethod
|
|
def build_project_request_payload(data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Map frontend rental request data to Rentman POST /projectrequests payload."""
|
|
date_start = data.get("date_start")
|
|
date_end = data.get("date_end")
|
|
|
|
contact_name = data.get("contact_name", "")
|
|
parts = contact_name.strip().split(" ", 1)
|
|
first_name = parts[0] if parts else ""
|
|
last_name = parts[1] if len(parts) > 1 else ""
|
|
|
|
street = data.get("contact_street", "")
|
|
house_number = ""
|
|
if street:
|
|
match = re.search(r"(\d+[a-zA-Z]*)", street)
|
|
if match:
|
|
house_number = match.group(1)
|
|
|
|
return {
|
|
"name": data.get("event_name", ""),
|
|
"planperiod_start": f"{date_start}T08:00:00+02:00" if date_start else None,
|
|
"planperiod_end": f"{date_end}T02:00:00+02:00" if date_end else None,
|
|
"usageperiod_start": f"{date_start}T18:00:00+02:00" if date_start else None,
|
|
"usageperiod_end": f"{date_end}T23:59:00+02:00" if date_end else None,
|
|
"contact_name": data.get("contact_company") or data.get("contact_name", ""),
|
|
"contact_person_first_name": first_name,
|
|
"contact_person_lastname": last_name,
|
|
"contact_person_email": data.get("contact_email", ""),
|
|
"contact_person_phone": data.get("contact_phone", ""),
|
|
"location_name": data.get("location", ""),
|
|
"location_mailing_street": street,
|
|
"location_mailing_number": house_number,
|
|
"location_mailing_postalcode": data.get("contact_postalcode", ""),
|
|
"location_mailing_city": data.get("contact_city") or data.get("location", ""),
|
|
"remark": data.get("message", ""),
|
|
}
|
|
|
|
@staticmethod
|
|
def build_equipment_payload(item: dict[str, Any]) -> dict[str, Any]:
|
|
"""Map a single rental request item to Rentman projectrequestequipment payload."""
|
|
return {
|
|
"name": item.get("equipment_name", ""),
|
|
"quantity": item.get("quantity", 1),
|
|
"quantity_total": item.get("quantity", 1),
|
|
"unit_price": item.get("unit_price", 0),
|
|
"linked_equipment": f"/equipment/{item.get('rentman_equipment_id', '')}" if item.get("rentman_equipment_id") else None,
|
|
}
|