| from pathlib import Path |
| import numpy as np |
| import pandas as pd |
|
|
| ENTRADA = Path("datos/externos") |
| SALIDA = Path("datos/peso/externo.csv") |
|
|
| |
| COL = {"peso": None, "largo": None, "ancho": None, "material": None} |
| A_GRAMOS = 1.0 |
| A_CM = 1.0 |
|
|
| def leer(p): |
| s = p.suffix.lower() |
| if s == ".parquet": |
| return pd.read_parquet(p) |
| if s == ".csv": |
| return pd.read_csv(p) |
| if s in (".json", ".jsonl"): |
| return pd.read_json(p, lines=(s == ".jsonl")) |
| return None |
|
|
| tablas = [] |
| for p in sorted(ENTRADA.rglob("*")): |
| if p.is_file(): |
| t = leer(p) |
| if t is not None: |
| print("Leído:", p, t.shape) |
| tablas.append(t) |
| if not tablas: |
| raise SystemExit("No hay archivos .csv/.parquet/.json en datos/externos") |
|
|
| df = pd.concat(tablas, ignore_index=True) |
| if not all(COL.values()): |
| print("\nColumnas disponibles:", list(df.columns)) |
| print(df.head(3).T) |
| raise SystemExit("\nEdita COL (y las unidades) en este script y vuelve a ejecutarlo.") |
|
|
| def primero(v): |
| if isinstance(v, str): |
| return v.lower() |
| if hasattr(v, "__len__") and len(v) > 0: |
| return str(v[0]).lower() |
| return "otro" |
|
|
| peso = pd.to_numeric(df[COL["peso"]], errors="coerce") * A_GRAMOS |
| a = pd.to_numeric(df[COL["largo"]], errors="coerce") * A_CM |
| b = pd.to_numeric(df[COL["ancho"]], errors="coerce") * A_CM |
| out = pd.DataFrame({ |
| "foto": "externo", |
| "sesion": "externo", |
| "tipo_objeto": "externo", |
| "material": df[COL["material"]].apply(primero), |
| "n_piezas": 1, |
| "area_cm2_total": a * b, |
| "largo_cm_medio": np.maximum(a, b), |
| "ancho_cm_medio": np.minimum(a, b), |
| "largo_cm_max": np.maximum(a, b), |
| "peso_real_g": peso, |
| }).replace([np.inf, -np.inf], np.nan).dropna() |
| out = out[(out.peso_real_g > 0) & (out.largo_cm_medio > 0) & (out.ancho_cm_medio > 0)] |
| out.to_csv(SALIDA, index=False) |
| print("Filas guardadas:", len(out), "->", SALIDA)
|
|
|
