482 lines
20 KiB
Python
482 lines
20 KiB
Python
"""
|
||
Utility functions to fetch auction items (lots) from different providers.
|
||
|
||
This module defines three functions that make HTTP calls to the public APIs of
|
||
Troostwijk Auctions (TWK), AuctionPort (AP) and Online Veilingmeester (OVM)
|
||
and normalises their responses into Python dictionaries. Each function
|
||
returns a list of dictionaries where each dictionary represents an
|
||
individual lot and includes standardised keys: ``title``, ``description``,
|
||
``bids`` (the number of bids if available), ``current_bid`` (current price
|
||
and currency if available), ``image_url`` and ``end_time``.
|
||
|
||
The implementations rely on the ``requests`` library for HTTP transport
|
||
and include basic error handling. They raise ``requests.HTTPError``
|
||
when the remote server responds with a non‑200 status code.
|
||
|
||
Note: the APIs these functions call are subject to change. Endpoints and
|
||
field names may differ depending on the auction status or provider version.
|
||
These functions are intended as a starting point for integrating with
|
||
multiple auction platforms; you may need to adjust query parameters,
|
||
header values or JSON field names if the provider updates their API.
|
||
|
||
Examples
|
||
--------
|
||
```
|
||
from auction_items import get_items_twk, get_items_ap, get_items_ovm
|
||
|
||
# Troostwijk Auctions (TWK): pass the visible auction identifier
|
||
lots = get_items_twk(display_id="35563")
|
||
for lot in lots:
|
||
print(lot['lot_number'], lot['title'], lot['current_bid'])
|
||
|
||
# AuctionPort (AP): pass the auction ID from the AuctionPort website
|
||
ap_lots = get_items_ap(auction_id=1323)
|
||
|
||
# Online Veilingmeester (OVM): the country code is required to build the
|
||
# endpoint path (e.g. ``'nederland'`` or ``'belgie'``) along with the
|
||
# numeric auction ID.
|
||
ovm_lots = get_items_ovm(country="nederland", auction_id=7713)
|
||
```
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from typing import Dict, List, Optional
|
||
|
||
import requests
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def get_items_twk(
|
||
display_id: str,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 200,
|
||
locale: str = "nl",
|
||
sort_by: str = "LOT_NUMBER_ASC",
|
||
platform: str = "TWK",
|
||
request_session: Optional[requests.Session] = None,
|
||
) -> List[Dict[str, Optional[str]]]:
|
||
"""Fetch lots (items) for a Troostwijk auction using the current GraphQL API.
|
||
|
||
Troostwijk’s public GraphQL API exposes a ``lotsByAuctionDisplayId`` query
|
||
that returns the lots belonging to a specific auction. The input
|
||
``LotsByAuctionDisplayIdInput`` requires an ``auctionDisplayId``,
|
||
``locale``, ``pageNumber``, ``pageSize`` and a non‑null ``sortBy``
|
||
parameter【566587544277629†screenshot】. The platform is provided as the
|
||
``Platform`` enum value and must be set to ``"TWK"`` for Troostwijk
|
||
auctions【622367855745945†screenshot】.
|
||
|
||
Each result in the ``results`` list is a ``ListingLot`` object that
|
||
includes fields such as ``number`` (the lot/kavel number), ``title``,
|
||
``description``, ``bidsCount``, ``currentBidAmount`` (Money object with
|
||
``cents`` and ``currency``), ``endDate`` (a scalar ``TbaDate``) and
|
||
``image`` with a ``url`` property【819766814773156†screenshot】. This
|
||
function builds a query that selects these fields and converts them
|
||
into a normalised Python dictionary per lot.
|
||
|
||
Parameters
|
||
----------
|
||
display_id: str
|
||
The auction display identifier (e.g., the number visible in the
|
||
auction URL).
|
||
page: int, optional
|
||
The page number to retrieve (1‑based). Defaults to 1.
|
||
page_size: int, optional
|
||
The number of lots per page. Defaults to 200.
|
||
locale: str, optional
|
||
Language code for the returned content (default ``"nl"``).
|
||
sort_by: str, optional
|
||
Sorting option from the ``LotsSortingOption`` enum (for example
|
||
``"LOT_NUMBER_ASC"`` or ``"END_DATE_DESC"``)【566587544277629†screenshot】.
|
||
Defaults to ``"LOT_NUMBER_ASC"``.
|
||
platform: str, optional
|
||
Platform code as defined in the ``Platform`` enum. Use ``"TWK"`` for
|
||
Troostwijk auctions【622367855745945†screenshot】.
|
||
request_session: Optional[requests.Session], optional
|
||
Optional session object for connection pooling.
|
||
|
||
Returns
|
||
-------
|
||
List[Dict[str, Optional[str]]]
|
||
A list of dictionaries representing lots. Each dictionary
|
||
contains the keys ``lot_number``, ``title``, ``description``,
|
||
``bids`` (number of bids), ``current_bid`` (amount in major units
|
||
and currency), ``image_url`` and ``end_time``.
|
||
|
||
Raises
|
||
------
|
||
requests.HTTPError
|
||
If the HTTP call does not return a 2xx status code.
|
||
Exception
|
||
If GraphQL returns errors or the response structure is unexpected.
|
||
"""
|
||
session = request_session or requests.Session()
|
||
url = "https://storefront.tbauctions.com/storefront/graphql"
|
||
# GraphQL query for lots by auction display ID. We request the
|
||
# ListingLots wrapper (hasNext, pageNumber etc.) and select relevant
|
||
# fields from each ListingLot【819766814773156†screenshot】. The money
|
||
# amounts are represented in cents; we convert them to major units.
|
||
graphql_query = """
|
||
query LotsByAuctionDisplayId($request: LotsByAuctionDisplayIdInput!, $platform: Platform!) {
|
||
lotsByAuctionDisplayId(request: $request, platform: $platform) {
|
||
pageNumber
|
||
pageSize
|
||
totalSize
|
||
hasNext
|
||
results {
|
||
number
|
||
title
|
||
description
|
||
bidsCount
|
||
currentBidAmount {
|
||
cents
|
||
currency
|
||
}
|
||
endDate
|
||
image {
|
||
url
|
||
}
|
||
}
|
||
}
|
||
}
|
||
"""
|
||
variables = {
|
||
"request": {
|
||
"auctionDisplayId": str(display_id),
|
||
"locale": locale,
|
||
"pageNumber": page,
|
||
"pageSize": page_size,
|
||
"sortBy": sort_by,
|
||
},
|
||
"platform": platform,
|
||
}
|
||
headers = {
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
}
|
||
# Use GET request because the GraphQL endpoint does not accept POST
|
||
# from unauthenticated clients. We encode the query and variables as
|
||
# query parameters.
|
||
try:
|
||
response = session.get(
|
||
url,
|
||
params={"query": graphql_query, "variables": json.dumps(variables)},
|
||
headers=headers,
|
||
timeout=30,
|
||
)
|
||
except Exception as exc:
|
||
logger.error("Error contacting Troostwijk GraphQL endpoint: %s", exc)
|
||
raise
|
||
try:
|
||
response.raise_for_status()
|
||
except requests.HTTPError:
|
||
logger.error("Troostwijk API returned status %s: %s", response.status_code, response.text)
|
||
raise
|
||
data = response.json()
|
||
if "errors" in data and data["errors"]:
|
||
message = data["errors"]
|
||
logger.error("GraphQL returned errors: %s", message)
|
||
raise Exception(f"GraphQL returned errors: {message}")
|
||
# Extract the list of lots
|
||
try:
|
||
lot_items = data["data"]["lotsByAuctionDisplayId"]["results"]
|
||
except (KeyError, TypeError) as e:
|
||
logger.error("Unexpected response structure from Troostwijk API: %s", data)
|
||
raise Exception(f"Unexpected response structure: {e}")
|
||
lots: List[Dict[str, Optional[str]]] = []
|
||
for item in lot_items:
|
||
lot_number = item.get("number")
|
||
title = item.get("title")
|
||
description = item.get("description")
|
||
bids = item.get("bidsCount")
|
||
bid_amount = item.get("currentBidAmount") or {}
|
||
# Convert cents to major units if available
|
||
current_bid = None
|
||
if bid_amount and isinstance(bid_amount, dict):
|
||
cents = bid_amount.get("cents")
|
||
currency = bid_amount.get("currency")
|
||
if cents is not None:
|
||
current_bid = {
|
||
"amount": cents / 100.0,
|
||
"currency": currency,
|
||
}
|
||
end_time = item.get("endDate")
|
||
image_obj = item.get("image") or {}
|
||
image_url = image_obj.get("url") if isinstance(image_obj, dict) else None
|
||
lots.append(
|
||
{
|
||
"lot_number": lot_number,
|
||
"title": title,
|
||
"description": description,
|
||
"bids": bids,
|
||
"current_bid": current_bid,
|
||
"image_url": image_url,
|
||
"end_time": end_time,
|
||
}
|
||
)
|
||
return lots
|
||
|
||
|
||
def get_items_ap(
|
||
auction_id: int,
|
||
*,
|
||
request_session: Optional[requests.Session] = None,
|
||
) -> List[Dict[str, Optional[str]]]:
|
||
"""Retrieve items (lots) from an AuctionPort auction.
|
||
|
||
AuctionPort operates a JSON API on ``https://api.auctionport.be``. While
|
||
official documentation for the lot endpoints is scarce, community code
|
||
suggests that auctions can be fetched via ``/auctions/small``【461010206788258†L10-L39】. The
|
||
corresponding lot information appears to reside under an
|
||
``/auctions/{id}/lots`` or ``/lots?auctionId={id}`` endpoint (the
|
||
platform uses XML internally for some pages as observed when visiting
|
||
``/auctions/{id}/lots`` in a browser). This function attempts to call
|
||
these endpoints in order and parse their JSON responses. If the
|
||
response is not JSON, it falls back to a simple text scrape looking
|
||
for lot numbers, titles, descriptions and current bid amounts.
|
||
|
||
Parameters
|
||
----------
|
||
auction_id: int
|
||
The numeric identifier of the auction on AuctionPort.
|
||
request_session: Optional[requests.Session], optional
|
||
An existing requests session.
|
||
|
||
Returns
|
||
-------
|
||
List[Dict[str, Optional[str]]]
|
||
A list of lot dictionaries with the keys ``lot_number``, ``title``,
|
||
``description``, ``bids`` (if available), ``current_bid`` (amount and
|
||
currency if provided), ``image_url`` and ``end_time``. If no lots
|
||
could be parsed, an empty list is returned.
|
||
|
||
Raises
|
||
------
|
||
requests.HTTPError
|
||
If both endpoint attempts return non‑200 responses.
|
||
"""
|
||
session = request_session or requests.Session()
|
||
# Candidate endpoints for AuctionPort lots. The first URL follows the
|
||
# pattern used by the AuctionPort website; the second is a query by
|
||
# parameter. Additional endpoints can be added if discovered.
|
||
url_candidates = [
|
||
f"https://api.auctionport.be/auctions/{auction_id}/lots",
|
||
f"https://api.auctionport.be/lots?auctionId={auction_id}",
|
||
]
|
||
last_error: Optional[Exception] = None
|
||
for url in url_candidates:
|
||
try:
|
||
response = session.get(url, headers={"Accept": "application/json"}, timeout=30)
|
||
except Exception as exc:
|
||
# Capture connection errors and continue with the next endpoint
|
||
last_error = exc
|
||
continue
|
||
if response.status_code == 404:
|
||
# Try the next candidate
|
||
continue
|
||
if response.status_code >= 400:
|
||
last_error = requests.HTTPError(
|
||
f"AuctionPort API error {response.status_code} for {url}",
|
||
response=response,
|
||
)
|
||
continue
|
||
# If the response is OK, attempt to parse JSON
|
||
try:
|
||
data = response.json()
|
||
except json.JSONDecodeError:
|
||
# Not JSON: fallback to naive parsing of plain text/XML. AuctionPort
|
||
# sometimes returns XML for lots pages. We'll attempt to extract
|
||
# structured information using simple patterns.
|
||
text = response.text
|
||
lots: List[Dict[str, Optional[str]]] = []
|
||
# Split by <div class="lot"> like markers (not guaranteed). In the
|
||
# absence of a stable API specification, heuristics must be used.
|
||
# Here we use a very simple split on "Lot " followed by a number.
|
||
import re
|
||
|
||
pattern = re.compile(r"\bLot\s+(\d+)\b", re.IGNORECASE)
|
||
for match in pattern.finditer(text):
|
||
lot_number = match.group(1)
|
||
# Attempt to extract the title and description following the
|
||
# lot number. This heuristic looks for a line break or
|
||
# sentence after the lot label; adjust as necessary.
|
||
start = match.end()
|
||
segment = text[start:start + 300] # arbitrary slice length
|
||
# Title is the first sentence or line
|
||
title_match = re.search(r"[:\-]\s*(.*?)\.(?=\s|<)", segment)
|
||
title = title_match.group(1).strip() if title_match else segment.strip()
|
||
lots.append({
|
||
"lot_number": lot_number,
|
||
"title": title,
|
||
"description": None,
|
||
"bids": None,
|
||
"current_bid": None,
|
||
"image_url": None,
|
||
"end_time": None,
|
||
})
|
||
if lots:
|
||
return lots
|
||
else:
|
||
# If no lots were extracted, continue to the next candidate
|
||
last_error = Exception("Unable to parse AuctionPort lots from non‑JSON response")
|
||
continue
|
||
# If JSON parsing succeeded, inspect the structure. Some endpoints
|
||
# return a top‑level object with a ``data`` field containing a list.
|
||
lots: List[Dict[str, Optional[str]]] = []
|
||
# Attempt to locate the list of lots: it might be in ``data``, ``lots`` or
|
||
# another property.
|
||
candidate_keys = ["lots", "data", "items"]
|
||
lot_list: Optional[List[Dict[str, any]]] = None
|
||
for key in candidate_keys:
|
||
if isinstance(data, dict) and isinstance(data.get(key), list):
|
||
lot_list = data[key]
|
||
break
|
||
# If the response is a list itself (not a dict), treat it as the lot list
|
||
if lot_list is None and isinstance(data, list):
|
||
lot_list = data
|
||
if lot_list is None:
|
||
# Unknown structure; return empty list
|
||
return []
|
||
for item in lot_list:
|
||
# Map fields according to common names; adjust if your endpoint
|
||
# uses different property names. Use dict.get to avoid KeyError.
|
||
lot_number = item.get("lotNumber") or item.get("lotnumber") or item.get("id")
|
||
title = item.get("title") or item.get("naam")
|
||
description = item.get("description") or item.get("beschrijving")
|
||
bids = item.get("numberOfBids") or item.get("bidCount")
|
||
# Determine current bid: AuctionPort might provide ``currentBid`` or
|
||
# ``currentPrice`` as an object or numeric value.
|
||
current_bid_obj = item.get("currentBid") or item.get("currentPrice")
|
||
current_bid: Optional[Dict[str, any]] = None
|
||
if isinstance(current_bid_obj, dict):
|
||
current_bid = {
|
||
"amount": current_bid_obj.get("amount"),
|
||
"currency": current_bid_obj.get("currency"),
|
||
}
|
||
elif current_bid_obj is not None:
|
||
current_bid = {"amount": current_bid_obj, "currency": None}
|
||
# Image
|
||
image_url = None
|
||
if isinstance(item.get("images"), list) and item["images"]:
|
||
image_url = item["images"][0].get("url")
|
||
elif isinstance(item.get("image"), str):
|
||
image_url = item.get("image")
|
||
# End time
|
||
end_time = item.get("endDateISO") or item.get("closingDateISO") or item.get("closingDate")
|
||
lots.append(
|
||
{
|
||
"lot_number": lot_number,
|
||
"title": title,
|
||
"description": description,
|
||
"bids": bids,
|
||
"current_bid": current_bid,
|
||
"image_url": image_url,
|
||
"end_time": end_time,
|
||
}
|
||
)
|
||
return lots
|
||
|
||
# All candidates failed
|
||
if last_error:
|
||
raise last_error
|
||
raise requests.HTTPError(f"Could not fetch lots for AuctionPort auction {auction_id}")
|
||
|
||
|
||
def get_items_ovm(
|
||
country: str,
|
||
auction_id: int,
|
||
*,
|
||
request_session: Optional[requests.Session] = None,
|
||
) -> List[Dict[str, Optional[str]]]:
|
||
"""Fetch lots from an Online Veilingmeester auction.
|
||
|
||
Online Veilingmeester's REST API exposes auctions and their lots via
|
||
endpoints under ``https://onlineveilingmeester.nl/rest/nl``. The
|
||
AuctionViewer project's source code constructs lot URLs as
|
||
``{land}/veilingen/{id}/kavels``, where ``land`` is the lower‑cased
|
||
country name (e.g. ``nederland`` or ``belgie``)【366543684390870†L13-L50】.
|
||
Therefore, the full path for retrieving the lots of a specific auction
|
||
is ``https://onlineveilingmeester.nl/rest/nl/{country}/veilingen/{auction_id}/kavels``.
|
||
|
||
Parameters
|
||
----------
|
||
country: str
|
||
Lower‑case country name used in the API path (for example
|
||
``"nederland"`` or ``"belgie"``). The value should correspond to
|
||
the ``land`` property returned by the OVM auctions endpoint【366543684390870†L13-L50】.
|
||
auction_id: int
|
||
The numeric identifier of the auction.
|
||
request_session: Optional[requests.Session], optional
|
||
A ``requests`` session to reuse connections.
|
||
|
||
Returns
|
||
-------
|
||
List[Dict[str, Optional[str]]]
|
||
A list of lot dictionaries with keys ``lot_number``, ``title``,
|
||
``description``, ``bids``, ``current_bid`` (if available),
|
||
``image_url`` and ``end_time``. If the endpoint returns no items,
|
||
an empty list is returned.
|
||
|
||
Raises
|
||
------
|
||
requests.HTTPError
|
||
If the HTTP call returns a non‑200 response.
|
||
Exception
|
||
If the response cannot be decoded as JSON.
|
||
"""
|
||
session = request_session or requests.Session()
|
||
base_url = "https://onlineveilingmeester.nl/rest/nl"
|
||
url = f"{base_url}/{country}/veilingen/{auction_id}/kavels"
|
||
response = session.get(url, headers={"Accept": "application/json"}, timeout=30)
|
||
try:
|
||
response.raise_for_status()
|
||
except requests.HTTPError:
|
||
logger.error("OVM API returned status %s: %s", response.status_code, response.text)
|
||
raise
|
||
# Parse the JSON body; expect a list of lots
|
||
data = response.json()
|
||
lots: List[Dict[str, Optional[str]]] = []
|
||
# The response may be a dictionary containing a ``kavels`` key or a list
|
||
if isinstance(data, dict) and isinstance(data.get("kavels"), list):
|
||
lot_list = data["kavels"]
|
||
elif isinstance(data, list):
|
||
lot_list = data
|
||
else:
|
||
logger.error("Unexpected response structure from OVM API: %s", data)
|
||
raise Exception("Unexpected response structure for OVM lots")
|
||
for item in lot_list:
|
||
lot_number = item.get("kavelnummer") or item.get("lotNumber") or item.get("id")
|
||
title = item.get("naam") or item.get("title")
|
||
description = item.get("beschrijving") or item.get("description")
|
||
bids = item.get("aantalBiedingen") or item.get("numberOfBids")
|
||
# Current bid is nested in ``hoogsteBod`` or ``currentBid``
|
||
current_bid_obj = item.get("hoogsteBod") or item.get("currentBid")
|
||
current_bid: Optional[Dict[str, any]] = None
|
||
if isinstance(current_bid_obj, dict):
|
||
current_bid = {
|
||
"amount": current_bid_obj.get("bodBedrag") or current_bid_obj.get("amount"),
|
||
"currency": current_bid_obj.get("valuta") or current_bid_obj.get("currency"),
|
||
}
|
||
image_url = None
|
||
# OVM may provide a list of image URLs under ``afbeeldingen`` or ``images``
|
||
if isinstance(item.get("afbeeldingen"), list) and item["afbeeldingen"]:
|
||
image_url = item["afbeeldingen"][0]
|
||
elif isinstance(item.get("images"), list) and item["images"]:
|
||
image_url = item["images"][0].get("url") if isinstance(item["images"][0], dict) else item["images"][0]
|
||
end_time = item.get("eindDatumISO") or item.get("endDateISO") or item.get("eindDatum")
|
||
lots.append(
|
||
{
|
||
"lot_number": lot_number,
|
||
"title": title,
|
||
"description": description,
|
||
"bids": bids,
|
||
"current_bid": current_bid,
|
||
"image_url": image_url,
|
||
"end_time": end_time,
|
||
}
|
||
)
|
||
return lots
|