pythonic structure, packaging, first script

.gitignore

+result

default.nix

@@ -5,11 +5,43 @@
   }) {}
 }:
 
-pkgs.mkShell {
-  buildInputs = with pkgs; [
-    python3Packages.ipython
-    python3Packages.requests
-    python3Packages.pandas
-    python3Packages.ics
-  ];
+let
+  pyPkgs = pkgs.python3Packages;
+in rec {
+  lflex_celcat_survival = pyPkgs.buildPythonPackage {
+    pname = "lflex_celcat_survival";
+    version = "local";
+    format = "pyproject";
+
+    src = pkgs.lib.sourceByRegex ./. [
+      "pyproject\.toml"
+      "LICENSE"
+      "lflex_celcat_survival"
+      "lflex_celcat_survival/.*\.py"
+      "lflex_celcat_survival/cmd"
+      "lflex_celcat_survival/cmd/.*\.py"
+    ];
+    buildInputs = with pyPkgs; [
+      flit
+    ];
+    propagatedBuildInputs = with pyPkgs; [
+      ics
+      pandas
+      requests
+      click
+    ];
+  };
+
+  user-shell = pkgs.mkShell {
+    buildInputs = with pyPkgs; [
+      ipython
+      lflex_celcat_survival
+    ];
+  };
+
+  dev-shell = pkgs.mkShell {
+    buildInputs = with pyPkgs; [
+      ipython
+    ] ++ lflex_celcat_survival.propagatedBuildInputs;
+  };
 }

lflex_celcat_survival/__init__.py

+from . import course_request
+from . import events
+from . import fetch
+from . import ics

lflex_celcat_survival/cmd/fetch_ics.py

+#!/usr/bin/env python3
+import click
+import logging
+import lflex_celcat_survival as lcs
+
+@click.command()
+@click.argument('course_request_file')
+@click.option('--output-file', '-o', default=None, help='Where to write the generated ICS (stdout if unset).')
+def main(course_request_file, output_file):
+    logging.basicConfig(level=logging.INFO)
+    req = lcs.course_request.CourseRequest(course_request_file)
+    celcat_raw_response = req.do_request()
+
+    celcat_events = lcs.events.CelcatEvents(celcat_raw_response)
+    filtered_celcat_events = lcs.events.FilteredCelcatEvents(req, celcat_events)
+    filtered_celcat_events.check_expected_nb_timeslots()
+
+    calendar = lcs.ics.course_df_to_ics(filtered_celcat_events.df)
+
+    if output_file is None:
+        print(calendar)
+    else:
+        with open(output_file, 'w') as f:
+            f.write(str(calendar))
+
+if __name__ == "__main__":
+    main()

lflex_celcat_survival/course_request.py

+import pandas as pd
+from . import fetch
+
+class CourseRequest:
+    def __init__(self, filename):
+        self.df = pd.read_csv(filename, parse_dates=['begin_date', 'end_date'])
+        self.df['course_request_id'] = self.df.index
+
+    def generate_request_input(self):
+        date_range_min = min(self.df['begin_date']).strftime("%Y-%m-%d")
+        date_range_max = (max(self.df['end_date']) + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
+        apogee_codes = self.df['module_apogee'].unique()
+
+        return (date_range_min, date_range_max, apogee_codes)
+
+    def do_request(self, url='https://edt.univ-tlse3.fr/calendar2/Home/GetCalendarData'):
+        (date_min, date_max, apogee_codes) = self.generate_request_input()
+        return fetch.do_celcat_calendar_request(date_min, date_max, apogee_codes, url)

src/script.py → lflex_celcat_survival/events.py

-#!/usr/bin/env python3
-import ics
-import requests
-import pandas as pd
 import logging
-
-class CourseRequest:
-    def __init__(self, filename):
-        self.df = pd.read_csv(filename, parse_dates=['begin_date', 'end_date'])
-        self.df['course_request_id'] = self.df.index
-
-    def generate_request_input(self):
-        date_range_min = min(self.df['begin_date']).strftime("%Y-%m-%d")
-        date_range_max = (max(self.df['end_date']) + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
-        apogee_codes = self.df['module_apogee'].unique()
-
-        return (date_range_min, date_range_max, apogee_codes)
-
-    def do_request(self, url='https://edt.univ-tlse3.fr/calendar2/Home/GetCalendarData'):
-        (date_min, date_max, apogee_codes) = self.generate_request_input()
-        return do_celcat_calendar_request(date_min, date_max, apogee_codes, url)
-
-def do_celcat_calendar_request(min_date, max_date, module_apogee_codes, url='https://edt.univ-tlse3.fr/calendar2/Home/GetCalendarData'):
-    headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
-    fields = [
-        f'start={min_date}',
-        f'end={max_date}',
-        'resType=100',
-        'calView=agendaWeek',
-    ] + ['federationIds%5B%5D={}'.format(apogee_code) for apogee_code in module_apogee_codes]
-    fields_str = '&'.join(fields)
-
-    logging.info(f'Sending a POST request with data={fields_str}')
-    response = requests.post(url, data=fields_str, headers=headers)
-    if not response.ok:
-        logging.error(f'POST HTTP request failed (status code {response.status_code}): {response.reason}')
-        logging.error(f'Request response text:\n---\n{response.text}\n---')
-        response.raise_for_status()
-    return response.text
+import pandas as pd
 
 class CelcatEvents:
     def __init__(self, celcat_raw_response):
@@ -132,30 +95,3 @@ class FilteredCelcatEvents:
             groups_joined = ' '.join(groups)
 
         return pd.Series([room, course_type, groups_joined], index=['room_parsed', 'course_type_parsed', 'groups_parsed'])
-
-def course_df_to_ics(df):
-    c = ics.Calendar()
-    for _, row in df.iterrows():
-        event = ics.Event(
-            name = f'{row["module_readable"]} - {row["course_type"]} - {row["groups_parsed"]}',
-            begin = row['start'].tz_localize(tz='Europe/Paris'),
-            end = row['end'].tz_localize(tz='Europe/Paris'),
-        )
-        if row['room_parsed'] != 'unset':
-            event.location = row['room_parsed']
-        c.events.add(event)
-
-    return c
-
-logging.basicConfig(level=logging.INFO)
-req = CourseRequest('input-data.csv')
-celcat_raw_response = req.do_request()
-
-celcat_events = CelcatEvents(celcat_raw_response)
-filtered_celcat_events = FilteredCelcatEvents(req, celcat_events)
-filtered_celcat_events.check_expected_nb_timeslots()
-
-c = course_df_to_ics(filtered_celcat_events.df)
-
-with open('out.ics', 'w') as f:
-    f.writelines(c)

lflex_celcat_survival/fetch.py

+import logging
+import requests
+
+def do_celcat_calendar_request(min_date, max_date, module_apogee_codes, url='https://edt.univ-tlse3.fr/calendar2/Home/GetCalendarData'):
+    headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
+    fields = [
+        f'start={min_date}',
+        f'end={max_date}',
+        'resType=100',
+        'calView=agendaWeek',
+    ] + ['federationIds%5B%5D={}'.format(apogee_code) for apogee_code in module_apogee_codes]
+    fields_str = '&'.join(fields)
+
+    logging.info(f'Fetching modules {module_apogee_codes} from {min_date} to {min_date} on url={url}')
+    response = requests.post(url, data=fields_str, headers=headers)
+    if not response.ok:
+        logging.error(f'POST HTTP request failed (status code {response.status_code}): {response.reason}')
+        logging.error(f'Request response text:\n---\n{response.text}\n---')
+        response.raise_for_status()
+    return response.text

lflex_celcat_survival/ics.py

+import ics
+
+def course_df_to_ics(df):
+    c = ics.Calendar()
+    for _, row in df.iterrows():
+        event = ics.Event(
+            name = f'{row["module_readable"]} - {row["course_type"]} - {row["groups_parsed"]}',
+            begin = row['start'].tz_localize(tz='Europe/Paris'),
+            end = row['end'].tz_localize(tz='Europe/Paris'),
+        )
+        if row['room_parsed'] != 'unset':
+            event.location = row['room_parsed']
+        c.events.add(event)
+
+    return c

pyproject.toml

+[build-system]
+build-backend = "flit_core.buildapi"
+requires = ["flit_core"]
+
+[project]
+name = "lflex_celcat_survival"
+version = "0.1.0"
+description = "Set of tools to stay sane while using a CELCAT calendar"
+authors = [
+    {name = "Millian Poquet", email="millian.poquet@irit.fr"},
+]
+license = {file = "LICENSE"}
+requires-python = ">=3.9"
+
+keywords = ["calendar", "celcat"]
+
+classifiers = [
+    "Topic :: Software Development",
+    "License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
+    "Programming Language :: Python",
+    "Programming Language :: Python :: 3.9",
+]
+
+dependencies = [
+    "ics>=0.7.0",
+    "pandas>=1.3.0",
+    "requests>=2.26.0",
+    "click>=8.0.0"
+]
+
+[project.scripts]
+fetch-ics = "lflex_celcat_survival.cmd.fetch_ics:main"