Merge pull request #20 from ParthBibekar/refactor

Refactor welearnbot.py

src/moodlews/service.py

@@ -18,27 +18,28 @@ def __init__(self, baseurl):
         self.login_url = urllib.parse.urljoin(baseurl, "login/token.php")
         self.server_url = urllib.parse.urljoin(baseurl, "webservice/rest/server.php")
         self.session = Session()
+        self.token = ""
 
     def response(self, url, **data):
         return self.session.post(url, data)
-    
+
     def response_json(self, url, **data):
         response = self.response(url, **data)
         return json.loads(response.content)
-    
+
     def authenticate(self, username, password):
         login = self.response_json(
             self.login_url,
             username=username,
             password=password,
-            service="moodle_mobile_app"
+            service="moodle_mobile_app",
         )
         try:
-            self.token = login['token']
+            self.token = login["token"]
             return self.token
         except KeyError:
             return False
-    
+
     def server(self, function, **data):
         return self.response_json(
             self.server_url,

src/welearnbot/action_handlers.py

@@ -0,0 +1,220 @@
+from argparse import Namespace
+from configparser import RawConfigParser
+import sys
+from typing import List
+from bs4 import BeautifulSoup as bs
+from datetime import datetime
+
+from moodlews.service import MoodleClient, ServerFunctions
+from welearnbot import resolvers
+from welearnbot.gcal import publish_gcal_event
+from welearnbot.utils import get_resource, read_cache, write_cache, show_file_statuses
+
+
+def handle_whoami(moodle: MoodleClient) -> None:
+    info = moodle.server(ServerFunctions.SITE_INFO)
+    print(info["fullname"])
+
+
+def handle_courses(moodle: MoodleClient) -> None:
+    # Get core user information
+    info = moodle.server(ServerFunctions.SITE_INFO)
+    userid = info["userid"]
+
+    # Get enrolled courses information
+    courses = moodle.server(ServerFunctions.USER_COURSES, userid=userid)
+    for course in courses:
+        course_name = course["fullname"]
+        star = " "
+        if course["isfavourite"]:
+            star = "*"
+        print(f" {star} {course_name}")
+
+
+def handle_assignments(
+    args: Namespace,
+    config: RawConfigParser,
+    moodle: MoodleClient,
+    ignore_types: List[str],
+    prefix_path: str,
+    link_cache_filepath: str,
+    token: str,
+) -> None:
+    link_cache = read_cache(link_cache_filepath)
+    # Get assignment data from server
+    assignments = moodle.server(ServerFunctions.ASSIGNMENTS)
+
+    file_statuses = []
+    # Assignments are grouped by course
+    for course in assignments["courses"]:
+        course_name = course["shortname"]
+        # Ignore unspecified courses
+        if course_name not in args.courses:
+            continue
+        no_assignments = True
+        for assignment in course["assignments"]:
+            # Get the assignment name, details, due date, and relative due date
+            assignment_id = assignment["id"]
+            name = assignment["name"]
+            duedate = datetime.fromtimestamp(int(assignment["duedate"]))
+            due_str = duedate.strftime("%a %d %b, %Y, %H:%M:%S")
+            duedelta = duedate - datetime.now()
+            # Calculate whether the due date is in the future
+            due = duedelta.total_seconds() > 0
+            if args.dueassignments and not due:
+                continue
+            no_assignments = False
+            if not no_assignments:
+                print(course_name)
+            # Show assignment details
+            duedelta_str = (
+                f"{abs(duedelta.days)} days, {duedelta.seconds // 3600} hours"
+            )
+            detail = bs(assignment["intro"], "html.parser").text
+            print(f"    {name} - {detail}")
+            for attachment in assignment["introattachments"]:
+                print(f"        Attachment     : {attachment['filename']}")
+                file_statuses.append(
+                    get_resource(
+                        args,
+                        moodle,
+                        ignore_types,
+                        attachment,
+                        prefix_path,
+                        course_name,
+                        link_cache,
+                        token,
+                        indent=8,
+                    )
+                )
+            if due:
+                print(f"        Due on         : {due_str}")
+                print(f"        Time remaining : {duedelta_str}")
+            else:
+                print(f"        Due on         : {due_str} ({duedelta_str} ago)")
+
+            # Get submission details
+            submission = moodle.server(
+                ServerFunctions.ASSIGNMENT_STATUS, assignid=assignment_id
+            )
+            submission_made = False
+            try:
+                for plugin in submission["lastattempt"]["submission"]["plugins"]:
+                    if plugin["name"] == "File submissions":
+                        for filearea in plugin["fileareas"]:
+                            if filearea["area"] == "submission_files":
+                                for submitted_file in filearea["files"]:
+                                    submission_made = True
+                                    filename = submitted_file["filename"]
+                                    submission_date = datetime.fromtimestamp(
+                                        int(submitted_file["timemodified"])
+                                    )
+                                    submission_date_str = submission_date.strftime(
+                                        "%a %d %b, %Y, %H:%M:%S"
+                                    )
+                                    print(
+                                        f"        Submission     : {filename} ({submission_date_str})"
+                                    )
+            except KeyError:
+                continue
+            if not submission_made:
+                print(f"        Submission     : NONE")
+
+            # Write event to calendar
+            if args.gcalendar and due:
+                publish_gcal_event(
+                    config, duedate, course_name, name, assignment_id, detail
+                )
+            print()
+
+    write_cache(link_cache_filepath, link_cache)
+    show_file_statuses(file_statuses, verbose=args.verbose)
+
+
+def handle_urls(args: Namespace, moodle: MoodleClient) -> None:
+    course_ids = resolvers.get_courses_by_id(moodle, args)
+
+    # Get a list of available urls
+    urls = moodle.server(ServerFunctions.URLS)
+
+    # Iterate through all urls, and build a dictionary
+    url_list = dict()
+    for url in urls["urls"]:
+        if url["course"] in course_ids:
+            course_name = course_ids[url["course"]]
+            if not course_name in url_list:
+                url_list[course_name] = []
+            url_list[course_name].append(url)
+
+    # Display all urls
+    for course_name in args.courses:
+        if not course_name in url_list:
+            continue
+        no_url = True
+        for url in url_list[course_name]:
+            if no_url:
+                print(course_name)
+            no_url = False
+            url_name = url["name"]
+            url_detail = bs(url["intro"], "html.parser").text
+            url_link = url["externalurl"]
+            print(f"    {url_name} - {url_detail}")
+            print(f"        Link : {url_link}")
+            print()
+        print()
+
+
+def handle_files(
+    args: Namespace,
+    moodle: MoodleClient,
+    ignore_types: List[str],
+    prefix_path: str,
+    link_cache_filepath: str,
+    token: str,
+) -> None:
+    link_cache = read_cache(link_cache_filepath)
+    course_ids = resolvers.get_courses_by_id(moodle, args)
+
+    file_statuses = []
+
+    # Iterate through each course, and fetch all modules
+    for courseid in course_ids:
+        course_name = course_ids[courseid]
+        page = moodle.server(ServerFunctions.COURSE_CONTENTS, courseid=courseid)
+        for item in page:
+            modules = item.get("modules", [])
+            for module in modules:
+                modname = module.get("modname", "")
+                if modname == "resource":
+                    for resource in module["contents"]:
+                        file_statuses.append(
+                            get_resource(
+                                args,
+                                moodle,
+                                ignore_types,
+                                resource,
+                                prefix_path,
+                                course_name,
+                                link_cache,
+                                token,
+                            )
+                        )
+                elif modname == "folder":
+                    folder_name = module.get("name", "")
+                    for resource in module["contents"]:
+                        file_statuses.append(
+                            get_resource(
+                                args,
+                                moodle,
+                                ignore_types,
+                                resource,
+                                prefix_path,
+                                course_name,
+                                link_cache,
+                                token,
+                                subfolder=folder_name,
+                            )
+                        )
+
+    write_cache(link_cache_filepath, link_cache)
+    show_file_statuses(file_statuses, verbose=args.verbose)

src/welearnbot/constants.py

@@ -0,0 +1,3 @@
+BASEURL = "https://welearn.iiserkol.ac.in"
+LINK_CACHE = ".link_cache"
+EVENT_CACHE = "~/.welearn_event_cache"

src/welearnbot/gcal.py

@@ -0,0 +1,117 @@
+from welearnbot.constants import EVENT_CACHE
+
+from typing import Any, Tuple
+
+from configparser import RawConfigParser
+from datetime import datetime, timedelta
+from googleapiclient.discovery import build
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+from google.oauth2.credentials import Credentials
+
+import errno
+import os
+import sys
+
+from welearnbot.utils import create_event, read_cache, write_cache
+
+
+def setup_gcal(config: RawConfigParser) -> Tuple[str, Any]:
+    """Handle Google Calender
+
+    Returns
+    -------
+    gcal_calendar_id: str
+    service: Any
+    """
+    try:
+        OAUTH_CLIENT_ID = config["gcal"]["client_id"]
+        OAUTH_CLIENT_SECRET = config["gcal"]["client_secret"]
+
+        gcal_client_config = {
+            "installed": {
+                "client_id": OAUTH_CLIENT_ID,
+                "client_secret": OAUTH_CLIENT_SECRET,
+                "redirect_uris": ["http://localhost", "urn:ietf:wg:oauth:2.0:oob"],
+                "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+                "token_uri": "https://accounts.google.com/o/oauth2/token",
+            }
+        }
+    except KeyError:
+        print("Invalid configuration!")
+        sys.exit(errno.ENODATA)
+
+    try:
+        gcal_calendar_id = config["gcal"]["calendar_id"]
+    except KeyError:
+        gcal_calendar_id = "primary"
+
+    # Connect to the Google Calendar API
+    SCOPES = ["https://www.googleapis.com/auth/calendar.events"]
+    creds = None
+    gcal_token_path = os.path.expanduser("~/.gcal_token")
+    if os.path.exists(gcal_token_path):
+        creds = Credentials.from_authorized_user_file(gcal_token_path, SCOPES)
+    # If there are no valid credentials, login
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            creds.refresh(Request())
+        else:
+            flow = InstalledAppFlow.from_client_config(gcal_client_config, SCOPES)
+            creds = flow.run_local_server(port=0)
+        # Save the credentials for the next run
+        with open(gcal_token_path, "w") as gcal_token:
+            gcal_token.write(creds.to_json())
+    service = build("calendar", "v3", credentials=creds)
+
+    return gcal_calendar_id, service
+
+
+def publish_gcal_event(
+    config: RawConfigParser,
+    duedate: datetime,
+    course_name: str,
+    name: str,
+    assignment_id: int,
+    detail: str,
+) -> None:
+    event_cache_filepath = os.path.expanduser(EVENT_CACHE)
+    event_cache = read_cache(event_cache_filepath)
+
+    # Put deadline at the *end* of the event
+    startdate = duedate - timedelta(hours=1)
+    start_time = startdate.isoformat()
+    end_time = duedate.isoformat()
+    event_name = f"{course_name} - {name}"
+
+    gcal_calendar_id, service = setup_gcal(config)
+
+    assignment_id = str(assignment_id)
+    if assignment_id not in event_cache:
+        # Create and push a new event
+        event = create_event(event_name, detail, start_time, end_time, False)
+        added_event = (
+            service.events().insert(calendarId=gcal_calendar_id, body=event).execute()
+        )
+        event_id = added_event["id"]
+        event_cache[assignment_id] = event_id
+        print(f"        Added event to calendar.")
+    else:
+        # Update event if necessary
+        event = (
+            service.events()
+            .get(calendarId=gcal_calendar_id, eventId=event_cache[assignment_id],)
+            .execute()
+        )
+        if event["start"]["dateTime"] != (start_time + "+05:30"):
+            event["start"]["dateTime"] = start_time
+            event["end"]["dateTime"] = end_time
+            updated_event = (
+                service.events()
+                .update(calendarId=gcal_calendar_id, eventId=event["id"], body=event,)
+                .execute()
+            )
+            event_cache[assignment_id] = updated_event["id"]
+            print(f"        Updated event in calendar.")
+    write_cache(event_cache_filepath, event_cache)
+