ahora se procesa de inicio a fin correo por correo
This commit is contained in:
George committed
1 parent
8cdd075d15
commit
603243b7e0
25 files changed
+2234
-334
No files matched your search
@@ -0,0 +1,450 @@
|
||||
import pdfplumber
|
||||
import os
|
||||
import re
|
||||
|
||||
|
||||
TIPO_MAP = {
|
||||
"FACTURA": "01",
|
||||
"BOLETA": "03",
|
||||
"GUIA DE REMISION": "09",
|
||||
"GUIA DE REMISION TRANSPORTISTA": "31",
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# HELPERS INTERNOS (desde leer_pdf.py)
|
||||
# ============================================================================
|
||||
|
||||
def _extract_name_ruc(text_input: str) -> tuple[str | None, str | None]:
|
||||
if not text_input: return None, None
|
||||
full_s = " ".join(text_input.split()).strip()
|
||||
name, ruc = full_s, None
|
||||
if " - " in full_s:
|
||||
name = full_s.split(" - ")[0].strip()
|
||||
r_match = re.search(r"(\d{11})", full_s)
|
||||
if r_match:
|
||||
ruc = r_match.group(1)
|
||||
return name, ruc
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extract_base_data(text: str, pdf_path: str) -> dict:
|
||||
basename = os.path.basename(pdf_path)
|
||||
data = {
|
||||
"archivo": basename,
|
||||
"ruc_emisor": None,
|
||||
"tipo_documento": None,
|
||||
"serie_numero": None,
|
||||
"fecha_emision": None,
|
||||
"ruc_receptor": None,
|
||||
"nombre_receptor": None,
|
||||
"monto_total": None,
|
||||
}
|
||||
ruc_match = re.search(r"R\.U\.C\.?\s*(\d{11})", text, re.IGNORECASE)
|
||||
if ruc_match:
|
||||
data["ruc_emisor"] = ruc_match.group(1)
|
||||
else:
|
||||
if "-" in basename:
|
||||
parts = basename.split('-')
|
||||
if len(parts[0]) == 11 and parts[0].isdigit():
|
||||
data["ruc_emisor"] = parts[0]
|
||||
if not data["ruc_emisor"]:
|
||||
standalone_ruc = re.search(r"(\d{11})", text)
|
||||
if standalone_ruc:
|
||||
data["ruc_emisor"] = standalone_ruc.group(1)
|
||||
|
||||
if "FACTURA ELECTRÓNICA" in text.upper():
|
||||
data["tipo_documento"] = "FACTURA"
|
||||
elif "BOLETA DE VENTA" in text.upper():
|
||||
data["tipo_documento"] = "BOLETA"
|
||||
elif "TRANSPORTISTA" in text.upper() and "GUÍA DE REMISIÓN" in text.upper():
|
||||
data["tipo_documento"] = "GUIA DE REMISION TRANSPORTISTA"
|
||||
elif "GUÍA DE REMISIÓN" in text.upper() or "GRRE:" in text.upper():
|
||||
data["tipo_documento"] = "GUIA DE REMISION"
|
||||
|
||||
serie_numero_match = re.search(r"(?:N°\s*)?([A-Z0-9]{4})\s*-\s*(\d+)", text)
|
||||
if serie_numero_match:
|
||||
data["serie_numero"] = f"{serie_numero_match.group(1)}-{serie_numero_match.group(2)}"
|
||||
elif "GRRE:" in text:
|
||||
grre_match = re.search(r"GRRE:\s*([A-Z0-9-]{4,})", text)
|
||||
if grre_match:
|
||||
data["serie_numero"] = grre_match.group(1)
|
||||
|
||||
date_match = re.search(r"(\d{2})\s*[/-]\s*(\d{2})\s*[/-]\s*(\d{4})", text)
|
||||
if date_match:
|
||||
data["fecha_emision"] = f"{date_match.group(1)}/{date_match.group(2)}/{date_match.group(3)}"
|
||||
|
||||
ruc_label_match = re.search(r"(?:Ruc|REGISTRO ÚNICO DE CONTRIBUYENTES N°)\s*:?\s*(\d{11})", text, re.IGNORECASE)
|
||||
if ruc_label_match and ruc_label_match.group(1) != data["ruc_emisor"]:
|
||||
data["ruc_receptor"] = ruc_label_match.group(1)
|
||||
else:
|
||||
all_rucs = re.findall(r"(?<!\d)(\d{11})(?!\d)", text)
|
||||
for r in all_rucs:
|
||||
if r != data["ruc_emisor"]:
|
||||
data["ruc_receptor"] = r
|
||||
break
|
||||
|
||||
lines = text.split('\n')
|
||||
for line in lines:
|
||||
match = re.search(r"\b(?:SEÑOR\(ES\)|CLIENTE|RAZÓ?N SOCIAL|ADQUIRIENTE|DATOS DEL DESTINATARIO)\s*:?\s*(.+)", line, re.IGNORECASE)
|
||||
if match:
|
||||
name = match.group(1)
|
||||
name = re.split(r"FECHA", name, flags=re.IGNORECASE)[0]
|
||||
name = name.split("-")[0].strip()
|
||||
if len(name) > 2:
|
||||
data["nombre_receptor"] = name
|
||||
break
|
||||
return data
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extract_factura_details(text: str, data: dict) -> None:
|
||||
amount_regex = r"((?:\d{1,3}(?:,\d{3})*|\d+)\.\d{2})"
|
||||
total_patterns = [
|
||||
rf"TOTAL\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
|
||||
rf"IMPORTE TOTAL\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
|
||||
rf"NETO A PAGAR\s*(?:S/|S/\.)?[:\s]*{amount_regex}",
|
||||
]
|
||||
total_found = False
|
||||
for pattern in total_patterns:
|
||||
match = re.search(pattern, text, re.IGNORECASE)
|
||||
if match:
|
||||
data["monto_total"] = match.group(1).replace(",", "")
|
||||
total_found = True
|
||||
break
|
||||
if not total_found:
|
||||
amounts = re.findall(amount_regex, text)
|
||||
if amounts:
|
||||
data["monto_total"] = amounts[-1].replace(",", "")
|
||||
|
||||
data.update({
|
||||
"items": [],
|
||||
"detraccion": {"tiene": False, "porcentaje": None, "monto": None},
|
||||
})
|
||||
|
||||
lines = text.split('\n')
|
||||
table_start_index = -1
|
||||
for i, line in enumerate(lines):
|
||||
line_up = line.upper()
|
||||
headers_found = sum(1 for h in ["CANT", "DESCRIP", "PRECIO", "TOTAL", "UNIT", "VALOR", "CÓDIGO", "ITEM"] if h in line_up)
|
||||
if headers_found >= 3:
|
||||
table_start_index = i
|
||||
break
|
||||
|
||||
if table_start_index != -1:
|
||||
items_lines = lines[table_start_index + 1:]
|
||||
current_item = None
|
||||
for line in items_lines:
|
||||
line = line.strip()
|
||||
if not line: continue
|
||||
if re.search(r"^(?:SUB|TOTAL|IGV|SON:|OP\.|RES|CUO|DET|BANCO|Representación|Podrá|Autorizado|FORMA|CUOTA|GRAVADO|I\.G\.V\.|DESEMBOLSO|OBSERVACION)", line, re.IGNORECASE):
|
||||
break
|
||||
amounts = re.findall(r"(?<![A-Z0-9-])(\d+(?:,\d{3})*(?:\.\d{2,}))(?![A-Z0-9-])", line)
|
||||
if len(amounts) >= 2:
|
||||
if current_item: data["items"].append(current_item)
|
||||
qty = amounts[0]
|
||||
total = amounts[-1]
|
||||
p_unit = amounts[-2]
|
||||
if len(amounts) >= 4:
|
||||
p_unit = amounts[1]
|
||||
desc_line = line
|
||||
for amt in [qty, p_unit, total]:
|
||||
desc_line = desc_line.replace(amt, "", 1)
|
||||
desc_parts = [p.strip() for p in desc_line.split() if p.strip()]
|
||||
codigo = None
|
||||
for i in range(min(3, len(desc_parts))):
|
||||
if re.search(r"[A-Z].*\d|\d.*[A-Z]|[A-Z]{2,}-\d+", desc_parts[i]):
|
||||
codigo = desc_parts[i]
|
||||
desc_parts.pop(i)
|
||||
break
|
||||
if desc_parts and desc_parts[0].isdigit() and len(desc_parts[0]) <= 2:
|
||||
desc_parts.pop(0)
|
||||
descripcion = " ".join(desc_parts)
|
||||
current_item = {
|
||||
"codigo": codigo,
|
||||
"cantidad": qty,
|
||||
"descripcion": descripcion,
|
||||
"precio_unitario": p_unit.replace(",", ""),
|
||||
"total": total.replace(",", ""),
|
||||
"guias_remision": [],
|
||||
"guia_transportista": None,
|
||||
"bolsas": None,
|
||||
"peso": None,
|
||||
"unidad_peso": None,
|
||||
}
|
||||
elif current_item and len(line) > 5:
|
||||
current_item["descripcion"] += " " + line
|
||||
if current_item: data["items"].append(current_item)
|
||||
|
||||
for item in data["items"]:
|
||||
desc = item["descripcion"]
|
||||
guias_remision_matches = re.findall(r"\b(?:GRR\w*|GR\b|G\.R\.R\.?|GUIA REMISI.N)\s*[:/\s]*((?:[TVE][A-Z0-9]{3})-\d+)", desc, re.IGNORECASE)
|
||||
item["guias_remision"] = [m.strip() for m in guias_remision_matches] if guias_remision_matches else []
|
||||
guia_transportista_match = re.search(r"\b(?:GRT\w*|GR\b|G\.R\.T\.?|GUIA TRANSPORTISTA)\s*[:/\s]*((?:[TVE][A-Z0-9]{3})-\d+)", desc, re.IGNORECASE)
|
||||
if guia_transportista_match:
|
||||
item["guia_transportista"] = guia_transportista_match.group(1).strip()
|
||||
bolsas_match = re.search(r"(\d+)\s*(?:BOLSAS|BLS|BOLS|SACOS)\b", desc, re.IGNORECASE)
|
||||
if bolsas_match:
|
||||
item["bolsas"] = bolsas_match.group(1)
|
||||
peso_match = re.search(r"(?:PESO(?: TOTAL)?|CARGA)\s*(?:(TN|KG|KILOS?))?\.?\s*[:\s]\s*([\d,.]+)\s*(TN|KG|KILOS?)?", desc, re.IGNORECASE)
|
||||
if peso_match:
|
||||
item["peso"] = peso_match.group(2).strip()
|
||||
item["unidad_peso"] = (peso_match.group(1) or peso_match.group(3) or "").upper()
|
||||
|
||||
det = data["detraccion"]
|
||||
if re.search(r"\bDETRACC(?:I[OÓ]N|\.)", text, re.IGNORECASE):
|
||||
det["tiene"] = True
|
||||
pct_match = re.search(r"PORCENTAJE\s*[:/]\s*([\d.]+)\s*%", text, re.IGNORECASE)
|
||||
if not pct_match:
|
||||
pct_match = re.search(r"DETRACC[^\n]*?([\d]+(?:\.\d+)?)\s*%", text, re.IGNORECASE)
|
||||
if pct_match:
|
||||
det["porcentaje"] = pct_match.group(1)
|
||||
monto_match = re.search(
|
||||
r"(?:MONTO\s*(?:DETRACC\.?|DE\s*DETRACCI[OÓ]N)|Monto\s*Detracc\.?)\s*[:/]?\s*S/\.?\s*([\d,.]+)",
|
||||
text, re.IGNORECASE,
|
||||
)
|
||||
if not monto_match:
|
||||
monto_match = re.search(r"MONTO\s*DETRACC\w*[:\s]+S?/?\.?\s*([\d,.]+)", text, re.IGNORECASE)
|
||||
if not monto_match:
|
||||
monto_match = re.search(r"DETRACCION\s+\d+%\s+([\d.]+)", text, re.IGNORECASE)
|
||||
if monto_match:
|
||||
raw = monto_match.group(1).replace(",", "")
|
||||
try:
|
||||
det["monto"] = f"{float(raw):.2f}"
|
||||
except ValueError:
|
||||
det["monto"] = raw
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extract_guia_remision_details(text: str, data: dict) -> None:
|
||||
data.update({
|
||||
"peso_total": None, "punto_partida": None, "punto_llegada": None, "placa": None,
|
||||
"marca": None, "chofer": None, "brevete": None, "configuracion_vehicular": None, "items": [],
|
||||
})
|
||||
peso_match = re.search(r"Peso Total (?:TN|KG)?:\s*([\d,.]+)", text, re.IGNORECASE)
|
||||
if not peso_match:
|
||||
peso_match = re.search(r"Peso Total:\s*([\d,.]+)", text, re.IGNORECASE)
|
||||
if peso_match:
|
||||
data["peso_total"] = peso_match.group(1)
|
||||
conf = re.search(r"(?:Configuración vehicular|C\. Vehc\.?):\s*([A-Z0-9/]+)", text, re.IGNORECASE)
|
||||
if conf: data["configuracion_vehicular"] = conf.group(1).strip()
|
||||
for field, pattern in {
|
||||
"placa": r"Placa:\s*([A-Z0-9\s/]+?)(?=\s*(?:C\.|Marca|Peso|$))",
|
||||
"marca": r"Marca:\s*([A-Z0-9\s-]+?)(?=\s*(?:Placa|Chofer|Peso|$))",
|
||||
"chofer": r"Chofer:\s*([A-Z\s.-]+?)(?=\s*(?:Brevete|Placa|Peso|$))",
|
||||
"brevete": r"Brevete:\s*([A-Z0-9]+)",
|
||||
}.items():
|
||||
match = re.search(pattern, text, re.IGNORECASE)
|
||||
if match: data[field] = match.group(1).strip()
|
||||
p_match = re.search(r"Punto de Partida:\s*(.*?)\s*Punto de Llegada:", text, re.DOTALL | re.IGNORECASE)
|
||||
if p_match: data["punto_partida"] = " ".join(p_match.group(1).split()).strip()
|
||||
l_match = re.search(r"Punto de Llegada:\s*(.*?)\s*(?:Fecha de inicio|T[0-9]{3}-|$)", text, re.DOTALL | re.IGNORECASE)
|
||||
if l_match: data["punto_llegada"] = " ".join(l_match.group(1).split()).strip()
|
||||
header = re.search(r"(?:CANT|CANTIDAD)\s+(?:UN|UNIDAD)\s+(?:DESCRIPCIÓN|DESCRIPCION|DETALLE)", text, re.IGNORECASE)
|
||||
if header:
|
||||
items_text = text[header.end():].split("Peso Aprox")[0]
|
||||
current_item = None
|
||||
for line in items_text.split('\n'):
|
||||
line = line.strip()
|
||||
if not line or re.search(r"^(?:Peso Aprox|Punto de|Fecha de|Representación|Placa|Chofer|C\. Vehc)", line, re.IGNORECASE):
|
||||
continue
|
||||
match = re.match(r"^(\d+(?:\.\d+)?)\s+([A-Z]{2,3})\s+(.+)", line)
|
||||
if match:
|
||||
if current_item: data["items"].append(current_item)
|
||||
current_item = {"cantidad": match.group(1), "unidad": match.group(2), "descripcion": match.group(3).strip()}
|
||||
elif current_item:
|
||||
current_item["descripcion"] += " " + line
|
||||
if current_item: data["items"].append(current_item)
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extract_guia_transportista_details(text: str, data: dict) -> None:
|
||||
data.update({
|
||||
"punto_partida": None, "punto_llegada": None, "nombre_remitente": None, "ruc_remitente": None,
|
||||
"nombre_destinatario": None, "ruc_destinatario": None, "documentos_relacionados": [],
|
||||
"unidad_medida_peso_bruto": None, "peso_bruto_total_carga": None, "indicador_pagador_flete": None,
|
||||
"nombre_pagador": None, "ruc_pagador": None, "datos_vehiculos": [], "datos_conductores": [], "observaciones": None,
|
||||
})
|
||||
p_match = re.search(r"Punto de Partida:\s*(.*?)\s*Punto de llegada:", text, re.DOTALL | re.IGNORECASE)
|
||||
if p_match: data["punto_partida"] = " ".join(p_match.group(1).split()).strip()
|
||||
l_match = re.search(r"Punto de llegada:\s*(.*?)\s*Datos del remitente:", text, re.DOTALL | re.IGNORECASE)
|
||||
if l_match: data["punto_llegada"] = " ".join(l_match.group(1).split()).strip()
|
||||
rem_match = re.search(r"Datos del remitente:\s*(.*?)\s*Datos del destinatario:", text, re.DOTALL | re.IGNORECASE)
|
||||
if rem_match: data["nombre_remitente"], data["ruc_remitente"] = _extract_name_ruc(rem_match.group(1))
|
||||
dest_match = re.search(r"Datos del destinatario:\s*(.*?)\s*Documentos Relacionados:", text, re.DOTALL | re.IGNORECASE)
|
||||
if dest_match: data["nombre_destinatario"], data["ruc_destinatario"] = _extract_name_ruc(dest_match.group(1))
|
||||
pag_match = re.search(r"Datos de pagador de flete:\s*(.*?)\s*(?:Observaciones|Esta es|$)", text, re.DOTALL | re.IGNORECASE)
|
||||
if pag_match: data["nombre_pagador"], data["ruc_pagador"] = _extract_name_ruc(pag_match.group(1))
|
||||
rel_match = re.search(r"Documentos Relacionados:\s*(.*?)\s*(?:Bienes por Transportar|Datos de|$)", text, re.DOTALL | re.IGNORECASE)
|
||||
if rel_match:
|
||||
for line in rel_match.group(1).split('\n'):
|
||||
line = line.strip()
|
||||
if not line: continue
|
||||
doc = re.search(r"(?:N°\s*)([A-Z0-9\s-]+)- RUC N°\s*(\d{11})", line, re.IGNORECASE)
|
||||
data["documentos_relacionados"].append({"documento": doc.group(1).strip(), "ruc_relacionado": doc.group(2)} if doc else line)
|
||||
u_match = re.search(r"Unidad de Medida del Peso Bruto:\s*(\w+)", text, re.IGNORECASE)
|
||||
if u_match: data["unidad_medida_peso_bruto"] = u_match.group(1)
|
||||
p_match = re.search(r"Peso Bruto total de la carga:\s*([\d,.]+)", text, re.IGNORECASE)
|
||||
if p_match: data["peso_bruto_total_carga"] = p_match.group(1)
|
||||
ind_match = re.search(r"Indicador del pagador del flete:\s*(\w+)", text, re.IGNORECASE)
|
||||
if ind_match: data["indicador_pagador_flete"] = ind_match.group(1)
|
||||
v_match = re.search(r"Datos de los vehículos:\s*(.*?)\s*Datos de los conductores:", text, re.DOTALL | re.IGNORECASE)
|
||||
if v_match: data["datos_vehiculos"] = [{"placa": p} for p in re.findall(r"Número de placa:\s*([A-Z0-9]+)", v_match.group(1), re.IGNORECASE)]
|
||||
c_match = re.search(r"Datos de los conductores:\s*(.*?)\s*(?:Datos de pagador|Observaciones|$)", text, re.DOTALL | re.IGNORECASE)
|
||||
if c_match:
|
||||
c_text = c_match.group(1)
|
||||
data["datos_conductores"] = [{"nombre": n.strip()} for n in re.findall(r"Principal:\s*([A-Z\s-]+)(?:\s*-|$)", c_text, re.IGNORECASE)]
|
||||
for i, b in enumerate(re.findall(r"Número de lincencia de conducir:\s*([A-Z0-9]+)", c_text, re.IGNORECASE)):
|
||||
if i < len(data["datos_conductores"]): data["datos_conductores"][i]["brevete"] = b
|
||||
obs = re.search(r"Observaciones\s*:\s*(.*?)(?=\s*(?:Esta es|$))", text, re.IGNORECASE | re.DOTALL)
|
||||
if obs: data["observaciones"] = obs.group(1).strip()
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extract_detraccion_spatial(pdf, data: dict) -> None:
|
||||
det = data.get("detraccion", {})
|
||||
if not det.get("tiene"):
|
||||
return
|
||||
if det.get("porcentaje") and det.get("monto"):
|
||||
return
|
||||
for page in pdf.pages:
|
||||
words = page.extract_words()
|
||||
if not words:
|
||||
continue
|
||||
det_words = [w for w in words if re.search(r"Detrac", w["text"], re.IGNORECASE)]
|
||||
if not det_words:
|
||||
continue
|
||||
region_text = None
|
||||
for rect in (page.rects or []):
|
||||
for dw in det_words:
|
||||
if (rect["x0"] <= dw["x0"] <= rect["x1"] and
|
||||
rect["top"] <= dw["top"] <= rect["bottom"]):
|
||||
cropped = page.crop((rect["x0"], rect["top"], rect["x1"], rect["bottom"]))
|
||||
region_text = cropped.extract_text() or ""
|
||||
break
|
||||
if region_text:
|
||||
break
|
||||
if not region_text and det_words:
|
||||
dw = det_words[0]
|
||||
cropped = page.crop((0, max(0, dw["top"] - 30), page.width, min(page.height, dw["bottom"] + 80)))
|
||||
region_text = cropped.extract_text() or ""
|
||||
if not region_text:
|
||||
continue
|
||||
if not det.get("porcentaje"):
|
||||
pct = re.search(r"(\d+(?:\.\d+)?)\s*%\s*Detrac", region_text, re.IGNORECASE)
|
||||
if not pct:
|
||||
pct = re.search(r"PORCENTAJE\s*[:/]\s*(\d+(?:\.\d+)?)\s*%", region_text, re.IGNORECASE)
|
||||
if pct:
|
||||
try:
|
||||
det["porcentaje"] = f"{float(pct.group(1)):.2f}"
|
||||
except ValueError:
|
||||
det["porcentaje"] = pct.group(1)
|
||||
if not det.get("monto"):
|
||||
monto = re.search(r"Detrac[^:]*:\s*([\d,.]+)", region_text, re.IGNORECASE)
|
||||
if not monto:
|
||||
monto = re.search(r"MONTO[^:]*:\s*S?/?\.?\s*([\d,.]+)", region_text, re.IGNORECASE)
|
||||
if monto:
|
||||
raw = monto.group(1).replace(",", "")
|
||||
try:
|
||||
det["monto"] = f"{float(raw):.2f}"
|
||||
except ValueError:
|
||||
det["monto"] = raw
|
||||
break
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# FUNCIONES PÚBLICAS
|
||||
# ============================================================================
|
||||
|
||||
def extraer_datos_pdf(pdf_path: str) -> dict:
|
||||
try:
|
||||
with pdfplumber.open(pdf_path) as pdf:
|
||||
text = "".join(page.extract_text() or "" for page in pdf.pages)
|
||||
data = _extract_base_data(text, pdf_path)
|
||||
if data["tipo_documento"] == "FACTURA":
|
||||
_extract_factura_details(text, data)
|
||||
_extract_detraccion_spatial(pdf, data)
|
||||
elif data["tipo_documento"] == "GUIA DE REMISION":
|
||||
_extract_guia_remision_details(text, data)
|
||||
elif data["tipo_documento"] == "GUIA DE REMISION TRANSPORTISTA":
|
||||
_extract_guia_transportista_details(text, data)
|
||||
return data
|
||||
except Exception as e:
|
||||
return {"archivo": os.path.basename(pdf_path), "error": str(e)}
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def generar_nombre_pdf(data: dict) -> str | None:
|
||||
ruc = data.get("ruc_emisor")
|
||||
tipo_doc = data.get("tipo_documento")
|
||||
serie_numero = data.get("serie_numero")
|
||||
if not ruc or not tipo_doc or not serie_numero:
|
||||
return None
|
||||
abrev = TIPO_MAP.get(tipo_doc.upper())
|
||||
if not abrev:
|
||||
return None
|
||||
if "-" in serie_numero:
|
||||
serie, numero = serie_numero.split("-", 1)
|
||||
numero = numero.lstrip("0") or "0"
|
||||
else:
|
||||
serie = serie_numero
|
||||
numero = ""
|
||||
return f"{ruc}-{abrev}-{serie}-{numero}.pdf"
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def renombrar_pdf(pdf_path: str, data: dict) -> str:
|
||||
dirname = os.path.dirname(pdf_path)
|
||||
nuevo_nombre = generar_nombre_pdf(data)
|
||||
if not nuevo_nombre:
|
||||
return pdf_path
|
||||
nuevo_path = os.path.join(dirname, nuevo_nombre)
|
||||
if pdf_path == nuevo_path:
|
||||
return pdf_path
|
||||
try:
|
||||
os.rename(pdf_path, nuevo_path)
|
||||
return nuevo_path
|
||||
except OSError:
|
||||
return pdf_path
|
||||
|
||||
# ============================================================================
|
||||
|
||||
def _extraer_serie_numero(texto: str) -> tuple[str, str] | None:
|
||||
m = re.search(r'\b([TVE][A-Z0-9]{3})\s*-\s*(\d+)\b', texto)
|
||||
if m:
|
||||
return m.group(1).upper(), m.group(2)
|
||||
return None
|
||||
|
||||
|
||||
def extraer_referencias(items: list) -> list:
|
||||
referencias = []
|
||||
vistos = set()
|
||||
for item in items:
|
||||
for gr_str in item.get("guias_remision", []):
|
||||
parsed = _extraer_serie_numero(gr_str)
|
||||
if not parsed:
|
||||
partes = gr_str.split("-", 1)
|
||||
if len(partes) == 2:
|
||||
parsed = (partes[0].strip(), partes[1].strip())
|
||||
else:
|
||||
continue
|
||||
serie, numero = parsed
|
||||
clave = f"09|{serie}|{numero}"
|
||||
if clave not in vistos:
|
||||
vistos.add(clave)
|
||||
referencias.append({"tipo": "09", "serie": serie, "numero": numero})
|
||||
gt_str = item.get("guia_transportista")
|
||||
if gt_str:
|
||||
parsed = _extraer_serie_numero(gt_str)
|
||||
if not parsed:
|
||||
partes = gt_str.split("-", 1)
|
||||
if len(partes) == 2:
|
||||
parsed = (partes[0].strip(), partes[1].strip())
|
||||
else:
|
||||
continue
|
||||
serie, numero = parsed
|
||||
clave = f"31|{serie}|{numero}"
|
||||
if clave not in vistos:
|
||||
vistos.add(clave)
|
||||
referencias.append({"tipo": "31", "serie": serie, "numero": numero})
|
||||
return referencias
|
||||
Reference in new issue
Block a user