Refactor filebrowser uploader to disable SSL verification and implement upload callback mechanism

Author: Jared-Gross

filebrowser_uploader.py

@@ -52,6 +52,7 @@ async def get_public_share_url(file_relative_path: str, token: str) -> str:
             f"{FILEBROWSER_URL}/api/share/{file_relative_path}",
             headers=headers,
             json={"path": f"/{file_relative_path}"},
+            ssl=False, # This explicitly tells aiohttp not to use SSL
         ) as response:
             if response.status != 200:
                 body = await response.text()
@@ -67,6 +68,7 @@ async def get_filebrowser_token():
         async with session.post(
             f"{FILEBROWSER_URL}/api/login",
             json={"username": FILEBROWSER_USERNAME, "password": FILEBROWSER_PASSWORD},
+            ssl=False, # This explicitly tells aiohttp not to use SSL
         ) as response:
             token = await response.text()
             return token.strip()
@@ -77,6 +79,7 @@ async def delete_existing_file(file_relative_path: str, token: str):
         async with session.delete(
             f"{FILEBROWSER_URL}/api/resources/{file_relative_path}",
             headers={"X-Auth": token},
+            ssl=False
         ) as response:
             if response.status not in [200, 204, 404]:
                 body = await response.text()
@@ -101,6 +104,7 @@ async def upload_to_filebrowser(file_path, file_name) -> str:
                 f"{FILEBROWSER_URL}/api/resources/{FILEBROWSER_UPLOAD_PATH}/{file_name}",
                 headers=headers,
                 data=form,
+                ssl=False, # This explicitly tells aiohttp not to use SSL
             ) as response:
                 if response.status != 200:
                     body = await response.text()
