90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
import json
|
|
import logging
|
|
import re
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ROASTDB_URL = "https://www.roastdb.com/beans/{slug}"
|
|
ROASTDB_SEARCH_URL = "https://www.roastdb.com/explore?q={query}"
|
|
|
|
|
|
def get_coffee_from_slug(slug: str) -> dict:
|
|
coffee_url = ROASTDB_URL.format(slug=slug)
|
|
headers = {"User-Agent": "Vrobbler 0.11.12"}
|
|
response = requests.get(coffee_url, headers=headers)
|
|
coffee_dict = {"slug": slug}
|
|
|
|
if response.status_code != 200:
|
|
logger.warning("Bad response from roastdb.com", extra={"response": response})
|
|
return coffee_dict
|
|
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
|
|
for script in soup.find_all("script", type="application/ld+json"):
|
|
try:
|
|
data = json.loads(script.string)
|
|
if data.get("@type") == "Product":
|
|
coffee_dict["title"] = data.get("name", "")
|
|
coffee_dict["description"] = data.get("description", "")
|
|
coffee_dict["image_url"] = data.get("image", "")
|
|
brand = data.get("brand", {})
|
|
coffee_dict["roaster__name"] = brand.get("name", "")
|
|
origin = data.get("countryOfOrigin", {})
|
|
coffee_dict["origin"] = origin.get("name", "")
|
|
break
|
|
except (json.JSONDecodeError, AttributeError):
|
|
continue
|
|
|
|
if not coffee_dict.get("title"):
|
|
bean_data = _extract_initial_bean(soup)
|
|
if bean_data:
|
|
coffee_dict["title"] = bean_data.get("name", "")
|
|
coffee_dict["description"] = bean_data.get("description", "")
|
|
coffee_dict["origin"] = bean_data.get("origin_country", "")
|
|
coffee_dict["roaster__name"] = bean_data.get("roaster_name", "")
|
|
image_urls = bean_data.get("image_urls", [])
|
|
if image_urls:
|
|
coffee_dict["image_url"] = image_urls[0]
|
|
|
|
return coffee_dict
|
|
|
|
|
|
def _extract_initial_bean(soup) -> dict | None:
|
|
for script in soup.find_all("script"):
|
|
if script.string and "initialBean" in (script.string or ""):
|
|
match = re.search(r'"initialBean"\s*:\s*(\{.*?\})\s*\}', script.string)
|
|
if match:
|
|
try:
|
|
return json.loads(match.group(1))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def search_coffee_on_roastdb(query: str) -> list[dict]:
|
|
search_url = ROASTDB_SEARCH_URL.format(query=query)
|
|
headers = {"User-Agent": "Vrobbler 0.11.12"}
|
|
response = requests.get(search_url, headers=headers)
|
|
|
|
if response.status_code != 200:
|
|
logger.warning(
|
|
"Bad response from roastdb.com search", extra={"response": response}
|
|
)
|
|
return []
|
|
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
results = []
|
|
|
|
for link in soup.find_all("a", href=True):
|
|
href = link["href"]
|
|
if href.startswith("/beans/"):
|
|
slug = href.split("/beans/", 1)[1].rstrip("/")
|
|
title = link.get_text(strip=True)
|
|
if slug and title:
|
|
results.append({"slug": slug, "title": title})
|
|
|
|
return results
|