myice/myice/myice.py

#!/usr/bin/env python3
"""
Tool to work with My Ice Hockey schedules
"""

import configparser
import datetime
import json
import os
import re
import sys
from enum import Enum
from pathlib import Path
from typing import Annotated
from typing import List, Tuple
from rl_ai_tools import utils  # type: ignore
import requests
import typer

user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:131.0) Gecko/20100101 Firefox/131.0"

app = typer.Typer(no_args_is_help=True)
session: requests.Session
userid: int


class AgeGroup(str, Enum):
    u13e = "U13 (Elite)"
    u13t = "U13 (Top)"
    u13a = "U13 (A)"
    u13p = "U13 (Prép)"
    u15t = "U15 (Top)"
    u15a = "U15 (A)"
    u111 = "U11 (1)"


class EventType(str, Enum):
    game = "game"
    practice = "practice"


def load_cookies(file: str = "cookies.txt") -> requests.cookies.RequestsCookieJar:
    cookie_jar_file = Path(file)
    cj_dict = {}
    if cookie_jar_file.exists():
        with cookie_jar_file.open("rb") as f:
            cj_dict = json.load(f)
    return requests.cookies.cookiejar_from_dict(cj_dict)


def save_cookies(file: str = "cookies.txt"):
    cookie_jar_file = Path(file)
    with cookie_jar_file.open("w") as f:
        f.write(json.dumps(requests.utils.dict_from_cookiejar(session.cookies)))


def get_login(local_file: str = "myice.ini") -> tuple[str, str, int]:
    config = configparser.ConfigParser()
    config.read(
        [
            Path("~/.config/myice.ini").expanduser(),
            "myice.ini",
            local_file,
        ]
    )
    if "default" in config.sections():
        default_config = config["default"]

        username = default_config.get("username")
        password = default_config.get("password")
        userid = default_config.getint("userid")
        if not username or not password:
            print(
                "Error: please configure username/password in ini file", file=sys.stderr
            )
            sys.exit(1)
    else:
        print("Error: please configure username/password in ini file", file=sys.stderr)
        sys.exit(1)
    return username, password, userid


def do_login():
    global session
    global userid
    username, password, userid = get_login()
    r = session.get("https://app.myice.hockey/", headers={"User-Agent": user_agent})
    r.raise_for_status()
    form_data = {
        "login_email": username,
        "login_password": password,
        "sublogin": 1,
        "login_submit": "",
    }
    r = session.post(
        "https://app.myice.hockey/classes/process.php",
        data=form_data,
        headers={
            "Referer": "https://app.myice.hockey/",
            "Origin": "https://app.myice.hockey",
            "User-Agent": user_agent,
        },
    )
    r.raise_for_status()


def get_userid():
    global session
    r = session.get(
        "https://app.myice.hockey/players/clubschedule/",
        headers={
            "User-Agent": user_agent,
            "Referer": "https://app.myice.hockey/classes/process.php",
        },
    )
    r.raise_for_status()
    for line in r.text.splitlines():
        m = re.search(r"^\s+userid: '(?P<userid>[0-9]+)'", line)
        if m:
            userid = m.group("userid")
            break
    return userid


def wrapper_session(func):
    def wrapper(*args, **kwargs):
        global session, userid
        session = requests.Session()
        session.cookies = load_cookies()
        session.cookies.clear_expired_cookies()
        if not session.cookies.get("mih_v3_cookname"):
            print("login...", file=sys.stderr)
            do_login()
            save_cookies()
        _, _, userid = get_login()
        if not userid:
            print("get userid...", file=sys.stderr)
            userid = get_userid()
        return func(*args, **kwargs)

    return wrapper


@wrapper_session
def get_schedule(num_days: int) -> str:
    global session
    global userid
    assert session and userid
    now = datetime.datetime.now()
    date_start = now
    date_end = date_start + datetime.timedelta(days=num_days)
    r = session.post(
        "https://app.myice.hockey/inc/processclubplanning.php",
        data={
            "type": "fetchmy",
            "userid": userid,
            "type_location": ["*"],
            "start": date_start.strftime("%Y-%m-%d"),
            "end": date_end.strftime("%Y-%m-%d"),
        },
        headers={
            "User-Agent": user_agent,
            "Referer": "https://app.myice.hockey/players/clubschedule/",
        },
    )
    r.raise_for_status()
    return r.text


@wrapper_session
def game_pdf(gameid: int, outfile: Path):
    global session, userid
    assert session and userid
    r = session.get(
        f"https://app.myice.hockey/pdfexports/playerplatgamepdfgenerator.php?id={gameid}",
        headers={
            "User-Agent": user_agent,
            "Referer": "https://app.myice.hockey/inc/processclubplanning.php",
        },
    )
    r.raise_for_status()
    with outfile.open("wb") as fd:
        for chunk in r.iter_content(chunk_size=1024 * 1024 * 4):
            fd.write(chunk)


