feat(dav,obsidian): grupo dav completo (CardDAV/CalDAV client + split vcf/ics + import pipelines) + build_obsidian_graph + dav_list_calendars

Funciones reutilizables creadas esta sesion para el sistema self-hosted de contactos/calendario (Xandikos) y la app osint_web:
- grupo dav (infra): split_vcards, split_vevents_to_vcalendars, extract_or_make_uid, carddav_put_vcard, caldav_put_event, dav_list_resources, dav_get_resource, dav_list_calendars
- pipelines: import_vcf_to_carddav, import_ics_to_caldav
- obsidian: build_obsidian_graph (grafo agregado del vault)
This commit is contained in:
2026-06-12 00:43:59 +02:00
parent 4a0f0e9dc0
commit a76760edba
32 changed files with 2814 additions and 0 deletions
@@ -0,0 +1,91 @@
---
name: caldav_put_event
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def caldav_put_event(base_url: str, username: str, password: str, collection_path: str, uid: str, vcalendar_text: str, *, timeout_s: float = 20.0, verify_tls: bool = True) -> dict"
description: "Sube (HTTP PUT) un VCALENDAR (con un VEVENT) a una coleccion CalDAV con HTTP Basic auth. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib. El nombre del recurso se deriva del UID saneado (safe(uid)+'.ics'). verify_tls=True por defecto. Idempotente por UID: re-subir el mismo UID sobrescribe el recurso. Maneja errores sin lanzar (HTTPError/URLError -> {status:'error'}). Solo stdlib (urllib, base64, re, ssl). Probado contra Xandikos."
tags: [dav, caldav, ical, ics, vevent, http, put, calendar, infra, upload]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [base64, re, ssl, urllib.error, urllib.request]
params:
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: collection_path
desc: "ruta de la coleccion CalDAV (p.ej. '/enmanuel/calendars/calendar/')."
- name: uid
desc: "UID del evento; se sanea ([^A-Za-z0-9_.-]->_ , max 120 chars) para formar el nombre del recurso .ics."
- name: vcalendar_text
desc: "texto completo del VCALENDAR (BEGIN:VCALENDAR..END:VCALENDAR) con un VEVENT. Se asegura terminacion en CRLF."
- name: timeout_s
desc: "timeout de la peticion HTTP en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
output: "dict. En exito: {status:'ok', http_status:int, url:str}. En error (sin lanzar): {status:'error', error:str, http_status:int|None}. http_status es el codigo HTTP devuelto (201 created / 204 no content tipico en CalDAV)."
tested: true
tests:
- "test_construye_request_put_con_headers_correctos"
- "test_url_se_forma_con_uid_saneado"
- "test_content_type_es_text_calendar"
- "test_extension_es_ics"
- "test_httperror_devuelve_status_error"
test_file_path: "python/functions/infra/caldav_put_event_test.py"
file_path: "python/functions/infra/caldav_put_event.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.caldav_put_event import caldav_put_event
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
cal = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//x//EN\r\nCALSCALE:GREGORIAN\r\n"
"BEGIN:VEVENT\r\nUID:evt-1@google.com\r\nSUMMARY:Reunion\r\n"
"DTSTART:20260101T100000Z\r\nDTEND:20260101T110000Z\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
res = caldav_put_event(
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
collection_path="/enmanuel/calendars/calendar/",
uid="evt-1@google.com",
vcalendar_text=cal,
)
print(res) # {"status": "ok", "http_status": 201, "url": ".../evt-1_google.com.ics"}
```
## Cuando usarla
Cuando quieres subir un evento individual a Xandikos (u otro servidor CalDAV)
por HTTP. Es la primitiva de escritura de calendario del grupo `dav`; el
pipeline `import_ics_to_caldav` la invoca por cada VCALENDAR producido por
`split_vevents_to_vcalendars`. Antes de llamarla, resuelve el UID con
`extract_or_make_uid` y la password con `pass_get_secret`.
## Gotchas
- Escritura remota real: re-subir el mismo UID SOBRESCRIBE el recurso
(idempotente, no duplica).
- El VCALENDAR debe ser completo y autonomo (header + VTIMEZONE necesarias +
un VEVENT). Subir un VEVENT suelto sin envolver en VCALENDAR fallara.
- Contrasena en header Basic sobre TLS; nunca hardcodear, leer de `pass`. No se
logea.
- `verify_tls=False` solo en pruebas; abre MITM.
- Devuelve dict (status/http_status/error), NO un int crudo: captura errores
HTTP/red sin lanzar.
@@ -0,0 +1,83 @@
"""Sube (PUT) un VCALENDAR a una coleccion CalDAV via HTTP Basic auth.
Funcion impura: hace una peticion HTTP PUT. Construye el header
`Authorization: Basic base64(user:pass)` a mano con stdlib. El nombre del
recurso se deriva del UID saneado (`safe(uid) + '.ics'`). Maneja errores sin
lanzar: devuelve {status: 'ok', http_status: int} en exito o
{status: 'error', error: str}. Solo usa stdlib (urllib, base64, re, ssl).
"""
import base64
import re
import ssl
import urllib.error
import urllib.request
_UNSAFE_RE = re.compile(r"[^A-Za-z0-9_.-]")
def _basic_auth_header(username: str, password: str) -> str:
raw = ("%s:%s" % (username, password)).encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
def _safe_resource_name(uid: str, ext: str) -> str:
safe = _UNSAFE_RE.sub("_", uid)[:120]
return safe + ext
def _join_url(base_url: str, collection_path: str, resource: str) -> str:
return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/" + resource
def caldav_put_event(
base_url: str,
username: str,
password: str,
collection_path: str,
uid: str,
vcalendar_text: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
) -> dict:
"""Sube un VCALENDAR (con un VEVENT) a una coleccion CalDAV (PUT).
Args:
base_url: URL base del servidor DAV (p.ej. 'https://dav-x.example.com').
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
collection_path: ruta de la coleccion CalDAV (p.ej.
'/enmanuel/calendars/calendar/').
uid: UID del evento; se sanea para formar el nombre del recurso.
vcalendar_text: texto completo del VCALENDAR (BEGIN:VCALENDAR..END).
timeout_s: timeout de la peticion en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS. No desactivar
salvo en entornos de prueba controlados.
Returns:
dict. En exito: {status: 'ok', http_status: int, url: str}. En error
(sin lanzar): {status: 'error', error: str, http_status: int|None}.
"""
resource = _safe_resource_name(uid, ".ics")
url = _join_url(base_url, collection_path, resource)
body = vcalendar_text
if not body.endswith("\r\n"):
body = body.rstrip("\r\n") + "\r\n"
data = body.encode("utf-8")
headers = {
"Authorization": _basic_auth_header(username, password),
"Content-Type": "text/calendar; charset=utf-8",
}
req = urllib.request.Request(url, data=data, method="PUT", headers=headers)
context = None if verify_tls else ssl._create_unverified_context()
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
return {"status": "ok", "http_status": resp.status, "url": url}
except urllib.error.HTTPError as e:
return {"status": "error", "error": "http %s" % e.code, "http_status": e.code}
except urllib.error.URLError as e:
return {"status": "error", "error": str(e.reason), "http_status": None}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e), "http_status": None}
@@ -0,0 +1,88 @@
"""Tests para caldav_put_event.
Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el
Request object (URL, method, headers) sin enviarlo a un servidor real.
"""
import sys
import infra.caldav_put_event # noqa: F401
mod = sys.modules["infra.caldav_put_event"]
caldav_put_event = mod.caldav_put_event
_CAL = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//x//EN\r\nCALSCALE:GREGORIAN\r\n"
"BEGIN:VEVENT\r\nUID:evt-1@google.com\r\nSUMMARY:Reunion\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
class _FakeResp:
status = 201
def __enter__(self):
return self
def __exit__(self, *a):
return False
def _capture(monkeypatch):
captured = {}
def fake_urlopen(req, timeout=None, context=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
return _FakeResp()
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
return captured
def _call():
return caldav_put_event(
base_url="https://dav.example.com",
username="enmanuel",
password="secret-pw",
collection_path="/enmanuel/calendars/calendar/",
uid="evt-1@google.com",
vcalendar_text=_CAL,
)
def test_construye_request_put_con_headers_correctos(monkeypatch):
cap = _capture(monkeypatch)
res = _call()
assert res["status"] == "ok"
assert res["http_status"] == 201
assert cap["method"] == "PUT"
def test_url_se_forma_con_uid_saneado(monkeypatch):
cap = _capture(monkeypatch)
_call()
assert cap["url"].endswith("/enmanuel/calendars/calendar/evt-1_google.com.ics")
def test_content_type_es_text_calendar(monkeypatch):
cap = _capture(monkeypatch)
_call()
assert cap["headers"]["content-type"] == "text/calendar; charset=utf-8"
def test_extension_es_ics(monkeypatch):
cap = _capture(monkeypatch)
_call()
assert cap["url"].endswith(".ics")
def test_httperror_devuelve_status_error(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.HTTPError(req.full_url, 403, "Forbidden", {}, None)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = _call()
assert res["status"] == "error"
assert res["http_status"] == 403
@@ -0,0 +1,85 @@
---
name: carddav_put_vcard
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def carddav_put_vcard(base_url: str, username: str, password: str, collection_path: str, uid: str, vcard_text: str, *, timeout_s: float = 20.0, verify_tls: bool = True) -> dict"
description: "Sube (HTTP PUT) un VCARD a una coleccion CardDAV con HTTP Basic auth. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib. El nombre del recurso se deriva del UID saneado (safe(uid)+'.vcf'). verify_tls=True por defecto. Idempotente por UID: re-subir el mismo UID sobrescribe el recurso. Maneja errores sin lanzar (HTTPError/URLError -> {status:'error'}). Solo stdlib (urllib, base64, re, ssl). Probado contra Xandikos."
tags: [dav, carddav, vcard, http, put, contacts, infra, upload]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [base64, re, ssl, urllib.error, urllib.request]
params:
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: collection_path
desc: "ruta de la coleccion CardDAV (p.ej. '/enmanuel/contacts/addressbook/')."
- name: uid
desc: "UID del contacto; se sanea ([^A-Za-z0-9_.-]->_ , max 120 chars) para formar el nombre del recurso .vcf."
- name: vcard_text
desc: "texto completo del VCARD (BEGIN:VCARD..END:VCARD). Se asegura terminacion en CRLF."
- name: timeout_s
desc: "timeout de la peticion HTTP en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
output: "dict. En exito: {status:'ok', http_status:int, url:str}. En error (sin lanzar): {status:'error', error:str, http_status:int|None}. http_status es el codigo HTTP devuelto por el servidor (201 created / 204 no content tipico en CardDAV)."
tested: true
tests:
- "test_construye_request_put_con_headers_correctos"
- "test_url_se_forma_con_uid_saneado"
- "test_content_type_es_text_vcard"
- "test_basic_auth_header_correcto"
- "test_httperror_devuelve_status_error"
test_file_path: "python/functions/infra/carddav_put_vcard_test.py"
file_path: "python/functions/infra/carddav_put_vcard.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.carddav_put_vcard import carddav_put_vcard
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
res = carddav_put_vcard(
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
collection_path="/enmanuel/contacts/addressbook/",
uid="abc-123@google.com",
vcard_text="BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Ada Lovelace\r\nUID:abc-123@google.com\r\nEND:VCARD\r\n",
)
print(res) # {"status": "ok", "http_status": 201, "url": ".../abc-123_google.com.vcf"}
```
## Cuando usarla
Cuando quieres subir un contacto individual a Xandikos (u otro servidor CardDAV)
por HTTP. Es la primitiva de escritura del grupo `dav`; el pipeline
`import_vcf_to_carddav` la invoca por cada tarjeta de un .vcf. Antes de llamarla,
resuelve el UID con `extract_or_make_uid` y la password con `pass_get_secret`.
## Gotchas
- Hace una escritura remota real: re-subir el mismo UID SOBRESCRIBE el recurso
en el servidor (idempotente, no acumula duplicados — esa es la intencion).
- La contrasena va en el header Basic en claro sobre TLS; nunca hardcodear, leer
de `pass`. La funcion no logea la password.
- `verify_tls=False` solo para entornos de prueba; deja un agujero MITM.
- El servidor puede rechazar (4xx) si el path de la coleccion no existe o el UID
del nombre del recurso no coincide con el UID dentro del VCARD: asegurate de
que el mismo UID se usa para el nombre del archivo y para el campo UID:.
- Devuelve dict (status/http_status/error), NO un int crudo: asi captura errores
HTTP/red sin lanzar. Consulta `res["http_status"]` para el codigo.
@@ -0,0 +1,83 @@
"""Sube (PUT) un VCARD a una coleccion CardDAV via HTTP Basic auth.
Funcion impura: hace una peticion HTTP PUT. Construye el header
`Authorization: Basic base64(user:pass)` a mano con stdlib. El nombre del
recurso se deriva del UID saneado (`safe(uid) + '.vcf'`). Maneja errores sin
lanzar: devuelve {status: 'ok', http_status: int} en exito o
{status: 'error', error: str}. Solo usa stdlib (urllib, base64, re, ssl).
"""
import base64
import re
import ssl
import urllib.error
import urllib.request
_UNSAFE_RE = re.compile(r"[^A-Za-z0-9_.-]")
def _basic_auth_header(username: str, password: str) -> str:
raw = ("%s:%s" % (username, password)).encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
def _safe_resource_name(uid: str, ext: str) -> str:
safe = _UNSAFE_RE.sub("_", uid)[:120]
return safe + ext
def _join_url(base_url: str, collection_path: str, resource: str) -> str:
return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/" + resource
def carddav_put_vcard(
base_url: str,
username: str,
password: str,
collection_path: str,
uid: str,
vcard_text: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
) -> dict:
"""Sube un VCARD a una coleccion CardDAV (PUT).
Args:
base_url: URL base del servidor DAV (p.ej. 'https://dav-x.example.com').
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
collection_path: ruta de la coleccion CardDAV (p.ej.
'/enmanuel/contacts/addressbook/').
uid: UID del contacto; se sanea para formar el nombre del recurso.
vcard_text: texto completo del VCARD (BEGIN:VCARD..END:VCARD).
timeout_s: timeout de la peticion en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS. No desactivar
salvo en entornos de prueba controlados.
Returns:
dict. En exito: {status: 'ok', http_status: int, url: str}. En error
(sin lanzar): {status: 'error', error: str, http_status: int|None}.
"""
resource = _safe_resource_name(uid, ".vcf")
url = _join_url(base_url, collection_path, resource)
body = vcard_text
if not body.endswith("\r\n"):
body = body.rstrip("\r\n") + "\r\n"
data = body.encode("utf-8")
headers = {
"Authorization": _basic_auth_header(username, password),
"Content-Type": "text/vcard; charset=utf-8",
}
req = urllib.request.Request(url, data=data, method="PUT", headers=headers)
context = None if verify_tls else ssl._create_unverified_context()
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
return {"status": "ok", "http_status": resp.status, "url": url}
except urllib.error.HTTPError as e:
return {"status": "error", "error": "http %s" % e.code, "http_status": e.code}
except urllib.error.URLError as e:
return {"status": "error", "error": str(e.reason), "http_status": None}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e), "http_status": None}
@@ -0,0 +1,88 @@
"""Tests para carddav_put_vcard.
Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el
Request object (URL, method, headers, body) sin enviarlo a un servidor real.
"""
import base64
import sys
import infra.carddav_put_vcard # noqa: F401
mod = sys.modules["infra.carddav_put_vcard"]
carddav_put_vcard = mod.carddav_put_vcard
class _FakeResp:
status = 201
def __enter__(self):
return self
def __exit__(self, *a):
return False
def _capture(monkeypatch):
captured = {}
def fake_urlopen(req, timeout=None, context=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["headers"] = dict(req.header_items())
captured["body"] = req.data
return _FakeResp()
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
return captured
def _call():
return carddav_put_vcard(
base_url="https://dav.example.com",
username="enmanuel",
password="secret-pw",
collection_path="/enmanuel/contacts/addressbook/",
uid="abc-123@google.com",
vcard_text="BEGIN:VCARD\r\nFN:Ada\r\nUID:abc-123@google.com\r\nEND:VCARD",
)
def test_construye_request_put_con_headers_correctos(monkeypatch):
cap = _capture(monkeypatch)
res = _call()
assert res == {"status": "ok", "http_status": 201, "url": cap["url"]}
assert cap["method"] == "PUT"
def test_url_se_forma_con_uid_saneado(monkeypatch):
cap = _capture(monkeypatch)
_call()
# El '@' del uid se sanea a '_'.
assert cap["url"].endswith("/enmanuel/contacts/addressbook/abc-123_google.com.vcf")
def test_content_type_es_text_vcard(monkeypatch):
cap = _capture(monkeypatch)
_call()
# urllib capitaliza las claves de header.
headers = {k.lower(): v for k, v in cap["headers"].items()}
assert headers["content-type"] == "text/vcard; charset=utf-8"
def test_basic_auth_header_correcto(monkeypatch):
cap = _capture(monkeypatch)
_call()
headers = {k.lower(): v for k, v in cap["headers"].items()}
expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii")
assert headers["authorization"] == expected
def test_httperror_devuelve_status_error(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.HTTPError(req.full_url, 409, "Conflict", {}, None)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = _call()
assert res["status"] == "error"
assert res["http_status"] == 409
@@ -0,0 +1,75 @@
---
name: dav_get_resource
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def dav_get_resource(base_url: str, username: str, password: str, resource_path: str, *, timeout_s: float = 20.0, verify_tls: bool = True) -> dict"
description: "Descarga (HTTP GET) el contenido de un recurso DAV individual (un VCARD o un VCALENDAR) con HTTP Basic auth. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib. El resource_path puede ser un href absoluto (como los que devuelve dav_list_resources) o una URL completa. verify_tls=True por defecto. Maneja errores sin lanzar. Solo stdlib (urllib, base64, ssl). Probado contra Xandikos."
tags: [dav, carddav, caldav, get, download, http, infra]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [base64, ssl, urllib.error, urllib.request]
params:
- name: base_url
desc: "URL base del servidor DAV. Se ignora si resource_path ya es una URL absoluta."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: resource_path
desc: "href absoluto (p.ej. '/enmanuel/contacts/addressbook/x.vcf') o URL completa del recurso a descargar. Acepta directamente los hrefs que devuelve dav_list_resources."
- name: timeout_s
desc: "timeout de la peticion HTTP en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
output: "dict. En exito: {status:'ok', http_status:int, text:str, url:str} donde text es el cuerpo del recurso (VCARD o VCALENDAR). En error (sin lanzar): {status:'error', error:str, http_status:int|None}."
tested: true
tests:
- "test_construye_request_get_con_auth"
- "test_resource_path_relativo_se_resuelve_con_base_url"
- "test_resource_path_absoluto_se_respeta"
- "test_devuelve_texto_del_recurso"
- "test_httperror_devuelve_status_error"
test_file_path: "python/functions/infra/dav_get_resource_test.py"
file_path: "python/functions/infra/dav_get_resource.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.dav_list_resources import dav_list_resources
from infra.dav_get_resource import dav_get_resource
base = "https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com"
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
listing = dav_list_resources(base, "enmanuel", pw, "/enmanuel/contacts/addressbook/")
first = listing["resources"][0]["href"]
res = dav_get_resource(base, "enmanuel", pw, first)
print(res["text"][:13]) # BEGIN:VCARD
```
## Cuando usarla
Cuando quieres leer el contenido de un recurso concreto cuyo href ya conoces
(por `dav_list_resources` o porque lo construyes tu). Util para hacer backup de
una coleccion (listar + get cada uno), validar que un import quedo bien escrito,
o comparar etags en un sync. Acepta directamente los hrefs del listing sin
reconstruir la URL.
## Gotchas
- Lectura remota real sobre TLS; password de `pass`, no se logea.
- Decodifica el cuerpo como UTF-8 con `errors='replace'`: bytes invalidos se
sustituyen por el caracter de reemplazo en vez de fallar.
- Si resource_path es relativo se concatena a base_url; si es absoluto
(http/https) se usa tal cual y base_url se ignora.
- `verify_tls=False` solo en pruebas; abre MITM.
@@ -0,0 +1,67 @@
"""Descarga (GET) un recurso DAV individual via HTTP Basic auth.
Funcion impura: hace una peticion HTTP GET. Construye el header
`Authorization: Basic base64(user:pass)` a mano con stdlib. Devuelve el cuerpo
del recurso como texto (un VCARD o un VCALENDAR). El resource_path puede ser un
href absoluto (como los que devuelve dav_list_resources) o una ruta relativa al
base_url. Maneja errores sin lanzar. Solo usa stdlib (urllib, base64, ssl).
"""
import base64
import ssl
import urllib.error
import urllib.request
def _basic_auth_header(username: str, password: str) -> str:
raw = ("%s:%s" % (username, password)).encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
def _resolve_url(base_url: str, resource_path: str) -> str:
if resource_path.startswith("http://") or resource_path.startswith("https://"):
return resource_path
return base_url.rstrip("/") + "/" + resource_path.lstrip("/")
def dav_get_resource(
base_url: str,
username: str,
password: str,
resource_path: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
) -> dict:
"""Descarga el contenido de un recurso DAV (GET).
Args:
base_url: URL base del servidor DAV. Se ignora si resource_path ya es
una URL absoluta.
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
resource_path: href absoluto (p.ej. '/enmanuel/contacts/addressbook/x.vcf')
o URL completa del recurso a descargar.
timeout_s: timeout de la peticion en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS.
Returns:
dict. En exito: {status:'ok', http_status:int, text:str, url:str} donde
text es el cuerpo del recurso (VCARD o VCALENDAR). En error (sin lanzar):
{status:'error', error:str, http_status:int|None}.
"""
url = _resolve_url(base_url, resource_path)
headers = {"Authorization": _basic_auth_header(username, password)}
req = urllib.request.Request(url, method="GET", headers=headers)
context = None if verify_tls else ssl._create_unverified_context()
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
text = resp.read().decode("utf-8", "replace")
return {"status": "ok", "http_status": resp.status, "text": text, "url": url}
except urllib.error.HTTPError as e:
return {"status": "error", "error": "http %s" % e.code, "http_status": e.code}
except urllib.error.URLError as e:
return {"status": "error", "error": str(e.reason), "http_status": None}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e), "http_status": None}
@@ -0,0 +1,88 @@
"""Tests para dav_get_resource.
Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el
Request (method GET, auth, URL) y devolver un cuerpo simulado.
"""
import base64
import sys
import infra.dav_get_resource # noqa: F401
mod = sys.modules["infra.dav_get_resource"]
dav_get_resource = mod.dav_get_resource
_BODY = "BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Ada Lovelace\r\nEND:VCARD\r\n"
class _FakeResp:
status = 200
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return _BODY.encode("utf-8")
def _capture(monkeypatch):
captured = {}
def fake_urlopen(req, timeout=None, context=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
return _FakeResp()
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
return captured
def test_construye_request_get_con_auth(monkeypatch):
cap = _capture(monkeypatch)
res = dav_get_resource(
"https://dav.example.com", "enmanuel", "secret-pw",
"/enmanuel/contacts/addressbook/ada.vcf",
)
assert res["status"] == "ok"
assert cap["method"] == "GET"
expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii")
assert cap["headers"]["authorization"] == expected
def test_resource_path_relativo_se_resuelve_con_base_url(monkeypatch):
cap = _capture(monkeypatch)
dav_get_resource(
"https://dav.example.com", "u", "p",
"/enmanuel/contacts/addressbook/ada.vcf",
)
assert cap["url"] == "https://dav.example.com/enmanuel/contacts/addressbook/ada.vcf"
def test_resource_path_absoluto_se_respeta(monkeypatch):
cap = _capture(monkeypatch)
abs_url = "https://otra.example.com/path/x.vcf"
dav_get_resource("https://dav.example.com", "u", "p", abs_url)
assert cap["url"] == abs_url
def test_devuelve_texto_del_recurso(monkeypatch):
_capture(monkeypatch)
res = dav_get_resource(
"https://dav.example.com", "u", "p", "/x.vcf",
)
assert res["text"] == _BODY
assert res["http_status"] == 200
def test_httperror_devuelve_status_error(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.HTTPError(req.full_url, 404, "Not Found", {}, None)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = dav_get_resource("https://dav.example.com", "u", "p", "/x.vcf")
assert res["status"] == "error"
assert res["http_status"] == 404
@@ -0,0 +1,85 @@
---
name: dav_list_calendars
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def dav_list_calendars(base_url: str, username: str, password: str, home_path: str, *, timeout_s: float = 15.0, verify_tls: bool = True) -> dict"
description: "Lista las colecciones de calendario CalDAV bajo un calendar-home en UNA peticion PROPFIND Depth:1. Devuelve solo las colecciones hijas que son calendarios CalDAV de verdad (resourcetype {urn:ietf:params:xml:ns:caldav}calendar), cada una con su href, su displayname (DAV) y su color (calendar-color de Apple, ej. #FF2968FF) si el servidor lo expone. El propio calendar-home (coleccion plana sin el resourcetype calendar) y cualquier coleccion no-calendario se excluyen. Considera solo los propstat con estado 2xx para no leer props marcadas 404. Construye Authorization: Basic base64(user:pass) a mano y parsea el multistatus con regex. verify_tls=True por defecto. Maneja errores sin lanzar. Solo stdlib (urllib, base64, re, ssl, html). Probado contra Xandikos."
tags: [dav, caldav, calendar, calendars, propfind, displayname, color, selector, http, infra]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [base64, html, re, ssl, urllib.error, urllib.request]
params:
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: home_path
desc: "ruta del calendar-home del usuario (p.ej. '/enmanuel/calendars/'). Las colecciones de calendario cuelgan de el."
- name: timeout_s
desc: "timeout de la peticion HTTP en segundos. Default 15.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
output: "dict. En exito: {status:'ok', http_status:int, calendars:[{href:str, name:str, color:str|None}, ...]} con un elemento por coleccion de calendario, ordenadas por nombre. color es el calendar-color de Apple (ej. '#FF2968FF') o None. En error (sin lanzar): {status:'error', error:str, http_status:int|None}."
tested: true
tests:
- "test_lista_solo_calendarios"
- "test_excluye_home_plano"
- "test_extrae_nombre_y_color"
- "test_color_ausente_es_none"
- "test_ignora_props_404"
- "test_httperror_devuelve_status_error"
- "test_urlerror_sin_red"
test_file_path: "python/functions/infra/dav_list_calendars_test.py"
file_path: "python/functions/infra/dav_list_calendars.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.dav_list_calendars import dav_list_calendars
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
res = dav_list_calendars(
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
home_path="/enmanuel/calendars/",
)
for c in res["calendars"]:
print(c["name"], c["href"], c["color"])
# calendar /enmanuel/calendars/calendar/ None
```
## Cuando usarla
Cuando una UI necesita un selector de calendario: el usuario tiene varias
colecciones CalDAV bajo su calendar-home y quiere elegir una, con su nombre y su
color para pintarla. Devuelve el href de cada calendario (lo que luego pasas a
`dav_get_collection` / `caldav_put_event` como `collection_path`) sin que el
caller tenga que conocerlos de antemano ni distinguir el calendar-home de los
calendarios reales.
## Gotchas
- Filtra por el resourcetype `caldav:calendar`: el propio calendar-home es una
coleccion plana (sin ese resourcetype) y queda fuera, igual que carpetas
intermedias. Si tu servidor anida calendarios mas profundo que Depth:1, llama
con `home_path` apuntando al nivel correcto.
- El `color` es el `calendar-color` de Apple (`http://apple.com/ns/ical/`), que
Xandikos puede no tener seteado (devuelve None). Apple usa formato `#RRGGBBAA`
(8 digitos, con alfa); recortalo a `#RRGGBB` si tu UI no soporta alfa.
- Solo lee los `<propstat>` con estado 2xx para no confundir un
`<calendar-color/>` vacio de un propstat 404 con un color real.
- Lectura remota real sobre TLS; password de `pass`, no se logea.
@@ -0,0 +1,180 @@
"""Lista las colecciones de calendario CalDAV bajo un calendar-home (PROPFIND).
Funcion impura: hace UNA peticion HTTP PROPFIND Depth:1 sobre el directorio
calendar-home de un usuario (p.ej. `/enmanuel/calendars/`) y devuelve solo las
colecciones hijas que son calendarios CalDAV de verdad — las que declaran el
resourcetype `{urn:ietf:params:xml:ns:caldav}calendar`. Por cada una extrae su
href, su `displayname` (DAV) y, si el servidor lo expone, su color
(`calendar-color` de Apple, ej. `#FF2968FF`). El propio calendar-home (que es una
coleccion plana sin el resourcetype calendar) se excluye.
Esto es lo que necesita un selector de calendario en una UI: el usuario tiene
varias colecciones bajo su calendar-home y quiere elegir una, con su nombre y su
color. `dav_list_resources` solo devuelve hrefs+etag de los recursos de UNA
coleccion (los eventos), no las colecciones hijas con su metadata.
Construye el header `Authorization: Basic base64(user:pass)` a mano con stdlib y
parsea el multistatus con regex simple (sin parser XML externo), considerando
solo los `<propstat>` con estado 2xx para no recoger props que el servidor marca
404. Maneja errores sin lanzar. Solo usa stdlib (urllib, base64, re, ssl, html).
Probado contra Xandikos.
"""
import base64
import html
import re
import ssl
import urllib.error
import urllib.request
_RESPONSE_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?response>(.*?)</(?:[A-Za-z0-9]+:)?response>",
re.DOTALL | re.IGNORECASE,
)
_HREF_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?href>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?href>",
re.DOTALL | re.IGNORECASE,
)
_PROPSTAT_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?propstat>(.*?)</(?:[A-Za-z0-9]+:)?propstat>",
re.DOTALL | re.IGNORECASE,
)
_STATUS_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?status>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?status>",
re.DOTALL | re.IGNORECASE,
)
_DISPLAYNAME_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?displayname>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?displayname>",
re.DOTALL | re.IGNORECASE,
)
# Color de Apple: <ns:calendar-color>#RRGGBBAA</ns:calendar-color> (o sin alfa).
_COLOR_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?calendar-color[^>]*>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?calendar-color>",
re.DOTALL | re.IGNORECASE,
)
# Marca de calendario CalDAV en el resourcetype. El elemento `<C:calendar/>`
# puede venir con o sin prefijo de namespace (`<ns2:calendar/>`, `<calendar/>`).
_CALENDAR_TYPE_RE = re.compile(
r"<(?:[A-Za-z0-9]+:)?calendar(?:\s[^>]*)?/?>", re.IGNORECASE
)
# El PROPFIND pide nombre, tipo y color (Apple). Declarar el namespace de Apple y
# el de CalDAV permite que el servidor responda esas props cuando existan.
_PROPFIND_BODY = (
'<?xml version="1.0" encoding="utf-8" ?>'
'<D:propfind xmlns:D="DAV:" '
'xmlns:A="http://apple.com/ns/ical/" '
'xmlns:C="urn:ietf:params:xml:ns:caldav">'
"<D:prop>"
"<D:displayname/><D:resourcetype/><A:calendar-color/>"
"</D:prop></D:propfind>"
)
def _basic_auth_header(username: str, password: str) -> str:
raw = ("%s:%s" % (username, password)).encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
def _join_url(base_url: str, collection_path: str) -> str:
return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/"
def _ok_propstats(response_block: str) -> str:
"""Concatena solo los `<propstat>` con estado 2xx de un `<response>`.
El servidor agrupa las props por estado: las presentes en un propstat 200 y
las ausentes en un propstat 404. Tomar solo los 2xx evita leer un
`<calendar-color/>` vacio del bloque 404 como si fuera el valor real.
"""
parts = []
for ps in _PROPSTAT_RE.findall(response_block):
status_m = _STATUS_RE.search(ps)
if status_m and " 2" in (" " + status_m.group(1)):
parts.append(ps)
# Si no hay propstat (servidor minimalista), usar el bloque entero.
return "".join(parts) if parts else response_block
def dav_list_calendars(
base_url: str,
username: str,
password: str,
home_path: str,
*,
timeout_s: float = 15.0,
verify_tls: bool = True,
) -> dict:
"""Lista las colecciones de calendario CalDAV bajo un calendar-home.
Hace un PROPFIND Depth:1 sobre `home_path` y devuelve solo las colecciones
hijas marcadas como calendario CalDAV (resourcetype `caldav:calendar`), con
su nombre y color. El propio `home_path` y cualquier coleccion no-calendario
se excluyen.
Args:
base_url: URL base del servidor DAV (p.ej. 'https://dav-x.example.com').
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth. Resolver desde pass.
home_path: ruta del calendar-home del usuario (p.ej.
'/enmanuel/calendars/'). Las colecciones de calendario cuelgan de el.
timeout_s: timeout de la peticion en segundos. Default 15.0.
verify_tls: si True (default) verifica el certificado TLS.
Returns:
dict. En exito: {status:'ok', http_status:int,
calendars:[{href:str, name:str, color:str|None}, ...]} con un elemento
por coleccion de calendario (ordenadas por nombre). `color` es el
`calendar-color` de Apple (ej. '#FF2968FF') si el servidor lo expone, o
None. En error (sin lanzar): {status:'error', error:str,
http_status:int|None}.
"""
url = _join_url(base_url, home_path)
headers = {
"Authorization": _basic_auth_header(username, password),
"Content-Type": "application/xml; charset=utf-8",
"Depth": "1",
}
req = urllib.request.Request(
url, data=_PROPFIND_BODY.encode("utf-8"), method="PROPFIND", headers=headers
)
context = None if verify_tls else ssl._create_unverified_context()
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
status = resp.status
xml = resp.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return {"status": "error", "error": "http %s" % e.code, "http_status": e.code}
except urllib.error.URLError as e:
return {"status": "error", "error": str(e.reason), "http_status": None}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e), "http_status": None}
home_tail = home_path.strip("/").rsplit("/", 1)[-1]
calendars = []
for block in _RESPONSE_RE.findall(xml):
href_m = _HREF_RE.search(block)
if not href_m:
continue
href = href_m.group(1).strip()
tail = href.rstrip("/").rsplit("/", 1)[-1]
# El propio calendar-home (o un href identico al home): se excluye.
if tail == home_tail:
continue
# Solo las colecciones marcadas como calendario CalDAV. El home plano no
# lleva el resourcetype `caldav:calendar` y queda fuera.
if not _CALENDAR_TYPE_RE.search(block):
continue
ok = _ok_propstats(block)
name_m = _DISPLAYNAME_RE.search(ok)
name = html.unescape(name_m.group(1).strip()) if name_m else tail
if not name:
name = tail
color_m = _COLOR_RE.search(ok)
color = html.unescape(color_m.group(1).strip()) if color_m else None
if color == "":
color = None
calendars.append({"href": href, "name": name, "color": color})
calendars.sort(key=lambda c: c["name"].lower())
return {"status": "ok", "http_status": status, "calendars": calendars}
@@ -0,0 +1,180 @@
"""Tests para dav_list_calendars.
Smoke deterministas: monkeypatchean urllib.request.urlopen para devolver un
multistatus simulado al estilo Xandikos (PROPFIND Depth:1 sobre un calendar-home
con el propio home plano + dos calendarios CalDAV, uno con color y otro sin) y
una coleccion no-calendario que debe quedar fuera. Cubren: filtrado por
resourcetype caldav:calendar, exclusion del home plano, extraccion de nombre y
color, color ausente -> None, props marcadas 404 ignoradas, y los paths de
error HTTP / sin red.
"""
import sys
import infra.dav_list_calendars # noqa: F401
mod = sys.modules["infra.dav_list_calendars"]
dav_list_calendars = mod.dav_list_calendars
# Multistatus estilo Xandikos:
# - /enmanuel/calendars/ -> home plano (collection, sin caldav:calendar)
# - /enmanuel/calendars/calendar/ -> calendario CalDAV sin color (color en 404)
# - /enmanuel/calendars/trabajo/ -> calendario CalDAV con calendar-color
# - /enmanuel/calendars/inbox/ -> coleccion NO-calendario (debe excluirse)
_XML_HOME = (
'<?xml version="1.0"?>'
'<ns0:multistatus xmlns:ns0="DAV:" '
'xmlns:ns1="http://apple.com/ns/ical/" '
'xmlns:ns2="urn:ietf:params:xml:ns:caldav">'
# Home plano: collection, sin <ns2:calendar/>.
"<ns0:response><ns0:href>/enmanuel/calendars/</ns0:href>"
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
"<ns0:displayname>calendars</ns0:displayname>"
"<ns0:resourcetype><ns0:collection /></ns0:resourcetype>"
"</ns0:prop></ns0:propstat>"
"<ns0:propstat><ns0:status>HTTP/1.1 404 Not Found</ns0:status>"
"<ns0:prop><ns1:calendar-color /></ns0:prop></ns0:propstat>"
"</ns0:response>"
# Calendario sin color: el color va en un propstat 404 (no debe leerse).
"<ns0:response><ns0:href>/enmanuel/calendars/calendar/</ns0:href>"
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
"<ns0:displayname>calendar</ns0:displayname>"
"<ns0:resourcetype><ns0:collection /><ns2:calendar /></ns0:resourcetype>"
"</ns0:prop></ns0:propstat>"
"<ns0:propstat><ns0:status>HTTP/1.1 404 Not Found</ns0:status>"
"<ns0:prop><ns1:calendar-color /></ns0:prop></ns0:propstat>"
"</ns0:response>"
# Calendario con color (Apple #RRGGBBAA) y displayname con acento.
"<ns0:response><ns0:href>/enmanuel/calendars/trabajo/</ns0:href>"
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
"<ns0:displayname>Trabajo &amp; ocio</ns0:displayname>"
"<ns0:resourcetype><ns0:collection /><ns2:calendar /></ns0:resourcetype>"
"<ns1:calendar-color>#FF2968FF</ns1:calendar-color>"
"</ns0:prop></ns0:propstat></ns0:response>"
# Coleccion NO-calendario (p.ej. un inbox de scheduling): excluida.
"<ns0:response><ns0:href>/enmanuel/calendars/inbox/</ns0:href>"
"<ns0:propstat><ns0:status>HTTP/1.1 200 OK</ns0:status><ns0:prop>"
"<ns0:displayname>inbox</ns0:displayname>"
"<ns0:resourcetype><ns0:collection /><ns2:schedule-inbox /></ns0:resourcetype>"
"</ns0:prop></ns0:propstat></ns0:response>"
"</ns0:multistatus>"
)
class _FakeResp:
def __init__(self, payload: str):
self._payload = payload
self.status = 207
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return self._payload.encode("utf-8")
def _capture(monkeypatch, payload: str):
captured = {}
def fake_urlopen(req, timeout=None, context=None):
captured["method"] = req.get_method()
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
return _FakeResp(payload)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
return captured
def _call(home="/enmanuel/calendars/"):
return dav_list_calendars(
"https://dav.example.com", "enmanuel", "secret-pw", home
)
def test_construye_propfind_depth_1(monkeypatch):
cap = _capture(monkeypatch, _XML_HOME)
_call()
assert cap["method"] == "PROPFIND"
assert cap["headers"]["depth"] == "1"
def test_lista_solo_calendarios(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
assert res["status"] == "ok"
hrefs = [c["href"] for c in res["calendars"]]
# calendar + trabajo (los dos caldav:calendar), nada mas.
assert "/enmanuel/calendars/calendar/" in hrefs
assert "/enmanuel/calendars/trabajo/" in hrefs
assert len(res["calendars"]) == 2
def test_excluye_home_plano(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
hrefs = [c["href"] for c in res["calendars"]]
assert "/enmanuel/calendars/" not in hrefs
def test_excluye_coleccion_no_calendario(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
hrefs = [c["href"] for c in res["calendars"]]
assert "/enmanuel/calendars/inbox/" not in hrefs
def test_extrae_nombre_y_color(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
by_href = {c["href"]: c for c in res["calendars"]}
trabajo = by_href["/enmanuel/calendars/trabajo/"]
assert trabajo["name"] == "Trabajo & ocio" # entidad XML des-escapada
assert trabajo["color"] == "#FF2968FF"
def test_color_ausente_es_none(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
by_href = {c["href"]: c for c in res["calendars"]}
cal = by_href["/enmanuel/calendars/calendar/"]
assert cal["name"] == "calendar"
assert cal["color"] is None
def test_ignora_props_404(monkeypatch):
"""El <calendar-color/> vacio de un propstat 404 NO se lee como color real."""
_capture(monkeypatch, _XML_HOME)
res = _call()
by_href = {c["href"]: c for c in res["calendars"]}
# calendar tiene calendar-color SOLO en el propstat 404 -> color None, no "".
assert by_href["/enmanuel/calendars/calendar/"]["color"] is None
def test_ordenado_por_nombre(monkeypatch):
_capture(monkeypatch, _XML_HOME)
res = _call()
names = [c["name"] for c in res["calendars"]]
assert names == sorted(names, key=str.lower)
def test_httperror_devuelve_status_error(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, None)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = _call()
assert res["status"] == "error"
assert res["http_status"] == 401
def test_urlerror_sin_red(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.URLError("sin red")
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = _call()
assert res["status"] == "error"
assert res["http_status"] is None
@@ -0,0 +1,82 @@
---
name: dav_list_resources
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def dav_list_resources(base_url: str, username: str, password: str, collection_path: str, *, timeout_s: float = 20.0, verify_tls: bool = True) -> dict"
description: "Lista los recursos de una coleccion DAV (CardDAV o CalDAV) via PROPFIND Depth:1 con HTTP Basic auth. Construye el header Authorization: Basic base64(user:pass) a mano con stdlib. Parsea el XML multistatus con regex simple (sin parser XML externo) y devuelve los hrefs + getetag de cada recurso, excluyendo la propia coleccion. verify_tls=True por defecto. Maneja errores sin lanzar. Solo stdlib (urllib, base64, re, ssl). Probado contra Xandikos."
tags: [dav, carddav, caldav, propfind, list, http, infra]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [base64, re, ssl, urllib.error, urllib.request]
params:
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: collection_path
desc: "ruta de la coleccion a listar (CardDAV '/enmanuel/contacts/addressbook/' o CalDAV '/enmanuel/calendars/calendar/')."
- name: timeout_s
desc: "timeout de la peticion HTTP en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
output: "dict. En exito: {status:'ok', http_status:int, resources:[{href:str, etag:str|None}, ...]} con un elemento por recurso de la coleccion (excluida la propia coleccion). En error (sin lanzar): {status:'error', error:str, http_status:int|None}."
tested: true
tests:
- "test_construye_request_propfind_depth_1"
- "test_basic_auth_header_correcto"
- "test_parsea_hrefs_y_etags_del_multistatus"
- "test_excluye_la_propia_coleccion"
- "test_httperror_devuelve_status_error"
test_file_path: "python/functions/infra/dav_list_resources_test.py"
file_path: "python/functions/infra/dav_list_resources.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.dav_list_resources import dav_list_resources
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
res = dav_list_resources(
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
collection_path="/enmanuel/contacts/addressbook/",
)
print(res["status"], len(res["resources"]))
# ok 820
print(res["resources"][0]) # {"href": "/enmanuel/contacts/addressbook/abc.vcf", "etag": '"..."'}
```
## Cuando usarla
Cuando quieres enumerar lo que ya hay en una coleccion CardDAV/CalDAV: contar
contactos/eventos importados, verificar una migracion, o construir un mapa
href->etag para sync incremental. Sirve igual para libretas de direcciones y
calendarios (PROPFIND es generico). Combinala con `dav_get_resource` para
descargar el contenido de cada href.
## Gotchas
- Usa PROPFIND Depth:1 (no addressbook-query REPORT): lista TODOS los recursos
hijos de la coleccion. Para colecciones enormes la respuesta XML puede ser
grande; el timeout default es 20s.
- El parseo es regex simple sobre el multistatus, no un parser XML completo: es
robusto para la salida estandar de Xandikos pero podria fallar con servidores
que devuelvan XML muy exotico. La intencion es KISS sin dependencias.
- Excluye la propia coleccion comparando el ultimo segmento del href; si tu
coleccion y un recurso comparten exactamente el ultimo segmento (raro), ese
recurso se omitiria.
- Lectura remota real sobre TLS; password de `pass`, no se logea.
@@ -0,0 +1,101 @@
"""Lista los recursos de una coleccion DAV via PROPFIND Depth:1.
Funcion impura: hace una peticion HTTP PROPFIND. Construye el header
`Authorization: Basic base64(user:pass)` a mano con stdlib. Devuelve los hrefs
(y getetag cuando el servidor los expone) de los recursos de la coleccion,
parseados del XML multistatus con regex simple (sin dependencias de parser XML
externas). Sirve tanto para colecciones CardDAV como CalDAV. Maneja errores sin
lanzar. Solo usa stdlib (urllib, base64, re, ssl).
"""
import base64
import re
import ssl
import urllib.error
import urllib.request
_HREF_RE = re.compile(r"<(?:[A-Za-z0-9]+:)?href>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?href>", re.DOTALL | re.IGNORECASE)
_RESPONSE_RE = re.compile(r"<(?:[A-Za-z0-9]+:)?response>(.*?)</(?:[A-Za-z0-9]+:)?response>", re.DOTALL | re.IGNORECASE)
_ETAG_RE = re.compile(r"<(?:[A-Za-z0-9]+:)?getetag>\s*(.*?)\s*</(?:[A-Za-z0-9]+:)?getetag>", re.DOTALL | re.IGNORECASE)
_PROPFIND_BODY = (
'<?xml version="1.0" encoding="utf-8" ?>'
'<D:propfind xmlns:D="DAV:"><D:prop>'
"<D:getetag/><D:resourcetype/>"
"</D:prop></D:propfind>"
)
def _basic_auth_header(username: str, password: str) -> str:
raw = ("%s:%s" % (username, password)).encode("utf-8")
return "Basic " + base64.b64encode(raw).decode("ascii")
def _join_url(base_url: str, collection_path: str) -> str:
return base_url.rstrip("/") + "/" + collection_path.strip("/") + "/"
def dav_list_resources(
base_url: str,
username: str,
password: str,
collection_path: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
) -> dict:
"""Lista los recursos de una coleccion DAV (PROPFIND Depth:1).
Args:
base_url: URL base del servidor DAV.
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
collection_path: ruta de la coleccion (CardDAV o CalDAV).
timeout_s: timeout de la peticion en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS.
Returns:
dict. En exito: {status:'ok', http_status:int,
resources:[{href:str, etag:str|None}, ...]}. El primer <response> suele
ser la propia coleccion; se excluye comparando su href con la ruta de la
coleccion. En error (sin lanzar): {status:'error', error:str,
http_status:int|None}.
"""
url = _join_url(base_url, collection_path)
headers = {
"Authorization": _basic_auth_header(username, password),
"Content-Type": "application/xml; charset=utf-8",
"Depth": "1",
}
req = urllib.request.Request(
url, data=_PROPFIND_BODY.encode("utf-8"), method="PROPFIND", headers=headers
)
context = None if verify_tls else ssl._create_unverified_context()
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
status = resp.status
xml = resp.read().decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return {"status": "error", "error": "http %s" % e.code, "http_status": e.code}
except urllib.error.URLError as e:
return {"status": "error", "error": str(e.reason), "http_status": None}
except Exception as e: # noqa: BLE001
return {"status": "error", "error": str(e), "http_status": None}
coll_tail = collection_path.strip("/").rsplit("/", 1)[-1]
resources = []
for block in _RESPONSE_RE.findall(xml):
href_m = _HREF_RE.search(block)
if not href_m:
continue
href = href_m.group(1).strip()
# El ultimo segmento del href identifica el recurso. Si coincide con el
# ultimo segmento de la coleccion, ese <response> ES la coleccion: skip.
tail = href.rstrip("/").rsplit("/", 1)[-1]
if tail == coll_tail:
continue
etag_m = _ETAG_RE.search(block)
etag = etag_m.group(1).strip() if etag_m else None
resources.append({"href": href, "etag": etag})
return {"status": "ok", "http_status": status, "resources": resources}
@@ -0,0 +1,107 @@
"""Tests para dav_list_resources.
Smoke deterministas: monkeypatchean urllib.request.urlopen para capturar el
Request (method PROPFIND, Depth, auth) y devolver un XML multistatus simulado.
"""
import base64
import io
import sys
import infra.dav_list_resources # noqa: F401
mod = sys.modules["infra.dav_list_resources"]
dav_list_resources = mod.dav_list_resources
_XML = (
'<?xml version="1.0"?>'
'<D:multistatus xmlns:D="DAV:">'
"<D:response><D:href>/enmanuel/contacts/addressbook/</D:href>"
"<D:propstat><D:prop><D:resourcetype><D:collection/></D:resourcetype>"
"</D:prop></D:propstat></D:response>"
"<D:response><D:href>/enmanuel/contacts/addressbook/ada.vcf</D:href>"
'<D:propstat><D:prop><D:getetag>"etag-ada"</D:getetag></D:prop></D:propstat>'
"</D:response>"
"<D:response><D:href>/enmanuel/contacts/addressbook/alan.vcf</D:href>"
'<D:propstat><D:prop><D:getetag>"etag-alan"</D:getetag></D:prop></D:propstat>'
"</D:response>"
"</D:multistatus>"
)
class _FakeResp:
status = 207
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return _XML.encode("utf-8")
def _capture(monkeypatch):
captured = {}
def fake_urlopen(req, timeout=None, context=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
return _FakeResp()
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
return captured
def _call():
return dav_list_resources(
base_url="https://dav.example.com",
username="enmanuel",
password="secret-pw",
collection_path="/enmanuel/contacts/addressbook/",
)
def test_construye_request_propfind_depth_1(monkeypatch):
cap = _capture(monkeypatch)
_call()
assert cap["method"] == "PROPFIND"
assert cap["headers"]["depth"] == "1"
def test_basic_auth_header_correcto(monkeypatch):
cap = _capture(monkeypatch)
_call()
expected = "Basic " + base64.b64encode(b"enmanuel:secret-pw").decode("ascii")
assert cap["headers"]["authorization"] == expected
def test_parsea_hrefs_y_etags_del_multistatus(monkeypatch):
_capture(monkeypatch)
res = _call()
assert res["status"] == "ok"
hrefs = [r["href"] for r in res["resources"]]
assert "/enmanuel/contacts/addressbook/ada.vcf" in hrefs
assert "/enmanuel/contacts/addressbook/alan.vcf" in hrefs
etags = {r["href"]: r["etag"] for r in res["resources"]}
assert etags["/enmanuel/contacts/addressbook/ada.vcf"] == '"etag-ada"'
def test_excluye_la_propia_coleccion(monkeypatch):
_capture(monkeypatch)
res = _call()
hrefs = [r["href"] for r in res["resources"]]
assert "/enmanuel/contacts/addressbook/" not in hrefs
assert len(res["resources"]) == 2
def test_httperror_devuelve_status_error(monkeypatch):
def fake_urlopen(req, timeout=None, context=None):
raise mod.urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, None)
monkeypatch.setattr(mod.urllib.request, "urlopen", fake_urlopen)
res = _call()
assert res["status"] == "error"
assert res["http_status"] == 401
@@ -0,0 +1,63 @@
---
name: extract_or_make_uid
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: pure
signature: "def extract_or_make_uid(component_text: str, prefix: str = 'goog-') -> str"
description: "Extrae el campo UID: de un componente iCal (VEVENT/VCALENDAR) o vCard (VCARD). Si el componente no declara UID, sintetiza uno determinista derivado del contenido: '<prefix><md5(text)[:16]>'. El mismo texto produce siempre el mismo UID. Pura, solo stdlib (re, hashlib). Imprescindible al importar a CardDAV/CalDAV: el nombre del recurso se deriva del UID y cada componente necesita uno estable para idempotencia (re-importar no duplica)."
tags: [dav, carddav, caldav, uid, vcard, vevent, ical, infra, extract]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re, hashlib]
params:
- name: component_text
desc: "texto de un componente VCARD o VEVENT/VCALENDAR del que extraer (o derivar) el UID."
- name: prefix
desc: "prefijo del UID sintetico cuando el componente no declara UID. Default 'goog-' (origen Google export)."
output: "str. El valor del campo UID stripeado si el componente lo declara. Si no, un UID determinista '<prefix><md5(component_text)[:16]>'. No lanza para input no vacio."
tested: true
tests:
- "test_uid_presente_se_extrae"
- "test_sin_uid_genera_determinista"
- "test_mismo_texto_mismo_uid_sintetico"
- "test_prefix_personalizado"
- "test_uid_con_espacios_se_stripea"
test_file_path: "python/functions/infra/extract_or_make_uid_test.py"
file_path: "python/functions/infra/extract_or_make_uid.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.extract_or_make_uid import extract_or_make_uid
with_uid = "BEGIN:VEVENT\r\nUID:abc-123@google.com\r\nSUMMARY:x\r\nEND:VEVENT"
print(extract_or_make_uid(with_uid)) # abc-123@google.com
no_uid = "BEGIN:VCARD\r\nFN:Ada\r\nEND:VCARD"
print(extract_or_make_uid(no_uid)) # goog-<md5 16 hex> (estable)
```
## Cuando usarla
Cuando importas un VCARD o un VEVENT a un servidor DAV y necesitas el UID para
(a) construir el nombre del recurso (`safe(uid) + .vcf/.ics`) y (b) garantizar
idempotencia: re-subir el mismo componente sobrescribe en vez de duplicar.
Usada por los pipelines `import_vcf_to_carddav` e `import_ics_to_caldav` por
cada elemento antes del PUT. Si el export no trae UID (comun en algunas tarjetas
de Google), esta funcion lo fabrica de forma estable a partir del contenido.
## Gotchas
Funcion pura. El UID sintetico depende del contenido EXACTO del componente: si
el texto cambia (aunque sea un espacio), el md5 cambia y se generaria un recurso
nuevo en vez de actualizar el existente. Para componentes con UID propio esto no
ocurre. md5 se usa solo como hash de identidad determinista, no como primitiva
de seguridad.
@@ -0,0 +1,32 @@
"""Extrae el UID de un componente iCal/vCard, o genera uno determinista.
Funcion pura: sin I/O, sin estado, determinista. Busca la propiedad `UID:` en
el texto del componente (VCARD o VEVENT/VCALENDAR). Si no existe, sintetiza un
UID estable derivado del contenido: `<prefix><md5(text)[:16]>`. El mismo texto
de entrada produce siempre el mismo UID sintetico. Solo usa stdlib (re, hashlib).
"""
import hashlib
import re
_UID_RE = re.compile(r"(?:^|\n)UID:(.+)")
def extract_or_make_uid(component_text: str, prefix: str = "goog-") -> str:
"""Devuelve el UID del componente, o uno determinista si no lo tiene.
Args:
component_text: texto de un componente VCARD o VEVENT/VCALENDAR.
prefix: prefijo del UID sintetico cuando el componente no declara UID.
Default 'goog-' (origen Google export).
Returns:
El valor del campo UID stripeado si existe. Si no, un UID determinista
'<prefix><md5(component_text)[:16]>'. Nunca lanza ni devuelve vacio para
un input no vacio.
"""
m = _UID_RE.search(component_text or "")
if m:
return m.group(1).strip()
digest = hashlib.md5((component_text or "").encode("utf-8")).hexdigest()[:16]
return "%s%s" % (prefix, digest)
@@ -0,0 +1,36 @@
"""Tests para extract_or_make_uid. Puros, deterministas, sin I/O."""
import sys
import infra.extract_or_make_uid # noqa: F401
mod = sys.modules["infra.extract_or_make_uid"]
extract = mod.extract_or_make_uid
def test_uid_presente_se_extrae():
txt = "BEGIN:VEVENT\r\nUID:abc-123@google.com\r\nSUMMARY:x\r\nEND:VEVENT"
assert extract(txt) == "abc-123@google.com"
def test_sin_uid_genera_determinista():
txt = "BEGIN:VCARD\r\nFN:Ada\r\nEND:VCARD"
uid = extract(txt)
assert uid.startswith("goog-")
assert len(uid) == len("goog-") + 16
def test_mismo_texto_mismo_uid_sintetico():
txt = "BEGIN:VCARD\r\nFN:Ada\r\nEND:VCARD"
assert extract(txt) == extract(txt)
def test_prefix_personalizado():
txt = "BEGIN:VCARD\r\nFN:Ada\r\nEND:VCARD"
uid = extract(txt, prefix="mig-")
assert uid.startswith("mig-")
def test_uid_con_espacios_se_stripea():
txt = "BEGIN:VCARD\nUID: spaced-uid \nEND:VCARD"
assert extract(txt) == "spaced-uid"
+61
View File
@@ -0,0 +1,61 @@
---
name: split_vcards
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: pure
signature: "def split_vcards(vcf_text: str) -> list"
description: "Divide el texto completo de un archivo .vcf en sus VCARDs individuales. Devuelve una lista de strings, cada uno un VCARD completo (BEGIN:VCARD..END:VCARD) stripeado. Pura, solo stdlib (re). Util para importar a CardDAV un .vcf exportado de Google Contacts que concatena N tarjetas en un solo archivo: cada string resultante se sube como recurso .vcf independiente."
tags: [dav, carddav, vcard, vcf, contacts, infra, split]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re]
params:
- name: vcf_text
desc: "contenido completo del archivo .vcf, con una o varias tarjetas VCARD concatenadas. Tolera saltos de linea LF o CRLF."
output: "list[str]. Cada elemento es un VCARD completo ('BEGIN:VCARD'..'END:VCARD') stripeado de espacios al inicio/fin. Lista vacia si el input es vacio o no contiene ninguna tarjeta."
tested: true
tests:
- "test_dos_vcards_devuelve_dos"
- "test_vcard_unico"
- "test_input_vacio_devuelve_lista_vacia"
- "test_crlf_se_tolera"
- "test_cada_card_es_begin_end"
test_file_path: "python/functions/infra/split_vcards_test.py"
file_path: "python/functions/infra/split_vcards.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.split_vcards import split_vcards
vcf = (
"BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Ada Lovelace\r\nEND:VCARD\r\n"
"BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Alan Turing\r\nEND:VCARD\r\n"
)
cards = split_vcards(vcf)
print(len(cards)) # 2
print(cards[0][:13]) # BEGIN:VCARD
```
## Cuando usarla
Cuando exportas contactos de Google (o cualquier fuente) a un unico `.vcf` con
muchas tarjetas concatenadas y necesitas subir cada una como recurso CardDAV
separado. Es el primer paso del pipeline `import_vcf_to_carddav`: split → extraer
UID por tarjeta → `carddav_put_vcard`. Tambien para contar/validar un `.vcf`
sin parsear cada campo.
## Gotchas
Funcion pura sin gotchas relevantes. No valida el contenido interno del VCARD
(no comprueba VERSION ni campos obligatorios); solo segmenta por
BEGIN:VCARD..END:VCARD. Si una tarjeta esta truncada (sin END:VCARD) no aparece
en la salida.
+28
View File
@@ -0,0 +1,28 @@
"""Divide un archivo .vcf en sus VCARDs individuales.
Funcion pura: sin I/O, sin estado, determinista. Recibe el texto completo de un
archivo .vcf (que puede contener N tarjetas concatenadas) y devuelve una lista
de strings, cada uno un VCARD completo (BEGIN:VCARD..END:VCARD) ya stripeado.
Solo usa stdlib (re).
"""
import re
_VCARD_RE = re.compile(r"BEGIN:VCARD.*?END:VCARD", re.DOTALL)
def split_vcards(vcf_text: str) -> list:
"""Divide el texto de un .vcf en VCARDs individuales.
Args:
vcf_text: contenido completo del archivo .vcf, con una o varias tarjetas
concatenadas. Tolera saltos de linea LF o CRLF.
Returns:
Lista de strings. Cada elemento es un VCARD completo
('BEGIN:VCARD'..'END:VCARD') stripeado de espacios al inicio/fin.
Lista vacia si no hay ninguna tarjeta.
"""
if not vcf_text:
return []
return [m.strip() for m in _VCARD_RE.findall(vcf_text)]
@@ -0,0 +1,40 @@
"""Tests para split_vcards. Puros, deterministas, sin I/O."""
import sys
import infra.split_vcards # noqa: F401
mod = sys.modules["infra.split_vcards"]
split_vcards = mod.split_vcards
_TWO = (
"BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Ada Lovelace\r\nEND:VCARD\r\n"
"BEGIN:VCARD\r\nVERSION:3.0\r\nFN:Alan Turing\r\nEND:VCARD\r\n"
)
def test_dos_vcards_devuelve_dos():
cards = split_vcards(_TWO)
assert len(cards) == 2
def test_vcard_unico():
cards = split_vcards("BEGIN:VCARD\nFN:Solo\nEND:VCARD\n")
assert len(cards) == 1
assert "Solo" in cards[0]
def test_input_vacio_devuelve_lista_vacia():
assert split_vcards("") == []
assert split_vcards("ruido sin tarjetas") == []
def test_crlf_se_tolera():
lf = _TWO.replace("\r\n", "\n")
assert len(split_vcards(lf)) == 2
def test_cada_card_es_begin_end():
for c in split_vcards(_TWO):
assert c.startswith("BEGIN:VCARD")
assert c.endswith("END:VCARD")
@@ -0,0 +1,68 @@
---
name: split_vevents_to_vcalendars
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: pure
signature: "def split_vevents_to_vcalendars(ics_text: str, prodid: str = '-//xandikos-migracion//google-export//EN') -> list"
description: "Divide un .ics (un VCALENDAR con N VEVENT) en N VCALENDARs independientes, cada uno con un unico VEVENT, header VERSION/PRODID/CALSCALE y las VTIMEZONE del original. Pura, solo stdlib (re). Util para importar a CalDAV un .ics exportado de Google Calendar que mete todos los eventos en un solo VCALENDAR: cada salida se sube como recurso .ics independiente. Normaliza saltos de linea a CRLF (RFC 5545)."
tags: [dav, caldav, ical, ics, vevent, vcalendar, calendar, infra, split]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [re]
params:
- name: ics_text
desc: "contenido completo del .ics: un VCALENDAR con uno o varios VEVENT. Tolera LF o CRLF."
- name: prodid
desc: "valor del campo PRODID del header de cada VCALENDAR de salida. Default identifica la migracion a Xandikos."
output: "list[str]. Cada elemento es un VCALENDAR completo y autonomo ('BEGIN:VCALENDAR'..'END:VCALENDAR' terminado en CRLF) con header VERSION:2.0 / PRODID / CALSCALE:GREGORIAN, las VTIMEZONE del original (si las habia, replicadas en cada salida) y un unico VEVENT. Lista vacia si no hay ningun VEVENT."
tested: true
tests:
- "test_dos_vevents_devuelve_dos_vcalendars"
- "test_cada_salida_tiene_un_solo_vevent"
- "test_header_vcalendar_correcto"
- "test_vtimezone_se_replica_en_cada_salida"
- "test_salida_termina_en_crlf"
- "test_input_vacio_devuelve_lista_vacia"
test_file_path: "python/functions/infra/split_vevents_to_vcalendars_test.py"
file_path: "python/functions/infra/split_vevents_to_vcalendars.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.split_vevents_to_vcalendars import split_vevents_to_vcalendars
ics = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//Google//EN\r\n"
"BEGIN:VEVENT\r\nUID:a@x\r\nSUMMARY:Reunion\r\nEND:VEVENT\r\n"
"BEGIN:VEVENT\r\nUID:b@x\r\nSUMMARY:Comida\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
cals = split_vevents_to_vcalendars(ics)
print(len(cals)) # 2
print(cals[0].count("BEGIN:VEVENT")) # 1 (un evento por VCALENDAR)
```
## Cuando usarla
Cuando exportas tu calendario de Google a un unico `.ics` (un VCALENDAR con
todos los eventos dentro) y necesitas subir cada evento como recurso CalDAV
separado a Xandikos. Es el primer paso del pipeline `import_ics_to_caldav`:
split → extraer UID por evento → `caldav_put_event`. Cada salida es un `.ics`
valido y autonomo que un cliente de calendario puede consumir por si solo.
## Gotchas
Funcion pura. Replica TODAS las VTIMEZONE del VCALENDAR original en cada salida
(conservador: garantiza que cualquier TZID referenciado por el VEVENT este
definido, aunque algun evento no use ninguna). No deduplica ni filtra
timezones por evento. No valida que el VEVENT este completo ni reescribe DTSTART
/DTEND. Si el .ics no contiene VEVENT (p.ej. solo VTODO o VJOURNAL) devuelve
lista vacia.
@@ -0,0 +1,56 @@
"""Divide un .ics (un VCALENDAR con N VEVENT) en N VCALENDARs independientes.
Funcion pura: sin I/O, sin estado, determinista. Cada salida es un VCALENDAR
completo y autonomo con un unico VEVENT, listo para subir como recurso CalDAV
individual. Las VTIMEZONE del original se incluyen en cada salida (un VEVENT
puede referenciar un TZID definido en el VCALENDAR padre). Solo usa stdlib (re).
"""
import re
_VEVENT_RE = re.compile(r"BEGIN:VEVENT.*?END:VEVENT", re.DOTALL)
_VTIMEZONE_RE = re.compile(r"BEGIN:VTIMEZONE.*?END:VTIMEZONE", re.DOTALL)
_DEFAULT_PRODID = "-//xandikos-migracion//google-export//EN"
def _to_crlf(text: str) -> str:
"""Normaliza saltos de linea a CRLF (RFC 5545)."""
return text.strip().replace("\r\n", "\n").replace("\n", "\r\n")
def split_vevents_to_vcalendars(ics_text: str, prodid: str = _DEFAULT_PRODID) -> list:
"""Divide un VCALENDAR con N VEVENT en N VCALENDARs independientes.
Args:
ics_text: contenido completo del .ics (un VCALENDAR con uno o varios
VEVENT). Tolera LF o CRLF.
prodid: valor del campo PRODID a usar en el header de cada VCALENDAR de
salida. Default: identifica la migracion a Xandikos.
Returns:
Lista de strings. Cada elemento es un VCALENDAR completo y autonomo
('BEGIN:VCALENDAR'..'END:VCALENDAR' terminado en CRLF) con header
VERSION/PRODID/CALSCALE, las VTIMEZONE del original (si las habia) y un
unico VEVENT. Lista vacia si no hay ningun VEVENT.
"""
if not ics_text:
return []
events = _VEVENT_RE.findall(ics_text)
timezones = _VTIMEZONE_RE.findall(ics_text)
tz_block = ""
for tz in timezones:
tz_block += _to_crlf(tz) + "\r\n"
header = (
"BEGIN:VCALENDAR\r\n"
"VERSION:2.0\r\n"
"PRODID:%s\r\n"
"CALSCALE:GREGORIAN\r\n" % prodid
)
out = []
for ev in events:
body = header + tz_block + _to_crlf(ev) + "\r\nEND:VCALENDAR\r\n"
out.append(body)
return out
@@ -0,0 +1,61 @@
"""Tests para split_vevents_to_vcalendars. Puros, deterministas, sin I/O."""
import sys
import infra.split_vevents_to_vcalendars # noqa: F401
mod = sys.modules["infra.split_vevents_to_vcalendars"]
split = mod.split_vevents_to_vcalendars
_TWO = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//Google//EN\r\n"
"BEGIN:VEVENT\r\nUID:a@x\r\nSUMMARY:Reunion\r\nEND:VEVENT\r\n"
"BEGIN:VEVENT\r\nUID:b@x\r\nSUMMARY:Comida\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
_WITH_TZ = (
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//Google//EN\r\n"
"BEGIN:VTIMEZONE\r\nTZID:Europe/Madrid\r\nEND:VTIMEZONE\r\n"
"BEGIN:VEVENT\r\nUID:a@x\r\nDTSTART;TZID=Europe/Madrid:20260101T100000\r\nEND:VEVENT\r\n"
"BEGIN:VEVENT\r\nUID:b@x\r\nDTSTART;TZID=Europe/Madrid:20260102T100000\r\nEND:VEVENT\r\n"
"END:VCALENDAR\r\n"
)
def test_dos_vevents_devuelve_dos_vcalendars():
cals = split(_TWO)
assert len(cals) == 2
def test_cada_salida_tiene_un_solo_vevent():
for cal in split(_TWO):
assert cal.count("BEGIN:VEVENT") == 1
assert cal.count("END:VEVENT") == 1
assert cal.startswith("BEGIN:VCALENDAR")
assert cal.rstrip("\r\n").endswith("END:VCALENDAR")
def test_header_vcalendar_correcto():
cal = split(_TWO)[0]
assert "VERSION:2.0" in cal
assert "PRODID:" in cal
assert "CALSCALE:GREGORIAN" in cal
def test_vtimezone_se_replica_en_cada_salida():
cals = split(_WITH_TZ)
assert len(cals) == 2
for cal in cals:
assert "BEGIN:VTIMEZONE" in cal
assert "TZID:Europe/Madrid" in cal
def test_salida_termina_en_crlf():
cal = split(_TWO)[0]
assert cal.endswith("END:VCALENDAR\r\n")
def test_input_vacio_devuelve_lista_vacia():
assert split("") == []
assert split("BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n") == []
+4
View File
@@ -19,6 +19,9 @@ from .slugify_obsidian_name import slugify_obsidian_name
from .extract_obsidian_embeds import extract_obsidian_embeds
from .resolve_obsidian_embed import resolve_obsidian_embed
# Grafo agregado del vault (grupo obsidian)
from .build_obsidian_graph import build_obsidian_graph
__all__ = [
"parse_obsidian_frontmatter",
"extract_obsidian_wikilinks",
@@ -34,4 +37,5 @@ __all__ = [
"slugify_obsidian_name",
"extract_obsidian_embeds",
"resolve_obsidian_embed",
"build_obsidian_graph",
]
@@ -0,0 +1,76 @@
---
name: build_obsidian_graph
kind: function
lang: py
domain: obsidian
version: "1.0.0"
purity: impure
signature: "def build_obsidian_graph(vault_dir: str, include_dangling: bool = True) -> dict"
description: "Construye el grafo agregado (nodos + aristas) de un vault de Obsidian leyendo todas sus notas .md. Cada nota es un nodo tipado (tipo por carpeta o frontmatter, id = slug, label = frontmatter['nombre'] o slug) y cada wikilink [[...]] del cuerpo es una arista dirigida resuelta por slug del ultimo segmento del destino. Los wikilinks rotos se incluyen como nodos fantasma dangling o se descartan segun include_dangling. Compone list_obsidian_notes, read_obsidian_note y slugify_obsidian_name del grupo obsidian. Es la pieza que cierra la frontera 'el grupo obsidian no indexa el grafo agregado'. Base de la vista grafo (sigma.js) de la app osint_web."
tags: [obsidian, graph, vault, nodes, edges, wikilinks, sigma, osint]
uses_functions: ["list_obsidian_notes_py_obsidian", "read_obsidian_note_py_obsidian", "slugify_obsidian_name_py_obsidian"]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: ["os", "re"]
params:
- name: vault_dir
desc: "ruta (absoluta o relativa) a la raiz del vault de Obsidian a indexar; se excluyen .obsidian/ y .trash/"
- name: include_dangling
desc: "si True (por defecto) los wikilinks que no resuelven a ninguna nota generan un nodo fantasma con dangling=True y su arista; si False, esos enlaces rotos se descartan"
output: "dict con 'nodes' (lista de {id: slug, tipo: str, label: str, frontmatter: dict}; los fantasma anaden dangling=True y frontmatter vacio) y 'edges' (lista de {source: slug, target: slug, kind: str} deduplicada; kind es relacion/lugar/documento segun la seccion ## donde aparece el wikilink, o 'wikilink' por defecto)"
tested: true
tests:
- "golden grafo del mini-vault con nodos y aristas esperados"
- "resuelve wikilink con acentos maria del mar al slug"
- "el kind de la arista sale de la seccion del cuerpo"
- "dangling marcado con true y excluido con false"
- "el tipo cae a la carpeta si falta en frontmatter"
- "wikilink sintacticamente roto no tumba el grafo"
- "vault inexistente lanza filenotfounderror"
test_file_path: "python/functions/obsidian/build_obsidian_graph_test.py"
file_path: "python/functions/obsidian/build_obsidian_graph.py"
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from obsidian.build_obsidian_graph import build_obsidian_graph
graph = build_obsidian_graph("/home/enmanuel/Obsidian/osint")
print(len(graph["nodes"]), "nodos,", len(graph["edges"]), "aristas")
# 1199 nodos (984 reales + 215 fantasma), 618 aristas
# Un nodo tipado listo para sigma.js: color por tipo, label visible.
n = next(x for x in graph["nodes"] if x["tipo"] == "persona")
print(n["id"], n["label"], n["tipo"]) # ana-gomez Ana Gómez persona
# Aristas con su kind (relacion / lugar / documento / wikilink).
for e in graph["edges"][:3]:
print(e["source"], "->", e["target"], f"({e['kind']})")
```
Lanzable directo sobre el vault real (imprime conteo de nodos/aristas/dangling):
```bash
cd /home/enmanuel/fn_registry
PYTHONPATH=python/functions python/.venv/bin/python3 \
python/functions/obsidian/build_obsidian_graph.py /home/enmanuel/Obsidian/osint
```
## Cuando usarla
Cuando necesites el grafo entero de un vault de Obsidian de una sola pasada: para pintar una vista de nodos navegable (sigma.js / graphology), para detectar objetivos OSINT aun sin fichar (nodos `dangling`), o para alimentar el endpoint `/api/graph` de una app que lee el vault sin BD intermedia. Es el agregado que las funciones atomicas del grupo `obsidian` (`list_obsidian_notes`, `read_obsidian_note`) no daban por si solas.
## Gotchas
- **Lee todo el vault de disco** (I/O impuro): el grafo refleja el estado de los `.md` en ese instante; vuelve a llamar para refrescar tras editar notas.
- **El `id` de un nodo es el slug = nombre de archivo sin `.md`**. Si dos notas en carpetas distintas comparten ese nombre (caso real en el vault osint: `dni.md`, `fotos.md`, `certificado-digital.md` repetidos dentro de cada subcarpeta `personas/<slug>/`), colapsan al **mismo nodo** y solo la primera en orden alfabetico sobrevive. Por eso el vault con 1022 `.md` produce 984 nodos reales (13 grupos de slugs colisionan). Para evitarlo habria que usar el path relativo como id — se dejo el slug por compatibilidad con la resolucion de wikilinks `[[slug]]`.
- **Resolucion de wikilinks por ultimo segmento**: `[[organizaciones/acme-sl|Acme SL]]` resuelve a la nota cuyo slug es `acme-sl`; el alias (`|...`) y el ancla (`#...`) se ignoran. Nombres con acentos/mayusculas (`[[María del Mar]]`) se slugifican con `slugify_obsidian_name` antes de buscar, asi que resuelven igual que `[[maria-del-mar]]`.
- **`kind` por seccion es heuristico**: depende del texto del encabezado `## ...` mas cercano por encima del wikilink (`Relaciones`/`Relacionado` -> `relacion`, `Lugares` -> `lugar`, `Documentos` -> `documento`, resto -> `wikilink`). Un wikilink fuera de cualquier seccion conocida es `wikilink`.
- **Auto-enlaces ignorados**: si una nota se enlaza a si misma, esa arista no se emite.
- **Nodos fantasma** (`dangling: true`) llevan `tipo: "desconocido"` y `frontmatter` vacio; no representan un `.md` en disco. Con `include_dangling=False` no aparecen ni ellos ni sus aristas.
- Lanza `FileNotFoundError` si `vault_dir` no existe y `NotADirectoryError` si no es un directorio (heredado de `list_obsidian_notes`). Una nota individual ilegible se omite sin tumbar el grafo.
@@ -0,0 +1,245 @@
"""Construye el grafo agregado de un vault de Obsidian: nodos + aristas.
Funcion impura (lee disco) del grupo de capacidad ``obsidian``. Es la pieza
que cierra la frontera "el grupo obsidian no indexa el grafo agregado":
recorre todas las notas del vault, las convierte en nodos tipados y resuelve
los wikilinks ``[[...]]`` del cuerpo a aristas entre nodos existentes.
Registry-first: compone funciones puras/impuras ya existentes del grupo
``obsidian`` (``list_obsidian_notes``, ``read_obsidian_note``,
``slugify_obsidian_name``) en vez de reimplementar el parseo del vault.
"""
import os
import re
from obsidian.list_obsidian_notes import list_obsidian_notes
from obsidian.read_obsidian_note import read_obsidian_note
from obsidian.slugify_obsidian_name import slugify_obsidian_name
# Carpetas de primer nivel del vault -> tipo de nodo por defecto. El tipo del
# frontmatter (campo ``tipo``) siempre tiene prioridad sobre la carpeta.
_FOLDER_TIPO = {
"personas": "persona",
"organizaciones": "organizacion",
"lugares": "lugar",
"dominios": "dominio",
"casos": "caso",
}
# Encabezados de seccion (## ...) -> kind de las aristas que aparecen debajo.
# Se normaliza el texto del encabezado a un slug para comparar de forma estable
# (acentos, mayusculas y plurales/sinonimos cercanos colapsan al mismo bucket).
_SECTION_KIND = {
"relaciones": "relacion",
"relacionado": "relacion",
"relacion": "relacion",
"lugares": "lugar",
"lugar": "lugar",
"documentos": "documento",
"documento": "documento",
}
# Captura un wikilink [[...]] o embed ![[...]]; el grupo es el interior.
_WIKILINK_RE = re.compile(r"!?\[\[([^\[\]]+?)\]\]")
# Captura un encabezado Markdown ATX (## Titulo). Grupo 1 = texto del titulo.
_HEADING_RE = re.compile(r"^#{1,6}\s+(.*?)\s*#*\s*$")
def _tipo_for_note(path: str, vault_dir: str, frontmatter: dict) -> str:
"""Determina el tipo de un nodo: frontmatter['tipo'] o carpeta de 1er nivel.
El campo ``tipo`` del frontmatter manda. Si falta, se usa la primera
carpeta del path relativo al vault mapeada por ``_FOLDER_TIPO``. Si no
encaja en ninguna, el tipo es la propia carpeta de primer nivel (o
``"nota"`` para notas en la raiz del vault).
"""
tipo = frontmatter.get("tipo")
if isinstance(tipo, str) and tipo.strip():
return tipo.strip()
rel = os.path.relpath(path, vault_dir)
parts = rel.split(os.sep)
if len(parts) >= 2:
top = parts[0]
return _FOLDER_TIPO.get(top, top)
return "nota"
def _wikilink_target_slug(raw_target: str) -> str:
"""Reduce el destino de un wikilink a un slug estable para indexar.
El destino que entrega ``extract_obsidian_wikilinks`` ya viene sin alias
(``|...``) ni ancla (``#...``). Aqui se toma el ultimo segmento del path
(Obsidian resuelve ``[[carpeta/nota]]`` por la nota, no por la carpeta) y
se slugifica con ``slugify_obsidian_name`` para que ``[[Maria del Mar]]``
y ``[[maria-del-mar]]`` apunten al mismo nodo.
"""
# Ultimo segmento del path (soporta tanto '/' como '\\' por robustez).
last = re.split(r"[\\/]", raw_target)[-1]
return slugify_obsidian_name(last)
def _iter_body_links_with_kind(body: str):
"""Itera (slug_destino, kind) por cada wikilink del cuerpo, con su seccion.
Recorre el body linea a linea llevando la cuenta de la ultima seccion
``## ...`` vista para asignar el ``kind`` de cada wikilink segun
``_SECTION_KIND`` (relaciones/lugares/documentos). Fuera de esas secciones
el kind por defecto es ``"wikilink"``. No deduplica: cada aparicion produce
un par (la deduplicacion se hace al construir las aristas).
"""
current_kind = "wikilink"
for line in body.splitlines():
heading = _HEADING_RE.match(line)
if heading:
heading_slug = slugify_obsidian_name(heading.group(1))
current_kind = _SECTION_KIND.get(heading_slug, "wikilink")
continue
for match in _WIKILINK_RE.finditer(line):
inner = match.group(1)
target = inner.split("|", 1)[0].split("#", 1)[0].strip()
if not target:
continue
slug = _wikilink_target_slug(target)
if not slug:
continue
yield slug, current_kind
def build_obsidian_graph(vault_dir: str, include_dangling: bool = True) -> dict:
"""Construye el grafo agregado (nodos + aristas) de un vault de Obsidian.
Recorre cada nota ``.md`` del vault (excluyendo ``.obsidian/`` y
``.trash/`` via ``list_obsidian_notes``; ``attachments/`` no contiene
``.md`` y queda fuera de forma natural). Cada nota es un nodo cuyo ``id``
es su slug (nombre de archivo sin ``.md``) y cuyo ``tipo`` sale del campo
``tipo`` del frontmatter o, en su defecto, de la carpeta de primer nivel.
Cada wikilink ``[[...]]`` del cuerpo es una arista dirigida del nodo origen
al nodo destino, resuelto por slug del ultimo segmento del destino.
Args:
vault_dir: Ruta (absoluta o relativa) a la raiz del vault de Obsidian.
include_dangling: Si es ``True`` (por defecto), los wikilinks que no
resuelven a ninguna nota del vault generan un nodo fantasma con
``dangling: true`` y su arista correspondiente. Si es ``False``,
esos wikilinks rotos se descartan (ni nodo fantasma ni arista).
Returns:
dict con dos claves:
- ``nodes``: lista de ``{"id": slug, "tipo": str, "label": str,
"frontmatter": dict}``. Los nodos fantasma anaden
``"dangling": True`` y llevan ``frontmatter`` vacio.
- ``edges``: lista de ``{"source": slug, "target": slug,
"kind": str}`` deduplicada por (source, target, kind). El
``kind`` se deduce de la seccion (``relacion``/``lugar``/
``documento``) o es ``"wikilink"`` por defecto.
Raises:
FileNotFoundError: si ``vault_dir`` no existe.
NotADirectoryError: si ``vault_dir`` no es un directorio.
"""
note_paths = list_obsidian_notes(vault_dir)
# Indice slug -> nodo real. Si dos notas colapsan al mismo slug (raro), la
# primera en orden alfabetico gana (list_obsidian_notes devuelve ordenado).
nodes_by_slug: dict[str, dict] = {}
# Conserva el orden de descubrimiento de los nodos reales para la salida.
real_order: list[str] = []
for path in note_paths:
slug = os.path.splitext(os.path.basename(path))[0]
if not slug or slug in nodes_by_slug:
continue
try:
note = read_obsidian_note(path)
except OSError:
# Una nota ilegible no debe tumbar el grafo entero: se omite.
continue
frontmatter = note.get("frontmatter", {}) or {}
tipo = _tipo_for_note(path, vault_dir, frontmatter)
nombre = frontmatter.get("nombre")
label = str(nombre).strip() if isinstance(nombre, str) and nombre.strip() else slug
nodes_by_slug[slug] = {
"id": slug,
"tipo": tipo,
"label": label,
"frontmatter": frontmatter,
"_path": path, # interno: para resolver enlaces del body
}
real_order.append(slug)
edges: list[dict] = []
seen_edges: set[tuple] = set()
dangling_slugs: list[str] = []
dangling_seen: set[str] = set()
for slug in real_order:
node = nodes_by_slug[slug]
try:
body = read_obsidian_note(node["_path"]).get("body", "") or ""
except OSError:
continue
for target_slug, kind in _iter_body_links_with_kind(body):
if target_slug == slug:
# Auto-enlace: se ignora (no aporta al grafo).
continue
resolved = target_slug in nodes_by_slug
if not resolved:
if not include_dangling:
continue
if target_slug not in dangling_seen:
dangling_seen.add(target_slug)
dangling_slugs.append(target_slug)
edge_key = (slug, target_slug, kind)
if edge_key in seen_edges:
continue
seen_edges.add(edge_key)
edges.append({"source": slug, "target": target_slug, "kind": kind})
# Serializa nodos reales (sin la clave interna _path) + nodos fantasma.
nodes: list[dict] = []
for slug in real_order:
node = nodes_by_slug[slug]
nodes.append(
{
"id": node["id"],
"tipo": node["tipo"],
"label": node["label"],
"frontmatter": node["frontmatter"],
}
)
if include_dangling:
for slug in dangling_slugs:
nodes.append(
{
"id": slug,
"tipo": "desconocido",
"label": slug,
"frontmatter": {},
"dangling": True,
}
)
return {"nodes": nodes, "edges": edges}
if __name__ == "__main__":
import json
import sys
target_vault = sys.argv[1] if len(sys.argv) > 1 else "/home/enmanuel/Obsidian/osint"
graph = build_obsidian_graph(target_vault)
print(
json.dumps(
{
"vault": target_vault,
"nodes": len(graph["nodes"]),
"edges": len(graph["edges"]),
"dangling": sum(1 for n in graph["nodes"] if n.get("dangling")),
},
ensure_ascii=False,
indent=2,
)
)
@@ -0,0 +1,216 @@
"""Tests para build_obsidian_graph.
Construyen mini-vaults temporales con notas en personas/ y organizaciones/,
wikilinks entre ellas (incluido uno roto y uno con acentos) y comprueban el
grafo resultante: numero de nodos/aristas, resolucion de wikilinks acentuados,
marcado de dangling y robustez ante enlaces rotos.
"""
import os
import tempfile
from obsidian.build_obsidian_graph import build_obsidian_graph
def _write(vault: str, rel_path: str, content: str) -> None:
"""Escribe una nota .md en el mini-vault, creando carpetas si hace falta."""
full = os.path.join(vault, rel_path)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "w", encoding="utf-8") as f:
f.write(content)
def _build_sample_vault(vault: str) -> None:
"""Crea un vault de prueba con 4 notas reales + 1 wikilink dangling.
Grafo esperado (aristas dirigidas):
ana -> bruno (## Relaciones, kind relacion)
ana -> maria-del-mar (## Relaciones, kind relacion, acento)
ana -> acme-sl (## Documentos, kind documento)
bruno -> persona-fantasma (## Relaciones, NO existe -> dangling)
acme-sl-> ana (cuerpo suelto, kind wikilink)
"""
# .obsidian/ debe ignorarse por completo.
_write(vault, ".obsidian/app.json", "{}")
_write(
vault,
"personas/ana.md",
"---\n"
"tipo: persona\n"
"nombre: Ana Gómez\n"
"---\n\n"
"## Relaciones\n"
"- [[bruno]] — amigo\n"
"- [[María del Mar]] — vecina\n\n"
"## Documentos\n"
"- [[organizaciones/acme-sl|Acme SL]]\n",
)
_write(
vault,
"personas/bruno.md",
"---\n"
"tipo: persona\n"
"nombre: Bruno Ruiz\n"
"---\n\n"
"## Relaciones\n"
"- [[Persona Fantasma]] — desconocido\n",
)
# Nota con acento en el nombre de archivo cuyo slug es maria-del-mar.
_write(
vault,
"personas/maria-del-mar.md",
"---\n"
"tipo: persona\n"
"nombre: María del Mar\n"
"---\n\n"
"## Notas\n"
"Sin enlaces.\n",
)
_write(
vault,
"organizaciones/acme-sl.md",
"---\n"
"tipo: organizacion\n"
"nombre: Acme SL\n"
"---\n\n"
"Cliente de [[ana]].\n",
)
def test_golden_graph_node_and_edge_counts():
"""Golden: el grafo del mini-vault tiene los nodos y aristas esperados."""
with tempfile.TemporaryDirectory() as vault:
_build_sample_vault(vault)
graph = build_obsidian_graph(vault, include_dangling=True)
real = [n for n in graph["nodes"] if not n.get("dangling")]
dangling = [n for n in graph["nodes"] if n.get("dangling")]
# 4 notas reales + 1 nodo fantasma (persona-fantasma).
assert len(real) == 4, [n["id"] for n in real]
assert len(dangling) == 1, [n["id"] for n in dangling]
# 5 aristas: ana->bruno, ana->maria-del-mar, ana->acme-sl,
# bruno->persona-fantasma, acme-sl->ana.
assert len(graph["edges"]) == 5, graph["edges"]
ids = {n["id"] for n in real}
assert ids == {"ana", "bruno", "maria-del-mar", "acme-sl"}, ids
# El tipo y el label salen del frontmatter.
ana = next(n for n in real if n["id"] == "ana")
assert ana["tipo"] == "persona"
assert ana["label"] == "Ana Gómez"
assert ana["frontmatter"]["nombre"] == "Ana Gómez"
def test_edge_resolves_wikilink_with_accents():
"""Edge: [[María del Mar]] resuelve al nodo slug maria-del-mar."""
with tempfile.TemporaryDirectory() as vault:
_build_sample_vault(vault)
graph = build_obsidian_graph(vault, include_dangling=True)
edge = next(
(e for e in graph["edges"] if e["source"] == "ana" and e["target"] == "maria-del-mar"),
None,
)
assert edge is not None, graph["edges"]
assert edge["kind"] == "relacion", edge
def test_edge_kind_from_section():
"""Edge: el kind de la arista se deduce de la seccion ## donde aparece."""
with tempfile.TemporaryDirectory() as vault:
_build_sample_vault(vault)
graph = build_obsidian_graph(vault, include_dangling=True)
kinds = {(e["source"], e["target"]): e["kind"] for e in graph["edges"]}
assert kinds[("ana", "bruno")] == "relacion", kinds
assert kinds[("ana", "acme-sl")] == "documento", kinds
# acme-sl -> ana esta fuera de cualquier seccion -> wikilink por defecto.
assert kinds[("acme-sl", "ana")] == "wikilink", kinds
def test_edge_dangling_marked_and_excluded():
"""Edge: dangling=True crea nodo fantasma; dangling=False lo descarta."""
with tempfile.TemporaryDirectory() as vault:
_build_sample_vault(vault)
with_dangling = build_obsidian_graph(vault, include_dangling=True)
ghost = next(
(n for n in with_dangling["nodes"] if n["id"] == "persona-fantasma"),
None,
)
assert ghost is not None and ghost.get("dangling") is True, with_dangling["nodes"]
assert ghost["tipo"] == "desconocido", ghost
# La arista hacia el fantasma existe.
assert any(
e["target"] == "persona-fantasma" for e in with_dangling["edges"]
), with_dangling["edges"]
without_dangling = build_obsidian_graph(vault, include_dangling=False)
assert all(
n["id"] != "persona-fantasma" for n in without_dangling["nodes"]
), without_dangling["nodes"]
# La arista rota se descarta junto con el nodo fantasma.
assert all(
e["target"] != "persona-fantasma" for e in without_dangling["edges"]
), without_dangling["edges"]
# Pasamos de 5 a 4 aristas (se elimina bruno->persona-fantasma).
assert len(without_dangling["edges"]) == 4, without_dangling["edges"]
def test_edge_tipo_falls_back_to_folder():
"""Edge: sin campo 'tipo' en frontmatter, el tipo sale de la carpeta."""
with tempfile.TemporaryDirectory() as vault:
_write(vault, "personas/sin-tipo.md", "---\nnombre: Sin Tipo\n---\n\nCuerpo.\n")
_write(vault, "organizaciones/org-sin-tipo.md", "Sin frontmatter.\n")
graph = build_obsidian_graph(vault)
by_id = {n["id"]: n for n in graph["nodes"]}
assert by_id["sin-tipo"]["tipo"] == "persona", by_id["sin-tipo"]
assert by_id["org-sin-tipo"]["tipo"] == "organizacion", by_id["org-sin-tipo"]
# Sin frontmatter ni nombre, el label cae al slug.
assert by_id["org-sin-tipo"]["label"] == "org-sin-tipo", by_id["org-sin-tipo"]
def test_error_path_broken_wikilink_no_crash():
"""Error path: un wikilink sintacticamente roto no tumba el grafo."""
with tempfile.TemporaryDirectory() as vault:
# Wikilink sin cerrar, doble corchete suelto y wikilink vacio.
_write(
vault,
"personas/raro.md",
"---\ntipo: persona\nnombre: Raro\n---\n\n"
"Texto con [[ roto y [[ ]] vacio y [[bien]] valido.\n",
)
_write(vault, "personas/bien.md", "---\ntipo: persona\nnombre: Bien\n---\n\nOK\n")
graph = build_obsidian_graph(vault, include_dangling=True)
# No crash; el unico enlace valido se resuelve a 'bien'.
edges = [(e["source"], e["target"]) for e in graph["edges"]]
assert ("raro", "bien") in edges, edges
# El wikilink vacio no genera arista ni nodo fantasma vacio.
assert all(n["id"] for n in graph["nodes"]), graph["nodes"]
def test_error_path_missing_vault_raises():
"""Error path: un vault inexistente lanza FileNotFoundError, no 500 mudo."""
raised = False
try:
build_obsidian_graph("/no/existe/vault/osint")
except FileNotFoundError:
raised = True
assert raised, "build_obsidian_graph deberia lanzar FileNotFoundError"
if __name__ == "__main__":
test_golden_graph_node_and_edge_counts()
test_edge_resolves_wikilink_with_accents()
test_edge_kind_from_section()
test_edge_dangling_marked_and_excluded()
test_edge_tipo_falls_back_to_folder()
test_error_path_broken_wikilink_no_crash()
test_error_path_missing_vault_raises()
print("build_obsidian_graph tests OK")
@@ -0,0 +1,87 @@
---
name: import_ics_to_caldav
kind: pipeline
lang: py
domain: pipelines
version: "1.0.0"
purity: impure
signature: "def import_ics_to_caldav(ics_path: str, base_url: str, username: str, password: str, collection_path: str, *, timeout_s: float = 20.0, verify_tls: bool = True, uid_prefix: str = 'goog-') -> dict"
description: "Pipeline que importa un archivo .ics completo a una coleccion CalDAV. Lee el .ics del disco, parte el VCALENDAR en N VCALENDARs independientes con un VEVENT cada uno (split_vevents_to_vcalendars, replicando las VTIMEZONE), por cada uno extrae o sintetiza el UID (extract_or_make_uid) y lo sube por HTTP PUT (caldav_put_event). Devuelve {ok, fail, total, errors}. Idempotente por UID: re-importar el mismo .ics sobrescribe en vez de duplicar. Formaliza el heredoc ad-hoc usado para migrar 98 eventos de Google Calendar a Xandikos. Solo stdlib."
tags: [dav, caldav, ical, ics, vevent, import, calendar, migration, pipelines]
uses_functions: [split_vevents_to_vcalendars_py_infra, extract_or_make_uid_py_infra, caldav_put_event_py_infra]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [os, sys, json]
params:
- name: ics_path
desc: "ruta en disco del archivo .ics a importar (export de Google Calendar: un VCALENDAR con N VEVENT)."
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: collection_path
desc: "ruta de la coleccion CalDAV destino (p.ej. '/enmanuel/calendars/calendar/')."
- name: timeout_s
desc: "timeout por PUT individual en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
- name: uid_prefix
desc: "prefijo del UID sintetico para eventos sin UID. Default 'goog-'."
output: "dict: {ok:int, fail:int, total:int, errors:[{uid, error}, ...]}. ok=eventos subidos con exito, fail=eventos que fallaron, total=VEVENT en el .ics, errors=detalle de los fallos (uid + mensaje, sin datos sensibles)."
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/pipelines/import_ics_to_caldav.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from pipelines.import_ics_to_caldav import import_ics_to_caldav
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
summary = import_ics_to_caldav(
ics_path="/home/enmanuel/Descargas/calendar.ics",
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
collection_path="/enmanuel/calendars/calendar/",
)
print(summary["ok"], summary["fail"], summary["total"]) # 98 0 98
```
Desde la CLI del registry (resuelve la pass tu mismo y pasala como arg):
```bash
PW=$(pass show dav/xandikos-enmanuel | head -n1)
./fn run import_ics_to_caldav /home/enmanuel/Descargas/calendar.ics \
https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com \
enmanuel "$PW" /enmanuel/calendars/calendar/
```
## Cuando usarla
Cuando tienes un `.ics` exportado (Google Calendar, otro CalDAV) con todos los
eventos en un unico VCALENDAR y quieres volcarlo entero a Xandikos en una sola
llamada. Reemplaza el flujo ad-hoc hecho a mano en la migracion. Re-ejecutable:
por idempotencia de UID, correrlo dos veces no duplica eventos.
## Gotchas
- Solo importa VEVENT. Si el .ics trae VTODO o VJOURNAL, esos componentes se
ignoran (split_vevents_to_vcalendars solo extrae VEVENT).
- Las VTIMEZONE del original se replican en CADA evento subido (conservador):
garantiza que cualquier TZID referenciado este definido.
- Escritura remota masiva secuencial: una request por evento. Cada PUT respeta
`timeout_s`.
- Password por HTTP Basic sobre TLS; leela de `pass`, no la hardcodees ni la
dejes en el historial del shell sin cuidado.
- `errors` lista solo uid + mensaje, nunca el contenido del evento.
@@ -0,0 +1,83 @@
"""Pipeline: importa un archivo .ics completo a una coleccion CalDAV.
Compone funciones del registry: lee el .ics del disco, parte el VCALENDAR en N
VCALENDARs independientes con un VEVENT cada uno (split_vevents_to_vcalendars),
por cada uno extrae o sintetiza el UID (extract_or_make_uid) y lo sube via HTTP
PUT (caldav_put_event). Devuelve un resumen {ok, fail, total, errors}. Impuro
(I/O de disco + red). Solo stdlib.
"""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from infra.split_vevents_to_vcalendars import split_vevents_to_vcalendars
from infra.extract_or_make_uid import extract_or_make_uid
from infra.caldav_put_event import caldav_put_event
def import_ics_to_caldav(
ics_path: str,
base_url: str,
username: str,
password: str,
collection_path: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
uid_prefix: str = "goog-",
) -> dict:
"""Importa todos los eventos de un .ics a una coleccion CalDAV.
Args:
ics_path: ruta en disco del archivo .ics a importar.
base_url: URL base del servidor DAV.
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
collection_path: ruta de la coleccion CalDAV destino.
timeout_s: timeout por PUT en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS.
uid_prefix: prefijo del UID sintetico cuando un evento no trae UID.
Returns:
dict: {ok:int, fail:int, total:int, errors:[{uid:str, error:str}, ...]}.
ok = eventos subidos con exito; fail = eventos que fallaron; total =
eventos (VEVENT) encontrados en el .ics.
"""
with open(ics_path, "r", encoding="utf-8", errors="replace") as fh:
data = fh.read()
vcalendars = split_vevents_to_vcalendars(data)
ok = 0
fail = 0
errors = []
for cal in vcalendars:
uid = extract_or_make_uid(cal, prefix=uid_prefix)
res = caldav_put_event(
base_url, username, password, collection_path, uid, cal,
timeout_s=timeout_s, verify_tls=verify_tls,
)
if res.get("status") == "ok":
ok += 1
else:
fail += 1
errors.append({"uid": uid, "error": res.get("error", "unknown")})
return {"ok": ok, "fail": fail, "total": len(vcalendars), "errors": errors}
if __name__ == "__main__":
import json
if len(sys.argv) < 6:
print(
"uso: import_ics_to_caldav.py <ics_path> <base_url> <username> "
"<password> <collection_path>",
file=sys.stderr,
)
sys.exit(2)
summary = import_ics_to_caldav(
sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5]
)
print(json.dumps({k: summary[k] for k in ("ok", "fail", "total")}))
@@ -0,0 +1,88 @@
---
name: import_vcf_to_carddav
kind: pipeline
lang: py
domain: pipelines
version: "1.0.0"
purity: impure
signature: "def import_vcf_to_carddav(vcf_path: str, base_url: str, username: str, password: str, collection_path: str, *, timeout_s: float = 20.0, verify_tls: bool = True, uid_prefix: str = 'goog-') -> dict"
description: "Pipeline que importa un archivo .vcf completo a una coleccion CardDAV. Lee el .vcf del disco, lo parte en VCARDs (split_vcards), por cada tarjeta extrae o sintetiza el UID (extract_or_make_uid), inyecta el UID en la tarjeta si faltaba, y la sube por HTTP PUT (carddav_put_vcard). Devuelve {ok, fail, total, errors}. Idempotente por UID: re-importar el mismo .vcf sobrescribe en vez de duplicar. Formaliza el heredoc ad-hoc usado para migrar 820 contactos de Google a Xandikos. Solo stdlib."
tags: [dav, carddav, vcard, import, contacts, migration, pipelines]
uses_functions: [split_vcards_py_infra, extract_or_make_uid_py_infra, carddav_put_vcard_py_infra]
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [os, sys, json]
params:
- name: vcf_path
desc: "ruta en disco del archivo .vcf a importar (export de Google Contacts con N tarjetas concatenadas)."
- name: base_url
desc: "URL base del servidor DAV (p.ej. 'https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com')."
- name: username
desc: "usuario para HTTP Basic auth (p.ej. 'enmanuel')."
- name: password
desc: "contrasena para HTTP Basic auth. Resolver desde pass con pass_get_secret, nunca hardcodear."
- name: collection_path
desc: "ruta de la coleccion CardDAV destino (p.ej. '/enmanuel/contacts/addressbook/')."
- name: timeout_s
desc: "timeout por PUT individual en segundos. Default 20.0."
- name: verify_tls
desc: "si True (default) verifica el certificado TLS. No desactivar salvo entorno de prueba."
- name: uid_prefix
desc: "prefijo del UID sintetico para tarjetas sin UID. Default 'goog-'."
output: "dict: {ok:int, fail:int, total:int, errors:[{uid, error}, ...]}. ok=tarjetas subidas con exito, fail=tarjetas que fallaron, total=tarjetas en el .vcf, errors=detalle de los fallos (uid + mensaje, sin datos sensibles)."
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/pipelines/import_vcf_to_carddav.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from pipelines.import_vcf_to_carddav import import_vcf_to_carddav
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
summary = import_vcf_to_carddav(
vcf_path="/home/enmanuel/Descargas/contacts.vcf",
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
collection_path="/enmanuel/contacts/addressbook/",
)
print(summary["ok"], summary["fail"], summary["total"]) # 820 0 820
```
Desde la CLI del registry (resuelve la pass tu mismo y pasala como arg):
```bash
PW=$(pass show dav/xandikos-enmanuel | head -n1)
./fn run import_vcf_to_carddav /home/enmanuel/Descargas/contacts.vcf \
https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com \
enmanuel "$PW" /enmanuel/contacts/addressbook/
```
## Cuando usarla
Cuando tienes un `.vcf` exportado (Google Contacts, iCloud, otro CardDAV) y
quieres volcarlo entero a Xandikos en una sola llamada en vez de subir tarjeta a
tarjeta con heredocs. Reemplaza el flujo ad-hoc que se hizo a mano para la
migracion. Re-ejecutable: por idempotencia de UID, correrlo dos veces no
duplica contactos.
## Gotchas
- Escritura remota masiva: sube una request por tarjeta secuencialmente. Para
miles de contactos puede tardar; cada PUT respeta `timeout_s`.
- Lee TODO el .vcf en memoria; para archivos de cientos de MB considera trocear.
- La password va por HTTP Basic sobre TLS; leela de `pass`, no la hardcodees ni
la pongas en el historial del shell sin cuidado (usa una variable como en el
ejemplo CLI).
- `errors` lista solo uid + mensaje, nunca el contenido de la tarjeta.
- Si una tarjeta no traia UID, el pipeline inyecta `UID:<goog-md5>` antes del
END:VCARD para que el campo UID: y el nombre del recurso queden consistentes.
@@ -0,0 +1,87 @@
"""Pipeline: importa un archivo .vcf completo a una coleccion CardDAV.
Compone funciones del registry: lee el .vcf del disco, lo parte en VCARDs
individuales (split_vcards), por cada tarjeta extrae o sintetiza el UID
(extract_or_make_uid) y la sube via HTTP PUT (carddav_put_vcard). Devuelve un
resumen {ok, fail, total, errors}. Impuro (I/O de disco + red). Solo stdlib.
"""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from infra.split_vcards import split_vcards
from infra.extract_or_make_uid import extract_or_make_uid
from infra.carddav_put_vcard import carddav_put_vcard
def import_vcf_to_carddav(
vcf_path: str,
base_url: str,
username: str,
password: str,
collection_path: str,
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
uid_prefix: str = "goog-",
) -> dict:
"""Importa todas las tarjetas de un .vcf a una coleccion CardDAV.
Args:
vcf_path: ruta en disco del archivo .vcf a importar.
base_url: URL base del servidor DAV.
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth.
collection_path: ruta de la coleccion CardDAV destino.
timeout_s: timeout por PUT en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS.
uid_prefix: prefijo del UID sintetico cuando una tarjeta no trae UID.
Returns:
dict: {ok:int, fail:int, total:int, errors:[{uid:str, error:str}, ...]}.
ok = tarjetas subidas con exito; fail = tarjetas que fallaron; total =
tarjetas encontradas en el .vcf.
"""
with open(vcf_path, "r", encoding="utf-8", errors="replace") as fh:
data = fh.read()
cards = split_vcards(data)
ok = 0
fail = 0
errors = []
for card in cards:
uid = extract_or_make_uid(card, prefix=uid_prefix)
# Si la tarjeta no declaraba UID, inyectarlo antes del END:VCARD para que
# el campo UID: y el nombre del recurso queden consistentes.
if "UID:" not in card:
card = card.replace("END:VCARD", "UID:%s\r\nEND:VCARD" % uid)
res = carddav_put_vcard(
base_url, username, password, collection_path, uid, card,
timeout_s=timeout_s, verify_tls=verify_tls,
)
if res.get("status") == "ok":
ok += 1
else:
fail += 1
errors.append({"uid": uid, "error": res.get("error", "unknown")})
return {"ok": ok, "fail": fail, "total": len(cards), "errors": errors}
if __name__ == "__main__":
import json
if len(sys.argv) < 6:
print(
"uso: import_vcf_to_carddav.py <vcf_path> <base_url> <username> "
"<password> <collection_path>",
file=sys.stderr,
)
sys.exit(2)
summary = import_vcf_to_carddav(
sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5]
)
# No imprimir errores con datos sensibles; solo conteos + uids.
print(json.dumps({k: summary[k] for k in ("ok", "fail", "total")}))