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)
68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
"""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}
|