dodalem backend dzialajacy
This commit is contained in:
2
backend/.gitignore
vendored
Normal file
2
backend/.gitignore
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
.venv-linux
|
||||
__pycache__
|
||||
15
backend/mayo/__init__.py
Normal file
15
backend/mayo/__init__.py
Normal file
@@ -0,0 +1,15 @@
|
||||
from .client import MayoClient
|
||||
from .models import MayoResponse
|
||||
from .exceptions import MayoError, MayoAuthError, MayoConnectionError, MayoOrderNotFound, MayoParseError, MayoResponseError, MayoSessionError
|
||||
|
||||
__all__ = [
|
||||
"MayoClient",
|
||||
"MayoResponse",
|
||||
"MayoError",
|
||||
"MayoAuthError",
|
||||
"MayoConnectionError",
|
||||
"MayoOrderNotFound",
|
||||
"MayoParseError",
|
||||
"MayoResponseError",
|
||||
"MayoSessionError",
|
||||
]
|
||||
139
backend/mayo/client.py
Normal file
139
backend/mayo/client.py
Normal file
@@ -0,0 +1,139 @@
|
||||
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,
|
||||
)
|
||||
26
backend/mayo/exceptions.py
Normal file
26
backend/mayo/exceptions.py
Normal file
@@ -0,0 +1,26 @@
|
||||
class MayoError(Exception):
|
||||
"""Bazowy wyjątek dla integracji z Mayo."""
|
||||
|
||||
# dla httpx.RequestError, timeoutów, DNS, resetu połączenia
|
||||
class MayoConnectionError(MayoError):
|
||||
"""Nie udało się połączyć z systemem Mayo."""
|
||||
|
||||
# dla złego loginu/hasła albo strony logowania po rzekomo udanym logowaniu
|
||||
class MayoAuthError(MayoError):
|
||||
"""Uwierzytelnienie w systemie Mayo nie powiodło się."""
|
||||
|
||||
# dla sytuacji, gdy sesja wygasła w trakcie pracy i trzeba się zalogować ponownie
|
||||
class MayoSessionError(MayoError):
|
||||
"""Sesja w systemie Mayo jest nieważna lub wygasła."""
|
||||
|
||||
# dla statusów HTTP typu 500, 403, 502 albo HTML-a, który nie pasuje do oczekiwanego flow
|
||||
class MayoResponseError(MayoError):
|
||||
"""System Mayo zwrócił nieoczekiwaną odpowiedź."""
|
||||
|
||||
# gdy odpowiedź przyszła, ale parser nie umie z niej wyciągnąć danych
|
||||
class MayoParseError(MayoError):
|
||||
"""Nie udało się sparsować odpowiedzi z systemu Mayo."""
|
||||
|
||||
# gdy biznesowo wszystko działa, ale danego zamówienia nie ma
|
||||
class MayoOrderNotFound(MayoError):
|
||||
"""Nie znaleziono zamówienia w systemie Mayo."""
|
||||
@@ -1,127 +0,0 @@
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
def clean(text):
|
||||
return " ".join(text.split())
|
||||
|
||||
def parse_html(path):
|
||||
with open(path, encoding="ISO-8859-2") as f:
|
||||
soup = BeautifulSoup(f, "html.parser")
|
||||
|
||||
tresc = soup.find("div", id="tresc")
|
||||
|
||||
result = {
|
||||
"meta": {},
|
||||
"sections": {}
|
||||
}
|
||||
|
||||
# -----------------------
|
||||
# 🔹 1. META (LEPSZE)
|
||||
# -----------------------
|
||||
|
||||
# 👉 Dot. zam.
|
||||
first_table = tresc.find("table")
|
||||
|
||||
if first_table:
|
||||
b_tags = first_table.find_all("b")
|
||||
print(first_table.getText())
|
||||
|
||||
if len(b_tags) >= 2:
|
||||
result["meta"]["nr_zamownia"] = clean(b_tags[1].get_text())
|
||||
result["meta"]["realizacja"] = clean(b_tags[3].get_text())
|
||||
|
||||
client = first_table.find('span', attrs={'style': "font-weight:bold;"} )
|
||||
if client:
|
||||
print(client.get_text())
|
||||
|
||||
|
||||
# 👉 formularz (Model, Odbiorca itd.)
|
||||
form_table = tresc.find("form")
|
||||
|
||||
if form_table:
|
||||
table = form_table.find_parent("table")
|
||||
|
||||
if table:
|
||||
# 🔥 przeszukujemy CAŁĄ tabelę (wszystkie tr)
|
||||
# Model
|
||||
model_input = table.find("input", {"name": "s_nr_kat"})
|
||||
if model_input:
|
||||
result["meta"]["Model"] = clean(model_input.get("value", ""))
|
||||
|
||||
# Odbiorca
|
||||
odb_input = table.find("input", {"name": "s_odbiorca"})
|
||||
if odb_input:
|
||||
result["meta"]["Odbiorca"] = clean(odb_input.get("value", ""))
|
||||
|
||||
# Grupa
|
||||
grupa_select = table.find("select", {"name": "s_grupa"})
|
||||
if grupa_select:
|
||||
selected = grupa_select.find("option", selected=True)
|
||||
if selected:
|
||||
result["meta"]["Grupa"] = clean(selected.get_text())
|
||||
|
||||
# -----------------------
|
||||
# 🔹 2. SEKCJE (SZYJKA itd.)
|
||||
# -----------------------
|
||||
current_section = None
|
||||
|
||||
for tr in tresc.find_all("tr"):
|
||||
tds = tr.find_all("td")
|
||||
|
||||
if not tds:
|
||||
continue
|
||||
|
||||
# 🔸 Sekcja (np. SZYJKA)
|
||||
if len(tds) == 1:
|
||||
text = clean(tds[0].get_text())
|
||||
|
||||
if text.isupper() and len(text) < 40:
|
||||
current_section = text
|
||||
result["sections"][current_section] = {}
|
||||
continue
|
||||
|
||||
# 🔸 Element w sekcji
|
||||
if len(tds) >= 2 and current_section:
|
||||
key_tag = tds[0].find("b")
|
||||
|
||||
if not key_tag:
|
||||
continue
|
||||
|
||||
key = clean(key_tag.get_text())
|
||||
|
||||
# usuń linki / śmieci
|
||||
key = key.replace("\xa0", "").strip()
|
||||
|
||||
value_td = tds[1]
|
||||
|
||||
# zbierz wszystkie teksty (ignorując "Notatka")
|
||||
texts = []
|
||||
|
||||
for x in value_td.stripped_strings:
|
||||
if "Notatka" in x:
|
||||
continue
|
||||
texts.append(x)
|
||||
|
||||
value = clean(" ".join(texts))
|
||||
|
||||
if key:
|
||||
result["sections"][current_section][key] = value
|
||||
|
||||
return result
|
||||
|
||||
import time
|
||||
|
||||
start = time.perf_counter()
|
||||
|
||||
data = parse_html("g.htm")
|
||||
|
||||
end = time.perf_counter()
|
||||
|
||||
print(f"Czas wykonania: {end - start:.6f} sekund")
|
||||
|
||||
# from pprint import pprint
|
||||
# pprint(data)
|
||||
|
||||
import json
|
||||
|
||||
with open("output.json", "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
@@ -1,176 +0,0 @@
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from urllib.parse import urljoin
|
||||
import re
|
||||
import logging
|
||||
from pprint import pprint
|
||||
|
||||
class MayoSession:
|
||||
def __init__(self, base_url, login, password, db="1"):
|
||||
"""
|
||||
base_url: np. 'http://192.168.0.152/mayo2'
|
||||
login, password: dane logowania
|
||||
db: numer bazy (np. "1" = Mayones 2)
|
||||
"""
|
||||
self.session = requests.Session()
|
||||
self.base_url = base_url
|
||||
self.login_url = f"{self.base_url}/login.php"
|
||||
self.credentials = {
|
||||
"login": login,
|
||||
"pass": password,
|
||||
"baza": db
|
||||
}
|
||||
|
||||
def login(self):
|
||||
"""Loguje się do systemu lokalnego."""
|
||||
r = self.session.post(self.login_url, data=self.credentials)
|
||||
if "Zaloguj się" in r.text or "login" in r.url:
|
||||
raise Exception("Nie udało się zalogować do Mayo.")
|
||||
logging.info("✅ Zalogowano poprawnie do systemu Mayo.")
|
||||
|
||||
def ensure_logged_in(self):
|
||||
test_url = f"{self.base_url}/index.php"
|
||||
|
||||
r = self.session.get(test_url)
|
||||
|
||||
if "Wyloguj" not in r.text:
|
||||
logging.info("🔐 Sesja wygasła — loguję ponownie...")
|
||||
self.login()
|
||||
|
||||
def get_order_page(self, url):
|
||||
self.ensure_logged_in()
|
||||
|
||||
r = self.session.get(url)
|
||||
|
||||
if "login" in r.url or "Zaloguj" in r.text:
|
||||
self.login()
|
||||
r = self.session.get(url)
|
||||
|
||||
return r.text
|
||||
|
||||
def search_order(self, order_number):
|
||||
self.ensure_logged_in()
|
||||
|
||||
url = f"{self.base_url}/index.php?filtr=1&strona=0&sort_order=1"
|
||||
|
||||
payload = {
|
||||
"zaw": "",
|
||||
"r_od": "",
|
||||
"nr_zam": str(order_number).zfill(4), # 🔥 ważne
|
||||
"typ_kl": "",
|
||||
"klient": "",
|
||||
"r_do": "",
|
||||
"row_count": "25"
|
||||
}
|
||||
|
||||
# headers = {
|
||||
# "Content-Type": "application/x-www-form-urlencoded",
|
||||
# "Referer": f"{self.base_url}/index.php",
|
||||
# "Origin": self.base_url
|
||||
# }
|
||||
|
||||
r = self.session.post(url, data=payload)
|
||||
|
||||
# 🔥 fallback jeśli sesja padła w trakcie
|
||||
if "login" in r.url or "Zaloguj" in r.text:
|
||||
logging.warning("⚠️ Sesja padła — ponawiam logowanie...")
|
||||
self.login()
|
||||
r = self.session.post(url, data=payload)
|
||||
|
||||
return r.text
|
||||
|
||||
|
||||
def parse_search_results(self, html):
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
|
||||
results = []
|
||||
|
||||
# tabela wyników
|
||||
table = soup.find("table", class_="std2")
|
||||
|
||||
if not table:
|
||||
return results
|
||||
|
||||
# rows = table.find_all("tr")
|
||||
tbody = table.find("tbody")
|
||||
rows = tbody.find_all("tr") if tbody else table.find_all("tr")
|
||||
|
||||
for row in rows:
|
||||
tds = row.find_all("td")
|
||||
|
||||
# pomijamy header / dziwne wiersze
|
||||
if len(tds) < 3:
|
||||
continue
|
||||
|
||||
link_tag = tds[0].find("a")
|
||||
|
||||
if not link_tag:
|
||||
continue
|
||||
|
||||
# 🔹 order_id
|
||||
order_id = link_tag.get_text(strip=True)
|
||||
|
||||
# 🔹 url (pełny)
|
||||
relative_url = link_tag.get("href")
|
||||
# full_url = urljoin(self.base_url, relative_url)
|
||||
full_url = f"{self.base_url}/{relative_url}"
|
||||
|
||||
# 🔹 prod_list
|
||||
prod_list = tds[1].get_text(strip=True).replace("\xa0", "")
|
||||
|
||||
# 🔹 client
|
||||
client = tds[2].get_text(strip=True)
|
||||
|
||||
results.append({
|
||||
"order_id": order_id,
|
||||
"client": client,
|
||||
"prod_list": prod_list,
|
||||
"url": full_url,
|
||||
"guitars_url": []
|
||||
})
|
||||
|
||||
return results
|
||||
|
||||
def parse_order_list(self, html):
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
|
||||
results = []
|
||||
|
||||
# tabela wyników
|
||||
table = soup.find("table", class_="std2")
|
||||
|
||||
if not table:
|
||||
return results
|
||||
|
||||
# rows = table.find_all("tr")
|
||||
tbody = table.find("tbody")
|
||||
rows = tbody.find_all("tr") if tbody else table.find_all("tr")
|
||||
|
||||
for row in rows:
|
||||
links = row.find_all("a", href=True)
|
||||
|
||||
for link in links:
|
||||
relative_url = link.get("href")
|
||||
|
||||
if "id_zestawu=" in relative_url:
|
||||
full_url = f"{self.base_url}/{relative_url}"
|
||||
results.append(full_url)
|
||||
break # jeden link na wiersz
|
||||
|
||||
return results
|
||||
|
||||
|
||||
if __name__ == "__main__" :
|
||||
mayo = MayoSession("http://10.8.0.6/mayo2", "nowakb", "def")
|
||||
req = mayo.search_order("0027")
|
||||
|
||||
orders = mayo.parse_search_results(req)
|
||||
pprint(orders)
|
||||
|
||||
for order in orders:
|
||||
html = mayo.get_order_page(order["url"])
|
||||
guitars = mayo.parse_order_list(html)
|
||||
order["guitars_url"] = guitars
|
||||
|
||||
print("---------------")
|
||||
pprint(orders)
|
||||
30
backend/mayo/models.py
Normal file
30
backend/mayo/models.py
Normal file
@@ -0,0 +1,30 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
class MayoSearchResult(BaseModel):
|
||||
order_id: str # eg 0001/2025
|
||||
client: str
|
||||
prod_list: str # eg STY-25
|
||||
url: str # url whole order
|
||||
guitars_url: List[str] = Field(default_factory=list) # urls concrete guitars
|
||||
|
||||
class MayoGuitarDetails(BaseModel):
|
||||
order_number: str # eg 0001/2025/1
|
||||
completion_date: str # eg 2025-01-31
|
||||
client: str
|
||||
model: str
|
||||
spec: Dict[str, Dict[str, List[str]]]
|
||||
|
||||
class MayoResponse(BaseModel):
|
||||
order_number: str # eg 0001/2025/1
|
||||
completion_date: str # eg 2025-01-31
|
||||
prod_list: str # eg STY-25
|
||||
url: str # url of a specific guitar
|
||||
client: str # reciever
|
||||
model: str # guitar model
|
||||
spec: Dict[str, Dict[str, List[str]]] # specification
|
||||
|
||||
|
||||
|
||||
|
||||
142
backend/mayo/parser.py
Normal file
142
backend/mayo/parser.py
Normal file
@@ -0,0 +1,142 @@
|
||||
from bs4 import BeautifulSoup
|
||||
from bs4.filter import SoupStrainer
|
||||
from typing import List
|
||||
from unidecode import unidecode
|
||||
import re
|
||||
from .models import MayoSearchResult, MayoGuitarDetails
|
||||
|
||||
class MayoParser:
|
||||
@staticmethod
|
||||
def clean_text(text: str) -> str:
|
||||
if not text:
|
||||
return ""
|
||||
return " ".join(text.split()).replace("\xa0", "").strip()
|
||||
|
||||
@classmethod
|
||||
def parse_search_results(cls, html: str, base_url: str) -> List[MayoSearchResult]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
results = []
|
||||
|
||||
table = soup.find("table", class_="std2")
|
||||
if not table:
|
||||
return results
|
||||
|
||||
tbody = table.find("tbody")
|
||||
rows = tbody.find_all("tr") if tbody else table.find_all("tr")
|
||||
|
||||
for row in rows:
|
||||
tds = row.find_all("td")
|
||||
if len(tds) < 3:
|
||||
continue
|
||||
|
||||
link_tag = tds[0].find("a")
|
||||
if not link_tag:
|
||||
continue
|
||||
|
||||
order_id = cls.clean_text(link_tag.get_text())
|
||||
relative_url = link_tag.get("href")
|
||||
full_url = f"{base_url}/{relative_url}" if relative_url else ""
|
||||
prod_list = cls.clean_text(tds[1].get_text())
|
||||
client = cls.clean_text(tds[2].get_text())
|
||||
|
||||
results.append(MayoSearchResult(
|
||||
order_id=order_id,
|
||||
client=client,
|
||||
prod_list=prod_list,
|
||||
url=full_url
|
||||
))
|
||||
|
||||
return results
|
||||
|
||||
@classmethod
|
||||
def parse_guitar_links(cls, html: str, base_url: str) -> List[str]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
links = []
|
||||
|
||||
table = soup.find("table", class_="std2")
|
||||
if not table:
|
||||
return links
|
||||
|
||||
tbody = table.find("tbody")
|
||||
rows = tbody.find_all("tr") if tbody else table.find_all("tr")
|
||||
|
||||
for row in rows:
|
||||
a_tags = row.find_all("a", href=True)
|
||||
for a in a_tags:
|
||||
href = a.get("href")
|
||||
if not isinstance(href, str):
|
||||
continue
|
||||
if not href.startswith("http") and not href.startswith("index.php"):
|
||||
continue
|
||||
if "id_zestawu=" in href:
|
||||
links.append(f"{base_url}/{href}")
|
||||
break
|
||||
return links
|
||||
|
||||
|
||||
@classmethod
|
||||
def parse_specification(cls, html: str) -> MayoGuitarDetails:
|
||||
result = {
|
||||
"order_number" : "",
|
||||
"completion_date" : "",
|
||||
"client" : "",
|
||||
"model" : "",
|
||||
"spec" : {}
|
||||
}
|
||||
|
||||
only_content = SoupStrainer("div", id="tresc")
|
||||
soup = BeautifulSoup(html, "html.parser", parse_only=only_content)
|
||||
center_tag = soup.find("center")
|
||||
if center_tag is None:
|
||||
return MayoGuitarDetails(**result)
|
||||
|
||||
table_tags = center_tag.find_all(lambda tag: tag.name == "table" and tag.get("class") == ["std"])
|
||||
|
||||
if not table_tags or len(table_tags) < 3:
|
||||
return MayoGuitarDetails(**result)
|
||||
|
||||
# order_id, date, client
|
||||
header = table_tags[0].get_text(strip=True)
|
||||
pattern = r"zam\.:(?P<order_id>\S+)\s*z datą realizacji:\s*(?P<date>\d{4}-\d{2}-\d{2})(?P<client>.*?)$"
|
||||
match = re.search(pattern=pattern, string=header)
|
||||
if match:
|
||||
data = match.groupdict()
|
||||
result["order_number"] = data["order_id"]
|
||||
result["completion_date"] = data["date"]
|
||||
result["client"] = data["client"].rstrip("- ")
|
||||
|
||||
# Model
|
||||
model_input = table_tags[1].find("input", attrs={"name": "s_nr_kat"})
|
||||
if model_input is not None:
|
||||
model = model_input.get("value")
|
||||
if isinstance(model, str):
|
||||
result["model"] = model
|
||||
|
||||
# spec
|
||||
current_section = None
|
||||
for row in table_tags[2].find_all("tr", recursive=False):
|
||||
cells = row.find_all("td", recursive=False)
|
||||
|
||||
if not cells:
|
||||
continue
|
||||
|
||||
if len(cells) == 1 and cells[0].get("colspan") == "4":
|
||||
text = cells[0].get_text(strip=True)
|
||||
|
||||
if text.isupper() and len(text) < 40:
|
||||
current_section = unidecode(text.lower())
|
||||
result["spec"][current_section] = {}
|
||||
continue
|
||||
|
||||
if len(cells) != 2:
|
||||
continue
|
||||
|
||||
key = unidecode( cells[0].get_text(strip=True).lower().replace(" ", "") )
|
||||
value = unidecode( cells[1].get_text(strip=True) )
|
||||
parts = re.split(r"notatka", value, flags=re.IGNORECASE)
|
||||
entries = [p.strip().lstrip("- ") for p in parts if p.strip()]
|
||||
|
||||
if key and value and current_section is not None:
|
||||
result["spec"][current_section][key] = entries
|
||||
|
||||
return MayoGuitarDetails(**result)
|
||||
114
backend/my_fastapi.py
Normal file
114
backend/my_fastapi.py
Normal file
@@ -0,0 +1,114 @@
|
||||
import re
|
||||
import os
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, Query, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from mayo.exceptions import (
|
||||
MayoAuthError,
|
||||
MayoConnectionError,
|
||||
MayoError,
|
||||
MayoOrderNotFound,
|
||||
MayoParseError,
|
||||
MayoResponseError,
|
||||
MayoSessionError,
|
||||
)
|
||||
from mayo import MayoClient, MayoResponse
|
||||
|
||||
app = FastAPI(title="Mayo Integration API")
|
||||
|
||||
MAYO_BASE_URL = os.getenv("MAYO_BASE_URL", "http://10.8.0.6/mayo2")
|
||||
MAYO_LOGIN = os.getenv("MAYO_LOGIN", "nowakb")
|
||||
MAYO_PASSWORD = os.getenv("MAYO_PASSWORD", "def")
|
||||
ORDER_ID_RE = re.compile(
|
||||
r"^(?P<order_num>\d{4})/(?P<year>\d{4})/(?P<item_idx>\d{1,2})$"
|
||||
)
|
||||
|
||||
|
||||
class OrderId(BaseModel):
|
||||
order_num: str
|
||||
year: str
|
||||
item_idx: str
|
||||
|
||||
|
||||
@app.exception_handler(MayoOrderNotFound)
|
||||
async def mayo_order_not_found_handler(request: Request, exc: MayoOrderNotFound):
|
||||
return JSONResponse(status_code=404, content={"detail": str(exc)})
|
||||
|
||||
|
||||
@app.exception_handler(MayoConnectionError)
|
||||
async def mayo_connection_error_handler(request: Request, exc: MayoConnectionError):
|
||||
return JSONResponse(
|
||||
status_code=503,
|
||||
content={"detail": "Mayo service is currently unavailable."},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(MayoAuthError)
|
||||
async def mayo_auth_error_handler(request: Request, exc: MayoAuthError):
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
content={"detail": "Authentication with Mayo failed."},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(MayoSessionError)
|
||||
async def mayo_session_error_handler(request: Request, exc: MayoSessionError):
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
content={"detail": "Mayo session is invalid or has expired."},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(MayoResponseError)
|
||||
async def mayo_response_error_handler(request: Request, exc: MayoResponseError):
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
content={"detail": "Mayo returned an unexpected response."},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(MayoParseError)
|
||||
async def mayo_parse_error_handler(request: Request, exc: MayoParseError):
|
||||
return JSONResponse(
|
||||
status_code=502,
|
||||
content={"detail": "Failed to parse data returned by Mayo."},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(MayoError)
|
||||
async def mayo_error_handler(request: Request, exc: MayoError):
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={"detail": "Unexpected Mayo integration error."},
|
||||
)
|
||||
|
||||
|
||||
def parse_order_id(
|
||||
order_id: Annotated[str, Query(description="Format: XXXX/YYYY/ZZ")],
|
||||
) -> OrderId:
|
||||
match = ORDER_ID_RE.fullmatch(order_id)
|
||||
if not match:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail="order_id must have format XXXX/YYYY/ZZ",
|
||||
)
|
||||
|
||||
return OrderId(**match.groupdict())
|
||||
|
||||
async def get_mayo_client():
|
||||
async with MayoClient(MAYO_BASE_URL, MAYO_LOGIN, MAYO_PASSWORD) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@app.get("/orders/", response_model=MayoResponse)
|
||||
async def get_order(order: Annotated[OrderId, Depends(parse_order_id)], client: MayoClient = Depends(get_mayo_client)):
|
||||
return await client.get_full_order_data(order.order_num, order.year, order.item_idx)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
uvicorn.run(app, host="0.0.0.0", port=8001)
|
||||
Reference in New Issue
Block a user