535 lines
20 KiB
Python
535 lines
20 KiB
Python
"""Календари: MCP-руки для модели и тот же ical-фид, читаемый gateway напрямую.
|
||
|
||
`CALENDAR_MCPS=name=url,name=url`; в url - приватный фид с токеном, и там же,
|
||
параметром `icsUrl`, лежит сам ical. Модель ходит в календари через MCP, а
|
||
справка конверта читает фид сама: канал, о котором никто не спрашивает, молчит
|
||
незаметно - так экзаменационная сессия и пролежала невидимой.
|
||
|
||
Парсер здесь свой и маленький: в образе gateway ical-библиотеки нет, а нужно
|
||
немного - когда начинается, как называется, где. Повторы разворачиваются по
|
||
FREQ=DAILY/WEEKLY/MONTHLY/YEARLY с INTERVAL, BYDAY, BYMONTH, COUNT и UNTIL;
|
||
EXDATE и перенесённые экземпляры (RECURRENCE-ID) вычитаются. Шаг считается по
|
||
настенным часам события, а не в абсолютном времени: еженедельная встреча
|
||
переживает перевод часов. Событие, порождённое ежедневным правилом, помечается
|
||
`routine`: в справке такие едут одной строкой, чтобы не вытеснять разовые.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import urllib.error
|
||
import urllib.request
|
||
from calendar import monthrange
|
||
from dataclasses import dataclass, field
|
||
from datetime import UTC, date, datetime, timedelta
|
||
from typing import TYPE_CHECKING
|
||
from urllib.parse import parse_qs, urlsplit
|
||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||
|
||
from beaver_gateway.mcp.types import HttpMcp, McpServer
|
||
|
||
if TYPE_CHECKING:
|
||
from collections.abc import Iterator
|
||
from datetime import tzinfo
|
||
|
||
__all__ = [
|
||
"Calendars",
|
||
"Event",
|
||
"Feed",
|
||
"Upcoming",
|
||
"calendars",
|
||
"feeds",
|
||
"is_routine",
|
||
"short",
|
||
]
|
||
|
||
_log = logging.getLogger(__name__)
|
||
|
||
USER_AGENT = "beaver-gateway (ical)"
|
||
MAX_BYTES = 4_000_000
|
||
"""Больше четырёх мегабайт фида - это уже не календарь, а чей-то экспорт всего."""
|
||
STEPS_CAP = 20_000
|
||
"""Потолок шагов разворачивания одного повтора: страховка от вечного цикла."""
|
||
PLACE_MAX = 70
|
||
TITLE_MAX = 90
|
||
ERROR_MAX = 90
|
||
NO_TITLE = "(без названия)"
|
||
|
||
WEEKDAYS = {"MO": 0, "TU": 1, "WE": 2, "TH": 3, "FR": 4, "SA": 5, "SU": 6}
|
||
DT_RE = re.compile(r"^(\d{4})(\d{2})(\d{2})(?:T(\d{2})(\d{2})(\d{2})(Z)?)?$")
|
||
BYDAY_RE = re.compile(r"^([+-]?\d{1,2})?(MO|TU|WE|TH|FR|SA|SU)$")
|
||
ESCAPE_RE = re.compile(r"\\([\\;,nN])")
|
||
PLACE_RE = re.compile(
|
||
r"^\s*(?:sala|room|зал|аудитория)\b\s*[:.]?\s*(.+)$", re.IGNORECASE | re.MULTILINE
|
||
)
|
||
"""Зал USOS живёт в описании (`Sala: 315`), а LOCATION - это адрес корпуса."""
|
||
|
||
|
||
def calendars(raw: str) -> list[HttpMcp]:
|
||
servers: list[HttpMcp] = []
|
||
for position, raw_entry in enumerate(raw.split(","), start=1):
|
||
entry = raw_entry.strip()
|
||
if not entry:
|
||
continue
|
||
name, sep, url = entry.partition("=")
|
||
if not sep or not url.strip():
|
||
msg = f"CALENDAR_MCPS: запись #{position} не вида `name=url`"
|
||
raise ValueError(msg)
|
||
servers.append(McpServer.http(name=f"calendar-{name.strip()}", url=url.strip()))
|
||
return servers
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Feed:
|
||
name: str
|
||
url: str
|
||
"""Ical по подписке: приватный, с токеном внутри - в логи не писать."""
|
||
|
||
|
||
def feeds(raw: str) -> tuple[Feed, ...]:
|
||
"""Ical-фиды из тех же записей `CALENDAR_MCPS`: урл лежит в `icsUrl`.
|
||
|
||
Запись без ical-урла просто не даёт фида: MCP-рука у модели остаётся,
|
||
справка про этот календарь молчит - конфигурация, а не отказ канала.
|
||
"""
|
||
out: list[Feed] = []
|
||
for raw_entry in raw.split(","):
|
||
name, sep, url = raw_entry.strip().partition("=")
|
||
if not sep or not url.strip():
|
||
continue
|
||
ics = _ics_url(url.strip())
|
||
if ics:
|
||
out.append(Feed(name=name.strip(), url=ics))
|
||
return tuple(out)
|
||
|
||
|
||
def _ics_url(url: str) -> str | None:
|
||
inner = parse_qs(urlsplit(url).query).get("icsUrl") or []
|
||
if inner and inner[0].strip():
|
||
return inner[0].strip()
|
||
return url if urlsplit(url).path.endswith(".ics") else None
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Event:
|
||
start: datetime
|
||
"""С зоной: у события своя (TZID), плавающее время читается в зоне агента."""
|
||
title: str
|
||
place: str
|
||
all_day: bool
|
||
routine: bool = False
|
||
"""Порождено правилом, повторяющимся раз в день или чаще: см. `is_routine`."""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Upcoming:
|
||
at: datetime
|
||
"""Момент запроса - его и показывает справка: «данные на …»."""
|
||
events: tuple[Event, ...] = ()
|
||
errors: tuple[str, ...] = ()
|
||
"""По строке на фид, который не ответил; пустое окно - это не ошибка."""
|
||
read: int = 0
|
||
"""Сколько фидов ответило: ноль - показывать нечего, кроме отказа."""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Calendars:
|
||
"""Ближайшие события всех фидов - одним списком, отсортированным по началу."""
|
||
|
||
feeds: tuple[Feed, ...]
|
||
tz: str = "Europe/Warsaw"
|
||
timeout: float = 8.0
|
||
days: int = 7
|
||
|
||
def upcoming(self, now: datetime) -> Upcoming:
|
||
"""Окно `days` от `now`; отказ фида - строка в `errors`, не исключение."""
|
||
zone = ZoneInfo(self.tz)
|
||
since = now.astimezone(zone)
|
||
events: list[Event] = []
|
||
errors: list[str] = []
|
||
read = 0
|
||
for feed in self.feeds:
|
||
try:
|
||
text = self.fetch(feed.url)
|
||
found = events_in(text, zone=zone, since=since, days=self.days)
|
||
except Exception as exc: # noqa: BLE001
|
||
_log.warning("календарь %s: %s", feed.name, short(exc), exc_info=True)
|
||
errors.append(f"{feed.name}: {short(exc)}")
|
||
continue
|
||
read += 1
|
||
events.extend(found)
|
||
events.sort(key=lambda e: (e.start, e.title))
|
||
return Upcoming(at=since, events=tuple(events), errors=tuple(errors), read=read)
|
||
|
||
def fetch(self, url: str) -> str:
|
||
if urlsplit(url).scheme not in ("http", "https"):
|
||
msg = "фид не по http(s)"
|
||
raise ValueError(msg)
|
||
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) # noqa: S310
|
||
with urllib.request.urlopen(request, timeout=self.timeout) as response: # noqa: S310
|
||
raw: bytes = response.read(MAX_BYTES)
|
||
return raw.decode("utf-8", errors="replace")
|
||
|
||
|
||
def short(exc: BaseException) -> str:
|
||
"""Причина отказа одной строкой: в конверт едет она, урл фида - никогда."""
|
||
if isinstance(exc, urllib.error.HTTPError):
|
||
reason = f"HTTP {exc.code}"
|
||
elif isinstance(exc, TimeoutError):
|
||
reason = "таймаут"
|
||
elif isinstance(exc, urllib.error.URLError):
|
||
reason = f"сеть: {exc.reason}"
|
||
else:
|
||
reason = f"{type(exc).__name__}: {exc}".strip(": ")
|
||
return " ".join(reason.split())[:ERROR_MAX]
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Prop:
|
||
value: str
|
||
params: dict[str, str] = field(default_factory=dict)
|
||
|
||
|
||
Vevent = dict[str, list[Prop]]
|
||
|
||
|
||
def unfold(text: str) -> list[str]:
|
||
"""Ical-строки, склеенные обратно: продолжение начинается с пробела или таба."""
|
||
lines: list[str] = []
|
||
for raw in text.replace("\r\n", "\n").replace("\r", "\n").split("\n"):
|
||
if lines and raw[:1] in (" ", "\t"):
|
||
lines[-1] += raw[1:]
|
||
else:
|
||
lines.append(raw)
|
||
return lines
|
||
|
||
|
||
def _split(line: str) -> tuple[str, str] | None:
|
||
"""`NAME;PARAM=x:значение` - двоеточие вне кавычек делит имя и значение."""
|
||
quoted = False
|
||
for i, char in enumerate(line):
|
||
if char == '"':
|
||
quoted = not quoted
|
||
elif char == ":" and not quoted:
|
||
return line[:i], line[i + 1 :]
|
||
return None
|
||
|
||
|
||
def _prop(line: str) -> tuple[str, Prop] | None:
|
||
parts = _split(line)
|
||
if parts is None:
|
||
return None
|
||
head, value = parts
|
||
chunks = head.split(";")
|
||
name = chunks[0].strip().upper()
|
||
if not name:
|
||
return None
|
||
params: dict[str, str] = {}
|
||
for chunk in chunks[1:]:
|
||
key, sep, raw = chunk.partition("=")
|
||
if sep:
|
||
params[key.strip().upper()] = raw.strip().strip('"')
|
||
return name, Prop(value=value, params=params)
|
||
|
||
|
||
def vevents(lines: list[str]) -> list[Vevent]:
|
||
"""Только свойства самих VEVENT: VALARM внутри них пропускается целиком."""
|
||
out: list[Vevent] = []
|
||
current: Vevent | None = None
|
||
depth = 0
|
||
for line in lines:
|
||
parsed = _prop(line)
|
||
if parsed is None:
|
||
continue
|
||
name, prop = parsed
|
||
block = prop.value.strip().upper()
|
||
if name == "BEGIN":
|
||
if block == "VEVENT" and current is None:
|
||
current = {}
|
||
elif current is not None:
|
||
depth += 1
|
||
continue
|
||
if name == "END":
|
||
if block == "VEVENT" and current is not None and depth == 0:
|
||
out.append(current)
|
||
current = None
|
||
elif current is not None and depth:
|
||
depth -= 1
|
||
continue
|
||
if current is not None and depth == 0:
|
||
current.setdefault(name, []).append(prop)
|
||
return out
|
||
|
||
|
||
def unescape(value: str) -> str:
|
||
return ESCAPE_RE.sub(lambda m: "\n" if m.group(1) in "nN" else m.group(1), value)
|
||
|
||
|
||
def _flat(value: str, limit: int) -> str:
|
||
flat = " ".join(unescape(value).split())
|
||
return flat if len(flat) <= limit else flat[: limit - 1] + "…"
|
||
|
||
|
||
def _first(event: Vevent, name: str) -> Prop | None:
|
||
props = event.get(name)
|
||
return props[0] if props else None
|
||
|
||
|
||
def _zone_of(prop: Prop, default: tzinfo) -> tzinfo:
|
||
tzid = prop.params.get("TZID")
|
||
if not tzid:
|
||
return default
|
||
try:
|
||
return ZoneInfo(tzid)
|
||
except (ZoneInfoNotFoundError, ValueError):
|
||
return default
|
||
|
||
|
||
def moment(prop: Prop, value: str, zone: tzinfo) -> tuple[datetime, bool] | None:
|
||
"""Значение DTSTART/EXDATE в момент времени; второе - «на весь день»."""
|
||
match = DT_RE.match(value.strip())
|
||
if match is None:
|
||
return None
|
||
year, month, day = (int(match.group(i)) for i in (1, 2, 3))
|
||
if match.group(4) is None or prop.params.get("VALUE", "").upper() == "DATE":
|
||
return datetime(year, month, day, tzinfo=zone), True
|
||
hour, minute, second = (int(match.group(i)) for i in (4, 5, 6))
|
||
where = UTC if match.group(7) else _zone_of(prop, zone)
|
||
return datetime(year, month, day, hour, minute, second, tzinfo=where), False
|
||
|
||
|
||
def _instants(event: Vevent, name: str, zone: tzinfo) -> set[datetime]:
|
||
"""Даты-исключения в UTC: значений в одной строке может быть несколько."""
|
||
out: set[datetime] = set()
|
||
for prop in event.get(name, []):
|
||
for chunk in prop.value.split(","):
|
||
parsed = moment(prop, chunk, zone)
|
||
if parsed is not None:
|
||
out.add(parsed[0].astimezone(UTC))
|
||
return out
|
||
|
||
|
||
def _rule(value: str) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
for chunk in value.split(";"):
|
||
key, sep, raw = chunk.partition("=")
|
||
if sep:
|
||
out[key.strip().upper()] = raw.strip()
|
||
return out
|
||
|
||
|
||
def _int(value: str | None, default: int) -> int:
|
||
try:
|
||
return int(value) if value else default
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def _byday(value: str) -> list[tuple[int, int]]:
|
||
"""`-1SU,2TU` → [(-1, воскресенье), (2, вторник)]; без числа - каждый такой день."""
|
||
out: list[tuple[int, int]] = []
|
||
for chunk in value.split(","):
|
||
match = BYDAY_RE.match(chunk.strip().upper())
|
||
if match:
|
||
out.append((_int(match.group(1), 0), WEEKDAYS[match.group(2)]))
|
||
return out
|
||
|
||
|
||
def _month_days(
|
||
year: int, month: int, days: list[tuple[int, int]], clock: datetime
|
||
) -> list[datetime]:
|
||
total = monthrange(year, month)[1]
|
||
out: list[datetime] = []
|
||
for nth, weekday in days:
|
||
matching = [
|
||
day
|
||
for day in range(1, total + 1)
|
||
if date(year, month, day).weekday() == weekday
|
||
]
|
||
if not nth:
|
||
picked = matching
|
||
elif abs(nth) <= len(matching):
|
||
picked = [matching[nth - 1] if nth > 0 else matching[nth]]
|
||
else:
|
||
picked = []
|
||
out += [
|
||
datetime(year, month, day, clock.hour, clock.minute, clock.second) # noqa: DTZ001
|
||
for day in picked
|
||
]
|
||
return sorted(out)
|
||
|
||
|
||
def _candidates(
|
||
naive: datetime, freq: str, interval: int, days: list[tuple[int, int]], month: int
|
||
) -> Iterator[datetime]:
|
||
"""Настенные времена повтора после первого - по возрастанию, бесконечно."""
|
||
if freq == "DAILY":
|
||
moving = naive
|
||
while True:
|
||
moving += timedelta(days=interval)
|
||
yield moving
|
||
elif freq == "WEEKLY" and not days:
|
||
moving = naive
|
||
while True:
|
||
moving += timedelta(weeks=interval)
|
||
yield moving
|
||
elif freq == "WEEKLY":
|
||
start = naive.date() - timedelta(days=naive.weekday())
|
||
step = 0
|
||
while True:
|
||
base = start + timedelta(weeks=interval * step)
|
||
for _, weekday in sorted(days, key=lambda d: d[1]):
|
||
yield datetime.combine(base + timedelta(days=weekday), naive.time())
|
||
step += 1
|
||
elif freq in ("MONTHLY", "YEARLY"):
|
||
step = 0
|
||
while True:
|
||
step += 1
|
||
if freq == "MONTHLY":
|
||
index = naive.year * 12 + naive.month - 1 + interval * step
|
||
year, number = divmod(index, 12)
|
||
number += 1
|
||
else:
|
||
year, number = naive.year + interval * step, month or naive.month
|
||
if days:
|
||
yield from _month_days(year, number, days, naive)
|
||
elif naive.day <= monthrange(year, number)[1]:
|
||
yield naive.replace(year=year, month=number)
|
||
|
||
|
||
def occurrences(
|
||
start: datetime, rule: dict[str, str], zone: tzinfo, *, end: datetime
|
||
) -> list[datetime]:
|
||
"""Начала повтора от `start` до `end`; правило, которого мы не знаем, - разовое."""
|
||
freq = rule.get("FREQ", "").upper()
|
||
if freq not in ("DAILY", "WEEKLY", "MONTHLY", "YEARLY"):
|
||
return [start]
|
||
where = start.tzinfo or zone
|
||
naive = start.replace(tzinfo=None)
|
||
interval = max(_int(rule.get("INTERVAL"), 1), 1)
|
||
count = _int(rule.get("COUNT"), 0)
|
||
until = _until(rule.get("UNTIL"), where)
|
||
days = _byday(rule.get("BYDAY", ""))
|
||
out = [start]
|
||
stream = _candidates(naive, freq, interval, days, _int(rule.get("BYMONTH"), 0))
|
||
for step, candidate in enumerate(stream):
|
||
if step > STEPS_CAP:
|
||
break
|
||
if candidate <= naive:
|
||
continue
|
||
moving = candidate.replace(tzinfo=where)
|
||
if moving > end or (until is not None and moving > until):
|
||
break
|
||
out.append(moving)
|
||
if count and len(out) >= count:
|
||
break
|
||
return out
|
||
|
||
|
||
def _until(value: str | None, zone: tzinfo) -> datetime | None:
|
||
if not value:
|
||
return None
|
||
parsed = moment(Prop(value=value), value, zone)
|
||
if parsed is None:
|
||
return None
|
||
stop, all_day = parsed
|
||
return stop + timedelta(days=1) if all_day else stop
|
||
|
||
|
||
ROUTINE_DAYS = 5
|
||
"""Со скольких дней недели в BYDAY повтор перестаёт быть событием и делается бытом."""
|
||
|
||
|
||
def is_routine(rule: dict[str, str]) -> bool:
|
||
"""Правило, срабатывающее раз в день или чаще: завтрак, зарядка, wind down.
|
||
|
||
Такие повторы за неделю дают десятки строк и топят в себе то единственное,
|
||
ради чего справку и читают, - экзамен, встречу, приём. В конверт они едут
|
||
одной свёрнутой строкой, а не списком.
|
||
"""
|
||
freq = rule.get("FREQ", "").upper()
|
||
if max(_int(rule.get("INTERVAL"), 1), 1) != 1:
|
||
return False
|
||
if freq == "DAILY":
|
||
return True
|
||
return freq == "WEEKLY" and len(_byday(rule.get("BYDAY", ""))) >= ROUTINE_DAYS
|
||
|
||
|
||
def place_of(event: Vevent) -> str:
|
||
"""Зал из описания и адрес из LOCATION - то, что помогает дойти."""
|
||
parts: list[str] = []
|
||
description = _first(event, "DESCRIPTION")
|
||
if description is not None:
|
||
match = PLACE_RE.search(unescape(description.value))
|
||
if match:
|
||
parts.append(" ".join(match.group(0).split()))
|
||
location = _first(event, "LOCATION")
|
||
if location is not None:
|
||
flat = " ".join(unescape(location.value).split())
|
||
if flat:
|
||
parts.append(flat)
|
||
return _flat(", ".join(parts), PLACE_MAX)
|
||
|
||
|
||
def events_in(text: str, *, zone: tzinfo, since: datetime, days: int) -> list[Event]:
|
||
"""События фида, начинающиеся в окне `[since, since + days]`, с повторами."""
|
||
end = since + timedelta(days=days)
|
||
day_start = since.astimezone(zone).replace(
|
||
hour=0, minute=0, second=0, microsecond=0
|
||
)
|
||
parsed = vevents(unfold(text))
|
||
moved: dict[str, set[datetime]] = {}
|
||
for event in parsed:
|
||
uid = _first(event, "UID")
|
||
if uid is not None and "RECURRENCE-ID" in event:
|
||
moved.setdefault(uid.value, set()).update(
|
||
_instants(event, "RECURRENCE-ID", zone)
|
||
)
|
||
out: list[Event] = []
|
||
for event in parsed:
|
||
status = _first(event, "STATUS")
|
||
dtstart = _first(event, "DTSTART")
|
||
if dtstart is None or (
|
||
status is not None and status.value.upper() == "CANCELLED"
|
||
):
|
||
continue
|
||
parsed_start = moment(dtstart, dtstart.value, zone)
|
||
if parsed_start is None:
|
||
continue
|
||
start, all_day = parsed_start
|
||
rrule = _first(event, "RRULE")
|
||
if rrule is None or "RECURRENCE-ID" in event:
|
||
starts = [start]
|
||
skip: set[datetime] = set()
|
||
routine = False
|
||
else:
|
||
rule = _rule(rrule.value)
|
||
starts = occurrences(start, rule, zone, end=end)
|
||
routine = is_routine(rule)
|
||
uid = _first(event, "UID")
|
||
skip = _instants(event, "EXDATE", zone) | moved.get(
|
||
uid.value if uid else "", set()
|
||
)
|
||
title = _flat(_text(event, "SUMMARY"), TITLE_MAX) or NO_TITLE
|
||
place = place_of(event)
|
||
for moving in starts:
|
||
floor = day_start if all_day else since
|
||
if moving < floor or moving > end or moving.astimezone(UTC) in skip:
|
||
continue
|
||
out.append(
|
||
Event(
|
||
start=moving,
|
||
title=title,
|
||
place=place,
|
||
all_day=all_day,
|
||
routine=routine,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _text(event: Vevent, name: str) -> str:
|
||
prop = _first(event, name)
|
||
return prop.value if prop is not None else ""
|