Spaces:
Running
Running
File size: 7,986 Bytes
eb0b2fd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 |
import requests
from datetime import datetime
import url64
import streamlit as st
from config import CONFIG
from excel.Formatter import RoutePlan
from .queries import CREATE_COLLECTION, DISABLE_COLLECTIONS_BY_DATE
class EstimatedContainer:
def __init__(self, amount: int, material_type_id: int, container_type_id: int, weight_kg: int | None):
self.amount = amount
self.material_type_id = material_type_id
self.container_type_id = container_type_id
self.weight_kg = weight_kg
def to_dict(self):
return {
"amount": self.amount,
"materialTypeId": self.material_type_id,
"containerTypeId": self.container_type_id,
"weightKg": self.weight_kg
}
class Collection:
def __init__(self, branch_id: int, driver_id: int, vehicle_id: int, estimated_containers: list[EstimatedContainer]):
self.driver_id = driver_id
self.vehicle_id = vehicle_id
self.branch_id = branch_id
self.estimated_containers = estimated_containers
class Route:
def __init__(self, id: int, ordered_collections: list[Collection]):
self.id = id
self.ordered_collections = ordered_collections
class Api:
_API_V1_BASE_URL = CONFIG.API_V1_BASE_URL
_API_V2_BASE_URL = CONFIG.API_V2_BASE_URL
_API_AUTH_URL = CONFIG.AUTH_URL
_DEFAULT_HEADERS = {'Content-Type': 'application/json; charset=utf-8' }
_DISABLE_COLLECTIONS_BY_DATE_QUERY = DISABLE_COLLECTIONS_BY_DATE
_CREATE_COLLECTION_QUERY = CREATE_COLLECTION
@staticmethod
def _get_auth_token(query_param_name="t") -> str:
# Obtener el token en crudo (en formato base64url) del query param 't'
query_params = st.experimental_get_query_params()
raw_token = query_params.get(query_param_name, [None])[0]
# Validar the el token esté presente
if raw_token is None or raw_token == "":
raise Exception("Query param t not found")
# Quitar formato base64url
token = url64.decode(raw_token)
return token
@staticmethod
def _format_headers(token: str) -> dict:
headers = dict(Api._DEFAULT_HEADERS)
headers["Authorization"] = f"Bearer {token}"
return headers
@staticmethod
def _get_recycler_id(token: str) -> int:
res = requests.post(Api._API_AUTH_URL, json={"authToken": token})
print(f"[AUTH] Status: {res.status_code}")
recycler_id = res.json().get("idClienteReciclador", None)
if recycler_id is None:
raise Exception("AUTHENTICATION FAILED")
return int(recycler_id)
def __init__(self):
self.token = Api._get_auth_token()
self.recycler_id = Api._get_recycler_id(self.token)
self.headers = Api._format_headers(self.token)
def _create_collection(self, route_id: int, driver_id: int, vehicle_id: int, branch_id: int, date: datetime, estimated_containers: list[EstimatedContainer]) -> int:
res = requests.post(Api._API_V2_BASE_URL, headers=self.headers, json={
"query": Api._CREATE_COLLECTION_QUERY,
"variables": {
"route_id" : int(route_id),
"recycler_id" : int(self.recycler_id),
"driver_id" : int(driver_id),
"vehicle_id" : int(vehicle_id),
"branch_id" : int(branch_id),
"date" : date.strftime("%Y-%m-%d"),
"estimated_containers": [ec.to_dict() for ec in estimated_containers]
}
})
print(f"[CREATE_COLLECTION] Status: {res.status_code}")
collection_id = res.json()["data"]["collection"]["id"]
if res.status_code != 200 or collection_id is None:
print("[ERROR] Input: ", {
"route_id": route_id, "driver_id": driver_id, "vehicle_id": vehicle_id, "branch_id": branch_id, "date": datetime, "estimated_containers": estimated_containers
})
print("[ERROR] Output: ", res.json())
raise Exception("Failed to create collection")
return collection_id
# def _create_collection(self, branch_id: int, driver_id: int, date: datetime, estimated_containers: list[EstimatedContainer]) -> int:
# payload = {
# "id_sucursal" : branch_id,
# "id_driver" : driver_id,
# "date" : str(date),
# "schedule_hr" : [],
# "materials" : [],
# "estimate_weight_kg" : [],
# "estimate_volume_m3" : None,
# "estimate_containers": [ec.to_dict() for ec in estimated_containers],
# "priority" : 0
# }
# # Crear nueva recolección
# res = requests.post(f"{Api._API_V1_BASE_URL}/clientes-recicladores/{self.recycler_id}/recollections", headers=self.headers, json=payload)
# print(f"[CREATE_COLLECTION] Status: {res.status_code}")
# # Validar resultado
# collection_id = res.json().get("id", None)
# if res.status != 201 or res.status != 200 or collection_id is None:
# raise Exception("Failed to create recollection with input:", payload)
# return collection_id
def _reorder_route(self, route_id: int, ordered_collection_ids: list[int]):
"""
collections is in format (collection_id, index)
"""
payload: list[dict] = []
for i in range(0, len(ordered_collection_ids)):
id = ordered_collection_ids[i]
payload.append({"id_recollection": id, "index": i})
res = requests.put(f"{Api._API_V1_BASE_URL}/clientes-recicladores/{self.recycler_id}/routes-order/{route_id}", headers=self.headers, json=payload)
print(f"[REORDER_ROUTES] Status: {res.status_code}")
if res.status_code != 200:
raise Exception("Failed to update route order")
def _disable_collections_by_date(self, date: datetime, excluded_collection_ids: list[int]):
res = requests.post(Api._API_V2_BASE_URL, headers=self.headers, json={
"query": Api._DISABLE_COLLECTIONS_BY_DATE_QUERY,
"variables": {
"recycler_id" : self.recycler_id,
"date" : str(date),
"excluded_ids": excluded_collection_ids
}
})
print(f"[DISABLE_COLLECTIONS] Status: {res.status_code}")
if res.status_code != 200:
raise Exception("Failed to disable collections by date")
def modify_daily_routes(self, date: datetime, routes: list[Route]) -> list[int]:
"""
Returns the ids of the newly created collections.
@TODO: Make this transaction-like.
"""
# Crear recolecciones nuevas del día indicado
created_collection_ids: list[int] = []
for r in routes:
ordered_collection_ids: list[int] = []
for c in r.ordered_collections:
new_collection_id = self._create_collection(
route_id=r.id,
vehicle_id=c.vehicle_id,
branch_id=c.branch_id,
driver_id=c.driver_id,
date=date,
estimated_containers=c.estimated_containers
)
ordered_collection_ids.append(new_collection_id)
created_collection_ids.append(new_collection_id)
# Ordenar recolecciones después de crearlas
self._reorder_route(route_id=r.id, ordered_collection_ids=ordered_collection_ids)
# Deshabilitar todas las recolecciones del mismo día
self._disable_collections_by_date(date=date, excluded_collection_ids=created_collection_ids)
return created_collection_ids |