from typing import NamedTuple import polars as pl from js import URL, Blob, document # type: ignore from pyodide.ffi import to_js # type: ignore from pyscript import display, web, when processed_data: str | None = None _COL_MAPPING: dict[str, str] = { "Número_Envío": "num_envio", "Fecha_Envío": "fecha_envio", "Abonado": "abonado", "CECO": "ceco", "Tipo_Cobro": "tipo_cobro", "Servicio": "servicio", "Tipo_Geográfico": "tipo_geo", "CENTRAL": "central", "NOSOTROS": "nosotros", "Diferencia": "dif", "Estado_Envío": "estado_envio", "Peso_Central": "peso_cen", "Peso_Nuestro": "peso_nos", "Medidas_Central": "dim_cen", "Medidas_Nuestras": "dim_nos", } _INPUT_SCHEMA = { "Número_Envío": pl.Utf8, "Fecha_Envío": pl.Utf8, "Abonado": pl.Utf8, "CECO": pl.Utf8, "Tipo_Cobro": pl.Utf8, "Servicio": pl.Utf8, "Tipo_Geográfico": pl.Utf8, "CENTRAL": pl.Utf8, "NOSOTROS": pl.Utf8, "Diferencia": pl.Utf8, "Estado_Envío": pl.Utf8, "Peso_Central": pl.Utf8, "Peso_Nuestro": pl.Utf8, "Medidas_Central": pl.Utf8, "Medidas_Nuestras": pl.Utf8, } _SERVICES: dict[str, list[str]] = { "exp": ["EXPEDICION", "EXPEDICIÓN"], "isl": ["ISLA", "MARITIMO", "MARÍTIMO", "ECONÓMICO", "ECONOMICO"], } # Service limits: weight (kg) and dimensions in cms (L + W + H) # EXP = EXPEDICION # ISL = ISLA | MARITIMO | ECONOMICO # OTH = OTROS (NONE OF THE ABOVE) class ExcessBrackets(NamedTuple): """Both end points are inclusive.""" lower: int upper: int _SERVICE_LIMITS: dict[str, dict[str, ExcessBrackets]] = { "exp": { "weight": ExcessBrackets(lower=0, upper=3), "dims": ExcessBrackets(lower=0, upper=150), }, "isl": { "weight": ExcessBrackets(lower=0, upper=3), "dims": ExcessBrackets(lower=0, upper=100), }, "oth": { "weight": ExcessBrackets(lower=0, upper=5), "dims": ExcessBrackets(lower=0, upper=100), }, } # Weigth and dimensions excess is billed in brackets of different sizes for each service. _EXCESS_BRACKETS: dict[str, dict[str, int]] = { "exp": { "weight": 3, "dims": 50, }, "isl": { "weight": 3, "dims": 50, }, "oth": { "weight": 5, "dims": 50, }, } def reset_state(): print("Resetting app...") global processed_data processed_data = None web.page["#csv-upload"].value = "" web.page["#run-btn"].textContent = "⚙️ Procesar Datos" web.page["#run-btn"].disabled = True web.page["#status-msg"].textContent = "Aplicación lista" web.page["#download-area"].style["display"] = "none" print("App reset completed.") def generate_report(raw_data: bytes) -> None: """Process the data and return a string with the output report in csv format.""" df = ( pl.read_csv(raw_data, separator=";", has_header=True, schema=_INPUT_SCHEMA) .rename(_COL_MAPPING) .with_columns( pl.col("peso_cen", "peso_nos", "dim_cen", "dim_nos") .str.replace(",", ".") .cast(pl.Float64, strict=True) .round() .cast(pl.Int16, strict=True) ) ) # print(df.glimpse()) result = df.with_columns( pl.when(pl.col("servicio").str.contains_any(_SERVICES["exp"])) .then(pl.col("peso_cen") - _SERVICE_LIMITS["exp"]["weight"].upper) .when(pl.col("servicio").str.contains_any(_SERVICES["isl"])) .then(pl.col("peso_cen") - _SERVICE_LIMITS["isl"]["weight"].upper) .otherwise(pl.col("peso_cen") - _SERVICE_LIMITS["oth"]["weight"].upper) .clip(lower_bound=0) .alias("w_excess_ref"), pl.when(pl.col("servicio").str.contains_any(_SERVICES["exp"])) .then(pl.col("dim_cen") - _SERVICE_LIMITS["exp"]["dims"].upper) .when(pl.col("servicio").str.contains_any(_SERVICES["isl"])) .then(pl.col("dim_cen") - _SERVICE_LIMITS["isl"]["dims"].upper) .otherwise(pl.col("dim_cen") - _SERVICE_LIMITS["oth"]["dims"].upper) .clip(lower_bound=0) .alias("dim_excess_ref"), pl.when(pl.col("servicio").str.contains_any(_SERVICES["exp"])) .then(pl.col("peso_nos") - _SERVICE_LIMITS["exp"]["weight"].upper) .when(pl.col("servicio").str.contains_any(_SERVICES["isl"])) .then(pl.col("peso_nos") - _SERVICE_LIMITS["isl"]["weight"].upper) .otherwise(pl.col("peso_nos") - _SERVICE_LIMITS["oth"]["weight"].upper) .clip(lower_bound=0) .alias("w_excess"), pl.when(pl.col("servicio").str.contains_any(_SERVICES["exp"])) .then(pl.col("dim_nos") - _SERVICE_LIMITS["exp"]["dims"].upper) .when(pl.col("servicio").str.contains_any(_SERVICES["isl"])) .then(pl.col("dim_nos") - _SERVICE_LIMITS["isl"]["dims"].upper) .otherwise(pl.col("dim_nos") - _SERVICE_LIMITS["oth"]["dims"].upper) .clip(lower_bound=0) .alias("dim_excess"), ).with_columns( (pl.col("w_excess") - pl.col("w_excess_ref")).alias("w_excess_diff"), (pl.col("dim_excess") - pl.col("dim_excess_ref")).alias("dim_excess_diff"), ) global processed_data processed_data = result.filter( pl.any_horizontal( pl.col("w_excess_diff") < 0, pl.col("dim_excess_diff") < 0, ) ).write_csv(include_header=True, separator=";") print(f"DEBUG: {processed_data[:150]}") def download_csv(filename: str = "resultado.csv", add_bom: bool = True) -> None: """Trigger a browser download of csv_str as a CSV file.""" if processed_data is None: return data = processed_data.encode("utf-8") if add_bom: data = b"\xef\xbb\xbf" + data # UTF-8 BOM so Excel opens it correctly blob = Blob.new( [to_js(data)], to_js({"type": "text/csv;charset=utf-8"}), ) url = URL.createObjectURL(blob) try: a = document.createElement("a") a.href = url a.download = filename a.style.display = "none" document.body.appendChild(a) a.click() document.body.removeChild(a) finally: URL.revokeObjectURL(url) # free memory immediately @when("click", "#run-btn") async def on_run(event): print("EVENT: process_data begin") try: btn = event.target btn.textContent = "Procesando..." btn.disabled = True file_input = web.page["#csv-upload"] # print(file_input) if file_input and file_input.files.length > 0: print("Loading data...") buffer = await file_input.files.item(0).arrayBuffer() generate_report(bytes(buffer.to_py())) is_success = True except Exception as e: # noqa: BLE001 is_success = False print(f"[process_data] error: {e}") display("Lo siento, Algo salió mal,", target="#status-msg", append=False) finally: if is_success: btn.textContent = "Procesado" web.page["#download-area"].style["display"] = "block" print("EVENT: process_data end") @when("change", "#csv-upload") def on_csv_upload(event): print("EVENT: select_file begin") try: file_list = event.target.files if file_list.length == 0: print("No file chosen") return print("File chosen") except Exception as e: # noqa: BLE001 print(f"[process_data] error: {e}") finally: web.page["#run-btn"].disabled = False print("EVENT: select_file end") @when("click", "#download-btn") async def on_download(event): print("EVENT: download_result begin") if processed_data is None: print("No results available to download") return try: download_csv() except Exception as e: # noqa: BLE001 # Keep state intact so the user can retry without re-processing. print(f"[download] error: {e}") return finally: reset_state() print("EVENT: download_result end")