dataforge/osint
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d98127115b916ac49d27e4971eeaf61fa2913a97
osint/tools/sync_dav_to_osint.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

646 lines
26 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Sincroniza el servidor CardDAV (Xandikos) HACIA el vault OSINT (sentido
inverso de sync_osint_to_dav.py).
El movil (DAVx5) edita la libreta de contactos de Xandikos: anade contactos
nuevos, corrige telefonos/nombres. Este tool trae esos cambios de vuelta al
vault SIN pisar la capa OSINT (relaciones, dni, contexto, fuente, tags) que es
autoritativa en el lado del vault.
MODELO DE RECONCILIACION
========================
- El vault es la FUENTE DE VERDAD de los campos OSINT.
- Xandikos es la fuente de los cambios de AGENDA (nombre, telefono, email,
aliases) que llegan del movil.
Por cada vCard del servidor decidimos su accion comparando contra:
1. el estado persistente (.sync_state.json): UID -> {etag, vault_slug, ...}
2. el vault (match por telefono/email normalizado, o por slug del UID osint-)
Acciones:
- CREATE : UID nuevo (no en estado) y sin ficha que case por tel/email.
-> crea personas/<slug>.md con contexto: movil.
- UPDATE : el etag cambio respecto al estado (el movil lo edito).
-> actualiza SOLO los campos de agenda (nombre, telefono,
email, aliases), PRESERVANDO los campos OSINT.
- LINK : UID nuevo pero ya hay ficha que case por tel/email (no es un
contacto nuevo, es el mismo que ya esta en el vault).
-> registra el UID en el estado y, si el etag implica edicion,
aplica el UPDATE de agenda. No crea ficha duplicada.
- SKIP : etag igual al del estado -> sin cambios desde el ultimo sync.
last-write-wins por timestamp en campos de agenda: cuando el etag de Xandikos
cambia, el movil gano la ultima escritura de ese contacto -> se aplica al
vault. El push posterior (sync_osint_to_dav --apply, en el DAG) re-emite el
vault ya autoritativo en OSINT.
MODOS
=====
--dry-run (DEFAULT) reporta que crearia/actualizaria. NO escribe el vault ni
el estado.
--apply escribe el vault (create/update notes) y reescribe
.sync_state.json.
Tool de PROYECTO (vive en projects/osint/tools/). NO es funcion del registry,
NO se indexa.
"""
import sys
import os
import re
import json
import time
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_get_collection import dav_get_collection # noqa: E402
from obsidian import ( # noqa: E402
slugify_obsidian_name,
list_obsidian_notes,
read_obsidian_note,
create_obsidian_note,
update_obsidian_note,
)
# --------------------------------------------------------------------------
# Configuracion (espejo de sync_osint_to_dav.py)
# --------------------------------------------------------------------------
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"
STATE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
".sync_state.json")
# Carpetas del vault que representan contactos (mismo set que el push).
SUBFOLDERS = (("personas", "persona"), ("organizaciones", "organizacion"))
# Frontmatter canonico de persona (CONVENTIONS.md 3b), en orden. Las fichas que
# creamos aqui lo respetan campo a campo.
PERSON_CANON = [
"tipo", "nombre", "slug", "aliases", "sexo", "fecha_nacimiento", "dni",
"telefono", "email", "direccion", "pais", "relaciones", "contexto",
"fuente", "tags",
]
# Campos de AGENDA que el movil puede editar y que actualizamos en un UPDATE.
# El resto del frontmatter (campos OSINT) NUNCA se toca en un UPDATE.
AGENDA_FIELDS = ("nombre", "telefono", "email", "aliases")
# --------------------------------------------------------------------------
# Parseo de vCard (entrada)
# --------------------------------------------------------------------------
def _unfold(vcard_text: str) -> str:
"""Deshace el folding de lineas de vCard (continuacion con espacio/tab)."""
return re.sub(r"\r?\n[ \t]", "", vcard_text)
def _vcard_values(vcard_text: str, prop: str) -> list:
"""Devuelve todos los valores de una propiedad (TEL, EMAIL, UID, ...).
Acepta `PROP;PARAMS:valor` y `PROP:valor`. Decodifica los escapes simples de
vCard (\\, , \\; , \\n) en el valor.
"""
vals = []
for line in vcard_text.splitlines():
m = re.match(rf"^(?:item\d+\.)?{prop}(?:;[^:]*)?:(.*)$", line, re.IGNORECASE)
if m:
v = m.group(1).strip()
v = (v.replace("\\n", "\n").replace("\\,", ",")
.replace("\\;", ";").replace("\\\\", "\\"))
v = v.strip()
if v:
vals.append(v)
return vals
def parse_vcard(vcard_text: str) -> dict:
"""Extrae los campos de agenda + OSINT de un vCard.
Devuelve {uid, fn, tels[], emails[], nicknames[], sexo, dni, pais,
bday, categories[], note}. Solo lee; no escribe.
"""
txt = _unfold(vcard_text)
nick_raw = _vcard_values(txt, "NICKNAME")
nicknames = []
for nr in nick_raw:
nicknames.extend([a.strip() for a in nr.split(",") if a.strip()])
cat_raw = _vcard_values(txt, "CATEGORIES")
categories = []
for cr in cat_raw:
categories.extend([c.strip() for c in cr.split(",") if c.strip()])
return {
"uid": (_vcard_values(txt, "UID") or [""])[0],
"fn": (_vcard_values(txt, "FN") or [""])[0],
"tels": _dedup_keep_order(_vcard_values(txt, "TEL")),
"emails": _dedup_keep_order(_vcard_values(txt, "EMAIL")),
"nicknames": _dedup_keep_order(nicknames),
"sexo": (_vcard_values(txt, "X-OSINT-SEXO") or [None])[0],
"dni": (_vcard_values(txt, "X-OSINT-DNI") or [None])[0],
"pais": (_vcard_values(txt, "X-OSINT-PAIS") or [None])[0],
"bday": (_vcard_values(txt, "BDAY") or [None])[0],
"categories": categories,
"note": (_vcard_values(txt, "NOTE") or [""])[0],
}
def _dedup_keep_order(items: list) -> list:
seen, out = set(), []
for it in items:
key = str(it).strip().lower()
if key and key not in seen:
seen.add(key)
out.append(str(it).strip())
return out
def _norm_phone(p) -> str:
"""Normaliza un telefono a sus ultimos 9 digitos (numero nacional ES)."""
if not p:
return ""
d = re.sub(r"\D", "", str(p))
return d[-9:] if len(d) >= 9 else d
# --------------------------------------------------------------------------
# Indice del vault: telefono/email/slug -> ficha existente
# --------------------------------------------------------------------------
def _norm(v):
"""Normaliza 'null'/''/None del frontmatter a None real."""
if v is None:
return None
if isinstance(v, str) and v.strip().lower() in ("null", "none", ""):
return None
return v
def load_vault_index() -> dict:
"""Recorre las fichas del vault y construye los indices de match.
Devuelve {by_phone, by_email, by_slug, by_path, count} donde cada indice
mapea la clave -> dict de la ficha {slug, path, nombre, telefono, email,
aliases, mtime}. by_slug mapea slug -> ficha (para casar UID osint-<slug>).
"""
by_phone, by_email, by_slug, by_path = {}, {}, {}, {}
count = 0
for subfolder, _tipo in SUBFOLDERS:
folder = os.path.join(OSINT, subfolder)
if not os.path.isdir(folder):
continue
for p in list_obsidian_notes(OSINT, subfolder=subfolder):
# Solo fichas de nivel-1 (personas/<slug>.md), no las sub-notas.
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
fm = nd.get("frontmatter") or {}
slug = _norm(fm.get("slug")) or base
tel = _norm(fm.get("telefono"))
em = _norm(fm.get("email"))
aliases = fm.get("aliases") or []
if not isinstance(aliases, list):
aliases = [aliases]
ficha = {
"slug": slug,
"path": p,
"nombre": _norm(fm.get("nombre")) or base.replace("-", " "),
"telefono": tel,
"email": em,
"aliases": aliases,
"mtime": os.path.getmtime(p),
}
count += 1
by_slug[slug] = ficha
by_path[p] = ficha
if tel:
by_phone.setdefault(_norm_phone(tel), ficha)
if em:
by_email.setdefault(str(em).strip().lower(), ficha)
return {"by_phone": by_phone, "by_email": by_email,
"by_slug": by_slug, "by_path": by_path, "count": count}
def match_vault(parsed: dict, index: dict):
"""Devuelve la ficha del vault que casa con un vCard (o None).
Match por telefono primero (mas fiable), luego email, luego por el slug del
UID si es del estilo osint-<slug> (creado por el push). Devuelve (ficha,
source) con source in {phone, email, slug_uid} o (None, None).
"""
for tel in parsed["tels"]:
hit = index["by_phone"].get(_norm_phone(tel))
if hit:
return hit, "phone"
for em in parsed["emails"]:
hit = index["by_email"].get(str(em).strip().lower())
if hit:
return hit, "email"
uid = parsed.get("uid") or ""
if uid.startswith("osint-"):
hit = index["by_slug"].get(uid[len("osint-"):])
if hit:
return hit, "slug_uid"
return None, None
# --------------------------------------------------------------------------
# Estado persistente
# --------------------------------------------------------------------------
def load_state() -> dict:
"""Carga .sync_state.json. {} si no existe o esta corrupto."""
try:
with open(STATE_PATH, "r", encoding="utf-8") as fh:
data = json.load(fh)
return data if isinstance(data, dict) else {}
except (FileNotFoundError, json.JSONDecodeError):
return {}
def save_state(state: dict):
"""Reescribe .sync_state.json de forma atomica (tmp + rename)."""
tmp = STATE_PATH + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(state, fh, ensure_ascii=False, indent=2, sort_keys=True)
os.replace(tmp, STATE_PATH)
# --------------------------------------------------------------------------
# Construccion de fichas y de updates de agenda
# --------------------------------------------------------------------------
def _slug_for(parsed: dict) -> str:
"""Slug estable para una ficha nueva: del FN, o del UID si no hay FN."""
if parsed.get("fn"):
s = slugify_obsidian_name(parsed["fn"])
if s:
return s
return slugify_obsidian_name(parsed.get("uid") or "contacto-movil") or "contacto-movil"
def _unique_slug(slug: str, index: dict, planned_slugs: set) -> str:
"""Evita colision de slug: si ya existe, sufija -2, -3, ..."""
if slug not in index["by_slug"] and slug not in planned_slugs:
return slug
i = 2
while f"{slug}-{i}" in index["by_slug"] or f"{slug}-{i}" in planned_slugs:
i += 1
return f"{slug}-{i}"
def build_new_frontmatter(parsed: dict, slug: str) -> dict:
"""Frontmatter canonico (3b) para una ficha creada desde el movil.
contexto: movil (marca el origen). fuente: el UID de Xandikos. Los campos
OSINT que el vCard NO trae se dejan en null/[] como exige el esquema.
"""
bday = parsed.get("bday")
if bday and re.match(r"^\d{8}$", str(bday)):
bday = f"{bday[:4]}-{bday[4:6]}-{bday[6:8]}"
fm = {
"tipo": "persona",
"nombre": parsed.get("fn") or slug.replace("-", " "),
"slug": slug,
"aliases": parsed.get("nicknames") or [],
"sexo": parsed.get("sexo"),
"fecha_nacimiento": bday,
"dni": parsed.get("dni"),
"telefono": parsed["tels"][0] if parsed["tels"] else None,
"email": parsed["emails"][0] if parsed["emails"] else None,
"direccion": None,
"pais": parsed.get("pais"),
"relaciones": [],
"contexto": "movil",
"fuente": f"Xandikos UID {parsed.get('uid')}".strip(),
"tags": ["persona", "osint", "movil"],
}
# Reordenar segun el canon (las claves extra van al final).
ordered = {k: fm.get(k) for k in PERSON_CANON if k in fm}
for k, v in fm.items():
if k not in ordered:
ordered[k] = v
return ordered
def agenda_update_from_vcard(parsed: dict, ficha: dict) -> dict:
"""Calcula el diff de AGENDA (solo campos de agenda) entre vCard y ficha.
Devuelve {changes: {campo: (viejo, nuevo)}, set_frontmatter: {...}} con SOLO
los campos de agenda que difieren. NUNCA toca campos OSINT. Si no hay cambios
de agenda, changes queda vacio.
"""
changes = {}
new_fm = {}
new_nombre = parsed.get("fn")
if new_nombre and new_nombre != ficha.get("nombre"):
changes["nombre"] = (ficha.get("nombre"), new_nombre)
new_fm["nombre"] = new_nombre
new_tel = parsed["tels"][0] if parsed["tels"] else None
cur_tel = ficha.get("telefono")
if new_tel and _norm_phone(new_tel) != _norm_phone(cur_tel or ""):
changes["telefono"] = (cur_tel, new_tel)
new_fm["telefono"] = new_tel
new_email = parsed["emails"][0] if parsed["emails"] else None
cur_email = ficha.get("email")
if new_email and str(new_email).strip().lower() != str(cur_email or "").strip().lower():
changes["email"] = (cur_email, new_email)
new_fm["email"] = new_email
# aliases: union (el movil puede anadir un NICKNAME nuevo). No borramos los
# que ya hubiera en el vault.
new_nicks = parsed.get("nicknames") or []
cur_aliases = ficha.get("aliases") or []
merged = list(cur_aliases)
added = []
for nk in new_nicks:
if nk not in merged:
merged.append(nk)
added.append(nk)
if added:
changes["aliases"] = (cur_aliases, merged)
new_fm["aliases"] = merged
return {"changes": changes, "set_frontmatter": new_fm}
# --------------------------------------------------------------------------
# Plan: clasificar cada vCard en CREATE / UPDATE / LINK / SKIP
# --------------------------------------------------------------------------
def plan_pull(resources: list, index: dict, state: dict) -> dict:
"""Clasifica cada vCard del servidor. Devuelve {actions, counts}.
Cada accion es un dict con {kind, uid, etag, parsed, ...}. kind in
{create, update, link, skip, skip_empty}.
"""
actions = []
counts = {"total": len(resources), "create": 0, "update": 0,
"link": 0, "skip": 0, "skip_empty": 0}
planned_slugs = set()
for r in resources:
parsed = parse_vcard(r["data"])
uid = parsed.get("uid") or os.path.splitext(
os.path.basename(r["href"]))[0]
etag = r.get("etag")
# vCard sin nada util (ni nombre ni tel ni email) -> ignorar.
if not parsed.get("fn") and not parsed["tels"] and not parsed["emails"]:
counts["skip_empty"] += 1
actions.append({"kind": "skip_empty", "uid": uid, "etag": etag,
"href": r["href"]})
continue
st = state.get(uid)
ficha, match_src = match_vault(parsed, index)
if st is None:
# UID nuevo en el estado.
if ficha is None:
# No casa con nada del vault -> contacto nuevo del movil.
slug = _unique_slug(_slug_for(parsed), index, planned_slugs)
planned_slugs.add(slug)
counts["create"] += 1
actions.append({
"kind": "create", "uid": uid, "etag": etag,
"href": r["href"], "parsed": parsed, "slug": slug,
"frontmatter": build_new_frontmatter(parsed, slug),
})
else:
# Ya hay ficha que casa -> el push ya lo subio o es el mismo
# contacto. Es el LINK BASELINE: registramos el mapeo
# UID -> ficha en el estado SIN tocar el vault. NO aplicamos
# agenda aqui: un match por telefono/email es ambiguo (varios
# contactos del movil comparten numero, o el vault ya dedupeo)
# y el vault es autoritativo en su nombre curado. Los cambios de
# agenda solo fluyen cuando, ya con el UID en el estado, su etag
# CAMBIA en un sync posterior (rama UPDATE de abajo). diff vacio.
counts["link"] += 1
actions.append({
"kind": "link", "uid": uid, "etag": etag,
"href": r["href"], "parsed": parsed,
"ficha": ficha, "match_src": match_src,
"diff": {"changes": {}, "set_frontmatter": {}},
})
else:
# UID ya conocido. Comparar etag.
if st.get("etag") == etag and etag is not None:
counts["skip"] += 1
actions.append({"kind": "skip", "uid": uid, "etag": etag,
"href": r["href"]})
continue
# etag cambio (o no teniamos etag) -> el movil edito. Update agenda.
if ficha is None:
# Conociamos el UID pero la ficha ya no casa por tel/email
# (¿cambio el numero?). Intentar por slug guardado en estado.
saved_slug = st.get("vault_slug")
if saved_slug and saved_slug in index["by_slug"]:
ficha = index["by_slug"][saved_slug]
if ficha is None:
# No encontramos la ficha -> tratar como create (raro).
slug = _unique_slug(_slug_for(parsed), index, planned_slugs)
planned_slugs.add(slug)
counts["create"] += 1
actions.append({
"kind": "create", "uid": uid, "etag": etag,
"href": r["href"], "parsed": parsed, "slug": slug,
"frontmatter": build_new_frontmatter(parsed, slug),
})
else:
diff = agenda_update_from_vcard(parsed, ficha)
counts["update"] += 1
actions.append({
"kind": "update", "uid": uid, "etag": etag,
"href": r["href"], "parsed": parsed,
"ficha": ficha, "diff": diff,
})
return {"actions": actions, "counts": counts}
# --------------------------------------------------------------------------
# Aplicar (solo --apply)
# --------------------------------------------------------------------------
def apply_pull(plan: dict, state: dict, index: dict) -> dict:
"""Ejecuta el plan: crea/actualiza fichas y reescribe el estado."""
res = {"created": 0, "updated": 0, "linked": 0, "errors": []}
now = int(time.time())
for a in plan["actions"]:
kind = a["kind"]
uid = a["uid"]
try:
if kind == "create":
rel = f"personas/{a['slug']}.md"
path = create_obsidian_note(
OSINT, rel, body="\n## Notas\n",
frontmatter=a["frontmatter"], overwrite=False)
res["created"] += 1
state[uid] = {"etag": a["etag"], "vault_slug": a["slug"],
"vault_path": path,
"vault_mtime": os.path.getmtime(path),
"last_sync": now}
elif kind in ("update", "link"):
ficha = a["ficha"]
diff = a["diff"]
if diff["set_frontmatter"]:
update_obsidian_note(
ficha["path"], set_frontmatter=diff["set_frontmatter"])
if kind == "update":
res["updated"] += 1
else:
res["linked"] += 1
else:
if kind == "link":
res["linked"] += 1
mtime = (os.path.getmtime(ficha["path"])
if os.path.exists(ficha["path"]) else now)
state[uid] = {"etag": a["etag"], "vault_slug": ficha["slug"],
"vault_path": ficha["path"],
"vault_mtime": mtime, "last_sync": now}
elif kind in ("skip", "skip_empty"):
# Mantener/actualizar el etag conocido sin tocar el vault.
if kind == "skip":
prev = state.get(uid, {})
prev["etag"] = a["etag"]
prev["last_sync"] = now
state[uid] = prev
except Exception as e: # noqa: BLE001
res["errors"].append({"uid": uid, "kind": kind, "error": str(e)})
return res
# --------------------------------------------------------------------------
# Reporte
# --------------------------------------------------------------------------
def report(plan: dict, index: dict, sample_n: int = 8):
c = plan["counts"]
print("=" * 72)
print("DRY-RUN — sync_dav_to_osint.py (Xandikos CardDAV -> vault OSINT)")
print("=" * 72)
print(f"servidor ........................ {DAV_BASE}")
print(f"coleccion ....................... {DAV_COLLECTION}")
print(f"fichas en el vault .............. {index['count']}")
print("-" * 72)
print(f"vCards en Xandikos .............. {c['total']}")
print(f" CREATE (contacto nuevo movil) {c['create']}")
print(f" UPDATE (movil edito agenda) .. {c['update']}")
print(f" LINK (ya en vault, enlazar) {c['link']}")
print(f" SKIP (sin cambios) ......... {c['skip']}")
print(f" SKIP (vCard vacio) ......... {c['skip_empty']}")
print("=" * 72)
# Muestra de CREATE.
creates = [a for a in plan["actions"] if a["kind"] == "create"]
if creates:
print(f"NUEVAS fichas a crear (muestra {min(sample_n, len(creates))} "
f"de {len(creates)}):")
print("-" * 72)
for a in creates[:sample_n]:
fm = a["frontmatter"]
print(f" + personas/{a['slug']}.md | UID {a['uid']}")
print(f" nombre={fm.get('nombre')!r} tel={fm.get('telefono')!r} "
f"email={fm.get('email')!r} contexto={fm.get('contexto')!r}")
print("=" * 72)
# Muestra de UPDATE con changes reales.
updates = [a for a in plan["actions"]
if a["kind"] in ("update", "link") and a["diff"]["changes"]]
if updates:
print(f"UPDATES de agenda (movil edito) (muestra "
f"{min(sample_n, len(updates))} de {len(updates)}):")
print("-" * 72)
for a in updates[:sample_n]:
f = a["ficha"]
print(f" ~ {f['path'].split('/osint/')[-1]} | UID {a['uid']} "
f"[{a['kind']}]")
for campo, (old, new) in a["diff"]["changes"].items():
print(f" {campo}: {old!r} -> {new!r} (OSINT preservado)")
print("=" * 72)
elif c["update"] or c["link"]:
print("(UPDATE/LINK detectados pero sin cambios de agenda netos)")
print("=" * 72)
# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(
description="Sincroniza Xandikos CardDAV -> vault OSINT (reverse). "
"Trae contactos nuevos y ediciones de agenda del movil "
"preservando la capa OSINT del vault.")
ap.add_argument("--apply", action="store_true",
help="Escribe el vault y el estado. Por defecto: dry-run.")
ap.add_argument("--dry-run", action="store_true",
help="No escribe nada (es el comportamiento por defecto; "
"el flag existe para ser explicito).")
ap.add_argument("--sample", type=int, default=8,
help="Numero de fichas a mostrar por categoria en el dry-run.")
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
print("Descargando coleccion de Xandikos (1 REPORT)...")
coll = dav_get_collection(DAV_BASE, DAV_USER, pwd, DAV_COLLECTION, "vcard")
if coll.get("status") != "ok":
print(f"ERROR: no se pudo leer la coleccion CardDAV: "
f"{coll.get('error')} (http {coll.get('http_status')})",
file=sys.stderr)
return 1
resources = coll.get("resources", [])
print(f" {len(resources)} vCards descargados.")
print("Indexando el vault OSINT...")
index = load_vault_index()
print(f" {index['count']} fichas indexadas.")
state = load_state()
print(f"Estado previo: {len(state)} UIDs conocidos "
f"({'existe' if os.path.exists(STATE_PATH) else 'nuevo'} .sync_state.json)")
plan = plan_pull(resources, index, state)
report(plan, index, sample_n=args.sample)
if args.apply:
print("\nAPLICANDO (escribiendo el vault + estado)...")
res = apply_pull(plan, state, index)
save_state(state)
print(f"APLICADO: created={res['created']} updated={res['updated']} "
f"linked={res['linked']} errors={len(res['errors'])}")
for e in res["errors"][:10]:
print(f" ERROR uid={e['uid']} [{e['kind']}]: {e['error']}")
print(f"Estado reescrito: {len(state)} UIDs en {STATE_PATH}")
else:
print("\n(dry-run: NO se toco el vault ni el estado. "
"Usa --apply para escribir.)")
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink