Changing moodle activity handling

MoodleClasses/MoodleFile.py

@@ -0,0 +1,56 @@
+from MoodleClasses.MoodleActivity import MoodleActivity
+from urllib.parse import unquote
+import pathvalidate
+import requests
+import os
+
+
+class MoodleFile:
+    def __init__(self, parent_activity: MoodleActivity, url: str):
+        self.parent_activity = parent_activity
+        self.url = url
+        self.directory = self._get_directory()
+        self.filepath = self._get_filepath()
+        self.extension = str(self.filepath).split(".")[-1]
+
+    def _get_directory(self) -> os.path:
+        directory = os.path.join(self.parent_activity.parent_dir, self.parent_activity.name)
+        return pathvalidate.sanitize_filepath(directory)
+
+    def _get_filepath(self, filename: str=None) -> os.path:
+        if filename is None:
+            file = os.path.join(self.directory, unquote(self.url.split('/')[-1]))
+        else:
+            file = os.path.join(self.directory, filename)
+        return pathvalidate.sanitize_filepath(file)
+
+    def _decide_download(self, ignore_extension: list[str]) -> bool:
+        if ignore_extension is not None and self.extension in ignore_extension:
+            return False
+        if os.path.exists(self.filepath):
+            return False
+        return True
+
+    def _download(self, s: requests.Session) -> bool:
+        if not os.path.exists(self.directory):
+            os.makedirs(self.directory)
+        try:
+            with s.get(self.url, stream=True) as r:
+                with open(self.filepath+".part", 'wb') as f:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        f.write(chunk)
+            os.rename(self.filepath+".part", self.filepath)
+            if os.path.exists(self.filepath + ".part"):
+                os.remove(self.filepath + ".part")
+        except Exception as e:
+            if os.path.exists(self.filepath+".part"):
+                os.remove(self.filepath+".part")
+            return False
+        finally:
+            return True
+
+    def download(self, s: requests.Session, ignore_extension: list[str]) -> bool:
+        if self._decide_download(ignore_extension):
+            return self._download(s)
+        else:
+            return True