egutierrez
a76760edba
Funciones reutilizables creadas esta sesion para el sistema self-hosted de contactos/calendario (Xandikos) y la app osint_web: - grupo dav (infra): split_vcards, split_vevents_to_vcalendars, extract_or_make_uid, carddav_put_vcard, caldav_put_event, dav_list_resources, dav_get_resource, dav_list_calendars - pipelines: import_vcf_to_carddav, import_ics_to_caldav - obsidian: build_obsidian_graph (grafo agregado del vault)
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
"""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}
|