dataforge/osint
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
db05c58893a100395ce1f65fc3a6767003e9112c
osint/tools/sync_osint_to_dav.py
T
egutierrez fe280ec8ac feat(tools): sync bidireccional vault OSINT <-> Xandikos CardDAV 
- sync_dav_to_osint.py (NUEVO): reverse sync Xandikos->vault. Trae contactos
  nuevos del movil (contexto: movil, dedup por tel/email) y ediciones de agenda
  (nombre/tel/email/aliases) PRESERVANDO la capa OSINT (relaciones/dni/contexto/
  fuente/tags). Estado persistente .sync_state.json (UID->etag/vault_mtime).
  Reconciliacion por etag; --dry-run (default) / --apply.
- sync_osint_to_dav.py: anade --check (audit read-only vault<->Xandikos: fichas
  sin vCard, vCards huerfanos, agendas divergentes) y optimiza build_existing_index
  con dav_get_collection (1 REPORT, ~9s->~0.5s) en vez del patron N+1.

Usa las funciones del registry: dav_get_collection, dav_delete_resource,
carddav_put_vcard, obsidian CRUD, pass_get_secret.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 00:30:27 +02:00

844 lines
34 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Sincroniza las fichas del vault OSINT hacia el servidor CardDAV (Xandikos),
enriqueciendo cada vCard con la capa de informacion extra de Obsidian: alias,
relaciones, contexto, direccion, lugares, DNI, fecha de nacimiento, etc.
Los contactos ya estan en Xandikos (vinieron de Google con tel/email). Este
sync AÑADE la capa OSINT a esos vCards (o crea los que falten). Es idempotente
por UID: re-ejecutar no duplica.
Tool de PROYECTO (vive en projects/osint/tools/). NO es funcion del registry,
NO se indexa.
=============================================================================
COMO PERSONALIZAR QUE CAMPOS OSINT VIAJAN AL VCARD -> edita FIELD_MAP
=============================================================================
FIELD_MAP es la unica fuente de verdad del mapeo frontmatter -> vCard. Cada
entrada es una tupla:
(campo_frontmatter, sensible, generador)
- campo_frontmatter : str. Clave del frontmatter de la ficha (p.ej. "dni").
Hay claves SINTETICAS calculadas en build_record()
("lugares", "nombre_completo") que tambien se mapean
aqui: no salen tal cual del YAML pero se exponen al
generador como si fueran un campo mas.
- sensible : bool. True marca el campo como dato sensible que viaja
al movil (DNI, direccion, lugares). El dry-run los lista
EXPLICITAMENTE para que confirmes antes de --apply.
- generador : callable(valor) -> list[str]. Recibe el valor del campo
y devuelve 0..N lineas vCard ya formateadas (sin CRLF).
Devuelve [] para omitir el campo (valor vacio/null).
PARA ACTIVAR / DESACTIVAR UN CAMPO:
- Desactivar: comenta (o borra) su linea en FIELD_MAP. Esa propiedad dejara
de generarse y de viajar al servidor.
- Activar uno nuevo: añade una tupla. El generador recibe el valor crudo del
frontmatter; usa las propiedades estandar vCard 3.0 (FN, N, TEL, EMAIL,
ADR, BDAY, NICKNAME, NOTE, CATEGORIES) o una extension X- propia
(X-OSINT-*) para datos que no tienen propiedad estandar.
REGLAS DE FORMATO QUE EL GENERADOR DEBE RESPETAR:
- Una propiedad NOTE por vCard como maximo es lo limpio: por eso TODOS los
textos largos (relaciones, contexto, notas-body, lugares) se acumulan en
NOTE_PARTS (ver helper note()) y se emiten al final como UNA sola linea
NOTE con saltos escapados (\\n). No emitas varias lineas NOTE sueltas.
- Escapa SIEMPRE los valores con vcard_escape() (\\ , ; , : de mas, saltos de
linea). Los generadores de abajo ya lo hacen.
- El orden de FIELD_MAP es el orden en que aparecen las propiedades en el
vCard resultante (salvo NOTE, que siempre va al final).
EJEMPLO — añadir el pais como propiedad X-:
("pais", False, lambda v: [f"X-OSINT-PAIS:{vcard_escape(v)}"] if v else []),
EJEMPLO — desactivar el envio del DNI al movil:
# ("dni", True, lambda v: [f"X-OSINT-DNI:{vcard_escape(v)}"] if v else []),
=============================================================================
"""
import sys
import os
import re
import argparse
sys.path.insert(0, "/home/enmanuel/fn_registry/python/functions")
from infra.pass_get_secret import pass_get_secret # noqa: E402
from infra.dav_list_resources import dav_list_resources # noqa: E402
from infra.dav_get_resource import dav_get_resource # noqa: E402
from infra.dav_get_collection import dav_get_collection # noqa: E402
from infra.carddav_put_vcard import carddav_put_vcard # noqa: E402
from obsidian import ( # noqa: E402
list_obsidian_notes,
read_obsidian_note,
)
# --------------------------------------------------------------------------
# Configuracion
# --------------------------------------------------------------------------
OSINT = "/home/enmanuel/Obsidian/osint"
DAV_BASE = "https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com"
DAV_USER = "enmanuel"
DAV_COLLECTION = "/enmanuel/contacts/addressbook/"
PASS_SECRET = "dav/xandikos-enmanuel"
# Carpetas del vault a sincronizar y el tipo que representan.
SUBFOLDERS = (("personas", "persona"), ("organizaciones", "organizacion"))
# Prefijo del UID para fichas que NO casan con un contacto existente de Google.
OSINT_UID_PREFIX = "osint-"
# --------------------------------------------------------------------------
# Helpers de formato vCard
# --------------------------------------------------------------------------
def vcard_escape(value) -> str:
"""Escapa un valor para una propiedad vCard 3.0 (RFC 2426).
Escapa backslash, coma, punto y coma y normaliza los saltos de linea a la
secuencia escapada \\n (un NOTE multilinea valido va en UNA sola linea
logica con \\n literales).
"""
if value is None:
return ""
s = str(value)
s = s.replace("\\", "\\\\")
s = s.replace("\n", "\\n").replace("\r", "")
s = s.replace(",", "\\,").replace(";", "\\;")
return s
def _date_to_bday(value) -> str:
"""Convierte una fecha ISO (YYYY-MM-DD) o un date a BDAY vCard (YYYYMMDD).
Devuelve "" si no parsea. vCard 3.0 acepta tanto YYYY-MM-DD como YYYYMMDD;
usamos el formato basico sin guiones por compatibilidad amplia.
"""
if not value:
return ""
s = str(value).strip()
m = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", s)
if m:
return f"{m.group(1)}{m.group(2)}{m.group(3)}"
m = re.match(r"^(\d{4})(\d{2})(\d{2})$", s)
if m:
return s
return ""
def _clean_relacion(rel: str) -> str:
"""Normaliza una entrada de `relaciones` para mostrarla legible en NOTE.
Las relaciones del vault son del estilo '[[isagri]] — contacto' o
'[[personas/x|Nombre]] — hermano'. Quitamos los corchetes wikilink y la
barra de alias, dejando texto plano legible en el movil.
"""
if not rel:
return ""
s = str(rel)
# [[target|alias]] -> alias ; [[target]] -> target
def _wl(m):
inner = m.group(1)
return inner.split("|", 1)[1] if "|" in inner else inner.split("/")[-1]
s = re.sub(r"\[\[([^\]]+)\]\]", _wl, s)
return s.strip()
# --------------------------------------------------------------------------
# NOTE acumulado: las propiedades de texto largo se juntan en UNA sola NOTE.
# Cada build_vcard() arranca con esta lista vacia (ver build_vcard()).
# --------------------------------------------------------------------------
_NOTE_PARTS: list = []
def note(text: str) -> list:
"""Acumula un fragmento en la NOTE global y devuelve [] (no emite linea).
Los generadores de FIELD_MAP que producen texto descriptivo (relaciones,
contexto, notas del body, lugares) llaman a note() en vez de emitir su
propia linea NOTE: asi el vCard final lleva una unica propiedad NOTE bien
formada con todos los fragmentos separados por \\n.
"""
if text and str(text).strip():
_NOTE_PARTS.append(str(text).strip())
return []
# --------------------------------------------------------------------------
# FIELD_MAP — EDITA AQUI para activar/desactivar campos (ver cabecera).
# Cada tupla: (campo_frontmatter, sensible, generador(valor) -> list[str])
# --------------------------------------------------------------------------
FIELD_MAP = [
# --- Identidad basica ---
# FN es obligatorio en vCard 3.0; el nombre completo de la ficha.
("nombre_completo", False, lambda v: [f"FN:{vcard_escape(v)}"] if v else []),
# N estructurado (apellidos;nombre;;;) — derivado del nombre completo.
("n_struct", False, lambda v: [f"N:{v}"] if v else []),
# Alias / otros nombres por los que aparece -> NICKNAME (CSV escapado).
("aliases", False, lambda v: [
"NICKNAME:" + ",".join(vcard_escape(a) for a in v)
] if v else []),
# --- Contacto (ya suelen estar en el vCard de Google; los reafirmamos) ---
("telefono", False, lambda v: [f"TEL;TYPE=CELL:{vcard_escape(v)}"] if v else []),
("email", False, lambda v: [f"EMAIL;TYPE=INTERNET:{vcard_escape(v)}"] if v else []),
# --- Localizacion ---
# ADR estructurado vCard: ;;<calle>;<ciudad>;;<cp>;<pais>. Metemos la
# direccion completa en el campo street (componente 3) por simplicidad.
("direccion", True, lambda v: [f"ADR;TYPE=HOME:;;{vcard_escape(v)};;;;"] if v else []),
("pais", False, lambda v: [f"X-OSINT-PAIS:{vcard_escape(v)}"] if v else []),
# --- Datos OSINT sin propiedad estandar -> extensiones X- ---
("dni", True, lambda v: [f"X-OSINT-DNI:{vcard_escape(v)}"] if v else []),
("fecha_nacimiento", False, lambda v: (
[f"BDAY:{_date_to_bday(v)}"] if _date_to_bday(v) else []
)),
("sexo", False, lambda v: [f"X-OSINT-SEXO:{vcard_escape(v)}"] if v else []),
("contexto", False, lambda v: note(f"Contexto: {v}") if v else []),
("fuente", False, lambda v: note(f"Fuente: {v}") if v else []),
# --- Texto descriptivo -> se acumula en la NOTE unica ---
("relaciones", False, lambda v: note(
"Relaciones: " + "; ".join(_clean_relacion(r) for r in v if _clean_relacion(r))
) if v else []),
# Lugares: sintetico (extraido de la seccion ## Lugares del body). Sensible.
("lugares", True, lambda v: note(
"Lugares: " + "; ".join(v)
) if v else []),
# Notas libres del body (seccion ## Notas), si las hay.
("notas_body", False, lambda v: note(v) if v else []),
# --- Categoria para agruparlos en la libreta del movil ---
("tags", False, lambda v: [
"CATEGORIES:" + ",".join(vcard_escape(t) for t in v)
] if v else []),
]
# Conjunto de campos marcados sensibles (para el reporte de privacidad).
SENSITIVE_FIELDS = sorted({campo for campo, sensible, _ in FIELD_MAP if sensible})
# --------------------------------------------------------------------------
# Extraccion de datos sinteticos del body de la ficha
# --------------------------------------------------------------------------
def _section(body: str, header: str) -> str:
"""Devuelve el texto crudo de una seccion `## <header>` del body (hasta la
siguiente cabecera `## ` o el final). "" si la seccion no existe o vacia.
"""
if not body:
return ""
pat = re.compile(rf"^##\s+{re.escape(header)}\s*$(.*?)(?=^##\s|\Z)",
re.MULTILINE | re.DOTALL)
m = pat.search(body)
return m.group(1).strip() if m else ""
def _extract_lugares(body: str) -> list:
"""Extrae los lugares de la seccion `## Lugares` como texto plano legible.
Las lineas son del estilo:
- [[lugares/calle-x|Calle X 1 Almachar Malaga]]
Devolvemos el alias legible (lo de despues del |) o el target slug.
"""
sec = _section(body, "Lugares")
out = []
for line in sec.splitlines():
line = line.strip().lstrip("-").strip()
if not line:
continue
m = re.search(r"\[\[([^\]]+)\]\]", line)
if m:
inner = m.group(1)
disp = inner.split("|", 1)[1] if "|" in inner else inner.split("/")[-1]
out.append(disp.strip())
elif line:
out.append(line)
return out
def _extract_notas(body: str) -> str:
"""Extrae el texto libre de la seccion `## Notas` (si tiene contenido)."""
sec = _section(body, "Notas")
# Limpiar wikilinks/embeds del texto de notas para que sea legible plano.
sec = re.sub(r"!?\[\[([^\]]+)\]\]",
lambda m: (m.group(1).split("|", 1)[1] if "|" in m.group(1)
else m.group(1).split("/")[-1]),
sec)
return sec.strip()
def _n_struct_from_nombre(nombre: str) -> str:
"""Construye el campo N (apellidos;nombre;;;) heuristico desde el nombre.
vCard N = Family;Given;Additional;Prefix;Suffix. Heuristica simple para
nombres espanoles: primer token = given, resto = family. No es perfecto
(apellidos compuestos) pero es suficiente para ordenar en el movil; el FN
sigue siendo el nombre completo canonico.
"""
if not nombre:
return ""
parts = [p for p in str(nombre).split() if p]
if not parts:
return ""
given = vcard_escape(parts[0])
family = vcard_escape(" ".join(parts[1:])) if len(parts) > 1 else ""
return f"{family};{given};;;"
# --------------------------------------------------------------------------
# Construccion del "record" enriquecido por ficha + del vCard
# --------------------------------------------------------------------------
def build_record(note_data: dict, tipo: str) -> dict:
"""Aplana una ficha (frontmatter + body) en un dict de campos para FIELD_MAP.
Incluye los campos del frontmatter mas los sinteticos:
- nombre_completo : el nombre canonico (FN).
- n_struct : el campo N estructurado.
- lugares : lista extraida de ## Lugares.
- notas_body : texto libre de ## Notas.
Normaliza los 'null' string/None a None y deja las listas vacias como [].
"""
fm = dict(note_data.get("frontmatter") or {})
body = note_data.get("body") or ""
def _norm(v):
if v is None:
return None
if isinstance(v, str) and v.strip().lower() in ("null", "none", ""):
return None
return v
nombre = _norm(fm.get("nombre")) or os.path.splitext(
os.path.basename(note_data["path"]))[0].replace("-", " ")
rec = {k: _norm(v) for k, v in fm.items()}
# Asegurar listas para los campos lista (evita None en los generadores).
for list_field in ("aliases", "relaciones", "tags"):
val = rec.get(list_field)
if val is None:
rec[list_field] = []
elif not isinstance(val, list):
rec[list_field] = [val]
rec["nombre_completo"] = nombre
rec["n_struct"] = _n_struct_from_nombre(nombre)
rec["lugares"] = _extract_lugares(body)
rec["notas_body"] = _extract_notas(body)
rec["_tipo"] = tipo
rec["_slug"] = _norm(fm.get("slug")) or os.path.splitext(
os.path.basename(note_data["path"]))[0]
rec["_path"] = note_data["path"]
return rec
def build_vcard(rec: dict, uid: str) -> dict:
"""Genera el texto vCard 3.0 de un record aplicando FIELD_MAP.
Devuelve {text, fields_emitted} donde fields_emitted es el conjunto de
campos de FIELD_MAP que produjeron alguna linea (incluye los sensibles que
viajaron). NOTE se emite UNA sola vez al final con todos los fragmentos.
"""
global _NOTE_PARTS
_NOTE_PARTS = [] # reset del acumulador de NOTE para este vCard
lines = ["BEGIN:VCARD", "VERSION:3.0"]
fields_emitted = set()
for campo, _sensible, gen in FIELD_MAP:
value = rec.get(campo)
try:
produced = gen(value)
except Exception: # noqa: BLE001 — un generador no debe tumbar el sync
produced = []
if produced:
lines.extend(produced)
fields_emitted.add(campo)
# Emitir la NOTE acumulada (relaciones + contexto + fuente + lugares + notas).
if _NOTE_PARTS:
joined = "\\n".join(vcard_escape(p) for p in _NOTE_PARTS)
lines.append(f"NOTE:{joined}")
fields_emitted.add("NOTE")
lines.append(f"UID:{uid}")
lines.append("END:VCARD")
text = "\r\n".join(lines) + "\r\n"
return {"text": text, "fields_emitted": fields_emitted}
# --------------------------------------------------------------------------
# Indice de dedup: telefono/email -> UID de un vCard ya existente en Xandikos
# --------------------------------------------------------------------------
def _norm_phone(p) -> str:
"""Normaliza un telefono a sus ultimos 9 digitos (numero nacional ES).
Quita espacios, prefijos +34/0034 y separadores. Dos formatos distintos del
mismo numero ('+34 680 43 88 89' y '680438889') colapsan a la misma clave.
"""
if not p:
return ""
d = re.sub(r"\D", "", str(p))
return d[-9:] if len(d) >= 9 else d
def _vcard_prop_values(text: str, prop: str) -> list:
"""Extrae los valores de una propiedad (TEL, EMAIL, UID...) de un vCard."""
out = []
for line in text.splitlines():
m = re.match(rf"^(?:item\d+\.)?{prop}(?:;[^:]*)?:(.*)$", line, re.IGNORECASE)
if m:
v = m.group(1).strip()
if v:
out.append(v)
return out
def build_existing_index(base, user, pwd, collection, *, verbose=True) -> dict:
"""Descarga los vCards existentes de Xandikos y construye los indices de dedup.
Devuelve {phone: {key->uid}, email: {key->uid}, total: int, fetched: int,
errors: int}. Las claves son telefono normalizado (9 digitos) y email en
minusculas. Cada clave apunta al UID del PRIMER vCard que la declara.
Usa dav_get_collection (1 sola peticion REPORT con el contenido inline) en
vez del patron N+1 (PROPFIND + un GET por recurso): para ~1064 contactos baja
de ~9s a ~1s, lo que mantiene el step PUSH del DAG dentro de su timeout.
"""
coll = dav_get_collection(base, user, pwd, collection, "vcard")
if coll.get("status") != "ok":
return {"phone": {}, "email": {}, "total": 0, "fetched": 0,
"errors": 1, "error": coll.get("error")}
resources = coll.get("resources", [])
phone_idx, email_idx = {}, {}
fetched = errors = 0
for r in resources:
text = r.get("data") or ""
if not text:
errors += 1
continue
fetched += 1
uid = (_vcard_prop_values(text, "UID") or [""])[0]
if not uid:
# UID derivado del nombre del recurso si el vCard no lo declara.
uid = os.path.splitext(os.path.basename(r["href"]))[0]
for tel in _vcard_prop_values(text, "TEL"):
k = _norm_phone(tel)
if k:
phone_idx.setdefault(k, uid)
for em in _vcard_prop_values(text, "EMAIL"):
k = em.strip().lower()
if k:
email_idx.setdefault(k, uid)
if verbose:
print(f" indice Xandikos: {fetched}/{len(resources)} vCards leidos, "
f"{len(phone_idx)} telefonos, {len(email_idx)} emails, "
f"{errors} errores de lectura")
return {"phone": phone_idx, "email": email_idx,
"total": len(resources), "fetched": fetched, "errors": errors}
def resolve_uid(rec: dict, index: dict) -> dict:
"""Resuelve el UID a usar para un record: reusa el existente o crea osint-<slug>.
Devuelve {uid, source} donde source es 'phone', 'email' o 'new'. Match por
telefono primero (mas fiable que email para estos contactos), luego email.
"""
tel = rec.get("telefono")
em = rec.get("email")
if tel:
hit = index["phone"].get(_norm_phone(tel))
if hit:
return {"uid": hit, "source": "phone"}
if em:
hit = index["email"].get(str(em).strip().lower())
if hit:
return {"uid": hit, "source": "email"}
return {"uid": OSINT_UID_PREFIX + rec["_slug"], "source": "new"}
# --------------------------------------------------------------------------
# Orquestacion: planificar todas las fichas
# --------------------------------------------------------------------------
def load_fichas() -> list:
"""Carga todas las fichas de las carpetas configuradas como (note_data, tipo)."""
out = []
for subfolder, tipo in SUBFOLDERS:
for p in list_obsidian_notes(OSINT, subfolder=subfolder):
# Solo fichas de nivel-1 (personas/<slug>.md). Excluye las sub-notas de
# documentos (personas/<slug>/dni.md, fotos.md, ...) que no son contactos.
if os.path.basename(os.path.dirname(p)) != subfolder:
continue
base = os.path.splitext(os.path.basename(p))[0]
if base.startswith("_"):
continue
try:
nd = read_obsidian_note(p)
except Exception: # noqa: BLE001
continue
out.append((nd, tipo))
return out
def plan_sync(index: dict) -> dict:
"""Construye el plan completo: un vCard enriquecido por ficha + su UID.
Devuelve {records: [...], counts: {...}} donde cada record del plan lleva
{rec, uid, uid_source, vcard, fields_emitted, sensitive_emitted}.
"""
fichas = load_fichas()
plan = []
counts = {
"fichas_total": len(fichas),
"persona": 0, "organizacion": 0,
"uid_reused_phone": 0, "uid_reused_email": 0, "uid_new": 0,
}
sensitive_carrying = {f: 0 for f in SENSITIVE_FIELDS}
for note_data, tipo in fichas:
rec = build_record(note_data, tipo)
counts[tipo] = counts.get(tipo, 0) + 1
resolved = resolve_uid(rec, index)
uid = resolved["uid"]
built = build_vcard(rec, uid)
emitted = built["fields_emitted"]
sens_here = sorted(f for f in SENSITIVE_FIELDS if f in emitted)
for f in sens_here:
sensitive_carrying[f] += 1
counts_key = {
"phone": "uid_reused_phone",
"email": "uid_reused_email",
"new": "uid_new",
}[resolved["source"]]
counts[counts_key] += 1
plan.append({
"rec": rec,
"uid": uid,
"uid_source": resolved["source"],
"vcard": built["text"],
"fields_emitted": emitted,
"sensitive_emitted": sens_here,
})
return {"records": plan, "counts": counts,
"sensitive_carrying": sensitive_carrying}
# --------------------------------------------------------------------------
# Aplicar (solo --apply)
# --------------------------------------------------------------------------
def apply_sync(plan, base, user, pwd, collection) -> dict:
"""Sube cada vCard del plan a Xandikos via carddav_put_vcard. Idempotente."""
ok = fail = 0
errors = []
for item in plan["records"]:
res = carddav_put_vcard(base, user, pwd, collection,
item["uid"], item["vcard"])
if res.get("status") == "ok":
ok += 1
else:
fail += 1
errors.append({"uid": item["uid"], "error": res.get("error"),
"http_status": res.get("http_status")})
return {"ok": ok, "fail": fail, "errors": errors}
# --------------------------------------------------------------------------
# Reporte dry-run
# --------------------------------------------------------------------------
def report(plan, index, sample_n=5):
c = plan["counts"]
print("=" * 70)
print("DRY-RUN — sync_osint_to_dav.py (vault OSINT -> Xandikos CardDAV)")
print("=" * 70)
print(f"servidor ........................ {DAV_BASE}")
print(f"coleccion ....................... {DAV_COLLECTION}")
print("-" * 70)
print(f"vCards ya en Xandikos ........... {index['total']} "
f"(leidos {index['fetched']}, errores {index['errors']})")
print(f"fichas OSINT a sincronizar ...... {c['fichas_total']}")
print(f" personas ...................... {c.get('persona', 0)}")
print(f" organizaciones ................ {c.get('organizacion', 0)}")
print("-" * 70)
print("Resolucion de UID (dedup contra Xandikos):")
print(f" reusa UID existente (telefono) {c['uid_reused_phone']}")
print(f" reusa UID existente (email) ... {c['uid_reused_email']}")
print(f" UID nuevo (osint-<slug>) ...... {c['uid_new']}")
reused = c['uid_reused_phone'] + c['uid_reused_email']
print(f" => {reused} fichas ENRIQUECEN un contacto existente, "
f"{c['uid_new']} CREAN uno nuevo")
print("=" * 70)
# ---- Privacidad: que campos sensibles viajarian al movil ----
print("PRIVACIDAD — campos SENSIBLES que viajarian al movil con --apply:")
print("-" * 70)
any_sensitive = False
for f in SENSITIVE_FIELDS:
n = plan["sensitive_carrying"].get(f, 0)
if n:
any_sensitive = True
print(f" [SENSIBLE] {f:<12} -> presente en {n} vCard(s)")
if not any_sensitive:
print(" (ningun campo sensible viajaria con el FIELD_MAP actual)")
else:
print("")
print(" Estos datos se escribirian en la libreta de contactos del movil")
print(" (que se sincroniza/respalda fuera de tu control). Revisa el")
print(" FIELD_MAP y comenta los campos que NO quieras enviar antes de")
print(" ejecutar --apply.")
print("=" * 70)
# ---- Campos OSINT no sensibles que tambien viajan ----
emitted_counter = {}
for item in plan["records"]:
for f in item["fields_emitted"]:
emitted_counter[f] = emitted_counter.get(f, 0) + 1
print("Cobertura de campos (cuantos vCards llevan cada propiedad):")
print("-" * 70)
for f in sorted(emitted_counter, key=lambda k: -emitted_counter[k]):
mark = " [SENSIBLE]" if f in SENSITIVE_FIELDS else ""
print(f" {f:<16} {emitted_counter[f]:>4}{mark}")
print("=" * 70)
# ---- Muestra de N vCards completos ----
print(f"MUESTRA de {sample_n} vCards completos (asi se veria el mapeo):")
print("=" * 70)
# Elegir muestras variadas: prioriza las que llevan mas campos OSINT.
enriched = sorted(plan["records"],
key=lambda it: -len(it["fields_emitted"]))
shown = 0
for item in enriched:
rec = item["rec"]
src = {"phone": "reusa UID (tel)", "email": "reusa UID (email)",
"new": "UID nuevo"}[item["uid_source"]]
print(f"--- {rec['nombre_completo']} [{rec['_tipo']}] | {src}: {item['uid']}")
print(item["vcard"].replace("\r\n", "\n").rstrip())
print("")
shown += 1
if shown >= sample_n:
break
print("=" * 70)
# --------------------------------------------------------------------------
# Audit de consistencia (--check) — read-only, vault <-> Xandikos
# --------------------------------------------------------------------------
def _norm_email(e) -> str:
return str(e).strip().lower() if e else ""
def audit_consistency(base, user, pwd, collection) -> dict:
"""Compara vault OSINT <-> Xandikos y devuelve el reporte de drift.
Read-only: descarga la coleccion en 1 REPORT (dav_get_collection) y la cruza
con las fichas del vault. Reporta:
- vault_sin_vcard : fichas del vault sin contraparte en Xandikos
(por telefono/email normalizado, ni por UID osint-).
- vcard_huerfano : vCards del servidor sin ficha en el vault (anadidos
en el movil que aun no se han traido / contactos no-OSINT).
- difiere : contactos que casan pero donde tel/email/alias difieren
entre vault y vCard.
Devuelve {vault_total, vcard_total, vault_sin_vcard:[...], vcard_huerfano:[...],
difiere:[...]}.
"""
coll = dav_get_collection(base, user, pwd, collection, "vcard")
if coll.get("status") != "ok":
return {"error": coll.get("error"), "http_status": coll.get("http_status")}
resources = coll.get("resources", [])
# Indice del servidor: telefono/email -> {uid, fn, tels, emails, nicks, href}
srv_by_phone, srv_by_email, srv_by_uid = {}, {}, {}
srv_records = []
for r in resources:
txt = re.sub(r"\r?\n[ \t]", "", r["data"]) # unfold
uid = (_vcard_prop_values(txt, "UID") or [""])[0] or \
os.path.splitext(os.path.basename(r["href"]))[0]
fn = (_vcard_prop_values(txt, "FN") or [""])[0]
tels = _vcard_prop_values(txt, "TEL")
emails = _vcard_prop_values(txt, "EMAIL")
nicks = []
for nk in _vcard_prop_values(txt, "NICKNAME"):
nicks.extend([a.strip() for a in nk.split(",") if a.strip()])
rec = {"uid": uid, "fn": fn, "tels": tels, "emails": emails,
"nicks": nicks, "href": r["href"], "matched": False}
srv_records.append(rec)
srv_by_uid[uid] = rec
for t in tels:
srv_by_phone.setdefault(_norm_phone(t), rec)
for e in emails:
srv_by_email.setdefault(_norm_email(e), rec)
# Recorrer el vault y cruzar.
vault_total = 0
vault_sin_vcard, difiere = [], []
for note_data, _tipo in load_fichas():
vault_total += 1
fm = note_data.get("frontmatter") or {}
slug = fm.get("slug") or os.path.splitext(
os.path.basename(note_data["path"]))[0]
nombre = fm.get("nombre") or slug.replace("-", " ")
tel = fm.get("telefono")
em = fm.get("email")
aliases = fm.get("aliases") or []
if not isinstance(aliases, list):
aliases = [aliases]
if isinstance(tel, str) and tel.strip().lower() in ("null", "none", ""):
tel = None
if isinstance(em, str) and em.strip().lower() in ("null", "none", ""):
em = None
rec = None
if tel and _norm_phone(tel) in srv_by_phone:
rec = srv_by_phone[_norm_phone(tel)]
elif em and _norm_email(em) in srv_by_email:
rec = srv_by_email[_norm_email(em)]
elif ("osint-" + slug) in srv_by_uid:
rec = srv_by_uid["osint-" + slug]
if rec is None:
vault_sin_vcard.append({"slug": slug, "nombre": nombre,
"telefono": tel, "email": em})
continue
rec["matched"] = True
# Comparar campos de agenda.
diffs = []
if tel and _norm_phone(tel) not in {_norm_phone(t) for t in rec["tels"]}:
diffs.append(("telefono", tel, ", ".join(rec["tels"]) or "-"))
if em and _norm_email(em) not in {_norm_email(e) for e in rec["emails"]}:
diffs.append(("email", em, ", ".join(rec["emails"]) or "-"))
srv_nick_set = {n.lower() for n in rec["nicks"]}
missing_alias = [a for a in aliases if a and a.lower() not in srv_nick_set]
if missing_alias and rec["nicks"]:
diffs.append(("aliases", "; ".join(aliases),
"; ".join(rec["nicks"]) or "-"))
if diffs:
difiere.append({"slug": slug, "nombre": nombre, "uid": rec["uid"],
"diffs": diffs})
vcard_huerfano = [
{"uid": r["uid"], "fn": r["fn"],
"telefono": ", ".join(r["tels"]) or None,
"email": ", ".join(r["emails"]) or None}
for r in srv_records if not r["matched"]
]
return {
"vault_total": vault_total,
"vcard_total": len(srv_records),
"vault_sin_vcard": vault_sin_vcard,
"vcard_huerfano": vcard_huerfano,
"difiere": difiere,
}
def report_audit(a: dict, sample_n: int = 15):
if a.get("error"):
print(f"ERROR audit: {a['error']} (http {a.get('http_status')})",
file=sys.stderr)
return
print("=" * 72)
print("AUDIT de consistencia — vault OSINT <-> Xandikos CardDAV (read-only)")
print("=" * 72)
print(f"fichas en el vault .............. {a['vault_total']}")
print(f"vCards en Xandikos .............. {a['vcard_total']}")
print("-" * 72)
print(f"fichas del vault SIN vCard ...... {len(a['vault_sin_vcard'])}")
print(f"vCards SIN ficha (huerfanos) .... {len(a['vcard_huerfano'])}")
print(f"contactos con AGENDA divergente . {len(a['difiere'])}")
print("=" * 72)
if a["vault_sin_vcard"]:
print(f"FICHAS DEL VAULT SIN vCard (muestra "
f"{min(sample_n, len(a['vault_sin_vcard']))}):")
print("-" * 72)
for f in a["vault_sin_vcard"][:sample_n]:
print(f" - {f['slug']:<36} tel={f['telefono']!r} email={f['email']!r}")
print("=" * 72)
if a["vcard_huerfano"]:
print(f"vCards HUERFANOS (en el movil, sin ficha) (muestra "
f"{min(sample_n, len(a['vcard_huerfano']))}):")
print("-" * 72)
for r in a["vcard_huerfano"][:sample_n]:
print(f" - {r['fn'][:34]:<34} | UID {r['uid']} tel={r['telefono']!r}")
print("=" * 72)
if a["difiere"]:
print(f"AGENDA DIVERGENTE vault vs movil (muestra "
f"{min(sample_n, len(a['difiere']))}):")
print("-" * 72)
for d in a["difiere"][:sample_n]:
print(f" ~ {d['slug']} (UID {d['uid']})")
for campo, vault_v, srv_v in d["diffs"]:
print(f" {campo}: vault={vault_v!r} movil={srv_v!r}")
print("=" * 72)
# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(
description="Sincroniza fichas OSINT (Obsidian) -> Xandikos CardDAV, "
"enriqueciendo los vCards con la capa OSINT.")
ap.add_argument("--apply", action="store_true",
help="Sube los vCards al servidor. Por defecto: dry-run "
"(no toca el servidor).")
ap.add_argument("--check", action="store_true",
help="Audit de consistencia READ-ONLY: compara vault <-> "
"Xandikos y reporta fichas sin vCard, vCards huerfanos "
"y agendas divergentes. No sube nada.")
ap.add_argument("--sample", type=int, default=5,
help="Numero de vCards completos a mostrar en el dry-run "
"(o de filas por categoria en --check).")
args = ap.parse_args()
secret = pass_get_secret(PASS_SECRET)
if secret.get("status") != "ok":
print(f"ERROR: no se pudo leer el secreto '{PASS_SECRET}' de pass: "
f"{secret.get('error')}", file=sys.stderr)
return 1
pwd = secret["value"] # sensible: NUNCA logear
if args.check:
a = audit_consistency(DAV_BASE, DAV_USER, pwd, DAV_COLLECTION)
report_audit(a, sample_n=max(args.sample, 15))
return 1 if a.get("error") else 0
print("Construyendo indice de contactos ya existentes en Xandikos...")
index = build_existing_index(DAV_BASE, DAV_USER, pwd, DAV_COLLECTION)
if index.get("errors") and index["fetched"] == 0:
print(f"ERROR: no se pudo listar/leer la coleccion CardDAV: "
f"{index.get('error')}", file=sys.stderr)
return 1
plan = plan_sync(index)
report(plan, index, sample_n=args.sample)
if args.apply:
print("\nAPLICANDO (PUT a Xandikos)...")
res = apply_sync(plan, DAV_BASE, DAV_USER, pwd, DAV_COLLECTION)
print(f"APLICADO: ok={res['ok']} fail={res['fail']}")
for e in res["errors"][:10]:
print(f" FALLO uid={e['uid']}: {e['error']} (http {e['http_status']})")
else:
print("\n(dry-run: NO se toco el servidor. Usa --apply para subir los vCards.)")
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink