mirror of
https://github.com/EDCD/EDMarketConnector.git
synced 2026-10-22 17:22:09 +03:00
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""Export various CSV formats."""
|
|
|
|
import csv
|
|
from datetime import datetime
|
|
from enum import IntEnum
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from collections.abc import Mapping
|
|
from config import config
|
|
from edmc_data import commodity_bracketmap as bracketmap
|
|
|
|
|
|
class CommodityExportKind(IntEnum):
|
|
"""List the various available output forms."""
|
|
|
|
SEMICOLON = 0
|
|
CSV = 1
|
|
CSV_NEW = 2
|
|
TAB = 3
|
|
PIPE = 4
|
|
|
|
|
|
# BACKWARD COMPATIBILITY
|
|
COMMODITY_SEMICOLON = CommodityExportKind.SEMICOLON
|
|
COMMODITY_CSV = CommodityExportKind.CSV
|
|
COMMODITY_CSV_NEW = CommodityExportKind.CSV_NEW
|
|
COMMODITY_TAB = CommodityExportKind.TAB
|
|
COMMODITY_PIPE = CommodityExportKind.PIPE
|
|
mkt_out_types = ("CSV", "CSV_NEW", "TAB", "PIPE", "SEMICOLON")
|
|
|
|
|
|
def export(
|
|
data: Mapping[str, Any],
|
|
kind: CommodityExportKind = CommodityExportKind.SEMICOLON,
|
|
filename: Path | str | None = None,
|
|
) -> None:
|
|
"""Export commodity data from the given CAPI data.
|
|
|
|
:param data: CAPI data.
|
|
:param kind: The type of file to write.
|
|
:param filename: Filename to write to, or None for a standard format name.
|
|
"""
|
|
query_time_raw = config.get_int("querytime", default=None)
|
|
query_datetime = (
|
|
datetime.fromtimestamp(query_time_raw)
|
|
if query_time_raw is not None
|
|
else datetime.now()
|
|
)
|
|
|
|
match kind:
|
|
case CommodityExportKind.CSV | CommodityExportKind.CSV_NEW:
|
|
mkt_out_delim = ","
|
|
case CommodityExportKind.TAB:
|
|
mkt_out_delim = "\t"
|
|
case CommodityExportKind.PIPE:
|
|
mkt_out_delim = "|"
|
|
case _:
|
|
mkt_out_delim = ";"
|
|
|
|
sysname = data["lastSystem"]["name"].strip()
|
|
station = data["lastStarport"]["name"].strip()
|
|
|
|
if not filename:
|
|
timestamp = query_datetime.strftime("%Y-%m-%dT%H.%M.%S")
|
|
# Use .csv for comma-separated files; use .txt for other text formats
|
|
ext = "csv" if mkt_out_delim == "," else "txt"
|
|
filename = (
|
|
Path(config.get_str("outdir")) / f"{sysname}.{station}.{timestamp}.{ext}"
|
|
)
|
|
else:
|
|
filename = Path(filename)
|
|
|
|
header = ["System", "Station", "Commodity", "Sell", "Buy", "Demand",
|
|
"demandBracket", "Supply", "stockBracket"]
|
|
if kind != CommodityExportKind.CSV:
|
|
header.extend(["Average", "FDevID"])
|
|
header.append("Date")
|
|
|
|
with open(filename, "w", newline="", encoding="utf-8") as output_file:
|
|
writer = csv.writer(output_file, delimiter=mkt_out_delim)
|
|
writer.writerow(header)
|
|
|
|
for commodity in data["lastStarport"]["commodities"]:
|
|
|
|
def get_int_field(
|
|
field_key: str, condition_key: str | None = None
|
|
) -> int | str:
|
|
"""Return the integer value of a field from the commodity."""
|
|
check_key = condition_key or field_key
|
|
if (
|
|
commodity.get(check_key) is not None
|
|
and commodity.get(field_key) is not None
|
|
):
|
|
return int(commodity[field_key])
|
|
return ""
|
|
|
|
row = [
|
|
sysname,
|
|
station,
|
|
commodity["name"],
|
|
get_int_field("sellPrice"),
|
|
get_int_field("buyPrice"),
|
|
get_int_field("demand", "demandBracket"),
|
|
bracketmap.get(commodity.get("demandBracket"), ""),
|
|
get_int_field("stock", "stockBracket"),
|
|
bracketmap.get(commodity.get("stockBracket"), ""),
|
|
]
|
|
|
|
# newer export fields format
|
|
if kind != CommodityExportKind.CSV:
|
|
mean = get_int_field("meanPrice")
|
|
row.extend([mean, commodity["id"]])
|
|
|
|
row.append(data["timestamp"])
|
|
writer.writerow(row)
|