387 lines
14 KiB
Python
387 lines
14 KiB
Python
"""Diálogo de selección de campos basado en `customtkinter`.
|
|
|
|
Permite buscar, seleccionar/deseleccionar y revisar una muestra de la primera
|
|
factura para facilitar la elección de paths del schema.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Dict, List, Callable, Optional
|
|
|
|
import customtkinter as ctk
|
|
import threading
|
|
import tkinter.messagebox as mb
|
|
from application.app_prefs import get_allowed_document_types
|
|
|
|
DOCUMENT_TYPE_HINT_ALIASES = {
|
|
"CC", "TI", "CE", "RC", "PA", "PPT", "PEP", "EPT", "NUIP",
|
|
"PT", "PAS", "PASS",
|
|
}
|
|
|
|
|
|
class FieldSelectorDialog(ctk.CTkToplevel):
|
|
"""Ventana modal para elegir campos del schema a extraer."""
|
|
def __init__(self, master, schema: Dict[str, str], sample: Optional[dict] = None, initial_selected: Optional[List[str]] = None):
|
|
"""Crea el diálogo con lista scrolleable y búsqueda.
|
|
|
|
Args:
|
|
master: Ventana padre.
|
|
schema: Mapa path->tipo (ayuda visual en etiqueta del checkbox).
|
|
sample: JSON de muestra para vista previa.
|
|
initial_selected: Paths seleccionados inicialmente.
|
|
"""
|
|
super().__init__(master)
|
|
self.title("Seleccionar campos del schema")
|
|
self.geometry("800x600")
|
|
self.transient(master)
|
|
self.grab_set()
|
|
|
|
self._schema = schema
|
|
self._sample = sample
|
|
self._on_submit: Optional[Callable[[List[str], Dict[str, str]], None]] = None
|
|
self._on_reprocess: Optional[Callable[[], tuple[Dict[str, str], dict]]] = None
|
|
|
|
self.search_entry = ctk.CTkEntry(self, placeholder_text="Buscar campo...")
|
|
self.search_entry.grid(row=0, column=0, columnspan=3, sticky="we", padx=10, pady=(10, 5))
|
|
self.search_entry.bind("<KeyRelease>", self._on_search)
|
|
|
|
self.scroll = ctk.CTkScrollableFrame(self)
|
|
self.scroll.grid(row=1, column=0, columnspan=3, sticky="nsew", padx=10, pady=5)
|
|
|
|
self.select_all_btn = ctk.CTkButton(self, text="Seleccionar todo", command=self._select_all)
|
|
self.select_all_btn.grid(row=2, column=0, padx=10, pady=10, sticky="w")
|
|
self.clear_btn = ctk.CTkButton(self, text="Limpiar", command=self._clear_all)
|
|
self.clear_btn.grid(row=2, column=1, padx=10, pady=10, sticky="w")
|
|
self._right_btns = ctk.CTkFrame(self)
|
|
self._right_btns.grid(row=2, column=2, padx=10, pady=10, sticky="e")
|
|
self.sample_btn = ctk.CTkButton(self._right_btns, text="Ver muestra", command=self._show_sample)
|
|
self.sample_btn.pack(side="left", padx=(0, 8))
|
|
self.reprocess_btn = ctk.CTkButton(self._right_btns, text="Reprocesar", command=self._reprocess)
|
|
self.reprocess_btn.pack(side="left")
|
|
|
|
self.cancel_btn = ctk.CTkButton(self, text="Cancelar", command=self._cancel)
|
|
self.cancel_btn.grid(row=3, column=0, padx=10, pady=(0, 10), sticky="w")
|
|
self.ok_btn = ctk.CTkButton(self, text="Aceptar", command=self._accept)
|
|
self.ok_btn.grid(row=3, column=2, padx=10, pady=(0, 10), sticky="e")
|
|
|
|
self.grid_columnconfigure(0, weight=1)
|
|
self.grid_columnconfigure(1, weight=0)
|
|
self.grid_columnconfigure(2, weight=1)
|
|
self.grid_rowconfigure(1, weight=1)
|
|
|
|
self._vars: Dict[str, ctk.BooleanVar] = {}
|
|
self._rows: Dict[str, ctk.CTkFrame] = {}
|
|
self._alias_entries: Dict[str, ctk.CTkEntry] = {}
|
|
self._group_labels: Dict[str, ctk.CTkLabel] = {}
|
|
self._group_paths: Dict[str, list[str]] = {}
|
|
|
|
initial_set = set(initial_selected or [])
|
|
# Construir grupos por prefijo (Tipo) cuando el path contiene '/'
|
|
grouped: Dict[str, list[tuple[str, str]]] = {}
|
|
for path, typ in sorted(self._schema.items(), key=lambda x: x[0]):
|
|
if '/' in path:
|
|
group, sub = path.split('/', 1)
|
|
else:
|
|
group, sub = "General", path
|
|
grouped.setdefault(group, []).append((path, typ))
|
|
# Crear filas agrupadas
|
|
rows_container = self.scroll
|
|
for group_name, items in grouped.items():
|
|
# Encabezado del grupo
|
|
glabel = ctk.CTkLabel(rows_container, text=group_name, anchor="w")
|
|
glabel.pack(anchor="w", fill="x", padx=8, pady=(6, 2))
|
|
self._group_labels[group_name] = glabel
|
|
self._group_paths[group_name] = []
|
|
# Filas del grupo
|
|
for path, typ in items:
|
|
var = ctk.BooleanVar(value=(path in initial_set))
|
|
row = ctk.CTkFrame(rows_container)
|
|
row.pack(anchor="w", fill="x", padx=8, pady=2)
|
|
cb = ctk.CTkCheckBox(row, text=f"{path} ({typ})", variable=var)
|
|
cb.grid(row=0, column=0, sticky="w")
|
|
alias_entry = ctk.CTkEntry(row, width=260, placeholder_text="Nombre en CSV")
|
|
try:
|
|
alias_entry.insert(0, path)
|
|
except Exception:
|
|
pass
|
|
alias_entry.grid(row=0, column=1, padx=(12, 0))
|
|
self._add_document_type_hint(row, path)
|
|
try:
|
|
alias_entry.bind("<KeyRelease>", lambda _e: self._validate_aliases())
|
|
except Exception:
|
|
pass
|
|
def _toggle(ent: ctk.CTkEntry, v: ctk.BooleanVar = var, p: str = path):
|
|
try:
|
|
if p == 'radicado':
|
|
ent.grid_remove()
|
|
else:
|
|
if v.get():
|
|
ent.grid(row=0, column=1, padx=(12, 0))
|
|
else:
|
|
ent.grid_remove()
|
|
except Exception:
|
|
pass
|
|
self._validate_aliases()
|
|
_toggle(alias_entry)
|
|
cb.configure(command=lambda e=alias_entry, v=var, p=path: _toggle(e, v, p))
|
|
self._vars[path] = var
|
|
self._rows[path] = row
|
|
self._alias_entries[path] = alias_entry
|
|
self._group_paths[group_name].append(path)
|
|
self._validate_aliases()
|
|
|
|
def set_on_submit(self, callback: Callable[[List[str], Dict[str, str]], None]) -> None:
|
|
"""Registra el callback a invocar con (selected_fields, alias_map)."""
|
|
self._on_submit = callback
|
|
|
|
def set_on_reprocess(self, callback: Callable[[], tuple[Dict[str, str], dict]]) -> None:
|
|
"""Registra el callback a invocar para reprocesar y actualizar (schema, sample)."""
|
|
self._on_reprocess = callback
|
|
|
|
def _on_search(self, _event=None):
|
|
"""Filtra filas visibles por coincidencia de texto en el path."""
|
|
query = self.search_entry.get().strip().lower()
|
|
|
|
# Lista ordenada de grupos según se crearon
|
|
ordered_groups = list(self._group_labels.keys())
|
|
|
|
for group_name in ordered_groups:
|
|
paths = self._group_paths.get(group_name, [])
|
|
# Determinar qué filas de este grupo coinciden
|
|
visible_paths = []
|
|
for p in paths:
|
|
if query in p.lower():
|
|
visible_paths.append(p)
|
|
|
|
# Si hay filas visibles, mostrar grupo y esas filas; si no, ocultar todo el grupo
|
|
lbl = self._group_labels.get(group_name)
|
|
if visible_paths:
|
|
if lbl:
|
|
lbl.pack(anchor="w", fill="x", padx=8, pady=(6, 2))
|
|
for p in paths:
|
|
row = self._rows.get(p)
|
|
if not row: continue
|
|
if p in visible_paths:
|
|
row.pack(anchor="w", fill="x", padx=8, pady=2)
|
|
else:
|
|
row.pack_forget()
|
|
else:
|
|
if lbl:
|
|
lbl.pack_forget()
|
|
for p in paths:
|
|
row = self._rows.get(p)
|
|
if row:
|
|
row.pack_forget()
|
|
|
|
def _select_all(self):
|
|
"""Marca todos los campos y muestra las entradas de alias (excepto radicado)."""
|
|
for path, var in self._vars.items():
|
|
var.set(True)
|
|
try:
|
|
ent = self._alias_entries.get(path)
|
|
if ent is not None:
|
|
if path == 'radicado':
|
|
ent.grid_remove()
|
|
else:
|
|
ent.grid(row=0, column=1, padx=(12, 0))
|
|
except Exception:
|
|
pass
|
|
self._validate_aliases()
|
|
|
|
def _clear_all(self):
|
|
"""Desmarca todos los campos y oculta entradas de alias."""
|
|
for path, var in self._vars.items():
|
|
var.set(False)
|
|
try:
|
|
ent = self._alias_entries.get(path)
|
|
if ent is not None:
|
|
ent.grid_remove()
|
|
except Exception:
|
|
pass
|
|
self._validate_aliases()
|
|
|
|
def _show_sample(self):
|
|
"""Muestra la muestra JSON en un toplevel de solo lectura (si existe)."""
|
|
if self._sample is None:
|
|
return
|
|
win = ctk.CTkToplevel(self)
|
|
win.title("Muestra de la primera factura")
|
|
win.geometry("700x500")
|
|
win.transient(self)
|
|
win.lift()
|
|
try:
|
|
win.attributes("-topmost", True)
|
|
# mantener al frente un instante y luego permitir comportamiento normal
|
|
win.after(300, lambda: win.attributes("-topmost", False))
|
|
except Exception:
|
|
pass
|
|
win.focus_force()
|
|
txt = ctk.CTkTextbox(win)
|
|
txt.pack(expand=True, fill="both", padx=10, pady=10)
|
|
txt.insert("end", json.dumps(self._sample, ensure_ascii=False, indent=2))
|
|
txt.configure(state="disabled")
|
|
|
|
def _accept(self):
|
|
"""Construye selección y alias y cierra devolviendo datos al callback."""
|
|
selected = [p for p, v in self._vars.items() if v.get()]
|
|
alias_map: Dict[str, str] = {}
|
|
for p in selected:
|
|
ent = self._alias_entries.get(p)
|
|
alias = None
|
|
try:
|
|
alias = (ent.get().strip() if ent is not None else "")
|
|
except Exception:
|
|
alias = ""
|
|
if not alias:
|
|
alias = p
|
|
# 'radicado' permanece fijo si aparece
|
|
if p == 'radicado':
|
|
alias = 'radicado'
|
|
alias_map[p] = alias
|
|
if self._on_submit:
|
|
self._on_submit(selected, alias_map)
|
|
self.destroy()
|
|
|
|
def _cancel(self):
|
|
"""Cancela la selección (devuelve listas vacías) y cierra el diálogo."""
|
|
if self._on_submit:
|
|
self._on_submit([], {})
|
|
self.destroy()
|
|
|
|
def _validate_aliases(self):
|
|
"""Valida alias. (Validación de duplicados eliminada para permitir fusión de columnas)."""
|
|
# Siempre permitir aceptar, ya que los duplicados se fusionarán en el backend.
|
|
try:
|
|
self.ok_btn.configure(state="normal")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# --- Reprocesar (reextraer primer documento y refrescar lista) ---
|
|
def _set_busy(self, busy: bool) -> None:
|
|
"""Habilita/deshabilita controles y cambia texto del botón Reprocesar."""
|
|
try:
|
|
state = "disabled" if busy else "normal"
|
|
self.select_all_btn.configure(state=state)
|
|
self.clear_btn.configure(state=state)
|
|
self.sample_btn.configure(state=state)
|
|
self.ok_btn.configure(state=state)
|
|
self.cancel_btn.configure(state=state)
|
|
self.search_entry.configure(state=state)
|
|
self.reprocess_btn.configure(text=("Reprocesando…" if busy else "Reprocesar"), state=("disabled" if busy else "normal"))
|
|
except Exception:
|
|
pass
|
|
|
|
def _reprocess(self):
|
|
"""Ejecuta reproceso en hilo de fondo y actualiza schema/muestra preservando selección y alias existentes."""
|
|
if self._on_reprocess is None:
|
|
return
|
|
self._set_busy(True)
|
|
|
|
def worker():
|
|
try:
|
|
new_schema, new_sample = self._on_reprocess()
|
|
def apply_updates():
|
|
self._apply_new_schema_sample(new_schema, new_sample)
|
|
self._set_busy(False)
|
|
self.after(0, apply_updates)
|
|
except Exception as e:
|
|
def on_err():
|
|
try:
|
|
mb.showerror("Error de reproceso", f"No se pudo reprocesar: {e}")
|
|
except Exception:
|
|
pass
|
|
self._set_busy(False)
|
|
self.after(0, on_err)
|
|
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
def _apply_new_schema_sample(self, new_schema: Dict[str, str], new_sample: dict) -> None:
|
|
"""Reconstruye la lista de campos preservando selección y alias cuando sea posible."""
|
|
# Guardar selección y alias actuales
|
|
prev_selected = {p for p, v in self._vars.items() if v.get()}
|
|
prev_alias: Dict[str, str] = {}
|
|
for p, ent in self._alias_entries.items():
|
|
try:
|
|
val = (ent.get().strip() if ent is not None else "") or p
|
|
except Exception:
|
|
val = p
|
|
if p == 'radicado':
|
|
val = 'radicado'
|
|
prev_alias[p] = val
|
|
|
|
# Limpiar filas actuales
|
|
try:
|
|
for _, frame in self._rows.items():
|
|
frame.destroy()
|
|
except Exception:
|
|
pass
|
|
self._vars.clear()
|
|
self._rows.clear()
|
|
self._alias_entries.clear()
|
|
|
|
# Asignar nuevos datos
|
|
self._schema = new_schema or {}
|
|
self._sample = new_sample
|
|
|
|
rows_container = self.scroll
|
|
for path in sorted(self._schema.keys()):
|
|
var = ctk.BooleanVar(value=(path in prev_selected))
|
|
row = ctk.CTkFrame(rows_container)
|
|
row.pack(anchor="w", fill="x", padx=8, pady=2)
|
|
cb = ctk.CTkCheckBox(row, text=f"{path} ({self._schema[path]})", variable=var)
|
|
cb.grid(row=0, column=0, sticky="w")
|
|
alias_entry = ctk.CTkEntry(row, width=260, placeholder_text="Nombre en CSV")
|
|
# Establecer alias previo si existe, si no por defecto al path
|
|
alias_value = prev_alias.get(path, path)
|
|
try:
|
|
alias_entry.insert(0, alias_value)
|
|
except Exception:
|
|
pass
|
|
alias_entry.grid(row=0, column=1, padx=(12, 0))
|
|
self._add_document_type_hint(row, path)
|
|
try:
|
|
alias_entry.bind("<KeyRelease>", lambda _e: self._validate_aliases())
|
|
except Exception:
|
|
pass
|
|
def _toggle(ent: ctk.CTkEntry, v: ctk.BooleanVar = var, p: str = path):
|
|
try:
|
|
if p == 'radicado':
|
|
ent.grid_remove()
|
|
else:
|
|
if v.get():
|
|
ent.grid(row=0, column=1, padx=(12, 0))
|
|
else:
|
|
ent.grid_remove()
|
|
except Exception:
|
|
pass
|
|
self._validate_aliases()
|
|
_toggle(alias_entry)
|
|
cb.configure(command=lambda e=alias_entry, v=var, p=path: _toggle(e, v, p))
|
|
self._vars[path] = var
|
|
self._rows[path] = row
|
|
self._alias_entries[path] = alias_entry
|
|
|
|
# Revalidar y re-aplicar filtro de búsqueda actual
|
|
self._validate_aliases()
|
|
try:
|
|
self._on_search()
|
|
except Exception:
|
|
pass
|
|
|
|
def _add_document_type_hint(self, row: ctk.CTkFrame, path: str) -> None:
|
|
"""Muestra una ayuda breve bajo claves tipo CC, TI, CE, PAS, etc."""
|
|
key_tail = str(path or "").split("/")[-1].strip().upper()
|
|
valid_hint_keys = set(get_allowed_document_types()) | DOCUMENT_TYPE_HINT_ALIASES
|
|
if key_tail not in valid_hint_keys:
|
|
return
|
|
|
|
hint = ctk.CTkLabel(
|
|
row,
|
|
text="Si a este campo le pones el alias 'documento_forzate', se guardará con la sigla exacta detectada, por ejemplo 'CC 123456789' o 'EPT 123456789'.",
|
|
anchor="w",
|
|
justify="left",
|
|
)
|
|
hint.grid(row=1, column=0, columnspan=2, sticky="w", padx=(28, 0), pady=(2, 0))
|
|
|