@wrapper_session
def practice_pdf(gameid: int, outfile: Path):
    global session, userid
    assert session and userid
    r = session.get(
        f"https://app.myice.hockey/pdfexports/playerplatpracticepdf.php?practice={gameid}",
        headers={
            "User-Agent": user_agent,
            "Referer": "https://app.myice.hockey/inc/processclubplanning.php",
        },
    )
    r.raise_for_status()
    with outfile.open("wb") as fd:
        for chunk in r.iter_content(chunk_size=1024 * 1024 * 4):
            fd.write(chunk)


@app.command()
def schedule(
    outfile: Annotated[
        Path | None,
        typer.Option(
            "--outfile", "-o", help="file to write result to, or stdout if none"
        ),
    ] = None,
    num_days: Annotated[int, typer.Option("--days")] = 7,
):
    """
    Fetch schedule as json
    """
    schedule = get_schedule(num_days)
    if outfile:
        with outfile.open("w") as f:
            f.write(schedule)
    else:
        print(schedule)


def os_open(file: str) -> None:
    if os.uname().sysname == "Linux":
        os.system(f"xdg-open {file}")
    else:
        print(f"Opening file {file}", file=sys.stderr)
        os.system(f"open {file}")


@app.command("game")
def get_game_pdf(
    game_id: Annotated[int, typer.Argument(help="ID of game to gen pdf for")],
):
    """
    Genate the pdf for the game invitation
    """
    output_filename = f"game_{game_id}.pdf"
    game_pdf(game_id, Path(output_filename))
    os_open(output_filename)


@app.command("practice")
def get_practice_pdf(
    game_id: Annotated[int, typer.Argument(help="ID of practice to gen pdf for")],
):
    """
    Genate the pdf for the practice invitation
    """
    output_filename = f"practice_{game_id}.pdf"
    practice_pdf(game_id, Path(output_filename))
    os_open(output_filename)


@app.command("search")
def parse_schedule(
    age_group: Annotated[AgeGroup | None, typer.Option(...)] = None,
    event_type_filter: Annotated[
        EventType | None,
        typer.Option("--type", help="Only display events of this type"),
    ] = None,
    schedule_file: Annotated[
        Path, typer.Option(help="schedule json file to parse")
    ] = Path("schedule.json"),
):
    """
    Parse schedule.json to look for specific games or practices
    """
    with schedule_file.open("r") as f:
        data = json.load(f)
    # age_group filter
    if age_group:
        events = [x for x in data if x["agegroup"] == age_group]
    else:
        events = [x for x in data if x["agegroup"] in AgeGroup]
    # event_type filter
    if event_type_filter:
        if event_type_filter.value == EventType.game:
            events = [x for x in events if "event" in x and x["event"] == "Jeu"]
        else:
            events = [x for x in events if "event" not in x or x["event"] == "Jeu"]
    for event in events:
        if age_group:
            raw_title = event["title"].removeprefix(age_group + "\n")
        else:
            raw_title = event["title"]
        title = " ".join(raw_title.split("\n"))
        start = datetime.datetime.fromisoformat(event["start"])
        start_fmt = start.strftime("%H:%M")
        end = datetime.datetime.fromisoformat(event["end"])
        end_fmt = end.strftime("%H:%M")
        if "event" in event and event["event"] == "Jeu":
            event_type = "game"
        else:
            event_type = "practice"
        print(
            f"[{event['id_event']}] {event['date']} {event_type} {start_fmt}-> {end_fmt}"
        )
        print(f"{event_type}: {title}\n")


@app.command("ai")
def check_with_ai(
    schedule_file: Annotated[
        Path, typer.Option(help="schedule json file to parse")
    ] = Path("schedule.json"),
):
    """
    Search through the schedule with natural language using Infomaniak LLM API
    """
    if not utils.init():
        sys.exit(1)
    with schedule_file.open("r") as f:
        schedule_data = json.load(f)
    schedule_data = [x for x in schedule_data if x["agegroup"] in AgeGroup]
    for event in schedule_data:
        event["team"] = event["agegroup"].replace("(", "").replace(")", "")
        del event["agegroup"]
    when = datetime.datetime.now().strftime("%d-%m-%Y et il est %H:%M")
    system = "\n".join(
        [
            "Tu es une IA connaissant bien les données suivantes, qui décrivent les match et entraînements d'équipes de hockey sur glace.",
            f"aujourd'hui, nous sommes le {when}"
            "attention: ce qu'il y a entre parenthèse après la catégorie est une catégorie à part entière, example, u13 a = U13 (A) et ça correspond aux agegroup",
            "assure-toi de ne pas confondre les catégories d'age dans tes réponses. ne donne pas une réponse pour la mauvaise équipe",
            "ne confond pas top, elite, a, prép",
            "```json",
            json.dumps(schedule_data),
            "```",
        ],
    )
    history: List[Tuple[str, str]] = []
    while True:
        try:
            question = input("> ")
        except EOFError:
            break
        answer = utils.llm_inference(
            question, history, system=system, model="mixtral8x22b"
        )
        print("<", answer)
        history.append((question, answer))


if __name__ == "__main__":
    app()