diff --git a/python/functions/infra/caldav_put_event.md b/python/functions/infra/caldav_put_event.md new file mode 100644 index 00000000..e90a205b --- /dev/null +++ b/python/functions/infra/caldav_put_event.md @@ -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. diff --git a/python/functions/infra/caldav_put_event.py b/python/functions/infra/caldav_put_event.py new file mode 100644 index 00000000..299e0f82 --- /dev/null +++ b/python/functions/infra/caldav_put_event.py @@ -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} diff --git a/python/functions/infra/caldav_put_event_test.py b/python/functions/infra/caldav_put_event_test.py new file mode 100644 index 00000000..f8f1aab2 --- /dev/null +++ b/python/functions/infra/caldav_put_event_test.py @@ -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 diff --git a/python/functions/infra/carddav_put_vcard.md b/python/functions/infra/carddav_put_vcard.md new file mode 100644 index 00000000..3703f082 --- /dev/null +++ b/python/functions/infra/carddav_put_vcard.md @@ -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. diff --git a/python/functions/infra/carddav_put_vcard.py b/python/functions/infra/carddav_put_vcard.py new file mode 100644 index 00000000..170ae925 --- /dev/null +++ b/python/functions/infra/carddav_put_vcard.py @@ -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} diff --git a/python/functions/infra/carddav_put_vcard_test.py b/python/functions/infra/carddav_put_vcard_test.py new file mode 100644 index 00000000..7f43c4a4 --- /dev/null +++ b/python/functions/infra/carddav_put_vcard_test.py @@ -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 diff --git a/python/functions/infra/dav_get_resource.md b/python/functions/infra/dav_get_resource.md new file mode 100644 index 00000000..969b61b0 --- /dev/null +++ b/python/functions/infra/dav_get_resource.md @@ -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. diff --git a/python/functions/infra/dav_get_resource.py b/python/functions/infra/dav_get_resource.py new file mode 100644 index 00000000..501d3d4f --- /dev/null +++ b/python/functions/infra/dav_get_resource.py @@ -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} diff --git a/python/functions/infra/dav_get_resource_test.py b/python/functions/infra/dav_get_resource_test.py new file mode 100644 index 00000000..696950a4 --- /dev/null +++ b/python/functions/infra/dav_get_resource_test.py @@ -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 diff --git a/python/functions/infra/dav_list_calendars.md b/python/functions/infra/dav_list_calendars.md new file mode 100644 index 00000000..10b23ca0 --- /dev/null +++ b/python/functions/infra/dav_list_calendars.md @@ -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 `` con estado 2xx para no confundir un + `` vacio de un propstat 404 con un color real. +- Lectura remota real sobre TLS; password de `pass`, no se logea. diff --git a/python/functions/infra/dav_list_calendars.py b/python/functions/infra/dav_list_calendars.py new file mode 100644 index 00000000..48d139e2 --- /dev/null +++ b/python/functions/infra/dav_list_calendars.py @@ -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 `` 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>(.*?)", + re.DOTALL | re.IGNORECASE, +) +_HREF_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?href>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +_PROPSTAT_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?propstat>(.*?)", + re.DOTALL | re.IGNORECASE, +) +_STATUS_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?status>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +_DISPLAYNAME_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?displayname>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +# Color de Apple: #RRGGBBAA (o sin alfa). +_COLOR_RE = re.compile( + r"<(?:[A-Za-z0-9]+:)?calendar-color[^>]*>\s*(.*?)\s*", + re.DOTALL | re.IGNORECASE, +) +# Marca de calendario CalDAV en el resourcetype. El elemento `` +# puede venir con o sin prefijo de namespace (``, ``). +_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 = ( + '' + '' + "" + "" + "" +) + + +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 `` con estado 2xx de un ``. + + 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 + `` 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} diff --git a/python/functions/infra/dav_list_calendars_test.py b/python/functions/infra/dav_list_calendars_test.py new file mode 100644 index 00000000..a06520d0 --- /dev/null +++ b/python/functions/infra/dav_list_calendars_test.py @@ -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 = ( + '' + '' + # Home plano: collection, sin . + "/enmanuel/calendars/" + "HTTP/1.1 200 OK" + "calendars" + "" + "" + "HTTP/1.1 404 Not Found" + "" + "" + # Calendario sin color: el color va en un propstat 404 (no debe leerse). + "/enmanuel/calendars/calendar/" + "HTTP/1.1 200 OK" + "calendar" + "" + "" + "HTTP/1.1 404 Not Found" + "" + "" + # Calendario con color (Apple #RRGGBBAA) y displayname con acento. + "/enmanuel/calendars/trabajo/" + "HTTP/1.1 200 OK" + "Trabajo & ocio" + "" + "#FF2968FF" + "" + # Coleccion NO-calendario (p.ej. un inbox de scheduling): excluida. + "/enmanuel/calendars/inbox/" + "HTTP/1.1 200 OK" + "inbox" + "" + "" + "" +) + + +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 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 diff --git a/python/functions/infra/dav_list_resources.md b/python/functions/infra/dav_list_resources.md new file mode 100644 index 00000000..645bb236 --- /dev/null +++ b/python/functions/infra/dav_list_resources.md @@ -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. diff --git a/python/functions/infra/dav_list_resources.py b/python/functions/infra/dav_list_resources.py new file mode 100644 index 00000000..a0091ddc --- /dev/null +++ b/python/functions/infra/dav_list_resources.py @@ -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*", re.DOTALL | re.IGNORECASE) +_RESPONSE_RE = re.compile(r"<(?:[A-Za-z0-9]+:)?response>(.*?)", re.DOTALL | re.IGNORECASE) +_ETAG_RE = re.compile(r"<(?:[A-Za-z0-9]+:)?getetag>\s*(.*?)\s*", re.DOTALL | re.IGNORECASE) + +_PROPFIND_BODY = ( + '' + '' + "" + "" +) + + +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 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 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} diff --git a/python/functions/infra/dav_list_resources_test.py b/python/functions/infra/dav_list_resources_test.py new file mode 100644 index 00000000..d1867184 --- /dev/null +++ b/python/functions/infra/dav_list_resources_test.py @@ -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 = ( + '' + '' + "/enmanuel/contacts/addressbook/" + "" + "" + "/enmanuel/contacts/addressbook/ada.vcf" + '"etag-ada"' + "" + "/enmanuel/contacts/addressbook/alan.vcf" + '"etag-alan"' + "" + "" +) + + +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 diff --git a/python/functions/infra/extract_or_make_uid.md b/python/functions/infra/extract_or_make_uid.md new file mode 100644 index 00000000..d8e7764d --- /dev/null +++ b/python/functions/infra/extract_or_make_uid.md @@ -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: ''. 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 ''. 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- (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. diff --git a/python/functions/infra/extract_or_make_uid.py b/python/functions/infra/extract_or_make_uid.py new file mode 100644 index 00000000..e9e0a03e --- /dev/null +++ b/python/functions/infra/extract_or_make_uid.py @@ -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: ``. 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 + ''. 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) diff --git a/python/functions/infra/extract_or_make_uid_test.py b/python/functions/infra/extract_or_make_uid_test.py new file mode 100644 index 00000000..2ab4c9d8 --- /dev/null +++ b/python/functions/infra/extract_or_make_uid_test.py @@ -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" diff --git a/python/functions/infra/split_vcards.md b/python/functions/infra/split_vcards.md new file mode 100644 index 00000000..27f6e24f --- /dev/null +++ b/python/functions/infra/split_vcards.md @@ -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. diff --git a/python/functions/infra/split_vcards.py b/python/functions/infra/split_vcards.py new file mode 100644 index 00000000..d0ff5516 --- /dev/null +++ b/python/functions/infra/split_vcards.py @@ -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)] diff --git a/python/functions/infra/split_vcards_test.py b/python/functions/infra/split_vcards_test.py new file mode 100644 index 00000000..b6f1f767 --- /dev/null +++ b/python/functions/infra/split_vcards_test.py @@ -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") diff --git a/python/functions/infra/split_vevents_to_vcalendars.md b/python/functions/infra/split_vevents_to_vcalendars.md new file mode 100644 index 00000000..f6d1ee27 --- /dev/null +++ b/python/functions/infra/split_vevents_to_vcalendars.md @@ -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. diff --git a/python/functions/infra/split_vevents_to_vcalendars.py b/python/functions/infra/split_vevents_to_vcalendars.py new file mode 100644 index 00000000..6f66b246 --- /dev/null +++ b/python/functions/infra/split_vevents_to_vcalendars.py @@ -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 diff --git a/python/functions/infra/split_vevents_to_vcalendars_test.py b/python/functions/infra/split_vevents_to_vcalendars_test.py new file mode 100644 index 00000000..700232df --- /dev/null +++ b/python/functions/infra/split_vevents_to_vcalendars_test.py @@ -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") == [] diff --git a/python/functions/obsidian/__init__.py b/python/functions/obsidian/__init__.py index 7e80e37b..0b0ab60a 100644 --- a/python/functions/obsidian/__init__.py +++ b/python/functions/obsidian/__init__.py @@ -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", ] diff --git a/python/functions/obsidian/build_obsidian_graph.md b/python/functions/obsidian/build_obsidian_graph.md new file mode 100644 index 00000000..735b8b47 --- /dev/null +++ b/python/functions/obsidian/build_obsidian_graph.md @@ -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//`), 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. diff --git a/python/functions/obsidian/build_obsidian_graph.py b/python/functions/obsidian/build_obsidian_graph.py new file mode 100644 index 00000000..5dbd68e5 --- /dev/null +++ b/python/functions/obsidian/build_obsidian_graph.py @@ -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, + ) + ) diff --git a/python/functions/obsidian/build_obsidian_graph_test.py b/python/functions/obsidian/build_obsidian_graph_test.py new file mode 100644 index 00000000..08d4ee9a --- /dev/null +++ b/python/functions/obsidian/build_obsidian_graph_test.py @@ -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") diff --git a/python/functions/pipelines/import_ics_to_caldav.md b/python/functions/pipelines/import_ics_to_caldav.md new file mode 100644 index 00000000..9a3bd0ef --- /dev/null +++ b/python/functions/pipelines/import_ics_to_caldav.md @@ -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. diff --git a/python/functions/pipelines/import_ics_to_caldav.py b/python/functions/pipelines/import_ics_to_caldav.py new file mode 100644 index 00000000..5af8009b --- /dev/null +++ b/python/functions/pipelines/import_ics_to_caldav.py @@ -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 " + " ", + 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")})) diff --git a/python/functions/pipelines/import_vcf_to_carddav.md b/python/functions/pipelines/import_vcf_to_carddav.md new file mode 100644 index 00000000..48c86bd4 --- /dev/null +++ b/python/functions/pipelines/import_vcf_to_carddav.md @@ -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:` antes del + END:VCARD para que el campo UID: y el nombre del recurso queden consistentes. diff --git a/python/functions/pipelines/import_vcf_to_carddav.py b/python/functions/pipelines/import_vcf_to_carddav.py new file mode 100644 index 00000000..0ac0a2de --- /dev/null +++ b/python/functions/pipelines/import_vcf_to_carddav.py @@ -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 " + " ", + 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")}))