ruff check
This commit is contained in:
@@ -27,3 +27,29 @@ include = ["*"]
|
||||
|
||||
[project.scripts]
|
||||
pvv-calendar-bot = "pvv_calendar_bot:main"
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = [
|
||||
"A", # flake8-builtins
|
||||
"B", # flake8-bugbear
|
||||
"C4", # flake8-comprehensions
|
||||
"COM", # flake8-commas
|
||||
"ANN",
|
||||
"E", # pycodestyle
|
||||
"F", # Pyflakes
|
||||
"FA", # flake8-future-annotations
|
||||
"I", # isort
|
||||
"S", # flake8-bandit
|
||||
"ICN", # flake8-import-conventions
|
||||
"ISC", # flake8-implicit-str-concat
|
||||
# "N", # pep8-naming
|
||||
"PTH", # flake8-use-pathlib
|
||||
"RET", # flake8-return
|
||||
"SIM", # flake8-simplify
|
||||
"TC", # flake8-type-checking
|
||||
"UP", # pyupgrade
|
||||
"YTT", # flake8-2020
|
||||
]
|
||||
|
||||
@@ -1 +1,3 @@
|
||||
from pvv_calendar_bot.main import main
|
||||
|
||||
__all__ = ["main"]
|
||||
|
||||
@@ -4,22 +4,22 @@ import datetime
|
||||
class Event:
|
||||
def __init__(
|
||||
self,
|
||||
id: int,
|
||||
id_: int,
|
||||
start: datetime.datetime,
|
||||
name: str,
|
||||
location: str,
|
||||
organizer: str,
|
||||
description: str,
|
||||
):
|
||||
self.id = id
|
||||
) -> None:
|
||||
self.id = id_
|
||||
self.start = start
|
||||
self.name = name
|
||||
self.location = location
|
||||
self.organizer = organizer
|
||||
self.description = description
|
||||
|
||||
def __str__(self):
|
||||
def __str__(self) -> str:
|
||||
return f"{self.name} ({self.id})"
|
||||
|
||||
def __repr__(self):
|
||||
def __repr__(self) -> str:
|
||||
return f"Event({self.id}, {self.start}, {self.name})"
|
||||
|
||||
@@ -3,25 +3,28 @@
|
||||
# from .event import Event
|
||||
# from .scraping import get_soup, process_soup, get_events_today
|
||||
|
||||
import os
|
||||
from nio import AsyncClient
|
||||
from textwrap import dedent
|
||||
from markdown2 import Markdown
|
||||
import asyncio
|
||||
from .sql_connector import fetch_events, Event
|
||||
import os
|
||||
from textwrap import dedent
|
||||
|
||||
from markdown2 import Markdown
|
||||
from nio import AsyncClient
|
||||
|
||||
from .sql_connector import Event, fetch_events
|
||||
|
||||
MATRIX_URL = os.environ.get("MATRIX_URL", "https://matrix.pvv.ntnu.no").strip()
|
||||
MATRIX_USER = os.environ.get("MATRIX_USER", "@bot_calendar:pvv.ntnu.no").strip()
|
||||
MATRIX_TOKEN = os.environ.get("MATRIX_TOKEN").strip()
|
||||
ANNOUNCEMENT_CHANNEL = os.environ.get(
|
||||
"ANNOUNCEMENT_CHANNEL", "!announcements:pvv.ntnu.no"
|
||||
"ANNOUNCEMENT_CHANNEL",
|
||||
"!announcements:pvv.ntnu.no",
|
||||
).strip()
|
||||
|
||||
client = None
|
||||
|
||||
|
||||
def create_announcement(event: Event, atEveryone: bool) -> str:
|
||||
url = "https://www.pvv.ntnu.no/hendelser/info.php?id={}".format(event.id)
|
||||
url = f"https://www.pvv.ntnu.no/hendelser/info.php?id={event.id}"
|
||||
msgText = dedent("""\
|
||||
## Dagens arrangement / Event of the Day: "{name}"
|
||||
- 🕒 **{start}**
|
||||
@@ -45,7 +48,9 @@ def create_announcement(event: Event, atEveryone: bool) -> str:
|
||||
|
||||
|
||||
async def sendMatrixAnnouncement(
|
||||
event: Event, channel: str = ANNOUNCEMENT_CHANNEL, atEveryone: bool = False
|
||||
event: Event,
|
||||
channel: str = ANNOUNCEMENT_CHANNEL,
|
||||
atEveryone: bool = False,
|
||||
) -> None:
|
||||
msgText = create_announcement(event, atEveryone)
|
||||
return await client.room_send(
|
||||
@@ -73,7 +78,7 @@ async def sendCalendarEvents() -> None:
|
||||
await client.close()
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
asyncio.run(sendCalendarEvents())
|
||||
|
||||
|
||||
|
||||
@@ -1,78 +1,69 @@
|
||||
from .event import Event
|
||||
|
||||
from typing import List
|
||||
from bs4 import BeautifulSoup
|
||||
import requests
|
||||
from operator import add
|
||||
from functools import reduce
|
||||
import datetime
|
||||
from functools import reduce
|
||||
from operator import add
|
||||
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
from .event import Event
|
||||
|
||||
|
||||
def get_soup() -> BeautifulSoup:
|
||||
r = requests.get("http://www.pvv.ntnu.no/hendelser/")
|
||||
soup = BeautifulSoup(r.text, "html.parser")
|
||||
return soup
|
||||
r = requests.get("http://www.pvv.ntnu.no/hendelser/", timeout=300)
|
||||
return BeautifulSoup(r.text, "html.parser")
|
||||
|
||||
|
||||
def process_soup(soup: BeautifulSoup) -> List[Event]:
|
||||
def process_soup(soup: BeautifulSoup) -> list[Event]:
|
||||
zips = []
|
||||
events = soup.find_all("ul", "events")
|
||||
|
||||
for event in events:
|
||||
times, locations, organizers = zip(
|
||||
*(
|
||||
list(
|
||||
map(
|
||||
lambda x: map(lambda y: y.find("strong").text, x),
|
||||
filter(
|
||||
lambda x: x != [],
|
||||
map(lambda x: x.find_all("li"), event.find_all("li")),
|
||||
),
|
||||
[
|
||||
(y.find("strong").text for y in x)
|
||||
for x in filter(
|
||||
lambda x: x != [],
|
||||
(x.find_all("li") for x in event.find_all("li")),
|
||||
)
|
||||
)
|
||||
)
|
||||
]
|
||||
),
|
||||
strict=False,
|
||||
)
|
||||
times = list(
|
||||
map(
|
||||
lambda x: datetime.datetime.strptime(x, "%A %d. %b %H.%M").replace(
|
||||
year=datetime.datetime.now().year
|
||||
),
|
||||
times,
|
||||
times = [
|
||||
datetime.datetime.strptime(x, "%A %d. %b %H.%M").replace(
|
||||
year=datetime.datetime.now().year,
|
||||
)
|
||||
)
|
||||
names = list(
|
||||
map(
|
||||
lambda x: x[0].text,
|
||||
filter(
|
||||
lambda x: x != [],
|
||||
map(lambda x: x.find_all("a"), event.find_all("li")),
|
||||
),
|
||||
for x in times
|
||||
]
|
||||
names = [
|
||||
x[0].text
|
||||
for x in filter(
|
||||
lambda x: x != [],
|
||||
(x.find_all("a") for x in event.find_all("li")),
|
||||
)
|
||||
)
|
||||
descriptions = list(
|
||||
map(
|
||||
lambda x: x.text,
|
||||
filter(
|
||||
lambda x: x is not None,
|
||||
map(lambda x: x.find("p"), event.find_all("li")),
|
||||
),
|
||||
]
|
||||
descriptions = [
|
||||
x.text
|
||||
for x in filter(
|
||||
lambda x: x is not None,
|
||||
(x.find("p") for x in event.find_all("li")),
|
||||
)
|
||||
)
|
||||
]
|
||||
ids = []
|
||||
for a in event.find_all("a", href=True):
|
||||
if a["href"][:10] == "/hendelser":
|
||||
ids.append(int(a["href"][a["href"].find("=") + 1 :]))
|
||||
|
||||
zips.append(list(zip(ids, times, names, organizers, locations, descriptions)))
|
||||
zips.append(list(zip(ids, times, names, organizers, locations, descriptions, strict=False)))
|
||||
|
||||
events = reduce(add, zips)
|
||||
events = list(map(lambda x: Event(*x), events))
|
||||
return events
|
||||
return [Event(*x) for x in events]
|
||||
|
||||
|
||||
def get_events_today(events: List[Event]) -> List[Event]:
|
||||
def get_events_today(events: list[Event]) -> list[Event]:
|
||||
return list(
|
||||
filter(lambda e: e.start.date() == datetime.datetime.today().date(), events)
|
||||
filter(lambda e: e.start.date() == datetime.datetime.today().date(), events),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
from dataclasses import dataclass
|
||||
import mysql.connector
|
||||
from pprint import pprint
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pprint import pprint
|
||||
|
||||
import mysql.connector
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -20,7 +21,7 @@ class Event:
|
||||
organiser: str
|
||||
|
||||
|
||||
def fetch_events():
|
||||
def fetch_events() -> list[Event]:
|
||||
mydb = mysql.connector.connect(
|
||||
host=os.environ.get("MYSQL_URL", "mysql.pvv.ntnu.no").strip(),
|
||||
user=os.environ.get("MYSQL_USER", "calendar-bot").strip(),
|
||||
@@ -48,7 +49,7 @@ def fetch_events():
|
||||
location=row[4],
|
||||
description=row[5],
|
||||
organiser=row[6],
|
||||
)
|
||||
),
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
Reference in New Issue
Block a user