#!/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: ;;;;;;. 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 `##
` 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-. 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/.md). Excluye las sub-notas de # documentos (personas//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-) ...... {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())