@@ -145,12 +149,12 @@ async def upload(
 if __name__ == "__main__":
     asyncio.run(
         upload(
-            "Glenwayprivate - Unspecified description - February 20 Thursday 2025 08_50 PM - 1h 36m 34s.mp3",
-            "CURRENTLY_RECORDING/Glenwayprivate - Unspecified description - February 20 Thursday 2025 08_50 PM - 1h 36m 34s.mp3",
-            "glenway",
-            "Unspecified description",
-            "February 20 Thursday 2025 08_50 PM",
-            96.53333333333334,
+            "Pineland - just seeing if stuff works. - April 05 Saturday 2025 03_34 PM - 1m 1s.mp3",
+            "CURRENTLY_RECORDING/Pineland - just seeing if stuff works. - April 05 Saturday 2025 03_34 PM - 1m 1s.mp3",
+            "pineland",
+            "just seeing if stuff works.",
+            "April 05 Saturday 2025 03_34 PM",
+            1.1,
         ),
         debug=True,
     )

main.py

@@ -84,7 +84,7 @@ def excepthook(exc_type, exc_value, exc_traceback):
 
 class Stream:
     def __init__(
-        self, title: str, url: str, host: str, description: str, recording_finished
+        self, title: str, url: str, host: str, description: str, recording_finished, upload_callback
     ) -> None:
         load_dotenv()
         self.title = title
@@ -107,6 +107,7 @@ def __init__(
         self.recording_file_path = (
             f"{FOLDER_LOCATION}/CURRENTLY_RECORDING/{self.recording_file_name}"
         )
+        self.upload_callback = upload_callback
         self.send_notification()
 
     def start_recording(self):
@@ -192,24 +193,28 @@ def process_file(self):
             )
             self.uploaded = True
 
-        self.backup_stream(final_recording_file_path)
+        # self.backup_stream(final_recording_file_path)
         os.remove(final_recording_file_path)
         app_log.info(f"Original copy deleted: {self.recording_file_name}")
 
     def upload_stream(self, file_name: str, file_path: str):
         app_log.info(f"Uploading {self.host}: {self.recording_file_name}")
-        asyncio.run(
-            filebrowser_uploader.upload(
-                file_name,
-                file_path,
-                self.host,
-                self.description,
-                self.starting_time.strftime("%B %d %A %Y %I_%M %p"),
-                self.audio_file_length,
-            )
-        )
-        self.uploaded = True
-        app_log.info(f"Uploaded {self.host}: {self.recording_file_name}")
+        try:
+            if self.upload_callback:
+                self.upload_callback(
+                    file_name,
+                    file_path,
+                    self.host,
+                    self.description,
+                    self.starting_time.strftime("%B %d %A %Y %I_%M %p"),
+                    self.audio_file_length,
+                )
+                self.uploaded = True
+                app_log.info(f"Uploaded {self.host}: {self.recording_file_name}")
+            else:
+                raise RuntimeError("No upload callback provided.")
+        except Exception as e:
+            app_log.error(f"Failed to upload stream: {e}")
 
     def backup_stream(self, file_path: str):
         app_log.info(f"Starting compression for {self.recording_file_name}")
@@ -250,30 +255,28 @@ class StreamRecorder:
     def __init__(self) -> None:
         load_dotenv()
         self.active_streams: dict[str, Stream] = {}
+        self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
+        threading.Thread(target=self.loop.run_forever, daemon=True).start()
 
-    def fetch_icecast_status_json(
-        self, icecast_source="https://hbniaudio.hbni.net"
+    def fetch_broadcast_data(
+        self, source="https://broadcasting.hbni.net/get_broadcast_data"
     ) -> (
-        dict[
-            Literal["icestats"],
-            None | str | dict[Literal["source"], dict[str, str] | list[dict[str, str]]],
-        ]
+        list[dict]
         | None
     ):
         try:
-            response = requests.get(f"{icecast_source}/status-json.xsl")
+            response = requests.get(source)
             if response.status_code == 200:
-                json_content = response.text.replace('"title": - ,', '"title": null,')
-                json_data = json.loads(json_content)
-                json_data["icestats"]["icecast_source"] = icecast_source
-                app_log.info(f"Icecast status: {json_data}")
+                json_data = json.loads(response.text)
+                app_log.info(f"Broadcast data: {json_data}")
                 return json_data
             else:
-                app_log.info(f"Error fetching Icecast status: {response.status_code}")
-                return self.fetch_icecast_status_json("http://hbniaudio.hbni.net:8000")
+                app_log.info(f"Error fetching Broadcast data: {response.status_code}")
+                return self.fetch_broadcast_data()
         except Exception as e:
-            app_log.error(f"Error fetching Icecast status: {e}")
-            return self.fetch_icecast_status_json("http://hbniaudio.hbni.net:8000")
+            app_log.error(f"Error fetching Broadcast data: {e}")
+            return self.fetch_broadcast_data()
 
     def process_sources(
         self, sources: dict[str, str] | list[dict[str, str]]
@@ -289,7 +292,7 @@ def remove_stream(self, host: str):
         )
         del self.active_streams[host]
         if not list(self.active_streams.keys()):
-            self.update_recording_status()
+            self.loop.create_task(self.update_recording_status())
 
     def run(self):
         self.send_notification()
@@ -300,26 +303,31 @@ def run(self):
 
         while True:
             try:
-                status_data = self.fetch_icecast_status_json()
+                status_data = self.fetch_broadcast_data()
                 if not status_data:
                     time.sleep(15)
                     continue
 
-                sources = status_data.get("icestats", {}).get("source", [])
-                sources = self.process_sources(sources)
-
-                for source in sources:
-                    host = source["listenurl"].split("/")[-1]
+                for source in status_data:
+                    host: str = source.get("host")
                     description = source.get("server_description", "No description")
-                    title = host.replace("/", "").title()
-                    icecast_source = status_data.get("icestats", {}).get("icecast_source", "https://hbniaudio.hbni.net")
+                    title = host.title()
+                    icecast_source = "https://hbniaudio.hbni.net"
+                    is_private = source.get("is_private", False)
                     is_recording = source.get("genre", "various") == "RECORDING"
 
                     if (
-                        host not in self.active_streams and "test" not in host.lower() and "test" not in description.lower()
+                        host not in self.active_streams and "test" not in host.lower() and "test" not in description.lower() and not is_private
                         # and not is_recording # It is being recorded by HBNI Audio
                     ):
-                        stream = Stream(title, icecast_source, host, description, self.remove_stream)
+                        stream = Stream(
+                            title,
+                            icecast_source,
+                            host,
+                            description,
+                            self.remove_stream,
+                            upload_callback=upload_sync
+                        )
                         self.active_streams[host] = stream
                         stream.start_recording()
                         app_log.info(
@@ -328,13 +336,13 @@ def run(self):
 
                 # Cleanup inactive streams
                 active_hosts = {
-                    source["listenurl"].split("/")[-1] for source in sources
+                    source["host"].split("/")[-1] for source in status_data
                 }
                 for host in list(self.active_streams.keys()):
                     if host not in active_hosts:
                         self.remove_stream(host)
 
-                    self.update_recording_status()
+                    self.loop.create_task(self.update_recording_status())
 
                 time.sleep(15)
             except Exception as e:
@@ -380,6 +388,7 @@ async def update_recording_status(self):
                     await conn.execute(query, host, link, length, description, starting_time)
 
         await pool.close()
+
     def send_notification(self):
         send_email.send(
             "HBNI Audio Stream Recorder Started Successfully",
@@ -417,16 +426,36 @@ def start_log_server():
         httpd.serve_forever()
 
 
+def upload_sync(
+    file_name: str,
+    file_path: str,
+    host: str,
+    description: str,
+    date: str,
+    length: float,
+):
+    return asyncio.run(
+        filebrowser_uploader.upload(
+            file_name,
+            file_path,
+            host,
+            description,
+            date,
+            length,
+        )
+    )
+
+
 def start_recorder():
     stream_recorder = StreamRecorder()
     app_log.info("Starting stream recorder")
     stream_recorder.run()
 
 
 def main() -> None:
-
-    threading.Thread(target=start_recorder).start()
+    # threading.Thread(target=start_recorder).start()
     threading.Thread(target=start_log_server).start()
+    start_recorder()
 
 
 if __name__ == "__main__":