diff --git a/app.py b/app.py index f40721f..b2c83c1 100644 --- a/app.py +++ b/app.py @@ -7,6 +7,7 @@ from flask import send_from_directory from calendar_connector.calendar_converter import CalendarConverter from calendar_connector.data_decoder import decode_data +from calendar_connector.sporteasy_connector import SporteasyConnector logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", @@ -21,9 +22,9 @@ def _list_teams_response() -> str: username, password, _ = decode_data() - calendar_converter = CalendarConverter() - calendar_converter.login(username, password) - teams = calendar_converter.list_teams() + connector = SporteasyConnector() + connector.login(username, password) + teams = connector.list_teams() return json.dumps(teams) @@ -69,7 +70,7 @@ def serve_static_index() -> flask.Response: # Redirect user to /api if data is passed to keep compatibility redirect_route = flask.url_for("main_request_api_handler") redirect_url = f"{redirect_route}?data={data}" - return flask.redirect(redirect_url) + return flask.redirect(redirect_url) # type: ignore return send_from_directory("web-app/build", "index.html") diff --git a/calendar_connector/calendar_converter.py b/calendar_connector/calendar_converter.py index 38cf903..9211345 100644 --- a/calendar_connector/calendar_converter.py +++ b/calendar_connector/calendar_converter.py @@ -1,225 +1,24 @@ -from datetime import datetime, timedelta -from typing import Any, cast, Literal - -import requests -from icalendar import Calendar, Event, vText - -from calendar_connector.consts import ( - EVENT_TYPE, - TIMEZONE, - MY_PRESENCE, - ORDER_PRESENT, - url_authenticate, - url_list_teams, - url_list_events, -) +from icalendar import Calendar, vText from calendar_connector.datetime_utils import ( - get_current_timestamp, get_current_datetime, ) from calendar_connector.env import load_env_data -from calendar_connector.normalize import normalize - - -def _extract_event_dates(event_data: EVENT_TYPE, event: Event) -> None: - start_date = datetime.fromisoformat(cast(str, event_data["start_at"])) - start_date = start_date.astimezone(TIMEZONE) - end_date_str = cast(str | None, event_data["end_at"]) - if end_date_str is not None: - end_date = datetime.fromisoformat(end_date_str) - else: - end_date = start_date + timedelta(hours=2) - end_date = end_date.astimezone(TIMEZONE) - - event.add("dtstart", start_date) - event.add("dtstamp", start_date) - event.add("dtend", end_date) - - -def _extract_event_location(event_data: EVENT_TYPE, event: Event) -> None: - if event_data is None or event_data["location"] is None: - return - - location = normalize( - cast( - dict[str, str | Any], - event_data["location"], - )["formatted_address"] - ) - event.add("location", location) - - -def _status_to_ics_status(sp_status: str) -> str: - sp_status = sp_status.upper() - if sp_status == "CONFIRMED": - return "CONFIRMED" - if sp_status == "CANCELLED": - return "CANCELLED" - if sp_status == "NEEDS-ACTION": - return "TENTATIVE" - return sp_status - - -def _extract_event_summary( - event_data: EVENT_TYPE, event: Event, team_name: str -) -> None: - name = normalize(cast(str, event_data["name"])) - if team_name not in name: - summary = f"{team_name} - {name}" - else: - summary = name - if "is_cancelled" in event_data and event_data["is_cancelled"]: - summary = f"CANCELLED | {summary}" - - # opponent_right = cast(dict[str, int | str | None] | None, event_data["opponent_right"]) - # opponent_left = cast(dict[str, int | str | None] | None, event_data["opponent_left"]) - # if opponent_left is not None and opponent_right is not None: - # if opponent_left["id"] == team_id: - # opponent_name = opponent_right["short_name"] - # else: - # opponent_name = opponent_left["short_name"] - # summary += f" - contre {opponent_name}" - - me_object = event_data.get("me", {}) - if me_object is not None and me_object.get("group") is not None: - group = cast(dict[str, str], me_object.get("group", {})) - localized_name_presence = group.get("localized_name") - if localized_name_presence is not None: - summary += f" - {normalize(localized_name_presence)}" - - slug_presence = group.get("slug_name") - presence = MY_PRESENCE.get(slug_presence) - event.add("status", _status_to_ics_status(presence)) - - event.add("summary", summary) - - -def _extract_attendee_description(event_data: EVENT_TYPE) -> str: - attendance_groups = cast( - list[dict[str, int | str]] | None, event_data["attendance_groups"] - ) - attendee = "" - if attendance_groups is not None: - attendance_group_list: list[tuple[int, str, int]] = [] - for ppc in attendance_groups: - slug_sort_value = ORDER_PRESENT.get(cast(str, ppc["slug_name"]), 0) - attendance_group_list.append( - ( - slug_sort_value, - cast(str, ppc["localized_name"]), - cast(int, ppc["count"]), - ) - ) - attendance_group_list.sort(reverse=True) +from calendar_connector.event_convertor import event_to_calendar_event +from calendar_connector.sporteasy_connector import SporteasyConnector - attendee = ", ".join( - [ - f"{localized_name}: {count}" - for _, localized_name, count in attendance_group_list - ] - ) - return attendee - - -def _extract_scores_for_opponent( - event_data: EVENT_TYPE, left_or_right: Literal["left", "right"] -) -> tuple[str, int] | None: - opponent = event_data.get(f"opponent_{left_or_right}") - if opponent is None: - return None - score_str = opponent.get("score") - if score_str is None: - return None - name = cast(str, opponent.get("short_name")) - score = int(cast(str, score_str)) - return name, score - - -def _extract_scores(event_data: EVENT_TYPE) -> str | None: - left = _extract_scores_for_opponent(event_data, "left") - right = _extract_scores_for_opponent(event_data, "right") - - if left is None or right is None: - return None - - left_name, left_score = left - right_name, right_score = right - return f"{left_name} {left_score} - {right_score} {right_name}" - - -def _extract_event_description(event_data: EVENT_TYPE, event: Event) -> None: - description = "" - score = _extract_scores(event_data) - if score is not None: - description += f"{score}\n" - attendee = _extract_attendee_description(event_data) - description += attendee - - event.add("description", description.strip()) - - -def event_to_calendar_event(team_name: str, event_data: EVENT_TYPE) -> Event: - event = Event() - event.add("uid", str(event_data["id"]) + f"@sporteasy.net") - _extract_event_location(event_data, event) - _extract_event_dates(event_data, event) - _extract_event_summary(event_data, event, team_name) - _extract_event_description(event_data, event) - - event.add("class", "PUBLIC") - current_timestamp = get_current_timestamp() - event.add("sequence", current_timestamp) - event.add("transp", "OPAQUE") - - # todo: change this if possible - event.add("created", datetime(2020, 1, 1, 1, 1, 1)) - event.add("last-modified", datetime(2020, 1, 1, 1, 1, 1)) - - category_name = normalize( - cast(dict[str, str | Any], event_data["category"])["localized_name"] - ) - - return event class CalendarConverter: def __init__(self) -> None: - self.session_requests = requests.Session() - - def login(self, username: str, password: str) -> str: - authenticate_response = self.session_requests.post( - url_authenticate, - { - "username": username, - "password": password, - }, - ) - if authenticate_response.status_code != 200: - raise Exception("Authentication error") - token: str = authenticate_response.cookies.get("sporteasy") - # csrf = authenticate_response.cookies.get(csrf_name) - return token - - def list_teams(self) -> list[tuple[int, str]]: - response = self.session_requests.get(url_list_teams) - data = response.json() - return [(d["id"], normalize(d["name"])) for d in data["results"]] - - def list_events(self, team_id: int) -> list[EVENT_TYPE]: - response = self.session_requests.get( - url_list_events.format(team_id=team_id), - headers={"Accept-Language": "fr-FR"}, - ) - data: list[EVENT_TYPE] = response.json()["results"] - return data + self.connector = SporteasyConnector() def get_calendar_text( self, username: str, password: str, team_id: str | None = None ) -> str: - self.login(username, password) - teams = self.list_teams() + self.connector.login(username, password) + teams = self.connector.list_teams() cal = Calendar() cal.add("summary", vText("SportEasyCalendar")) @@ -244,7 +43,7 @@ def get_calendar_text( and int(team_id) != current_team_id ): continue - events = self.list_events(current_team_id) + events = self.connector.list_events(current_team_id) for event in events: cal.add_component(event_to_calendar_event(team_name, event)) diff --git a/calendar_connector/env.py b/calendar_connector/env.py index abae616..f9e5fc7 100644 --- a/calendar_connector/env.py +++ b/calendar_connector/env.py @@ -1,9 +1,11 @@ +from typing import Optional + from dotenv import dotenv_values env = dotenv_values(".env") -def load_env_data() -> tuple[str, str, str]: +def load_env_data() -> tuple[str, str, Optional[str]]: username = env.get("username") password = env.get("password") team_id = env.get("team_id") diff --git a/calendar_connector/event_convertor.py b/calendar_connector/event_convertor.py new file mode 100644 index 0000000..2c2b396 --- /dev/null +++ b/calendar_connector/event_convertor.py @@ -0,0 +1,36 @@ +from datetime import datetime +from typing import Any, cast + +from icalendar import Event + +from calendar_connector.consts import EVENT_TYPE +from calendar_connector.datetime_utils import get_current_timestamp +from calendar_connector.event_utils.date import extract_event_dates +from calendar_connector.event_utils.description import extract_event_description +from calendar_connector.event_utils.location import extract_event_location +from calendar_connector.event_utils.summary import extract_event_summary +from calendar_connector.normalize import normalize + + +def event_to_calendar_event(team_name: str, event_data: EVENT_TYPE) -> Event: + event = Event() + event.add("uid", str(event_data["id"]) + f"@sporteasy.net") + extract_event_location(event_data, event) + extract_event_dates(event_data, event) + extract_event_summary(event_data, event, team_name) + extract_event_description(event_data, event) + + event.add("class", "PUBLIC") + current_timestamp = get_current_timestamp() + event.add("sequence", current_timestamp) + event.add("transp", "OPAQUE") + + # todo: change this if possible + event.add("created", datetime(2020, 1, 1, 1, 1, 1)) + event.add("last-modified", datetime(2020, 1, 1, 1, 1, 1)) + + category_name = normalize( + cast(dict[str, str | Any], event_data["category"])["localized_name"] + ) + + return event diff --git a/calendar_connector/event_utils/date.py b/calendar_connector/event_utils/date.py new file mode 100644 index 0000000..9c88861 --- /dev/null +++ b/calendar_connector/event_utils/date.py @@ -0,0 +1,21 @@ +from datetime import datetime, timedelta +from typing import cast + +from icalendar import Event + +from calendar_connector.consts import EVENT_TYPE, TIMEZONE + + +def extract_event_dates(event_data: EVENT_TYPE, event: Event) -> None: + start_date = datetime.fromisoformat(cast(str, event_data["start_at"])) + start_date = start_date.astimezone(TIMEZONE) + end_date_str = cast(str | None, event_data["end_at"]) + if end_date_str is not None: + end_date = datetime.fromisoformat(end_date_str) + else: + end_date = start_date + timedelta(hours=2) + end_date = end_date.astimezone(TIMEZONE) + + event.add("dtstart", start_date) + event.add("dtstamp", start_date) + event.add("dtend", end_date) diff --git a/calendar_connector/event_utils/description.py b/calendar_connector/event_utils/description.py new file mode 100644 index 0000000..eb550f8 --- /dev/null +++ b/calendar_connector/event_utils/description.py @@ -0,0 +1,45 @@ +from typing import cast + +from icalendar import Event + +from calendar_connector.consts import EVENT_TYPE, ORDER_PRESENT +from calendar_connector.event_utils.score import extract_scores + + +def _extract_attendee_description(event_data: EVENT_TYPE) -> str: + attendance_groups = cast( + list[dict[str, int | str]] | None, event_data["attendance_groups"] + ) + attendee = "" + if attendance_groups is not None: + attendance_group_list: list[tuple[int, str, int]] = [] + for ppc in attendance_groups: + slug_sort_value = ORDER_PRESENT.get(cast(str, ppc["slug_name"]), 0) + attendance_group_list.append( + ( + slug_sort_value, + cast(str, ppc["localized_name"]), + cast(int, ppc["count"]), + ) + ) + attendance_group_list.sort(reverse=True) + + attendee = ", ".join( + [ + f"{localized_name}: {count}" + for _, localized_name, count in attendance_group_list + ] + ) + + return attendee + + +def extract_event_description(event_data: EVENT_TYPE, event: Event) -> None: + description = "" + score = extract_scores(event_data) + if score is not None: + description += f"{score}\n" + attendee = _extract_attendee_description(event_data) + description += attendee + + event.add("description", description.strip()) diff --git a/calendar_connector/event_utils/location.py b/calendar_connector/event_utils/location.py new file mode 100644 index 0000000..9b7a51f --- /dev/null +++ b/calendar_connector/event_utils/location.py @@ -0,0 +1,19 @@ +from typing import cast, Any + +from icalendar import Event + +from calendar_connector.consts import EVENT_TYPE +from calendar_connector.normalize import normalize + + +def extract_event_location(event_data: EVENT_TYPE, event: Event) -> None: + if event_data is None or event_data["location"] is None: + return + + location = normalize( + cast( + dict[str, str | Any], + event_data["location"], + )["formatted_address"] + ) + event.add("location", location) diff --git a/calendar_connector/event_utils/score.py b/calendar_connector/event_utils/score.py new file mode 100644 index 0000000..d3690eb --- /dev/null +++ b/calendar_connector/event_utils/score.py @@ -0,0 +1,30 @@ +from typing import Literal, cast, Any + +from calendar_connector.consts import EVENT_TYPE + + +def _extract_scores_for_opponent( + event_data: EVENT_TYPE, left_or_right: Literal["left", "right"] +) -> tuple[str, int] | None: + opponent = event_data.get(f"opponent_{left_or_right}") + if opponent is None: + return None + opponent = cast(dict[str, Any], opponent) + score_str = opponent.get("score") + if score_str is None: + return None + name = cast(str, opponent.get("short_name")) + score = int(cast(str, score_str)) + return name, score + + +def extract_scores(event_data: EVENT_TYPE) -> str | None: + left = _extract_scores_for_opponent(event_data, "left") + right = _extract_scores_for_opponent(event_data, "right") + + if left is None or right is None: + return None + + left_name, left_score = left + right_name, right_score = right + return f"{left_name} {left_score} - {right_score} {right_name}" diff --git a/calendar_connector/event_utils/summary.py b/calendar_connector/event_utils/summary.py new file mode 100644 index 0000000..66bd35a --- /dev/null +++ b/calendar_connector/event_utils/summary.py @@ -0,0 +1,55 @@ +from typing import cast, Any + +from icalendar import Event + +from calendar_connector.consts import EVENT_TYPE, MY_PRESENCE +from calendar_connector.exceptions import AttributeNotFoundException +from calendar_connector.normalize import normalize + + +def _status_to_ics_status(sp_status: str) -> str: + sp_status = sp_status.upper() + if sp_status == "CONFIRMED": + return "CONFIRMED" + if sp_status == "CANCELLED": + return "CANCELLED" + if sp_status == "NEEDS-ACTION": + return "TENTATIVE" + return sp_status + + +def extract_event_summary(event_data: EVENT_TYPE, event: Event, team_name: str) -> None: + name = normalize(cast(str, event_data["name"])) + if team_name not in name: + summary = f"{team_name} - {name}" + else: + summary = name + if "is_cancelled" in event_data and event_data["is_cancelled"]: + summary = f"CANCELLED | {summary}" + + # opponent_right = cast(dict[str, int | str | None] | None, event_data["opponent_right"]) + # opponent_left = cast(dict[str, int | str | None] | None, event_data["opponent_left"]) + # if opponent_left is not None and opponent_right is not None: + # if opponent_left["id"] == team_id: + # opponent_name = opponent_right["short_name"] + # else: + # opponent_name = opponent_left["short_name"] + # summary += f" - contre {opponent_name}" + + me_object = event_data.get("me", {}) + me_object = cast(dict[str, Any], me_object) + if me_object is not None and me_object.get("group") is not None: + group = cast(dict[str, str], me_object.get("group", {})) + localized_name_presence = group.get("localized_name") + if localized_name_presence is not None: + summary += f" - {normalize(localized_name_presence)}" + + slug_presence = group.get("slug_name") + if slug_presence is None: + raise AttributeNotFoundException("slug_name") + presence = MY_PRESENCE.get(slug_presence) + if presence is None: + raise AttributeNotFoundException("slug_presence") + event.add("status", _status_to_ics_status(presence)) + + event.add("summary", summary) diff --git a/calendar_connector/exceptions.py b/calendar_connector/exceptions.py new file mode 100644 index 0000000..f73db06 --- /dev/null +++ b/calendar_connector/exceptions.py @@ -0,0 +1,3 @@ +class AttributeNotFoundException(Exception): + def __init__(self, name: str) -> None: + super().__init__(f"{name} is not found") diff --git a/calendar_connector/sporteasy_connector.py b/calendar_connector/sporteasy_connector.py new file mode 100644 index 0000000..605481f --- /dev/null +++ b/calendar_connector/sporteasy_connector.py @@ -0,0 +1,41 @@ +import requests + +from calendar_connector.consts import ( + url_authenticate, + url_list_teams, + EVENT_TYPE, + url_list_events, +) +from calendar_connector.normalize import normalize + + +class SporteasyConnector: + def __init__(self) -> None: + self.session_requests = requests.Session() + + def login(self, username: str, password: str) -> str: + authenticate_response = self.session_requests.post( + url_authenticate, + { + "username": username, + "password": password, + }, + ) + if authenticate_response.status_code != 200: + raise Exception("Authentication error") + token: str = authenticate_response.cookies.get("sporteasy") + # csrf = authenticate_response.cookies.get(csrf_name) + return token + + def list_teams(self) -> list[tuple[int, str]]: + response = self.session_requests.get(url_list_teams) + data = response.json() + return [(d["id"], normalize(d["name"])) for d in data["results"]] + + def list_events(self, team_id: int) -> list[EVENT_TYPE]: + response = self.session_requests.get( + url_list_events.format(team_id=team_id), + headers={"Accept-Language": "fr-FR"}, + ) + data: list[EVENT_TYPE] = response.json()["results"] + return data diff --git a/list_teams.py b/list_teams.py index d19af23..b90f611 100644 --- a/list_teams.py +++ b/list_teams.py @@ -1,11 +1,11 @@ from calendar_connector.calendar_converter import CalendarConverter, load_env_data - +from calendar_connector.sporteasy_connector import SporteasyConnector username, password, _ = load_env_data() -calendar_converter = CalendarConverter() -calendar_converter.login(username, password) +connector = SporteasyConnector() +connector.login(username, password) -teams = calendar_converter.list_teams() +teams = connector.list_teams() team_list = [ ("TEAM NAME", "TEAM ID"), diff --git a/test/test_get_calendar_text.py b/test/test_get_calendar_text.py index 7b5c4a0..6b6408f 100644 --- a/test/test_get_calendar_text.py +++ b/test/test_get_calendar_text.py @@ -4,6 +4,7 @@ import requests_mock +import calendar_connector.event_convertor import calendar_connector.calendar_converter from calendar_connector.consts import url_list_events, url_list_teams, url_authenticate from .test_utils import ( @@ -24,6 +25,7 @@ def test_get_calendar_text( datetime_mock: MagicMock, ) -> None: importlib.reload(calendar_connector.calendar_converter) + importlib.reload(calendar_connector.event_convertor) mocked_response_teams = read_text_by_name("list_teams.json") mocked_response_events = read_text_by_name("list_events.json") diff --git a/test/test_list_teams.py b/test/test_list_teams.py index 4935494..2d8cb6a 100644 --- a/test/test_list_teams.py +++ b/test/test_list_teams.py @@ -1,5 +1,6 @@ import requests_mock +from calendar_connector.sporteasy_connector import SporteasyConnector from test.test_utils import read_text_by_name from calendar_connector.calendar_converter import CalendarConverter from calendar_connector.consts import url_list_teams @@ -10,8 +11,8 @@ def test_list_teams() -> None: with requests_mock.Mocker() as m: m.get(url_list_teams, text=mocked_response) - converter = CalendarConverter() - result = converter.list_teams() + connector = SporteasyConnector() + result = connector.list_teams() assert result == [ (1, "Equipe 1"), diff --git a/test/test_login.py b/test/test_login.py index 482381d..ffc84a3 100644 --- a/test/test_login.py +++ b/test/test_login.py @@ -2,13 +2,14 @@ from calendar_connector.calendar_converter import CalendarConverter from calendar_connector.consts import url_authenticate +from calendar_connector.sporteasy_connector import SporteasyConnector def test_login() -> None: with requests_mock.Mocker() as m: m.post(url_authenticate, status_code=200, cookies={"sporteasy": "token test"}) - converter = CalendarConverter() - token = converter.login("username", "password") + connector = SporteasyConnector() + token = connector.login("username", "password") assert token == "token test" diff --git a/test/test_utils.py b/test/test_utils.py index 776d53d..b435f22 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -16,7 +16,7 @@ ("phrase normale", "phrase normale"), ], ) -def test_normalize(test_input, expected) -> None: +def test_normalize(test_input: str, expected: str) -> None: result = normalize(test_input) assert expected == result