📅 Calendar file generator for triathlonlive.tv upcoming events
triathlon-live-calendar.fly.dev
Adds logger
+43
-9
+4
-2
triathlon_live_calendar/__main__.py
+4
-2
triathlon_live_calendar/__main__.py
···
7
7
from uvicorn import run # type: ignore
8
8
9
9
from triathlon_live_calendar.calendar import calendar
10
+
from triathlon_live_calendar.logger import Logger
10
11
11
12
12
13
DEFAULT_HOST = "0.0.0.0"
···
45
46
46
47
47
48
@app.command()
48
-
def generate(path: Path):
49
+
def generate(path: Path, verbose: bool = False):
49
50
"""Generates the calendar .ics file"""
50
-
contents = asyncio.run(calendar())
51
+
logger = Logger(use_typer_echo=True) if verbose else None
52
+
contents = asyncio.run(calendar(logger))
51
53
path.write_text(str(contents))
52
54
53
55
+4
-2
triathlon_live_calendar/calendar.py
+4
-2
triathlon_live_calendar/calendar.py
···
1
1
from asyncio import gather
2
+
from typing import Optional
2
3
3
4
from httpx import AsyncClient
4
5
from ics import Calendar # type: ignore
5
6
6
7
from triathlon_live_calendar.scraper import event_from, event_urls
8
+
from triathlon_live_calendar.logger import Logger
7
9
8
10
9
-
async def calendar() -> Calendar:
11
+
async def calendar(logger: Optional[Logger]) -> Calendar:
10
12
async with AsyncClient() as client:
11
13
urls = await event_urls(client)
12
-
requests = tuple(event_from(client, url) for url in urls)
14
+
requests = tuple(event_from(client, url, logger) for url in urls)
13
15
events = await gather(*requests)
14
16
return Calendar(events=set(events))
+19
triathlon_live_calendar/logger.py
+19
triathlon_live_calendar/logger.py
···
1
+
import logging
2
+
from dataclasses import dataclass
3
+
from typing import Iterable, Union
4
+
5
+
from typer import echo
6
+
7
+
8
+
@dataclass
9
+
class Logger:
10
+
use_typer_echo: bool = False
11
+
12
+
def info(self, text: Union[str, Iterable[str]]) -> None:
13
+
if not isinstance(text, str):
14
+
text = "\n".join(text)
15
+
16
+
if self.use_typer_echo:
17
+
echo(text)
18
+
else:
19
+
logging.info(text)
+13
-4
triathlon_live_calendar/scraper.py
+13
-4
triathlon_live_calendar/scraper.py
···
1
1
from datetime import timedelta
2
2
from hashlib import md5
3
3
from re import compile, findall
4
-
from typing import Tuple
4
+
from typing import Optional, Tuple
5
5
6
6
from arrow import get # type: ignore
7
7
from httpx import AsyncClient
8
8
from ics import Event # type: ignore
9
9
from pyquery import PyQuery # type: ignore
10
+
11
+
from triathlon_live_calendar.logger import Logger
10
12
11
13
12
14
BASE_URL = "https://www.triathlonlive.tv/upcoming-live-streams"
···
22
24
return tuple(str(url) for url in urls if url)
23
25
24
26
25
-
async def event_from(client: AsyncClient, url: str) -> Event:
27
+
async def event_from(client: AsyncClient, url: str, logger: Optional[Logger]) -> Event:
26
28
response = await client.get(url)
27
29
dom = PyQuery(response.content)
28
30
title, *_ = dom("h1 strong")
29
31
begin, *_ = findall(DATETIME_REGEX, str(response.content))
32
+
33
+
begin = get(begin, DATETIME_FORMAT)
34
+
title = title.text.strip()
35
+
36
+
if logger:
37
+
logger.info((f"Parsed {url}", f" Title: {title}", f" Begin: {begin}"))
38
+
30
39
return Event(
31
-
name=title.text,
32
-
begin=get(begin, DATETIME_FORMAT),
40
+
name=title,
41
+
begin=begin,
33
42
duration=DEFAULT_DURATION,
34
43
url=url,
35
44
uid=md5(url.encode("utf-8")).hexdigest(),
+3
-1
triathlon_live_calendar/server.py
+3
-1
triathlon_live_calendar/server.py
···
3
3
4
4
from triathlon_live_calendar.cache import Cache
5
5
from triathlon_live_calendar.calendar import calendar
6
+
from triathlon_live_calendar.logger import Logger
6
7
7
8
8
9
DEFAULT_HEADERS = {
···
13
14
14
15
app = FastAPI()
15
16
cache = Cache()
17
+
logger = Logger()
16
18
17
19
18
20
@app.get("/", response_class=PlainTextResponse)
···
23
25
if cached:
24
26
return cached
25
27
26
-
contents = str(await calendar())
28
+
contents = str(await calendar(logger))
27
29
cache.save(contents)
28
30
return contents