Spaces:
Running
Running
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 | |
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 | |
def _format_headers(token: str) -> dict: | |
headers = dict(Api._DEFAULT_HEADERS) | |
headers["Authorization"] = f"Bearer {token}" | |
return headers | |
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 |