Files
eve_structure/webapp/services/cookbook.py
2025-08-27 18:55:45 +02:00

122 lines
4.2 KiB
Python

from __future__ import annotations
from typing import Any, Dict, List, Tuple, Union
from urllib.parse import urljoin
import requests
from requests.adapters import HTTPAdapter, Retry
__all__ = ["call_cookbook_build_cost", "call_eve_ref_materials"]
_SESSION = requests.Session()
_RETRY = Retry(total=3, connect=3, read=3, backoff_factor=0.3,
status_forcelist=(429,500,502,503,504),
allowed_methods=frozenset(["GET","POST"]))
_ADAPTER = HTTPAdapter(max_retries=_RETRY)
_SESSION.mount("http://", _ADAPTER)
_SESSION.mount("https://", _ADAPTER)
def call_cookbook_build_cost(
blueprint_type_ids: Union[List[str], List[int]],
*,
quantity: int = 1,
price_mode: str = "buy",
additional_costs: int = 0,
base_me: int = 0,
components_me: int = 0,
system: str = "Jita",
facility_tax: int = 0,
industry_structure_type: str = "Station",
industry_rig: str = "None",
reaction_structure_type: str = "Athanor",
reaction_rig: str = "None",
reaction_flag: str = "",
blueprint_version: str = "tq",
timeout: Tuple[float, float] = (5.0, 25.0),
) -> Dict[str, Any]:
base = "https://evecookbook.com/api/"
url = urljoin(base, "buildCost")
params: List[Tuple[str, Any]] = []
for bid in blueprint_type_ids:
params.append(("blueprintTypeId", str(bid)))
params.extend([
("quantity", str(int(quantity))),
("priceMode", price_mode),
("additionalCosts", str(int(additional_costs))),
("baseMe", str(int(base_me))),
("componentsMe", str(int(components_me))),
("system", system),
("facilityTax", str(int(facility_tax))),
("industryStructureType", industry_structure_type),
("industryRig", industry_rig),
("reactionStructureType", reaction_structure_type),
("reactionRig", reaction_rig),
("reactionFlag", reaction_flag),
("blueprintVersion", blueprint_version),
])
headers = {"Accept": "application/json"}
resp = _SESSION.get(url, params=params, headers=headers, timeout=timeout)
resp.raise_for_status()
try:
return resp.json() if resp.content else {}
except Exception:
return {"raw": resp.text}
EVEREF_COST_URL = "https://api.everef.net/v1/industry/cost"
REFDATA_TYPE_URL = "https://ref-data.everef.net/types/{type_id}"
_TYPE_NAME_CACHE: Dict[int, str] = {}
def _type_name(type_id: int) -> str:
if type_id in _TYPE_NAME_CACHE:
return _TYPE_NAME_CACHE[type_id]
try:
r = _SESSION.get(REFDATA_TYPE_URL.format(type_id=type_id), timeout=10)
r.raise_for_status()
data = r.json()
name = (data.get("name", {}) or {}).get("en") or data.get("name") or str(type_id)
except Exception:
name = str(type_id)
_TYPE_NAME_CACHE[type_id] = name
return name
def call_eve_ref_materials(params: Dict[str, Any]) -> Dict[str, Any]:
bp_id = int(params.get("blueprintTypeId"))
runs = int(params.get("quantity", 1))
me = int(params.get("baseMe", 0))
te = int(params.get("baseTe", 0)) if params.get("baseTe") is not None else 0
structure_map = {"Station": None, "Raitaru": 35825, "Azbel": 35826, "Sotiyo": 35827}
st_name = params.get("industryStructureType", "Station")
st_id = structure_map.get(st_name)
q: Dict[str, Any] = {"blueprint_id": bp_id, "runs": runs, "me": me, "te": te}
if st_id:
q["structure_type_id"] = st_id
r = _SESSION.get(EVEREF_COST_URL, params=q, timeout=20)
r.raise_for_status()
payload = r.json()
manuf = (payload.get("manufacturing") or {})
if not manuf:
return {"materials": []}
first_key = next(iter(manuf.keys()))
mats = manuf[first_key].get("materials", {}) or {}
out: List[Dict[str, Any]] = []
for _, row in mats.items():
try:
tid = int(row.get("type_id"))
except Exception:
continue
out.append({
"type_id": tid,
"name": _type_name(tid),
"quantity": row.get("quantity"),
"cost_per_unit": row.get("cost_per_unit"),
"cost": row.get("cost"),
})
out.sort(key=lambda x: (x.get("cost") or 0), reverse=True)
return {"materials": out}