diff options
Diffstat (limited to 'backend/src/market.py')
| -rw-r--r-- | backend/src/market.py | 203 |
1 files changed, 203 insertions, 0 deletions
diff --git a/backend/src/market.py b/backend/src/market.py new file mode 100644 index 0000000..311b181 --- /dev/null +++ b/backend/src/market.py @@ -0,0 +1,203 @@ +import json +import logging +from dataclasses import dataclass, asdict +from datetime import datetime +from pathlib import Path +from typing import List, Dict + +from loader import load_items +from esi.api import get_region_orders + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +ORDERS_CACHE = Path(__file__).resolve().parent.parent / "cache" / "orders_snapshot_orders.json" + +REGION_IDS = [ + 10000002, # The Forge + 10000016, # Lonetrek + 10000030, # Heimatar + 10000033, # The Citadel + 10000032, # Sinq Laison + 10000042, # Metropolis + 10000064, # Essence + 10000037, # Everyshore + 10000043, # Domain + 10000036, # Devoid + 10000052, # Kador + 10000067, # Genesis + 10000068, # Verge Vendor + 10000001, # Derelik + 10000020, # Tash-Murkon +] + +@dataclass +class Order: + order_id: int + region_id: int + location_id: int + type_id: int + is_buy_order: bool + price: float + issued: datetime + volume_remain: int + item_name: str = '' + item_cargo_volume: float = 1.0 + +class Market: + """ + Manages market orders, including fetching, updating, and caching. + """ + + def __init__(self): + self.orders: List[Order] = [] + self.expires_times: Dict[int, datetime] = {} + + def get_orders(self) -> List[Order]: + """Retrieves the current list of market orders.""" + self.update_orders() + return self.orders + + def update_orders(self) -> List[int]: + """ + Updates market orders by fetching new data for expired regions. + + Returns: + A list of region IDs that were updated. + """ + updated_orders: List[Order] = [] + regions_updated: List[int] = [] + + for region_id in REGION_IDS: + if self.check_expired(region_id): + logger.info(f"Downloading orders for region {region_id}") + orders = self.download_orders(region_id) + updated_orders.extend(orders) + regions_updated.append(region_id) + else: + # Include existing orders for this region + updated_orders.extend( + [order for order in self.orders if order.region_id == region_id] + ) + + self.orders = updated_orders + return regions_updated + + def check_expired(self, region_id: int) -> bool: + """ + Checks if the cached orders for a region have expired. + + Args: + region_id: The ID of the region to check. + + Returns: + True if the orders have expired or are not cached, False otherwise. + """ + current_time = datetime.utcnow() + expiry_time = self.expires_times.get(region_id) + return expiry_time is None or current_time >= expiry_time + + def download_orders(self, region_id: int) -> List[Order]: + """ + Downloads orders for a given region and updates the expiry time. + + Args: + region_id: The ID of the region to download orders for. + + Returns: + A list of Order instances for the region. + """ + raw_orders, expiry = get_region_orders(region_id) + expiry_time = datetime.strptime(expiry, "%a, %d %b %Y %H:%M:%S %Z") + self.expires_times[region_id] = expiry_time + + orders = self.add_item_info_to_orders(raw_orders) + return orders + + def add_item_info_to_orders(self, raw_orders: List[Dict]) -> List[Order]: + """ + Enhances raw order data with item information and converts them into Order instances. + + Args: + raw_orders: A list of raw order dictionaries from the API. + + Returns: + A list of Order instances with additional item information. + """ + items = load_items() + orders: List[Order] = [] + + for raw_order in raw_orders: + item_info = items.get(raw_order['type_id'], {}) + order = Order( + order_id=raw_order['order_id'], + region_id=raw_order['region_id'], + location_id=raw_order['location_id'], + type_id=raw_order['type_id'], + is_buy_order=raw_order['is_buy_order'], + price=raw_order['price'], + issued=datetime.strptime(raw_order['issued'], "%Y-%m-%dT%H:%M:%S%z"), + volume_remain=raw_order['volume_remain'], + item_name=item_info.get('type_name', f"Not found: {raw_order['type_id']}"), + item_cargo_volume=item_info.get('volume', 1.0), + ) + orders.append(order) + + return orders + + def cache_orders(self) -> None: + """Caches the current list of orders to a JSON file.""" + try: + with ORDERS_CACHE.open("w", encoding="utf-8") as f: + # Serialize orders to dictionaries and handle datetime serialization + orders_data = [self._serialize_order(order) for order in self.orders] + json.dump(orders_data, f, ensure_ascii=False, indent=4) + logger.info("Orders cached successfully.") + except IOError as e: + logger.error(f"Error caching orders: {e}") + + def load_orders(self) -> List[Order]: + """ + Loads orders from the cache file. + + Returns: + A list of Order instances loaded from the cache. + """ + try: + with ORDERS_CACHE.open("r", encoding="utf-8") as f: + orders_data = json.load(f) + self.orders = [self._deserialize_order(data) for data in orders_data] + logger.info("Orders loaded from cache successfully.") + except (IOError, json.JSONDecodeError) as e: + logger.error(f"Error loading orders from cache: {e}") + self.orders = [] + + return self.orders + + def _serialize_order(self, order: Order) -> Dict: + """ + Serializes an Order instance to a dictionary suitable for JSON serialization. + + Args: + order: The Order instance to serialize. + + Returns: + A dictionary representation of the Order. + """ + order_dict = asdict(order) + # Convert datetime objects to ISO format strings + order_dict['issued'] = order.issued.isoformat() + return order_dict + + def _deserialize_order(self, data: Dict) -> Order: + """ + Deserializes a dictionary into an Order instance. + + Args: + data: The dictionary representation of an Order. + + Returns: + An Order instance. + """ + data['issued'] = datetime.fromisoformat(data['issued']) + return Order(**data) |