Mateusz Gruszczyński
71 lines
3.2 KiB
Python
71 lines
3.2 KiB
Python
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from flask import current_app
|
|
from .http import get_json
|
|
|
|
PRODUCT_IDS = {"PB95": 41, "PB98": 42, "DIESEL": 43}
|
|
POLISH_REGIONS = (
|
|
"dolnośląskie", "kujawsko-pomorskie", "lubelskie", "lubuskie", "łódzkie",
|
|
"małopolskie", "mazowieckie", "opolskie", "podkarpackie", "podlaskie",
|
|
"pomorskie", "śląskie", "świętokrzyskie", "warmińsko-mazurskie",
|
|
"wielkopolskie", "zachodnio-pomorskie",
|
|
)
|
|
|
|
|
|
def _rows(url, params=None):
|
|
data = get_json(url, params=params, timeout_key="ORLEN_TIMEOUT_SECONDS")
|
|
if not isinstance(data, list):
|
|
raise ValueError("Nieprawidłowa odpowiedź API Orlen")
|
|
return data
|
|
|
|
|
|
def fetch_orlen_range(fuel_type, date_from, date_to, region="mazowieckie"):
|
|
fuel_type = fuel_type.upper()
|
|
if fuel_type == "LPG":
|
|
result, seen = [], set()
|
|
current = date_from
|
|
while current <= date_to:
|
|
rows = _rows(current_app.config["ORLEN_LPG_BY_DATE_URL"], {"date": current.isoformat()})
|
|
for row in rows:
|
|
row_region = (row.get("locationName") or "").strip().lower()
|
|
if region and str(region).lower() not in ("all", "*") and row_region != str(region).lower():
|
|
continue
|
|
effective = date.fromisoformat(row["effectiveDate"][:10])
|
|
if not (date_from <= effective <= date_to):
|
|
continue
|
|
key = (effective, row_region)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
value = Decimal(str(row["value"]))
|
|
result.append({"fuel_type": "LPG", "effective_date": effective, "price_per_liter": value,
|
|
"raw_value": value, "region": row_region, "product_name": row.get("productName") or "LPG",
|
|
"source": f"Orlen LPG ByDate, {row_region}"})
|
|
current += timedelta(days=1)
|
|
return result
|
|
|
|
product_id = PRODUCT_IDS.get(fuel_type)
|
|
if not product_id:
|
|
raise ValueError("Nieobsługiwany rodzaj paliwa")
|
|
rows = _rows(current_app.config["ORLEN_WHOLESALE_URL"], {
|
|
"productId": product_id, "from": date_from.isoformat(), "to": date_to.isoformat()
|
|
})
|
|
result = []
|
|
for row in rows:
|
|
effective = date.fromisoformat(row["effectiveDate"][:10])
|
|
raw = Decimal(str(row["value"]))
|
|
result.append({"fuel_type": fuel_type, "effective_date": effective, "price_per_liter": raw / Decimal("1000"),
|
|
"raw_value": raw, "region": "", "product_name": row.get("productName") or fuel_type,
|
|
"source": f"Orlen {row.get('productName') or fuel_type}, netto/m³ → zł/l"})
|
|
return result
|
|
|
|
|
|
def fetch_orlen_price(fuel_type, day=None, region="mazowieckie"):
|
|
day = day or date.today()
|
|
rows = fetch_orlen_range(fuel_type, day.replace(day=1), day, region)
|
|
candidates = [row for row in rows if row["effective_date"] <= day]
|
|
if not candidates:
|
|
raise ValueError("Brak ceny Orlen dla wybranej daty")
|
|
row = max(candidates, key=lambda item: item["effective_date"])
|
|
return {"price_per_liter": float(row["price_per_liter"]), "date": row["effective_date"].isoformat(), "source": row["source"]}
|