"""
Core business logic for graph tools.
This module contains the logic for exploring connections between notes,
finding backlinks, searching tags, and analyzing graph structure.
"""
from typing import Dict, List
from ..config import get_vault_path
from ..result import Result
from ..utils import extract_internal_links, extract_tags_from_content
def get_backlinks(nombre_nota: str) -> Result[str]:
"""
Get all notes that link to the specified note.
Args:
nombre_nota: Name of the note (with or without .md)
Returns:
Result with list of backlinks.
"""
try:
vault_path = get_vault_path()
if not vault_path:
return Result.fail("La ruta del vault no está configurada.")
# Normalize name
nombre_limpio = nombre_nota.replace(".md", "")
backlinks: List[Dict[str, str]] = []
for archivo in vault_path.rglob("*.md"):
# Ignore self
if archivo.stem == nombre_limpio:
continue
try:
with open(archivo, "r", encoding="utf-8") as f:
contenido = f.read()
enlaces = extract_internal_links(contenido)
for enlace in enlaces:
enlace_limpio = enlace.split("|")[0].strip()
if enlace_limpio == nombre_limpio:
ruta_rel = archivo.relative_to(vault_path)
backlinks.append(
{
"nota": archivo.stem,
"ruta": str(ruta_rel),
}
)
break
except Exception:
continue
if not backlinks:
return Result.ok(f"🔗 No se encontraron backlinks hacia '{nombre_nota}'")
resultado = (
f"🔗 **Backlinks hacia '{nombre_nota}'** ({len(backlinks)} notas):\n\n"
)
for bl in backlinks:
resultado += f" • [[{bl['nota']}]] - {bl['ruta']}\n"
return Result.ok(resultado)
except Exception as e:
return Result.fail(f"Error al obtener backlinks: {e}")
def get_notes_by_tag(tag: str) -> Result[str]:
"""
Find all notes containing a specific tag.
Args:
tag: Tag to search for (with or without #)
Returns:
Result with list of notes.
"""
try:
vault_path = get_vault_path()
if not vault_path:
return Result.fail("La ruta del vault no está configurada.")
tag_limpia = tag.lstrip("#")
notas_con_tag: List[Dict[str, str]] = []
for archivo in vault_path.rglob("*.md"):
try:
with open(archivo, "r", encoding="utf-8") as f:
contenido = f.read()
tags = extract_tags_from_content(contenido)
if tag_limpia in tags:
ruta_rel = archivo.relative_to(vault_path)
notas_con_tag.append(
{
"nota": archivo.stem,
"ruta": str(ruta_rel),
}
)
except Exception:
continue
if not notas_con_tag:
return Result.ok(f"🏷️ No se encontraron notas con la etiqueta #{tag_limpia}")
resultado = (
f"🏷️ **Notas con #{tag_limpia}** ({len(notas_con_tag)} encontradas):\n\n"
)
por_carpeta: Dict[str, List[str]] = {}
for nota in notas_con_tag:
carpeta = (
str(nota["ruta"]).rsplit("/", 1)[0] if "/" in nota["ruta"] else "Raíz"
)
if carpeta not in por_carpeta:
por_carpeta[carpeta] = []
por_carpeta[carpeta].append(nota["nota"])
for carpeta, notas in sorted(por_carpeta.items()):
resultado += f"📁 {carpeta}:\n"
for nombre_nota in sorted(notas):
resultado += f" • [[{nombre_nota}]]\n"
resultado += "\n"
return Result.ok(resultado)
except Exception as e:
return Result.fail(f"Error al buscar por tag: {e}")
def get_local_graph(nombre_nota: str, profundidad: int = 1) -> Result[str]:
"""
Get local graph for a note: outgoing and incoming links.
Args:
nombre_nota: Name of the central note.
profundidad: Depth levels (1 = direct connections only).
Returns:
Result with graph visualization.
"""
try:
vault_path = get_vault_path()
if not vault_path:
return Result.fail("La ruta del vault no está configurada.")
nombre_limpio = nombre_nota.replace(".md", "")
# Find note
nota_path = None
for archivo in vault_path.rglob("*.md"):
if archivo.stem == nombre_limpio:
nota_path = archivo
break
if not nota_path:
return Result.fail(f"No se encontró la nota '{nombre_nota}'")
# Get outgoing links
with open(nota_path, "r", encoding="utf-8") as f:
contenido = f.read()
enlaces_salientes = extract_internal_links(contenido)
enlaces_salientes = [e.split("|")[0].strip() for e in enlaces_salientes]
enlaces_salientes = list(set(enlaces_salientes))
# Get backlinks
backlinks = []
for archivo in vault_path.rglob("*.md"):
if archivo.stem == nombre_limpio:
continue
try:
with open(archivo, "r", encoding="utf-8") as f:
cont = f.read()
enlaces = extract_internal_links(cont)
for enlace in enlaces:
enlace_limpio = enlace.split("|")[0].strip()
if enlace_limpio == nombre_limpio:
backlinks.append(archivo.stem)
break
except Exception:
continue
resultado = f"🕸️ **Grafo Local de '{nombre_nota}'**\n\n"
resultado += f"📤 **Enlaces salientes** ({len(enlaces_salientes)}):\n"
if enlaces_salientes:
for enlace in sorted(enlaces_salientes)[:15]:
resultado += f" → [[{enlace}]]\n"
if len(enlaces_salientes) > 15:
resultado += f" ... y {len(enlaces_salientes) - 15} más\n"
else:
resultado += " (ninguno)\n"
resultado += f"\n📥 **Backlinks** ({len(backlinks)}):\n"
if backlinks:
for bl in sorted(backlinks)[:15]:
resultado += f" ← [[{bl}]]\n"
if len(backlinks) > 15:
resultado += f" ... y {len(backlinks) - 15} más\n"
else:
resultado += " (ninguno)\n"
total = len(enlaces_salientes) + len(backlinks)
resultado += f"\n📊 **Conectividad total**: {total} conexiones"
return Result.ok(resultado)
except Exception as e:
return Result.fail(f"Error al obtener grafo: {e}")
def find_orphan_notes() -> Result[str]:
"""
Find orphan notes: those without incoming or outgoing links.
Returns:
Result with list of orphan notes.
"""
try:
vault_path = get_vault_path()
if not vault_path:
return Result.fail("La ruta del vault no está configurada.")
enlaces_salientes_por_nota: Dict[str, List[str]] = {}
todos_los_enlaces: set = set()
for archivo in vault_path.rglob("*.md"):
try:
with open(archivo, "r", encoding="utf-8") as f:
contenido = f.read()
enlaces = extract_internal_links(contenido)
enlaces_limpios = [e.split("|")[0].strip() for e in enlaces]
enlaces_salientes_por_nota[archivo.stem] = enlaces_limpios
todos_los_enlaces.update(enlaces_limpios)
except Exception:
continue
notas_huerfanas = []
for archivo in vault_path.rglob("*.md"):
nombre = archivo.stem
if any(x in str(archivo) for x in [".git", ".obsidian", "ZZ_"]):
continue
tiene_salientes = bool(enlaces_salientes_por_nota.get(nombre, []))
recibe_enlaces = nombre in todos_los_enlaces
if not tiene_salientes and not recibe_enlaces:
ruta_rel = archivo.relative_to(vault_path)
notas_huerfanas.append(
{
"nota": nombre,
"ruta": str(ruta_rel),
}
)
if not notas_huerfanas:
return Result.ok(
"✅ No hay notas huérfanas. Todas están conectadas al grafo."
)
resultado = f"🔍 **Notas Huérfanas** ({len(notas_huerfanas)}):\n\n"
resultado += "Estas notas no tienen enlaces entrantes ni salientes:\n\n"
for nota in notas_huerfanas[:30]:
resultado += f" • {nota['ruta']}\n"
if len(notas_huerfanas) > 30:
resultado += f"\n... y {len(notas_huerfanas) - 30} más"
return Result.ok(resultado)
except Exception as e:
return Result.fail(f"Error al buscar huérfanas: {e}")
