dataforge/fn_registry
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat(dav): expand_rrule + dav_make_calendar para recurrencia y multi-calendario

Browse Source 
Dos funciones nuevas del grupo de capacidad `dav`:
- expand_rrule_py_infra (pure): expande una RRULE iCalendar a las fechas de
  cada ocurrencia dentro de un rango [from, to]. Solo stdlib (datetime, re).
  Soporta FREQ DAILY/WEEKLY/MONTHLY/YEARLY, INTERVAL, COUNT, UNTIL, BYDAY. 9 tests.
- dav_make_calendar_py_infra (impure): crea una coleccion de calendario nueva
  via MKCALENDAR + PROPPATCH de nombre/color. Idempotente si ya existe. 11 tests.

Consumidas por la app osint_web (eventos recurrentes + creacion de agendas).
Pagina del grupo dav actualizada con ambas.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-06-12 23:30:01 +02:00
parent a76760edba
commit 1c8a86594f
7 changed files with 919 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/infra/dav_make_calendar.md
+106
View File
@@ -0,0 +1,106 @@
---
name: dav_make_calendar
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def dav_make_calendar(base_url: str, username: str, password: str, calendar_home: str, slug: str, display_name: str = \"\", color: str = \"\", description: str = \"\", *, timeout_s: float = 20.0, verify_tls: bool = True) -> dict"
description: "Crea una nueva coleccion de calendario CalDAV (una agenda nueva) bajo el calendar-home de un principal via MKCALENDAR, fijando el displayname en el cuerpo, y opcionalmente fija color (Apple calendar-color) y descripcion (CalDAV calendar-description) con un PROPPATCH posterior. La coleccion se crea en <calendar_home><slug>/. El slug se sanea a [a-z0-9_-] (minusculas, espacios->guion); si queda vacio devuelve error de validacion. Idempotente: 201 Created es exito; 405/301 (ya existe) devuelve {status:'ok', existed:True}. Escapa display_name/description para XML. Construye Authorization: Basic base64(user:pass) a mano. Maneja errores sin lanzar (salvo validacion de args). Solo stdlib (urllib, base64, re, ssl, xml.sax.saxutils). Probado contra Xandikos."
tags: [dav, caldav, calendar, mkcalendar, proppatch, create, collection, color, http, infra]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: [base64, re, ssl, urllib.error, urllib.request, xml.sax.saxutils]
params:
- name: base_url
desc: "URL base del servidor DAV sin barra final (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: calendar_home
desc: "ruta del calendar-home del principal con barra final (p.ej. '/enmanuel/calendars/'). La nueva coleccion cuelga de el."
- name: slug
desc: "segmento de path de la coleccion en la URL (p.ej. 'trabajo'); se sanea a [a-z0-9_-]. La coleccion se crea en <calendar_home><slug>/. Si queda vacio tras sanear, devuelve error de validacion."
- name: display_name
desc: "nombre visible de la coleccion (DAV:displayname). Si vacio, usa el slug saneado."
- name: color
desc: "color de la coleccion como hex '#rrggbb' (propiedad calendar-color de Apple, http://apple.com/ns/ical/). Opcional; '' lo omite."
- name: description
desc: "descripcion de la coleccion (calendar-description de CalDAV). Opcional; '' lo omite."
- name: timeout_s
desc: "timeout de cada 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, href:str} y, si la coleccion ya existia, ademas existed:True. En error (sin lanzar): {status:'error', http_status:int|None, href:str, error:str}. href es la ruta de la coleccion (calendar_home + slug saneado + '/')."
tested: true
tests:
- "test_sanitize_slug_minusculas"
- "test_sanitize_slug_espacios_a_guion"
- "test_sanitize_slug_elimina_caracteres_raros"
- "test_sanitize_slug_colapsa_guiones_y_recorta"
- "test_sanitize_slug_vacio"
- "test_join_url_compone_la_coleccion"
- "test_mkcalendar_xml_incluye_displayname"
- "test_mkcalendar_xml_escapa_displayname"
- "test_proppatch_xml_color_y_descripcion"
- "test_proppatch_xml_solo_color"
- "test_proppatch_xml_escapa_descripcion"
test_file_path: "python/functions/infra/dav_make_calendar_test.py"
file_path: "python/functions/infra/dav_make_calendar.py"
---
## Ejemplo
```python
import sys
sys.path.insert(0, "python/functions")
from infra.pass_get_secret import pass_get_secret
from infra.dav_make_calendar import dav_make_calendar
pw = pass_get_secret("dav/xandikos-enmanuel")["value"] # NO logear
res = dav_make_calendar(
base_url="https://dav-eedeb681c4ab89ab8e444ac9.organic-machine.com",
username="enmanuel",
password=pw,
calendar_home="/enmanuel/calendars/",
slug="trabajo",
display_name="Trabajo",
color="#e8590c",
)
print(res)
# {'status': 'ok', 'http_status': 201, 'href': '/enmanuel/calendars/trabajo/'}
# Volver a llamar con el mismo slug:
# {'status': 'ok', 'http_status': 405, 'href': '/enmanuel/calendars/trabajo/', 'existed': True}
```
## Cuando usarla
Cuando el usuario quiere anadir una agenda/calendario nuevo ademas del
principal: una coleccion CalDAV separada ("Trabajo", "Personal", "Cumpleanos")
con su propio nombre visible y color, bajo el calendar-home del principal. El
`href` devuelto es lo que luego pasas como `collection_path` a
`caldav_put_event` para crear eventos en esa agenda, o a `dav_list_calendars`
para verla en el selector.
## Gotchas
- Impura: requiere red + Basic auth contra el servidor DAV. El password viene de
`pass`, no se logea ni se hardcodea.
- Idempotente: si la coleccion ya existe en ese path el servidor responde 405
(Method Not Allowed) o 301; ambos se traducen a `{status:'ok', existed:True}`
en vez de error, asi que es seguro reintentar.
- El PROPPATCH de color usa el `calendar-color` de Apple
(`http://apple.com/ns/ical/`). Servidores que no lo soporten pueden ignorarlo:
el fallo del PROPPATCH NO es fatal (el calendario ya quedo creado) y se ignora
silenciosamente; el color simplemente no se aplica. Si necesitas confirmar el
color, leelo despues con `dav_list_calendars`.
- El `slug` se sanea a `[a-z0-9_-]` (minusculas, espacios->guion, resto fuera).
Un slug que queda vacio tras sanear (p.ej. solo simbolos) devuelve error de
validacion sin tocar la red. El `display_name` y la `description` se escapan
para XML, pero el `slug` que va en la URL ya esta restringido al charset
seguro.
python/functions/infra/dav_make_calendar.py
+202
View File
@@ -0,0 +1,202 @@
"""Crea una nueva coleccion de calendario CalDAV bajo un calendar-home.
Funcion impura: hace una peticion HTTP MKCALENDAR (metodo HTTP literal) para
crear una "agenda" nueva bajo el calendar-home de un principal, y opcionalmente
un PROPPATCH posterior para fijarle el color (Apple `calendar-color`) y la
descripcion (`{urn:ietf:params:xml:ns:caldav}calendar-description`). El nombre
visible (DAV:displayname) se setea ya en el cuerpo del MKCALENDAR.
El slug (segmento de path de la coleccion) se sanea a `[a-z0-9_-]` (minusculas,
espacios -> '-'); si queda vacio se devuelve un error de validacion. La
coleccion se crea en `<calendar_home><slug>/`.
Idempotente: un 201 (Created) es exito; un 405 (Method Not Allowed) o un 301
(la coleccion ya existe en ese path) se devuelven como
{status:'ok', existed:True}. El display_name y la description se escapan para
XML. Construye `Authorization: Basic base64(user:pass)` a mano con stdlib.
Maneja errores sin lanzar (salvo validacion de args). Solo usa stdlib
(urllib, base64, re, ssl, xml.sax.saxutils). Probado contra Xandikos.
"""
import base64
import re
import ssl
import urllib.error
import urllib.request
from xml.sax.saxutils import escape as _xml_escape
_UNSAFE_SLUG_RE = re.compile(r"[^a-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 _sanitize_slug(slug: str) -> str:
"""Sanea un slug a `[a-z0-9_-]`.
Pasa a minusculas, convierte espacios (y runs de espacios) en un guion, y
elimina cualquier otro caracter no permitido. Colapsa guiones repetidos y
recorta guiones de los extremos. Puede devolver "" si no queda nada usable;
el caller trata "" como error de validacion.
"""
s = slug.strip().lower()
s = re.sub(r"\s+", "-", s)
s = _UNSAFE_SLUG_RE.sub("", s)
s = re.sub(r"-{2,}", "-", s).strip("-")
return s
def _build_mkcalendar_xml(display_name: str) -> str:
"""Cuerpo XML minimo del MKCALENDAR que setea el displayname (escapado)."""
name = _xml_escape(display_name)
return (
'<?xml version="1.0" encoding="utf-8" ?>'
'<C:mkcalendar xmlns:C="urn:ietf:params:xml:ns:caldav" xmlns:D="DAV:">'
"<D:set><D:prop>"
"<D:displayname>%s</D:displayname>"
"</D:prop></D:set>"
"</C:mkcalendar>"
) % name
def _build_proppatch_xml(color: str = "", description: str = "") -> str:
"""Cuerpo XML del PROPPATCH que fija color (Apple) y/o descripcion (CalDAV).
Solo incluye las props no vacias. El color va como `calendar-color` del
namespace `http://apple.com/ns/ical/` con el hex tal cual lo pasa el caller
(p.ej. '#RRGGBB'). La descripcion es `calendar-description` de CalDAV. Ambos
valores se escapan para XML.
"""
props = []
if color:
props.append("<A:calendar-color>%s</A:calendar-color>" % _xml_escape(color))
if description:
props.append(
"<C:calendar-description>%s</C:calendar-description>"
% _xml_escape(description)
)
return (
'<?xml version="1.0" encoding="utf-8" ?>'
'<D:propertyupdate xmlns:D="DAV:" '
'xmlns:A="http://apple.com/ns/ical/" '
'xmlns:C="urn:ietf:params:xml:ns:caldav">'
"<D:set><D:prop>%s</D:prop></D:set>"
"</D:propertyupdate>"
) % "".join(props)
def _join_url(base_url: str, calendar_home: str, slug: str) -> str:
return base_url.rstrip("/") + "/" + calendar_home.strip("/") + "/" + slug + "/"
def dav_make_calendar(
base_url: str,
username: str,
password: str,
calendar_home: str,
slug: str,
display_name: str = "",
color: str = "",
description: str = "",
*,
timeout_s: float = 20.0,
verify_tls: bool = True,
) -> dict:
"""Crea una nueva coleccion de calendario CalDAV (MKCALENDAR + PROPPATCH).
Crea la coleccion en `<calendar_home><slug>/` via MKCALENDAR, fijando el
displayname en el propio cuerpo. Si se pasa `color` y/o `description`, hace
un PROPPATCH posterior para setearlos. Idempotente: si la coleccion ya
existe (405/301) devuelve {status:'ok', existed:True}.
Args:
base_url: URL base del servidor DAV (sin barra final), p.ej.
'https://dav-x.organic-machine.com'.
username: usuario para HTTP Basic auth.
password: contrasena para HTTP Basic auth. Resolver desde pass.
calendar_home: ruta del calendar-home del principal (con barra final),
p.ej. '/enmanuel/calendars/'. La coleccion cuelga de el.
slug: segmento de path de la coleccion (p.ej. 'trabajo'); se sanea a
[a-z0-9_-]. Si queda vacio tras sanear, error de validacion.
display_name: nombre visible (DAV:displayname). Si vacio, usa el slug.
color: color de la coleccion como hex '#rrggbb' (Apple calendar-color).
Opcional.
description: descripcion (CalDAV calendar-description). Opcional.
timeout_s: timeout de cada peticion en segundos. Default 20.0.
verify_tls: si True (default) verifica el certificado TLS.
Returns:
dict. En exito: {status:'ok', http_status:int, href:str} (y existed:True
si ya existia). En error (sin lanzar): {status:'error', http_status:
int|None, href:str, error:str}.
"""
clean = _sanitize_slug(slug)
href = (calendar_home.rstrip("/") + "/" + clean + "/") if clean else ""
if not clean:
return {
"status": "error",
"http_status": None,
"href": href,
"error": "slug invalido: queda vacio tras sanear a [a-z0-9_-]",
}
name = display_name if display_name else clean
url = _join_url(base_url, calendar_home, clean)
context = None if verify_tls else ssl._create_unverified_context()
headers = {
"Authorization": _basic_auth_header(username, password),
"Content-Type": "application/xml; charset=utf-8",
}
# 1) MKCALENDAR — crea la coleccion + displayname.
mk_body = _build_mkcalendar_xml(name).encode("utf-8")
req = urllib.request.Request(
url, data=mk_body, method="MKCALENDAR", headers=headers
)
existed = False
http_status = None
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=context) as resp:
http_status = resp.status
except urllib.error.HTTPError as e:
# 405/301: la coleccion ya existe en ese path -> idempotente.
if e.code in (301, 405):
existed = True
http_status = e.code
else:
return {
"status": "error",
"http_status": e.code,
"href": href,
"error": "http %s" % e.code,
}
except urllib.error.URLError as e:
return {
"status": "error",
"http_status": None,
"href": href,
"error": str(e.reason),
}
except Exception as e: # noqa: BLE001
return {"status": "error", "http_status": None, "href": href, "error": str(e)}
# 2) PROPPATCH — color y/o descripcion. Fallo aqui no es fatal: la coleccion
# ya existe. Servidores que no soporten calendar-color pueden ignorarlo.
if color or description:
pp_body = _build_proppatch_xml(color, description).encode("utf-8")
pp_req = urllib.request.Request(
url, data=pp_body, method="PROPPATCH", headers=headers
)
try:
urllib.request.urlopen(pp_req, timeout=timeout_s, context=context).close()
except Exception: # noqa: BLE001
# No fatal: el calendario quedo creado; el color/desc puede no
# soportarse en este servidor. Se ignora silenciosamente.
pass
result = {"status": "ok", "http_status": http_status, "href": href}
if existed:
result["existed"] = True
return result
python/functions/infra/dav_make_calendar_test.py
+73
View File
@@ -0,0 +1,73 @@
"""Tests para dav_make_calendar.
La funcion publica es impura (hace HTTP), asi que no se prueba contra un servidor
real. Se ejercitan los helpers puros extraidos a nivel de modulo: la
sanitizacion del slug, la construccion de la URL de la coleccion y la generacion
de los cuerpos XML del MKCALENDAR y del PROPPATCH (displayname/color/descripcion
escapados). Sin red.
"""
from infra.dav_make_calendar import (
_build_mkcalendar_xml,
_build_proppatch_xml,
_join_url,
_sanitize_slug,
)
def test_sanitize_slug_minusculas():
assert _sanitize_slug("Trabajo") == "trabajo"
def test_sanitize_slug_espacios_a_guion():
assert _sanitize_slug("agenda de trabajo") == "agenda-de-trabajo"
def test_sanitize_slug_elimina_caracteres_raros():
assert _sanitize_slug("Casa/Ocio!! 2026") == "casaocio-2026"
def test_sanitize_slug_colapsa_guiones_y_recorta():
assert _sanitize_slug(" --Foo Bar-- ") == "foo-bar"
def test_sanitize_slug_vacio():
assert _sanitize_slug(" !!! ") == ""
def test_join_url_compone_la_coleccion():
url = _join_url(
"https://dav-x.organic-machine.com",
"/enmanuel/calendars/",
"trabajo",
)
assert url == "https://dav-x.organic-machine.com/enmanuel/calendars/trabajo/"
def test_mkcalendar_xml_incluye_displayname():
xml = _build_mkcalendar_xml("Trabajo")
assert "<C:mkcalendar" in xml
assert "<D:displayname>Trabajo</D:displayname>" in xml
def test_mkcalendar_xml_escapa_displayname():
xml = _build_mkcalendar_xml("Casa & <Ocio>")
assert "Casa &amp; &lt;Ocio&gt;" in xml
assert "<Ocio>" not in xml
def test_proppatch_xml_color_y_descripcion():
xml = _build_proppatch_xml(color="#e8590c", description="Mi agenda")
assert "<A:calendar-color>#e8590c</A:calendar-color>" in xml
assert "<C:calendar-description>Mi agenda</C:calendar-description>" in xml
def test_proppatch_xml_solo_color():
xml = _build_proppatch_xml(color="#e8590c")
assert "<A:calendar-color>#e8590c</A:calendar-color>" in xml
assert "calendar-description" not in xml
def test_proppatch_xml_escapa_descripcion():
xml = _build_proppatch_xml(description="A & B <c>")
assert "A &amp; B &lt;c&gt;" in xml
python/functions/infra/expand_rrule.md
+89
View File
@@ -0,0 +1,89 @@
---
name: expand_rrule
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: pure
signature: "def expand_rrule(dtstart_ical: str, rrule: str, range_start: str, range_end: str, all_day: bool = False) -> list[str]"
description: "Expande una RRULE iCalendar a la lista ordenada de fechas DTSTART de cada ocurrencia que cae dentro de un rango [range_start, range_end]. Pura, determinista, solo stdlib (sin python-dateutil). Soporta FREQ DAILY/WEEKLY/MONTHLY/YEARLY, INTERVAL, COUNT, UNTIL y BYDAY (para WEEKLY)."
tags: [dav, calendar, ical, rrule, recurrence, caldav]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: ""
imports: []
tested: true
tests: ["test_golden_weekly_count_4", "test_edge_monthly_interval_2", "test_edge_weekly_byday_two_days", "test_edge_all_day_vs_with_time", "test_until_recorta", "test_filtro_por_rango_excluye_fuera", "test_dtstart_anterior_al_rango_pero_serie_entra", "test_componente_no_soportado_se_ignora", "test_sin_count_ni_until_acota_a_range_end"]
test_file_path: "python/functions/infra/expand_rrule_test.py"
file_path: "python/functions/infra/expand_rrule.py"
params:
- name: dtstart_ical
desc: "Fecha de inicio del evento maestro en formato iCal crudo: YYYYMMDD (all-day), YYYYMMDDTHHMMSS o YYYYMMDDTHHMMSSZ. Es la primera ocurrencia (la serie la incluye si cae en rango)."
- name: rrule
desc: "Cuerpo de la RRULE SIN el prefijo 'RRULE:', p.ej. 'FREQ=WEEKLY;INTERVAL=1;COUNT=10' o 'FREQ=MONTHLY;UNTIL=20261231;BYDAY=MO,WE'."
- name: range_start
desc: "Limite inferior del rango como YYYYMMDD (inclusive). Solo se devuelven ocurrencias cuya fecha YYYYMMDD del DTSTART cae en [range_start, range_end]."
- name: range_end
desc: "Limite superior del rango como YYYYMMDD (inclusive). Tambien acota la generacion cuando faltan COUNT y UNTIL en la RRULE."
- name: all_day
desc: "Si True las ocurrencias se devuelven como YYYYMMDD; si False conservan la parte de hora del dtstart original (misma hora local en cada ocurrencia) y devuelven YYYYMMDDTHHMMSS sin sufijo Z. Default False."
output: "Lista ordenada de strings DTSTART iCal, una por ocurrencia en rango. Lista vacia si la RRULE no produce ninguna en [range_start, range_end]."
---
## Ejemplo
```python
import sys, os
sys.path.insert(0, os.path.join("python", "functions"))
from infra.expand_rrule import expand_rrule
# Reunion semanal los lunes a las 09:00, 4 ocurrencias, ventana de enero 2026.
fechas = expand_rrule(
"20260105T090000",
"FREQ=WEEKLY;COUNT=4",
"20260101",
"20261231",
)
print(fechas)
# ['20260105T090000', '20260112T090000', '20260119T090000', '20260126T090000']
# Evento all-day mensual cada 2 meses, solo las que caen en el primer semestre.
fechas = expand_rrule(
"20260115",
"FREQ=MONTHLY;INTERVAL=2;COUNT=4",
"20260101",
"20260630",
all_day=True,
)
print(fechas)
# ['20260115', '20260315', '20260515']
```
## Cuando usarla
Cuando un cliente CalDAV necesita mostrar las ocurrencias de un evento
recurrente dentro de la ventana visible del calendario: tienes el DTSTART y la
RRULE del VEVENT maestro y quieres la lista concreta de fechas de inicio que
caen entre dos limites para pintarlas en la agenda. Tambien para contar o
iterar instancias de una serie sin instanciar todo el iCal.
## Gotchas
- **No implementa el RFC 5545 completo.** Componentes soportados:
- `FREQ` (obligatorio): `DAILY`, `WEEKLY`, `MONTHLY`, `YEARLY`.
- `INTERVAL` (default 1).
- `COUNT` (incluye la primera ocurrencia = dtstart).
- `UNTIL` (`YYYYMMDD` o `YYYYMMDDTHHMMSSZ`, inclusive).
- `BYDAY` solo para `FREQ=WEEKLY` (`MO,TU,WE,TH,FR,SA,SU`).
- Cualquier otro componente (`BYMONTHDAY`, `BYSETPOS`, `BYMONTH`, `WKST`
avanzado, EXDATE, RDATE, etc.) se **ignora silenciosamente** — no falla, pero
el resultado puede diferir del esperado por el RFC en esos casos.
- Si faltan **COUNT y UNTIL** a la vez, la generacion se acota por `range_end`
con un tope de seguridad duro de 1000 ocurrencias para no colgar.
- En `FREQ=MONTHLY`/`YEARLY` con dia 29/30/31, los meses sin ese dia recortan al
ultimo dia valido del mes destino.
- No gestiona zonas horarias: con `all_day=False` conserva la hora local del
dtstart sin sufijo `Z`; el llamador es responsable de la tz (TZID/VTIMEZONE).
- El filtro de rango compara solo la parte `YYYYMMDD` del DTSTART, no la hora.
python/functions/infra/expand_rrule.py
+202
View File
@@ -0,0 +1,202 @@
"""Expandir una RRULE iCalendar a las fechas de inicio de cada ocurrencia.
Implementacion pura y determinista, solo stdlib (datetime, re). NO usa
python-dateutil (no disponible en los venvs del ecosistema). Soporta un
subconjunto pragmatico de RFC 5545 suficiente para una agenda personal:
FREQ (DAILY/WEEKLY/MONTHLY/YEARLY), INTERVAL, COUNT, UNTIL y BYDAY (solo
para FREQ=WEEKLY). Cualquier otro componente se ignora silenciosamente.
"""
import re
from datetime import date, datetime, timedelta
# Tope de seguridad para no colgar cuando faltan COUNT y UNTIL.
_MAX_OCCURRENCES = 1000
# Mapeo de codigos BYDAY iCal -> weekday() de Python (lunes=0 .. domingo=6).
_BYDAY_TO_WEEKDAY = {
"MO": 0,
"TU": 1,
"WE": 2,
"TH": 3,
"FR": 4,
"SA": 5,
"SU": 6,
}
def _parse_ical_date(value: str) -> date:
"""Extrae la parte YYYYMMDD de un valor iCal y la devuelve como date.
Acepta YYYYMMDD, YYYYMMDDTHHMMSS y YYYYMMDDTHHMMSSZ.
"""
digits = value.strip()[:8]
return date(int(digits[:4]), int(digits[4:6]), int(digits[6:8]))
def _parse_time_suffix(dtstart_ical: str) -> str:
"""Devuelve la parte de hora 'THHMMSS' del dtstart, o '' si es all-day."""
m = re.search(r"T(\d{6})", dtstart_ical.strip())
return f"T{m.group(1)}" if m else ""
def _parse_rrule(rrule: str) -> dict:
"""Parsea el cuerpo de una RRULE a un dict de componentes en mayusculas."""
parts: dict[str, str] = {}
for token in rrule.strip().split(";"):
if not token or "=" not in token:
continue
key, _, val = token.partition("=")
parts[key.strip().upper()] = val.strip()
return parts
def _add_months(d: date, months: int) -> date:
"""Suma `months` meses a una fecha, recortando el dia al ultimo valido."""
total = (d.year * 12 + (d.month - 1)) + months
year, month = divmod(total, 12)
month += 1
# Recorte de dia: si el dia no existe en el mes destino, usar el ultimo.
if month == 12:
next_first = date(year + 1, 1, 1)
else:
next_first = date(year, month + 1, 1)
last_day = (next_first - timedelta(days=1)).day
return date(year, month, min(d.day, last_day))
def _format_occurrence(d: date, time_suffix: str, all_day: bool) -> str:
"""Formatea una fecha de ocurrencia como string DTSTART iCal."""
ymd = f"{d.year:04d}{d.month:02d}{d.day:02d}"
if all_day or not time_suffix:
return ymd
return f"{ymd}{time_suffix}"
def _generate_dates(dtstart: date, rule: dict) -> list[date]:
"""Genera las fechas de ocurrencia segun la RRULE (antes de filtrar rango).
El recorte por rango lo hace el llamador; aqui se respeta COUNT, UNTIL y
el tope de seguridad _MAX_OCCURRENCES.
"""
freq = rule.get("FREQ", "").upper()
interval = max(1, int(rule["INTERVAL"])) if rule.get("INTERVAL", "").isdigit() else 1
count = int(rule["COUNT"]) if rule.get("COUNT", "").isdigit() else None
until = _parse_ical_date(rule["UNTIL"]) if rule.get("UNTIL") else None
results: list[date] = []
def reached_limit() -> bool:
if count is not None and len(results) >= count:
return True
return len(results) >= _MAX_OCCURRENCES
def past_until(d: date) -> bool:
return until is not None and d > until
if freq == "WEEKLY":
byday = rule.get("BYDAY", "")
weekdays = [
_BYDAY_TO_WEEKDAY[code.strip().upper()]
for code in byday.split(",")
if code.strip().upper() in _BYDAY_TO_WEEKDAY
]
if not weekdays:
weekdays = [dtstart.weekday()]
weekdays = sorted(set(weekdays))
# Lunes de la semana del dtstart.
week_anchor = dtstart - timedelta(days=dtstart.weekday())
week_index = 0
while True:
week_start = week_anchor + timedelta(weeks=week_index * interval)
for wd in weekdays:
occ = week_start + timedelta(days=wd)
if occ < dtstart:
continue
if past_until(occ):
return results
if reached_limit():
return results
results.append(occ)
if reached_limit():
return results
# Corte: si no hay COUNT ni UNTIL, el llamador acota por range_end,
# pero el tope duro evita el bucle infinito.
if until is None and count is None and len(results) >= _MAX_OCCURRENCES:
return results
week_index += 1
# Salvaguarda extra si UNTIL/COUNT no recortan a tiempo.
if week_index > _MAX_OCCURRENCES:
return results
return results
# FREQ por periodos simples (DAILY / MONTHLY / YEARLY).
step = 0
while True:
if freq == "DAILY":
occ = dtstart + timedelta(days=step * interval)
elif freq == "MONTHLY":
occ = _add_months(dtstart, step * interval)
elif freq == "YEARLY":
occ = _add_months(dtstart, step * interval * 12)
else:
# FREQ desconocido o ausente: solo la primera ocurrencia.
occ = dtstart
if not past_until(occ):
results.append(occ)
return results
if past_until(occ):
return results
if reached_limit():
return results
results.append(occ)
if reached_limit():
return results
if until is None and count is None and len(results) >= _MAX_OCCURRENCES:
return results
step += 1
if step > _MAX_OCCURRENCES:
return results
def expand_rrule(
dtstart_ical: str,
rrule: str,
range_start: str,
range_end: str,
all_day: bool = False,
) -> list[str]:
"""Expande una RRULE iCal a las fechas DTSTART de cada ocurrencia en rango.
Args:
dtstart_ical: fecha de inicio del evento maestro en formato iCal crudo.
Puede ser YYYYMMDD (all-day), YYYYMMDDTHHMMSS o YYYYMMDDTHHMMSSZ.
Es la primera ocurrencia (la serie incluye dtstart si cae en rango).
rrule: cuerpo de la RRULE SIN el prefijo 'RRULE:', p.ej.
'FREQ=WEEKLY;INTERVAL=1;COUNT=10' o 'FREQ=MONTHLY;UNTIL=20261231;BYDAY=MO,WE'.
range_start: limite inferior del rango como YYYYMMDD (inclusive).
range_end: limite superior del rango como YYYYMMDD (inclusive).
all_day: si True las ocurrencias se devuelven como YYYYMMDD; si False
conservan la parte de hora del dtstart original (misma hora local en
cada ocurrencia) y devuelven YYYYMMDDTHHMMSS (sin sufijo Z).
Returns:
Lista ordenada de strings DTSTART iCal, una por ocurrencia cuya fecha
(parte YYYYMMDD del DTSTART) cae en [range_start, range_end]. Lista
vacia si la RRULE no produce ninguna en rango.
"""
rule = _parse_rrule(rrule)
dtstart = _parse_ical_date(dtstart_ical)
time_suffix = _parse_time_suffix(dtstart_ical)
lo = _parse_ical_date(range_start)
hi = _parse_ical_date(range_end)
occurrences = _generate_dates(dtstart, rule)
out: list[str] = []
for occ in occurrences:
if lo <= occ <= hi:
out.append(_format_occurrence(occ, time_suffix, all_day))
out.sort()
return out
python/functions/infra/expand_rrule_test.py
+144
View File
@@ -0,0 +1,144 @@
"""Tests para expand_rrule."""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from functions.infra.expand_rrule import expand_rrule
def test_golden_weekly_count_4():
# FREQ=WEEKLY;COUNT=4 a partir del 2026-01-05 (lunes) -> 4 fechas a 7 dias.
got = expand_rrule(
"20260105T090000",
"FREQ=WEEKLY;COUNT=4",
"20260101",
"20261231",
)
assert got == [
"20260105T090000",
"20260112T090000",
"20260119T090000",
"20260126T090000",
]
def test_edge_monthly_interval_2():
# INTERVAL=2 mensual: cada dos meses, dia 15.
got = expand_rrule(
"20260115",
"FREQ=MONTHLY;INTERVAL=2;COUNT=4",
"20260101",
"20261231",
all_day=True,
)
assert got == ["20260115", "20260315", "20260515", "20260715"]
def test_edge_weekly_byday_two_days():
# BYDAY con 2 dias en WEEKLY: MO y WE, dentro de cada semana del intervalo.
# dtstart 2026-01-05 (lunes). COUNT=4 -> MO,WE de la sem 1 y MO,WE de la sem 2.
got = expand_rrule(
"20260105",
"FREQ=WEEKLY;BYDAY=MO,WE;COUNT=4",
"20260101",
"20261231",
all_day=True,
)
assert got == ["20260105", "20260107", "20260112", "20260114"]
def test_edge_all_day_vs_with_time():
# all_day=True -> YYYYMMDD; all_day=False conserva la hora del dtstart.
all_day = expand_rrule(
"20260105T143000",
"FREQ=DAILY;COUNT=2",
"20260101",
"20261231",
all_day=True,
)
assert all_day == ["20260105", "20260106"]
with_time = expand_rrule(
"20260105T143000",
"FREQ=DAILY;COUNT=2",
"20260101",
"20261231",
all_day=False,
)
assert with_time == ["20260105T143000", "20260106T143000"]
def test_until_recorta():
# UNTIL=20260120 recorta la serie semanal inclusive en esa fecha.
got = expand_rrule(
"20260105T090000",
"FREQ=WEEKLY;UNTIL=20260120T235959Z",
"20260101",
"20261231",
)
assert got == [
"20260105T090000",
"20260112T090000",
"20260119T090000",
]
def test_filtro_por_rango_excluye_fuera():
# COUNT=10 semanal, pero el rango solo cubre 3 ocurrencias intermedias.
got = expand_rrule(
"20260105T090000",
"FREQ=WEEKLY;COUNT=10",
"20260112",
"20260131",
)
assert got == [
"20260112T090000",
"20260119T090000",
"20260126T090000",
]
def test_dtstart_anterior_al_rango_pero_serie_entra():
# dtstart en 2025-12-29 (antes del rango), serie semanal entra en enero 2026.
got = expand_rrule(
"20251229T090000",
"FREQ=WEEKLY;COUNT=8",
"20260101",
"20260115",
)
assert got == [
"20260105T090000",
"20260112T090000",
]
def test_componente_no_soportado_se_ignora():
# BYMONTHDAY no soportado -> se ignora, no falla. Serie mensual normal.
got = expand_rrule(
"20260110",
"FREQ=MONTHLY;BYMONTHDAY=10;COUNT=3",
"20260101",
"20261231",
all_day=True,
)
assert got == ["20260110", "20260210", "20260310"]
def test_sin_count_ni_until_acota_a_range_end():
# Sin COUNT ni UNTIL: la generacion debe acotarse por range_end.
got = expand_rrule(
"20260101",
"FREQ=DAILY",
"20260101",
"20260105",
all_day=True,
)
assert got == [
"20260101",
"20260102",
"20260103",
"20260104",
"20260105",
]