import httpx import logging from typing import List, Optional from .exceptions import MayoConnectionError, MayoResponseError, MayoAuthError, MayoOrderNotFound from .parser import MayoParser from .models import MayoSearchResult, MayoGuitarDetails, MayoResponse logger = logging.getLogger(__name__) class MayoClient: def __init__(self, base_url: str, login: str, password: str, db: str = "1"): self.base_url = base_url.rstrip("/") self.credentials = { "login": login, "pass": password, "baza": db, } self.client: httpx.AsyncClient | None = None async def __aenter__(self): self.client = httpx.AsyncClient( base_url=self.base_url, follow_redirects=True, timeout=30.0, ) try: await self.login() return self except Exception: await self.client.aclose() self.client = None raise async def __aexit__(self, exc_type, exc_val, exc_tb): if self.client is not None: await self.client.aclose() self.client = None def _require_client(self) -> httpx.AsyncClient: if self.client is None: raise RuntimeError( "HTTP client is not initialized. Use 'async with MayoClient(...) as client:' first." ) return self.client async def login(self): client = self._require_client() try: response = await client.post("/login.php", data=self.credentials) response.raise_for_status() except httpx.TimeoutException as exc: raise MayoConnectionError( f"Mayo connection failed during login: timeout for {self.base_url}/login.php" ) from exc except httpx.RequestError as exc: raise MayoConnectionError( f"Mayo connection failed during login: {exc}" ) from exc except httpx.HTTPStatusError as exc: raise MayoResponseError( f"Mayo returned HTTP {exc.response.status_code} during login" ) from exc if "Zaloguj się" in response.text or "login" in str(response.url): raise MayoAuthError( "Mayo login failed: invalid credentials or login page was returned again" ) logger.info("✅ Zalogowano poprawnie do systemu Mayo.") async def search_order(self, order_number: str) -> List[MayoSearchResult]: """Wyszukuje zamówienia na podstawie numeru.""" # Formatowanie numeru zamówienia (np. 0027) formatted_nr = str(order_number).zfill(4) params = { "filtr": "1", "strona": "0", "sort_order": "1" } payload = { "zaw": "", "r_od": "", "nr_zam": formatted_nr, "typ_kl": "", "klient": "", "r_do": "", "row_count": "25" } client = self._require_client() response = await client.post("/index.php", params=params, data=payload) return MayoParser.parse_search_results(response.text, self.base_url) async def get_guitar_links(self, order_url: str) -> List[str]: """Pobiera listę linków do konkretnych instrumentów w zamówieniu.""" client = self._require_client() response = await client.get(order_url) return MayoParser.parse_guitar_links(response.text, self.base_url) async def get_guitar_details(self, guitar_url: str) -> MayoGuitarDetails: """Pobiera szczegóły konkretnego instrumentu.""" client = self._require_client() response = await client.get(guitar_url) return MayoParser.parse_specification(response.text) async def get_full_order_data(self, order_num: str, year: str, item_idx: str) -> MayoResponse: orders = await self.search_order(order_num) order = next((order for order in orders if order.order_id == f"{order_num}/{year}"), None) if order is None: raise MayoOrderNotFound(f"Order {order_num}/{year} was not found") guitars_url = await self.get_guitar_links(order.url) guitar_index = int(item_idx) - 1 if guitar_index < 0 or guitar_index >= len(guitars_url): raise MayoOrderNotFound( f"Product {order_num}/{year}/{item_idx} was not found" ) guitar_url = guitars_url[guitar_index] details = await self.get_guitar_details(guitar_url) return MayoResponse( order_number=details.order_number, completion_date=details.completion_date, prod_list=order.prod_list, url=guitar_url, client=details.client, model=details.model, spec=details.spec, )