import sys
import os
import re
import json
import requests
import unicodedata
import math
import hashlib
import argparse
from collections import defaultdict
from pdf2image import convert_from_path
import pytesseract
from tqdm import tqdm
# =========================================================
# IMPORTACIÓN PDF DIGITAL
# =========================================================
try:
import pypdf
PYPDF_DISPONIBLE = True
except ImportError:
PYPDF_DISPONIBLE = False
# =========================================================
# CONFIGURACIÓN
# =========================================================
MODEL_NAME = "huihui_ai/phi4-mini-abliterated:3.8b"
EMBEDDING_MODEL = "nomic-embed-text:latest"
OLLAMA_URL = "http://127.0.0.1:11434/api/chat"
OLLAMA_EMBED_URL = "http://127.0.0.1:11434/api/embeddings"
pytesseract.pytesseract.tesseract_cmd = "/data/data/com.termux/files/usr/bin/tesseract"
MAX_CARACTERES_BLOQUE = 1500
OVERLAP = 250
TOP_N_BLOQUES = 4
MIN_SCORE_THRESHOLD = 0.32
MAX_SIMILARIDAD_DUPLICADOS = 0.92
CACHE_VERSION = 2
STOPWORDS = {
"que", "para", "con", "por", "una", "uno", "unos", "unas",
"este", "esta", "estos", "estas", "como", "pero", "del",
"los", "las", "sus", "desde", "hasta", "entre", "sobre",
"donde", "cuando", "quien", "quienes", "porque", "segun",
"más", "mas", "muy", "sin", "sobre", "tambien", "también"
}
# =========================================================
# UTILIDADES DE CACHE
# =========================================================
def calcular_hash_archivo(ruta: str) -> str:
sha = hashlib.sha256()
with open(ruta, "rb") as f:
while True:
data = f.read(8192)
if not data:
break
sha.update(data)
return sha.hexdigest()
def cargar_cache_json(ruta: str):
if not os.path.exists(ruta):
return None, None
try:
with open(ruta, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict) and "meta" in data and "data" in data:
return data["meta"], data["data"]
return None, data
except Exception:
return None, None
def guardar_cache_json(ruta: str, meta: dict, data):
with open(ruta, "w", encoding="utf-8") as f:
json.dump({"meta": meta, "data": data}, f, ensure_ascii=False)
# =========================================================
# NORMALIZACIÓN
# =========================================================
def normalizar_texto(texto: str) -> str:
texto = unicodedata.normalize("NFKD", texto)
texto = "".join(c for c in texto if not unicodedata.combining(c))
texto = texto.lower()
texto = re.sub(r"[^\w\s]", " ", texto, flags=re.UNICODE)
texto = re.sub(r"\s+", " ", texto).strip()
return texto
# =========================================================
# EMBEDDINGS
# =========================================================
def get_embedding(text: str):
try:
texto = re.sub(r"\s+", " ", text).strip()
payload = {
"model": EMBEDDING_MODEL,
"prompt": texto[:7000]
}
r = requests.post(OLLAMA_EMBED_URL, json=payload, timeout=60)
if r.status_code != 200:
return None
return r.json().get("embedding")
except Exception:
return None
# =========================================================
# SIMILITUD COSENO
# =========================================================
def cosine_similarity(a, b):
if not a or not b:
return 0.0
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(x * x for x in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
# =========================================================
# OCR
# =========================================================
def extraer_texto_con_ocr(ruta_pdf: str):
print(" PDF escaneado detectado. Ejecutando OCR completo...")
try:
images = convert_from_path(ruta_pdf, dpi=250, thread_count=2)
texto_completo = []
for i, image in enumerate(tqdm(images, desc="OCR Total")):
texto = pytesseract.image_to_string(image, lang="spa")
texto_completo.append((i + 1, texto.strip()))
return texto_completo
except Exception as e:
print(f"❌ Error OCR: {e}")
return []
# =========================================================
# EXTRACCIÓN INTELIGENTE
# =========================================================
def extraer_texto_inteligente(ruta_pdf: str):
if not PYPDF_DISPONIBLE:
print("⚠️ pypdf no instalado. Saltando a OCR directo.")
return extraer_texto_con_ocr(ruta_pdf)
print(" Analizando páginas del PDF...")
try:
reader = pypdf.PdfReader(ruta_pdf)
texto_paginas = []
for i, pagina in enumerate(reader.pages):
texto = pagina.extract_text() or ""
texto = texto.strip()
if len(texto) < 20:
print(f" Página {i+1} parece imagen o escaneada. OCR parcial...")
try:
images = convert_from_path(
ruta_pdf,
dpi=250,
first_page=i + 1,
last_page=i + 1
)
if images:
texto = pytesseract.image_to_string(images[0], lang="spa")
except Exception as e:
print(f"⚠️ Error en OCR página {i+1}: {e}")
texto_paginas.append((i + 1, texto.strip()))
return texto_paginas
except Exception as e:
print(f"⚠️ Error lectura digital PDF: {e}. Activando fallback OCR...")
return extraer_texto_con_ocr(ruta_pdf)
# =========================================================
# TROCEADO CON PRE-PROCESAMIENTO DE KEYWORDS
# =========================================================
def _crear_bloque(texto_bloque: str, pagina_inicio: int, pagina_fin: int, bloque_id: int):
texto_bloque = texto_bloque.strip()
texto_norm = normalizar_texto(texto_bloque)
return {
"texto": texto_bloque,
"texto_norm": texto_norm,
"palabras_set": set(texto_norm.split()), # OPTIMIZACIÓN: Cacheado en RAM O(1)
"pagina_inicio": pagina_inicio,
"pagina_fin": pagina_fin,
"primeras_palabras": " ".join(texto_bloque.split()[:12]),
"id_bloque": bloque_id
}
def _normalizar_solape(texto: str, overlap: int) -> str:
if len(texto) <= overlap:
return texto.strip()
solape = texto[-overlap:]
corte = solape.find(" ")
if corte != -1 and corte < max(8, overlap // 4):
solape = solape[corte + 1 :]
return solape.strip()
def trocear_con_overlap(paginas_texto, max_chars, overlap):
bloques = []
bloque_actual = []
caracteres = 0
pagina_inicio_bloque = None
pagina_fin_bloque = None
for pagina, texto_pagina in paginas_texto:
parrafos = [p.strip() for p in texto_pagina.split("\n\n") if p.strip()]
if not parrafos and texto_pagina.strip():
parrafos = [texto_pagina.strip()]
for parrafo in parrafos:
if pagina_inicio_bloque is None:
pagina_inicio_bloque = pagina
pagina_fin_bloque = pagina
if caracteres + len(parrafo) > max_chars and bloque_actual:
texto_bloque = "\n\n".join(bloque_actual)
bloques.append(
_crear_bloque(
texto_bloque,
pagina_inicio_bloque,
pagina_fin_bloque,
len(bloques) + 1
)
)
solape = _normalizar_solape(texto_bloque, overlap)
bloque_actual = [solape, parrafo] if solape else [parrafo]
caracteres = len(solape) + len(parrafo)
pagina_inicio_bloque = pagina
else:
bloque_actual.append(parrafo)
caracteres += len(parrafo) + 2
if bloque_actual:
texto_bloque = "\n\n".join(bloque_actual)
bloques.append(
_crear_bloque(
texto_bloque,
pagina_inicio_bloque if pagina_inicio_bloque is not None else 1,
pagina_fin_bloque if pagina_fin_bloque is not None else 1,
len(bloques) + 1
)
)
return bloques
# =========================================================
# IDF (Optimizado con el set precalculado)
# =========================================================
def calcular_idf(bloques):
df = defaultdict(int)
for bloque in bloques:
# Consumo directo del set cacheado
for palabra in bloque["palabras_set"]:
if len(palabra) > 3:
df[palabra] += 1
total = len(bloques)
return {
word: math.log((total + 1) / (count + 1)) + 1
for word, count in df.items()
}
# =========================================================
# BÚSQUEDA HÍBRIDA (Filtro por sets e Invalidation de Duplicados)
# =========================================================
def buscar_bloques_relevantes(query, bloques, embeddings, idf, top_n):
query_norm = normalizar_texto(query)
palabras_query = [p for p in query_norm.split() if len(p) > 2 and p not in STOPWORDS]
query_emb = get_embedding(query)
ranking = []
for idx, bloque in enumerate(bloques):
# OPTIMIZACIÓN: Carga limpia sin llamadas a metodos de string recurrentes
palabras_bloque = bloque["palabras_set"]
# Componente semántica
score_semantico = 0.0
if embeddings and idx < len(embeddings) and embeddings[idx] and query_emb:
score_semantico = cosine_similarity(query_emb, embeddings[idx])
# Componente léxica
score_keywords = 0.0
if palabras_query:
coincidencias = [p for p in palabras_query if p in palabras_bloque]
if coincidencias:
acumulado = sum(idf.get(palabra, 1.0) for palabra in coincidencias)
total_posible = sum(idf.get(palabra, 1.0) for palabra in palabras_query)
ratio = acumulado / total_posible if total_posible > 0 else 0.0
score_keywords = ratio * 0.35
score_final = (score_semantico * 0.65) + score_keywords
ranking.append({
"idx": idx,
"score": score_final,
"bloque": bloque
})
ranking.sort(key=lambda x: x["score"], reverse=True)
resultados = []
embeddings_seleccionados = []
# Selección dinámica aplicando el umbral de de-duplicación semántica
for item in ranking:
if len(resultados) >= top_n:
break
bloque = item["bloque"]
idx = item["idx"]
emb_candidato = embeddings[idx] if embeddings and idx < len(embeddings) else None
duplicado = False
if emb_candidato:
for emb_sel in embeddings_seleccionados:
if cosine_similarity(emb_candidato, emb_sel) > MAX_SIMILARIDAD_DUPLICADOS:
duplicado = True
break
if duplicado:
continue
resultados.append({
**bloque,
"score": round(item["score"], 3)
})
if emb_candidato:
embeddings_seleccionados.append(emb_candidato)
return resultados
# =========================================================
# LLM
# =========================================================
def consultar_llm_con_contexto(query, bloques_seleccionados):
contexto = ""
for b in bloques_seleccionados:
contexto += (
f"--- FRAGMENTO {b['id_bloque']} "
f"(Páginas: {b['pagina_inicio']}–{b['pagina_fin']}) "
f"[Score: {b['score']}]\n"
)
contexto += f"{b['texto']}\n"
contexto += "--- FIN FRAGMENTO ---\n\n"
system_prompt = (
"Eres un asistente preciso.\n"
"Usa SOLO la información de los fragmentos.\n\n"
"Formato obligatorio:\n"
"FRAGMENTOS USADOS: [números]\n"
"CONCLUSIÓN: [respuesta]\n\n"
"Si no hay información suficiente responde exactamente:\n"
"'No lo sé, la información solicitada no está disponible en los fragmentos indexados.'"
)
payload = {
"model": MODEL_NAME,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Fragmentos:\n\n{contexto}\nPregunta: {query}"}
],
"stream": False,
"options": {
"temperature": 0.0,
"num_ctx": 8192
}
}
try:
r = requests.post(OLLAMA_URL, json=payload, timeout=900)
if r.status_code != 200:
return f"❌ Error Ollama: {r.text}"
return r.json()["message"]["content"]
except Exception as e:
return f"❌ Error conexión: {e}"
# =========================================================
# MAIN
# =========================================================
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--force", action="store_true")
args = parser.parse_args()
print(" RAG Híbrido Inteligente Profesional\n")
nombre = input("Nombre PDF: ").strip()
ruta_pdf = f"/storage/emulated/0/Documents/{nombre}"
if not nombre.lower().endswith(".pdf"):
ruta_pdf += ".pdf"
if not os.path.exists(ruta_pdf):
print("❌ Archivo no encontrado.")
sys.exit(1)
base = os.path.splitext(ruta_pdf)[0]
ruta_json_texto = base + "_texto.json"
ruta_emb = base + "_embeddings.json"
pdf_hash = calcular_hash_archivo(ruta_pdf)
# =====================================================
# TEXTO CACHEADO
# =====================================================
texto_paginas = None
meta_texto, data_texto = cargar_cache_json(ruta_json_texto)
cache_texto_valida = (
not args.force and
meta_texto and
meta_texto.get("version") == CACHE_VERSION and
meta_texto.get("pdf_hash") == pdf_hash and
isinstance(data_texto, list)
)
if cache_texto_valida:
texto_paginas = data_texto
print(" Cargando texto desde caché.")
else:
texto_paginas = extraer_texto_inteligente(ruta_pdf)
guardar_cache_json(
ruta_json_texto,
{
"version": CACHE_VERSION,
"pdf_hash": pdf_hash
},
texto_paginas
)
print(" Texto guardado en caché.")
# =====================================================
# BLOQUES
# =====================================================
bloques = trocear_con_overlap(texto_paginas, MAX_CARACTERES_BLOQUE, OVERLAP)
print(f" Bloques generados: {len(bloques)}")
# =====================================================
# EMBEDDINGS CACHEADOS
# =====================================================
embeddings = None
meta_emb, data_emb = cargar_cache_json(ruta_emb)
cache_emb_valida = (
not args.force and
meta_emb and
meta_emb.get("version") == CACHE_VERSION and
meta_emb.get("pdf_hash") == pdf_hash and
meta_emb.get("embedding_model") == EMBEDDING_MODEL and
meta_emb.get("max_chars") == MAX_CARACTERES_BLOQUE and
meta_emb.get("overlap") == OVERLAP and
isinstance(data_emb, list) and
len(data_emb) == len(bloques)
)
if cache_emb_valida:
embeddings = data_emb
print(" Cargando embeddings desde caché.")
else:
embeddings = []
print(" Generando embeddings...")
for bloque in tqdm(bloques):
emb = get_embedding(bloque["texto"])
embeddings.append(emb)
guardar_cache_json(
ruta_emb,
{
"version": CACHE_VERSION,
"pdf_hash": pdf_hash,
"embedding_model": EMBEDDING_MODEL,
"max_chars": MAX_CARACTERES_BLOQUE,
"overlap": OVERLAP
},
embeddings
)
print(" Embeddings guardados en caché.")
# =====================================================
# IDF
# =====================================================
idf = calcular_idf(bloques)
print("\n Sistema híbrido listo para consultas.\n")
while True:
pregunta = input("→ ").strip()
if pregunta.lower() in ["salir", "exit", "q"]:
break
if not pregunta:
continue
if pregunta.lower() == "resumen":
bloques_top = [{**b, "score": 1.0} for b in bloques[:6]]
else:
print(" Buscando...")
bloques_top = buscar_bloques_relevantes(
pregunta,
bloques,
embeddings,
idf,
TOP_N_BLOQUES
)
mejor_score = bloques_top[0]["score"] if bloques_top else 0.0
print(f" Mejor score obtenido: {mejor_score}")
if mejor_score < MIN_SCORE_THRESHOLD:
print("\n⚠️ No encuentro información relevante en el documento para esta consulta.")
continue
print(" Consultando IA...")
respuesta = consultar_llm_con_contexto(pregunta, bloques_top)
print("\n" + "=" * 90)
print(" FRAGMENTOS RECUPERADOS ORIGINALES")
print("=" * 90)
for b in bloques_top:
print(
f"\n Bloque {b['id_bloque']} | "
f"Páginas {b['pagina_inicio']}–{b['pagina_fin']} | "
f"Score {b['score']}"
)
print("-" * 50)
print(b["texto"][:1200])
print()
print("=" * 90)
print("\n RESPUESTA IA\n")
print(respuesta)
print("\n" + "=" * 90)
if __name__ == "__main__":
main()
