86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
from collections import defaultdict
|
|
|
|
from fastapi import FastAPI, Depends, Path, Query
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from typing import List, Annotated, Tuple, Literal, Dict, TypedDict
|
|
from sqlmodel import SQLModel, Session, select
|
|
|
|
from eveal.schemas import Evepraisal, PriceReprocess
|
|
from eveal.database import engine, get_session
|
|
from eveal import models_sde, esi
|
|
|
|
SQLModel.metadata.create_all(engine) # use alembic?
|
|
|
|
app = FastAPI()
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
|
|
@app.post("/reprocess/")
|
|
async def reprocess(ep_items: Evepraisal, ep_mat: Evepraisal, efficiency: float = .55, db: Session = Depends(get_session)) -> List[PriceReprocess]:
|
|
matprices = {item.typeID: {'sell': item.prices.sell.min, 'buy': item.prices.buy.max} for item in ep_mat.items}
|
|
|
|
item_reprocess: List[PriceReprocess] = []
|
|
for rawitem in ep_items.items:
|
|
item = db.get(models_sde.SDEType, rawitem.typeID)
|
|
buy_reprocess = sell_reprocess = 0.0
|
|
for mat in item.materials.all():
|
|
buy_reprocess += matprices[mat.material_type_id]['buy'] * mat.quantity/mat.type.portionSize * efficiency
|
|
sell_reprocess += matprices[mat.material_type_id]['sell'] * mat.quantity/mat.type.portionSize * efficiency
|
|
item_reprocess.append(PriceReprocess(typeID=rawitem.typeID,
|
|
buy=rawitem.prices.buy.max,
|
|
sell=rawitem.prices.sell.min,
|
|
buy_reprocess=buy_reprocess,
|
|
sell_reprocess=sell_reprocess,
|
|
name=rawitem.name
|
|
))
|
|
return item_reprocess
|
|
|
|
|
|
@app.get("/sde/types/{sde_type}/")
|
|
async def sde_types(sde_type: int | str, db: Session = Depends(get_session)) -> models_sde.SDEType:
|
|
try:
|
|
item = db.get(models_sde.SDEType, int(sde_type))
|
|
except ValueError:
|
|
item = db.query(models_sde.SDEType).filter(models_sde.SDEType.name == sde_type).one()
|
|
return item
|
|
|
|
|
|
@app.post("/sde/types/search")
|
|
async def sde_types_search(query: List[Dict[Literal["id", "name", "name_i", "marketgroup_id", "metagroup_id"], int | str]], db: Session = Depends(get_session)) -> List[models_sde.SDEType]:
|
|
items = []
|
|
for q in query:
|
|
qitems = db.query(models_sde.SDEType)
|
|
for k in q.keys():
|
|
if k.endswith("_i"):
|
|
qitems = qitems.filter(getattr(models_sde.SDEType, k[:-2]).contains(q[k]))
|
|
else:
|
|
qitems = qitems.filter(getattr(models_sde.SDEType, k) == q[k])
|
|
items.extend(qitems.all())
|
|
return items
|
|
|
|
|
|
@app.get("/esi/types/{sde_type}/market/{region_id}/")
|
|
async def sde_types_market(sde_type: int | str, region_id: int | str, db: Session = Depends(get_session)):
|
|
"""Get market orders for a type in a region. example: /esi/types/22291/market/10000002/"""
|
|
"""TODO: use ESIMarketOrder"""
|
|
return list(esi.esi_client.Market.get_markets_region_id_orders(order_type="all", type_id=sde_type, region_id=region_id))
|
|
|
|
|
|
@app.get("/_tools/get_all_mat")
|
|
async def _get_all_mat(db: Session = Depends(get_session)):
|
|
materials = db.query(models_sde.SDETypeMaterial).filter(models_sde.SDETypeMaterial.type.has(models_sde.SDEType.published == True)).all()
|
|
allmat = set()
|
|
for mat in materials:
|
|
allmat.add(mat.material_type.name)
|
|
return allmat
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run("eveal.main:app", host="0.0.0.0", port=8000, reload=True)
|