diff --git a/services/cleaning/api/main.py b/services/cleaning/api/main.py index eadceaf..1e1c3f8 100644 --- a/services/cleaning/api/main.py +++ b/services/cleaning/api/main.py @@ -29,6 +29,9 @@ from infrastructure.database import ( utcnow, ) +_SOURCE_BATCH_KEY = "_cleaning_source_batch_id" +_SOURCE_FILE_KEY = "_cleaning_source_file" + @asynccontextmanager async def lifespan(app: FastAPI): @@ -45,6 +48,7 @@ app = FastAPI( class CleaningRequest(BaseModel): extraction_batch_id: str | None = None + extraction_batch_ids: list[str] = Field(default_factory=list) job_id: str | None = None rules_per_column: dict[str, list[str]] = Field(default_factory=dict) cop_mod_fecha_column: dict[str, str] = Field(default_factory=dict) @@ -53,9 +57,15 @@ class CleaningRequest(BaseModel): cop_mod_config: dict[str, Any] | None = None +class AnalyzeBatchesRequest(BaseModel): + extraction_batch_ids: list[str] = Field(default_factory=list) + extraction_batch_id: str | None = None + + class PreviewRequest(BaseModel): rows: list[dict[str, Any]] = Field(default_factory=list) extraction_batch_id: str | None = None + extraction_batch_ids: list[str] = Field(default_factory=list) rules_per_column: dict[str, list[str]] = Field(default_factory=dict) cop_mod_fecha_column: dict[str, str] = Field(default_factory=dict) cop_mod_bruto_column: dict[str, str] = Field(default_factory=dict) @@ -98,6 +108,23 @@ def _validate_job_id(session: Any, job_id: str | None) -> None: raise HTTPException(status_code=404, detail="Trabajo (job) no encontrado") +def _normalize_batch_ids( + extraction_batch_id: str | None, + extraction_batch_ids: list[str] | None, +) -> list[str]: + ids = [str(x).strip() for x in (extraction_batch_ids or []) if str(x).strip()] + if not ids and extraction_batch_id and extraction_batch_id.strip(): + ids = [extraction_batch_id.strip()] + # dedupe preservando orden + seen: set[str] = set() + out: list[str] = [] + for bid in ids: + if bid not in seen: + seen.add(bid) + out.append(bid) + return out + + def _create_cleaning_task( session: Any, *, @@ -140,12 +167,15 @@ def _persist_cleaned_rows( ) -> None: session.execute(delete(CleanedRow).where(CleanedRow.cleaning_job_id == job_id)) for idx, row in enumerate(output_rows, start=1): + row_batch = row.get(_SOURCE_BATCH_KEY) + batch_for_row = str(row_batch).strip() if row_batch else extraction_batch_id + cleaned = {k: v for k, v in row.items() if k not in {_SOURCE_BATCH_KEY, _SOURCE_FILE_KEY}} session.add( CleanedRow( cleaning_job_id=job_id, - extraction_batch_id=extraction_batch_id, + extraction_batch_id=batch_for_row or None, row_index=idx, - row_json=row, + row_json=cleaned, ) ) @@ -158,14 +188,20 @@ def _run_rules_on_dataframe( cop_mod_bruto_column: dict[str, str] | None = None, cop_mod_neto_column: dict[str, str] | None = None, ) -> tuple[list[dict[str, Any]], list[str]]: + # Conservar trazabilidad fuera de las reglas + meta_cols = [c for c in (_SOURCE_BATCH_KEY, _SOURCE_FILE_KEY) if c in df.columns] + meta = df[meta_cols].copy() if meta_cols else None + work = df.drop(columns=meta_cols, errors="ignore") result_df, log = apply_rules( - df, + work, rules_per_column, cop_mod_fecha_column=cop_mod_fecha_column, cop_mod_config=cop_mod_config, cop_mod_bruto_column=cop_mod_bruto_column, cop_mod_neto_column=cop_mod_neto_column, ) + if meta is not None and not meta.empty: + result_df = pd.concat([result_df.reset_index(drop=True), meta.reset_index(drop=True)], axis=1) return dataframe_to_records(result_df), log @@ -179,11 +215,12 @@ def _parse_json_form(raw: str | None, default: Any) -> Any: def _analysis_payload(df: pd.DataFrame, **extra: Any) -> dict[str, Any]: + work = df.drop(columns=[_SOURCE_BATCH_KEY, _SOURCE_FILE_KEY], errors="ignore") return { **extra, - "row_count": len(df), - "columns": [summary.__dict__ for summary in analyze_dataframe(df)], - "sample_rows": sample_dataframe_rows(df, 8), + "row_count": len(work), + "columns": [summary.__dict__ for summary in analyze_dataframe(work)], + "sample_rows": sample_dataframe_rows(work, 8), } @@ -206,9 +243,10 @@ def _preview_from_dataframe( cop_mod_bruto_column: dict[str, str] | None = None, cop_mod_neto_column: dict[str, str] | None = None, ) -> dict[str, Any]: - rows_before = dataframe_to_records(df) + work = df.drop(columns=[_SOURCE_BATCH_KEY, _SOURCE_FILE_KEY], errors="ignore") + rows_before = dataframe_to_records(work) result_df, _log = apply_rules( - df, + work, rules_per_column, cop_mod_fecha_column=cop_mod_fecha_column, cop_mod_config=cop_mod_config, @@ -217,9 +255,8 @@ def _preview_from_dataframe( ) rows_after = dataframe_to_records(result_df) - # Emparejar por índice de fila de muestra (misma posición) n = min(len(rows_before), len(rows_after)) - all_cols = list(dict.fromkeys([*df.columns.astype(str), *result_df.columns.astype(str)])) + all_cols = list(dict.fromkeys([*work.columns.astype(str), *result_df.columns.astype(str)])) columns_preview: list[dict[str, Any]] = [] for col in all_cols: before_vals = [_cell_str(rows_before[i].get(col)) for i in range(n)] @@ -238,6 +275,72 @@ def _preview_from_dataframe( } +def _load_batch_records(session: Any, batch_id: str) -> list[dict[str, Any]]: + batch = session.get(ExtractionBatch, batch_id) + if batch is None: + raise HTTPException(status_code=404, detail=f"Lote de extracción no encontrado: {batch_id}") + rows = session.execute( + select(ExtractionRow.row_json) + .where(ExtractionRow.batch_id == batch_id) + .order_by(ExtractionRow.row_index, ExtractionRow.id) + ).all() + records = [row[0] for row in rows] + if not records: + raise HTTPException(status_code=400, detail=f"El lote {batch_id} no tiene filas extraídas") + return records + + +def _concat_extraction_batches(session: Any, batch_ids: list[str]) -> pd.DataFrame: + frames: list[pd.DataFrame] = [] + for bid in batch_ids: + records = _load_batch_records(session, bid) + part = pd.DataFrame(records) + part[_SOURCE_BATCH_KEY] = bid + frames.append(part) + return pd.concat(frames, ignore_index=True, sort=False) + + +async def _load_csv_uploads(uploads: list[UploadFile]) -> tuple[pd.DataFrame, list[str], list[str]]: + if not uploads: + raise HTTPException(status_code=400, detail="Se requiere al menos un archivo CSV") + frames: list[pd.DataFrame] = [] + filenames: list[str] = [] + delimiters: list[str] = [] + for upload in uploads: + filename = upload.filename or "upload.csv" + content = await upload.read() + if not content: + raise HTTPException(status_code=400, detail=f"El archivo {filename} está vacío") + try: + df, delimiter = load_csv_bytes(content, filename) + except Exception as exc: # noqa: BLE001 + raise HTTPException( + status_code=400, + detail=f"No se pudo leer {filename}: {exc}", + ) from exc + if df.empty: + raise HTTPException(status_code=400, detail=f"El CSV {filename} no tiene filas") + part = df.copy() + part[_SOURCE_FILE_KEY] = filename + frames.append(part) + filenames.append(filename) + delimiters.append(delimiter) + return pd.concat(frames, ignore_index=True, sort=False), filenames, delimiters + + +def _cleaning_title_extraction(n: int) -> str: + if n <= 1: + return "Limpieza desde extracción" + return f"Limpieza ({n} extracciones)" + + +def _cleaning_title_csv(filenames: list[str]) -> str: + n = len(filenames) + if n <= 1: + return f"Limpieza CSV — {filenames[0] if filenames else 'upload.csv'}" + return f"Limpieza ({n} CSV)" + + @app.get("/health") def health() -> dict[str, str]: return {"status": "ok"} @@ -258,59 +361,62 @@ def list_rules() -> list[dict[str, Any]]: @app.get("/batches/{batch_id}/analysis") def analyze_batch(batch_id: str) -> dict[str, Any]: with session_scope() as session: - batch = session.get(ExtractionBatch, batch_id) - if batch is None: - raise HTTPException(status_code=404, detail="Lote de extraccion no encontrado") - rows = session.execute( - select(ExtractionRow.row_json) - .where(ExtractionRow.batch_id == batch_id) - .order_by(ExtractionRow.row_index, ExtractionRow.id) - ).all() - records = [row[0] for row in rows] - + records = _load_batch_records(session, batch_id) df = pd.DataFrame(records) - return _analysis_payload(df, batch_id=batch_id) + return _analysis_payload(df, batch_id=batch_id, extraction_batch_ids=[batch_id]) + + +@app.post("/analyze-batches") +def analyze_batches(request: AnalyzeBatchesRequest) -> dict[str, Any]: + batch_ids = _normalize_batch_ids(request.extraction_batch_id, request.extraction_batch_ids) + if not batch_ids: + raise HTTPException(status_code=400, detail="extraction_batch_ids es requerido (mínimo 1)") + with session_scope() as session: + df = _concat_extraction_batches(session, batch_ids) + return _analysis_payload( + df, + batch_id=batch_ids[0], + extraction_batch_ids=batch_ids, + ) @app.post("/analyze-csv") -async def analyze_csv(file: UploadFile = File(...)) -> dict[str, Any]: - content = await file.read() - if not content: - raise HTTPException(status_code=400, detail="El archivo CSV está vacío") - try: - df, delimiter = load_csv_bytes(content, file.filename or "upload.csv") - except Exception as exc: # noqa: BLE001 - raise HTTPException(status_code=400, detail=f"No se pudo leer el CSV: {exc}") from exc - if df.empty: - raise HTTPException(status_code=400, detail="El CSV no tiene filas") - return _analysis_payload(df, filename=file.filename, delimiter=delimiter) +async def analyze_csv( + file: UploadFile | None = File(None), + files: list[UploadFile] | None = File(None), +) -> dict[str, Any]: + uploads: list[UploadFile] = [] + if files: + uploads.extend(files) + if file is not None: + uploads.append(file) + df, filenames, delimiters = await _load_csv_uploads(uploads) + return _analysis_payload( + df, + filename=filenames[0] if filenames else None, + filenames=filenames, + delimiter=delimiters[0] if delimiters else None, + delimiters=delimiters, + ) @app.post("/preview") def preview_cleaning(request: PreviewRequest) -> dict[str, Any]: """Aplica reglas sobre una muestra pequeña y devuelve Antes/Después.""" records: list[dict[str, Any]] = list(request.rows or []) + batch_ids = _normalize_batch_ids(request.extraction_batch_id, request.extraction_batch_ids) - if not records and request.extraction_batch_id: + if not records and batch_ids: with session_scope() as session: - batch = session.get(ExtractionBatch, request.extraction_batch_id) - if batch is None: - raise HTTPException(status_code=404, detail="Lote de extraccion no encontrado") - rows = session.execute( - select(ExtractionRow.row_json) - .where(ExtractionRow.batch_id == request.extraction_batch_id) - .order_by(ExtractionRow.row_index, ExtractionRow.id) - ).all() - all_records = [row[0] for row in rows] - if not all_records: - raise HTTPException(status_code=400, detail="El lote no tiene filas") - df_full = pd.DataFrame(all_records) + df_full = _concat_extraction_batches(session, batch_ids) + if df_full.empty: + raise HTTPException(status_code=400, detail="Los lotes no tienen filas") records = sample_dataframe_rows(df_full, max(1, min(request.sample_size, 20))) if not records: raise HTTPException( status_code=400, - detail="Se requieren rows o extraction_batch_id con filas", + detail="Se requieren rows o extraction_batch_ids con filas", ) df = pd.DataFrame(records) @@ -329,44 +435,37 @@ def preview_cleaning(request: PreviewRequest) -> dict[str, Any]: @app.post("/jobs") def run_cleaning(request: CleaningRequest) -> dict[str, Any]: - if not request.extraction_batch_id: + batch_ids = _normalize_batch_ids(request.extraction_batch_id, request.extraction_batch_ids) + if not batch_ids: raise HTTPException( status_code=400, - detail="extraction_batch_id es requerido (use /jobs/from-csv para CSV)", + detail="extraction_batch_ids es requerido (use /jobs/from-csv para CSV)", ) with session_scope() as session: - batch = session.get(ExtractionBatch, request.extraction_batch_id) - if batch is None: - raise HTTPException(status_code=404, detail="Lote de extraccion no encontrado") _validate_job_id(session, request.job_id) + df = _concat_extraction_batches(session, batch_ids) - extracted_rows = session.execute( - select(ExtractionRow.row_json) - .where(ExtractionRow.batch_id == request.extraction_batch_id) - .order_by(ExtractionRow.row_index, ExtractionRow.id) - ).all() - records = [row[0] for row in extracted_rows] - if not records: - raise HTTPException(status_code=400, detail="El lote no tiene filas extraidas") + request_dump = request.model_dump() + request_dump["extraction_batch_ids"] = batch_ids + request_dump["extraction_batch_id"] = batch_ids[0] job = CleaningJob( - extraction_batch_id=request.extraction_batch_id, + extraction_batch_id=batch_ids[0], job_id=request.job_id, source="extraction", status="running", - request=request.model_dump(), + request=request_dump, updated_at=utcnow(), ) session.add(job) session.flush() job_id = job.id - batch_id = request.extraction_batch_id + primary_batch_id = batch_ids[0] desktop_job_id = request.job_id - request_dump = request.model_dump() + title = _cleaning_title_extraction(len(batch_ids)) try: - df = pd.DataFrame(records) output_rows, log = _run_rules_on_dataframe( df, request.rules_per_column, @@ -380,7 +479,7 @@ def run_cleaning(request: CleaningRequest) -> dict[str, Any]: _persist_cleaned_rows( session, job_id=job_id, - extraction_batch_id=batch_id, + extraction_batch_id=primary_batch_id, output_rows=output_rows, ) job = session.get(CleaningJob, job_id) @@ -396,7 +495,7 @@ def run_cleaning(request: CleaningRequest) -> dict[str, Any]: source="extraction", status="completed", request=request_dump, - title="Limpieza desde extracción", + title=title, started_at=job.created_at, ) session.flush() @@ -419,7 +518,7 @@ def run_cleaning(request: CleaningRequest) -> dict[str, Any]: status="failed", request=request_dump, error=str(exc), - title="Limpieza desde extracción", + title=title, started_at=started, ) raise HTTPException(status_code=500, detail=str(exc)) from exc @@ -427,7 +526,8 @@ def run_cleaning(request: CleaningRequest) -> dict[str, Any]: @app.post("/jobs/from-csv") async def run_cleaning_from_csv( - file: UploadFile = File(...), + file: UploadFile | None = File(None), + files: list[UploadFile] | None = File(None), job_id: str | None = Form(None), rules_per_column: str | None = Form(None), cop_mod_fecha_column: str | None = Form(None), @@ -435,9 +535,13 @@ async def run_cleaning_from_csv( cop_mod_neto_column: str | None = Form(None), cop_mod_config: str | None = Form(None), ) -> dict[str, Any]: - content = await file.read() - if not content: - raise HTTPException(status_code=400, detail="El archivo CSV está vacío") + uploads: list[UploadFile] = [] + if files: + uploads.extend(files) + if file is not None: + uploads.append(file) + + df, filenames, delimiters = await _load_csv_uploads(uploads) rules = _parse_json_form(rules_per_column, {}) fecha_map = _parse_json_form(cop_mod_fecha_column, {}) @@ -453,17 +557,12 @@ async def run_cleaning_from_csv( if not isinstance(neto_map, dict): raise HTTPException(status_code=400, detail="cop_mod_neto_column debe ser un objeto JSON") - try: - df, delimiter = load_csv_bytes(content, file.filename or "upload.csv") - except Exception as exc: # noqa: BLE001 - raise HTTPException(status_code=400, detail=f"No se pudo leer el CSV: {exc}") from exc - if df.empty: - raise HTTPException(status_code=400, detail="El CSV no tiene filas") - request_dump: dict[str, Any] = { "source": "csv", - "filename": file.filename, - "delimiter": delimiter, + "filename": filenames[0] if filenames else None, + "filenames": filenames, + "delimiter": delimiters[0] if delimiters else None, + "delimiters": delimiters, "job_id": job_id, "rules_per_column": rules, "cop_mod_fecha_column": fecha_map, @@ -471,6 +570,7 @@ async def run_cleaning_from_csv( "cop_mod_neto_column": neto_map, "cop_mod_config": cop_cfg, } + title = _cleaning_title_csv(filenames) with session_scope() as session: _validate_job_id(session, job_id) @@ -512,7 +612,7 @@ async def run_cleaning_from_csv( source="csv", status="completed", request=request_dump, - title=f"Limpieza CSV — {file.filename or 'upload.csv'}", + title=title, started_at=job.created_at, ) session.flush() @@ -535,7 +635,7 @@ async def run_cleaning_from_csv( status="failed", request=request_dump, error=str(exc), - title=f"Limpieza CSV — {file.filename or 'upload.csv'}", + title=title, started_at=started, ) raise HTTPException(status_code=500, detail=str(exc)) from exc @@ -553,6 +653,8 @@ def list_jobs(job_id: str | None = None) -> list[dict[str, Any]]: @app.get("/jobs/{job_id}") def get_job(job_id: str) -> dict[str, Any]: + if job_id in {"from-csv", "analyze-batches"}: + raise HTTPException(status_code=405, detail=f"Use POST /jobs/{job_id}") with session_scope() as session: job = session.get(CleaningJob, job_id) if job is None: diff --git a/services/desktop_app/src/renderer/lib/apiClient.ts b/services/desktop_app/src/renderer/lib/apiClient.ts index 2743e97..1e7aeb1 100644 --- a/services/desktop_app/src/renderer/lib/apiClient.ts +++ b/services/desktop_app/src/renderer/lib/apiClient.ts @@ -364,12 +364,32 @@ export class ApiClient { ); } + async analyzeCleaningBatches(batchIds: string[]): Promise { + if (batchIds.length === 1) { + return this.analyzeCleaningBatch(batchIds[0]); + } + return this.postJson("/cleaning/analyze-batches", { + extraction_batch_ids: batchIds, + }); + } + async analyzeCleaningCsv(file: File | Blob, filename = "upload.csv"): Promise { const form = new FormData(); form.append("file", file, filename); return this.postForm("/cleaning/analyze-csv", form); } + async analyzeCleaningCsvFiles(files: File[]): Promise { + if (files.length === 1) { + return this.analyzeCleaningCsv(files[0], files[0].name); + } + const form = new FormData(); + for (const file of files) { + form.append("files", file, file.name); + } + return this.postForm("/cleaning/analyze-csv", form); + } + async runCleaning(payload: RunCleaningPayload): Promise { return this.postJson("/cleaning/jobs", payload); } diff --git a/services/desktop_app/src/renderer/modules/cleaning/CleaningWizard.tsx b/services/desktop_app/src/renderer/modules/cleaning/CleaningWizard.tsx index fe3129c..176b1bf 100644 --- a/services/desktop_app/src/renderer/modules/cleaning/CleaningWizard.tsx +++ b/services/desktop_app/src/renderer/modules/cleaning/CleaningWizard.tsx @@ -2,13 +2,16 @@ import { AlertCircle, ArrowRight, CheckCircle2, + CheckSquare, Cpu, Download, FileSpreadsheet, Loader2, Play, Sparkles, + Square, Upload, + X, } from "lucide-react"; import { useEffect, useRef, useState } from "react"; import type { @@ -44,6 +47,12 @@ function batchIdFromTask(task: Task): string { return String((task.result as Record)?.extraction_batch_id ?? ""); } +function formatBytes(n: number): string { + if (n < 1024) return `${n} B`; + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB`; + return `${(n / (1024 * 1024)).toFixed(1)} MB`; +} + export function CleaningWizard({ client, job, @@ -55,12 +64,13 @@ export function CleaningWizard({ const hasHandoff = Boolean(initialBatchId || initialTaskId); const [step, setStep] = useState(1); const [maxReached, setMaxReached] = useState(1); - const [source, setSource] = useState(hasHandoff ? "extraction" : "extraction"); + const [source, setSource] = useState("extraction"); const [extractionTasks, setExtractionTasks] = useState([]); const [loadingTasks, setLoadingTasks] = useState(true); - const [selectedTaskId, setSelectedTaskId] = useState(initialTaskId ?? null); - const [batchId, setBatchId] = useState(initialBatchId ?? ""); - const [csvFile, setCsvFile] = useState(null); + const [selectedTaskIds, setSelectedTaskIds] = useState>( + () => (initialTaskId ? new Set([initialTaskId]) : new Set()), + ); + const [csvFiles, setCsvFiles] = useState([]); const [rulesCatalog, setRulesCatalog] = useState([]); const [analysis, setAnalysis] = useState(null); const [rulesPerColumn, setRulesPerColumn] = useState>({}); @@ -81,6 +91,11 @@ export function CleaningWizard({ const fileInputRef = useRef(null); const previewTimerRef = useRef | null>(null); + const selectedBatchIds = extractionTasks + .filter((t) => selectedTaskIds.has(t.id)) + .map(batchIdFromTask) + .filter(Boolean); + function goToStep(next: CleaningStep) { setStep(next); setMaxReached((m) => (next > m ? next : m)); @@ -102,25 +117,20 @@ export function CleaningWizard({ .sort((a, b) => new Date(b.created_at).getTime() - new Date(a.created_at).getTime()); setExtractionTasks(extractions); - // Preselección por handoff o la más reciente if (initialTaskId && extractions.some((t) => t.id === initialTaskId)) { - const t = extractions.find((x) => x.id === initialTaskId)!; - setSelectedTaskId(t.id); - setBatchId(batchIdFromTask(t) || initialBatchId || ""); + setSelectedTaskIds(new Set([initialTaskId])); setSource("extraction"); } else if (initialBatchId) { const match = extractions.find((t) => batchIdFromTask(t) === initialBatchId); if (match) { - setSelectedTaskId(match.id); - setBatchId(initialBatchId); - } else { - setBatchId(initialBatchId); + setSelectedTaskIds(new Set([match.id])); + } else if (extractions.length === 0) { + // handoff solo con batch id sin task listable — se analiza por batch en handoff + setSelectedTaskIds(new Set()); } setSource("extraction"); - } else if (extractions.length > 0 && !selectedTaskId) { - const newest = extractions[0]; - setSelectedTaskId(newest.id); - setBatchId(batchIdFromTask(newest)); + } else if (extractions.length > 0 && selectedTaskIds.size === 0) { + setSelectedTaskIds(new Set([extractions[0].id])); } } catch { if (!cancelled) setExtractionTasks([]); @@ -136,20 +146,64 @@ export function CleaningWizard({ // Handoff: analizar automáticamente al llegar desde Resultados useEffect(() => { if (!hasHandoff || handoffAnalyzed || loadingTasks) return; - const id = batchId || initialBatchId; - if (!id) return; + const ids = + selectedBatchIds.length > 0 + ? selectedBatchIds + : initialBatchId + ? [initialBatchId] + : []; + if (ids.length === 0) return; setHandoffAnalyzed(true); - void loadAnalysis("extraction", id, null); + void loadAnalysis("extraction", ids, []); // eslint-disable-next-line react-hooks/exhaustive-deps - }, [hasHandoff, handoffAnalyzed, loadingTasks, batchId, initialBatchId]); + }, [hasHandoff, handoffAnalyzed, loadingTasks, selectedBatchIds.join(","), initialBatchId]); - function selectExtraction(task: Task) { - setSelectedTaskId(task.id); - setBatchId(batchIdFromTask(task)); + function toggleExtraction(task: Task) { + setSelectedTaskIds((prev) => { + const next = new Set(prev); + if (next.has(task.id)) next.delete(task.id); + else next.add(task.id); + return next; + }); setSource("extraction"); setError(null); } + function toggleAllExtractions() { + if (selectedTaskIds.size === extractionTasks.length) { + setSelectedTaskIds(new Set()); + } else { + setSelectedTaskIds(new Set(extractionTasks.map((t) => t.id))); + } + } + + function onCsvPicked(list: FileList | null) { + if (!list || list.length === 0) return; + const next = Array.from(list).filter( + (f) => + /\.csv$/i.test(f.name) || + f.type === "text/csv" || + f.type === "application/vnd.ms-excel", + ); + setCsvFiles((prev) => { + const keys = new Set(prev.map((f) => `${f.name}:${f.size}`)); + const merged = [...prev]; + for (const f of next) { + const key = `${f.name}:${f.size}`; + if (!keys.has(key)) { + merged.push(f); + keys.add(key); + } + } + return merged; + }); + if (fileInputRef.current) fileInputRef.current.value = ""; + } + + function removeCsv(index: number) { + setCsvFiles((prev) => prev.filter((_, i) => i !== index)); + } + function ruleName(id: string): string { return rulesCatalog.find((r) => r.id === id)?.name ?? id; } @@ -340,9 +394,9 @@ export function CleaningWizard({ const payload = sampleRows.length > 0 ? { rows: sampleRows, rules_per_column: rulesPerColumn, ...extra } - : batchId + : selectedBatchIds.length > 0 ? { - extraction_batch_id: batchId, + extraction_batch_ids: selectedBatchIds, rules_per_column: rulesPerColumn, ...extra, } @@ -364,29 +418,37 @@ export function CleaningWizard({ return () => { if (previewTimerRef.current) clearTimeout(previewTimerRef.current); }; - }, [step, analysis, rulesPerColumn, copModFechaColumn, copModBrutoColumn, copModNetoColumn, batchId, client]); + }, [step, analysis, rulesPerColumn, copModFechaColumn, copModBrutoColumn, copModNetoColumn, selectedBatchIds.join(","), client]); - async function loadAnalysis(mode: SourceMode, batch?: string, file?: File | null) { + async function loadAnalysis(mode: SourceMode, batchIds: string[], files: File[]) { setLoading(true); setError(null); setAnalysis(null); + const total = mode === "csv" ? Math.max(files.length, 1) : Math.max(batchIds.length, 1); onProgress?.({ module: "cleaning", - total: 1, + total, processed: 0, failed: 0, - currentFile: mode === "csv" ? (file?.name ?? "CSV") : "extracción", + currentFile: + mode === "csv" + ? files.length === 1 + ? files[0].name + : `${files.length} CSV` + : batchIds.length === 1 + ? "extracción" + : `${batchIds.length} extracciones`, running: true, label: "Analizando columnas…", }); try { let data: CleaningAnalysisResponse; if (mode === "extraction") { - if (!batch) throw new Error("Selecciona una extracción de la lista"); - data = await client.analyzeCleaningBatch(batch); + if (batchIds.length === 0) throw new Error("Selecciona al menos una extracción"); + data = await client.analyzeCleaningBatches(batchIds); } else { - if (!file) throw new Error("Selecciona un archivo CSV"); - data = await client.analyzeCleaningCsv(file, file.name); + if (files.length === 0) throw new Error("Selecciona al menos un archivo CSV"); + data = await client.analyzeCleaningCsvFiles(files); } setAnalysis(data); suggestRules(data.columns); @@ -400,8 +462,8 @@ export function CleaningWizard({ setLoading(false); onProgress?.({ module: "cleaning", - total: 1, - processed: 1, + total, + processed: total, failed: 0, currentFile: "", running: false, @@ -411,7 +473,7 @@ export function CleaningWizard({ } async function handleAnalyze() { - await loadAnalysis(source, batchId, csvFile); + await loadAnalysis(source, selectedBatchIds, csvFiles); } async function handleRun() { @@ -419,30 +481,41 @@ export function CleaningWizard({ setError(null); setResultJob(null); setResultRows([]); + const total = + source === "csv" ? Math.max(csvFiles.length, 1) : Math.max(selectedBatchIds.length, 1); onProgress?.({ module: "cleaning", - total: 1, + total, processed: 0, failed: 0, - currentFile: source === "csv" ? (csvFile?.name ?? "CSV") : "extracción", + currentFile: + source === "csv" + ? csvFiles.length === 1 + ? csvFiles[0]?.name ?? "CSV" + : `${csvFiles.length} CSV` + : selectedBatchIds.length === 1 + ? "extracción" + : `${selectedBatchIds.length} extracciones`, running: true, label: "Aplicando reglas de limpieza…", }); try { let cleaningJob: CleaningJob; if (source === "extraction") { - if (!batchId) throw new Error("Selecciona una extracción"); + if (selectedBatchIds.length === 0) throw new Error("Selecciona al menos una extracción"); cleaningJob = await client.runCleaning({ - extraction_batch_id: batchId, + extraction_batch_ids: selectedBatchIds, job_id: job.id, rules_per_column: rulesPerColumn, ...cleaningExtraPayload(), }); } else { - if (!csvFile) throw new Error("Selecciona un archivo CSV"); + if (csvFiles.length === 0) throw new Error("Selecciona al menos un archivo CSV"); const form = new FormData(); const extra = cleaningExtraPayload(); - form.append("file", csvFile, csvFile.name); + for (const file of csvFiles) { + form.append("files", file, file.name); + } form.append("job_id", job.id); form.append("rules_per_column", JSON.stringify(rulesPerColumn)); if (extra.cop_mod_fecha_column) { @@ -476,8 +549,8 @@ export function CleaningWizard({ setRunning(false); onProgress?.({ module: "cleaning", - total: 1, - processed: 1, + total, + processed: total, failed: 0, currentFile: "", running: false, @@ -511,9 +584,8 @@ export function CleaningWizard({ ? Array.from(new Set(resultRows.flatMap((r) => Object.keys(r)))) : []; - const selectedExtraction = extractionTasks.find((t) => t.id === selectedTaskId) ?? null; const canAnalyze = - source === "extraction" ? Boolean(batchId) : Boolean(csvFile); + source === "extraction" ? selectedBatchIds.length > 0 : csvFiles.length > 0; return (
@@ -532,7 +604,8 @@ export function CleaningWizard({

Origen de los datos

- Elige una extracción completada de este trabajo o sube un CSV para limpiar. + Elige una o varias extracciones completadas, o sube uno o varios CSV. Se limpian + juntos con las mismas reglas en un solo resultado.

@@ -557,7 +630,18 @@ export function CleaningWizard({ {source === "extraction" ? (
- +
+ + {extractionTasks.length > 0 && ( + + )} +
{loadingTasks ? (
Cargando extracciones… @@ -566,13 +650,13 @@ export function CleaningWizard({
No hay extracciones completadas en este trabajo.
) : ( -
+
{extractionTasks.map((task, index) => { - const selected = task.id === selectedTaskId; + const selected = selectedTaskIds.has(task.id); return (
+ {csvFiles.length > 0 && ( +
    + {csvFiles.map((file, index) => ( +
  • +
    + +
    +
    {file.name}
    +
    {formatBytes(file.size)}
    +
    +
    + +
    • + ))} +
+ )} +

+ Puedes seleccionar varios CSV; se consolidan en un solo resultado de limpieza. +

)} @@ -657,6 +763,12 @@ export function CleaningWizard({

Elige una columna, aplica reglas y mira el cambio al instante. {analysis?.row_count != null ? ` · ${analysis.row_count} filas` : ""} + {analysis?.extraction_batch_ids && analysis.extraction_batch_ids.length > 1 + ? ` · ${analysis.extraction_batch_ids.length} extracciones` + : ""} + {analysis?.filenames && analysis.filenames.length > 1 + ? ` · ${analysis.filenames.length} CSV` + : ""} {previewLoading ? " · actualizando…" : ""}

diff --git a/services/desktop_app/src/shared/types.ts b/services/desktop_app/src/shared/types.ts index 3c1fdef..d41f1f0 100644 --- a/services/desktop_app/src/shared/types.ts +++ b/services/desktop_app/src/shared/types.ts @@ -196,8 +196,11 @@ export interface CleaningColumnAnalysis { export interface CleaningAnalysisResponse { batch_id?: string; + extraction_batch_ids?: string[]; filename?: string; + filenames?: string[]; delimiter?: string; + delimiters?: string[]; row_count?: number; columns: CleaningColumnAnalysis[]; sample_rows?: Array>; @@ -219,6 +222,7 @@ export interface CleaningPreviewResponse { export interface CleaningPreviewPayload { rows?: Array>; extraction_batch_id?: string; + extraction_batch_ids?: string[]; rules_per_column: Record; cop_mod_fecha_column?: Record; cop_mod_bruto_column?: Record; @@ -243,7 +247,9 @@ export interface CleaningJob { } export interface RunCleaningPayload { - extraction_batch_id: string; + /** @deprecated Prefer extraction_batch_ids */ + extraction_batch_id?: string; + extraction_batch_ids: string[]; job_id?: string; rules_per_column: Record; cop_mod_fecha_column?: Record;