Code Updates & Optimisations Clearing UP ACTIVE_CLIENTS

FileStream/server/Functions/downloader.py

@@ -1,8 +1,8 @@
 import os
 import time
 import math
-import asyncio
 import logging
+import asyncio
 import traceback
 from aiohttp import web
 from pyrogram import raw
@@ -14,17 +14,14 @@ from FileStream.config import Telegram
 from FileStream.bot import req_client, FileStream
 from FileStream import utils, StartTime, __version__
 from FileStream.Tools import mime_identifier, Time_ISTKolNow
-from FileStream.utils.FileProcessors.custom_ul import TeleUploader
-from FileStream.server.exceptions import FIleNotFound, InvalidHash
 from FileStream.bot import MULTI_CLIENTS, WORK_LOADS, ACTIVE_CLIENTS
+from FileStream.server.exceptions import FIleNotFound, InvalidHash
 from FileStream.server.render_template import render_page, render_upload
-
+from FileStream.utils.FileProcessors.custom_ul import  TeleUploader
 
 
 async def media_streamer(request: web.Request, db_id: str, speed: str):
-    """
-    Handle streaming of media files to the client.
-    """
+    # Get the Range header from the request, default to 0 if not present
     range_header = request.headers.get("Range", "bytes=0-")
     client = await req_client()
 
@@ -32,23 +29,22 @@ async def media_streamer(request: web.Request, db_id: str, speed: str):
     if Telegram.MULTI_CLIENT:
         logging.info(f"Client {client['index']} is now serving {request.headers.get('X-FORWARDED-FOR', request.remote)}")
 
-    # Get or create the ByteStreamer object
+    # Use an existing ByteStreamer or create a new one
     tg_connect = ACTIVE_CLIENTS.get(client['client'], None)
     
     if tg_connect is None:
         logging.debug(f"Creating new ByteStreamer object for client {client['index']}")
         tg_connect = utils.ByteStreamer(client['client'])
-        ACTIVE_CLIENTS[client['client']] = {
-            "last_activity": asyncio.get_event_loop().time(),  # Store current timestamp
-            "streamer": tg_connect
-        }
+        ACTIVE_CLIENTS[client['client']] = tg_connect
+        
     else:
+        tg_connect.update_last_activity()
         logging.debug(f"Using cached ByteStreamer object for client {client['index']}")
 
     try:
         # Fetch file properties once and use it throughout
         logging.debug("Fetching file properties")
-        file_id = await tg_connect["streamer"].get_file_properties(db_id, MULTI_CLIENTS)
+        file_id = await tg_connect.get_file_properties(db_id, MULTI_CLIENTS)
         file_size = file_id.file_size
 
         # Parse range header efficiently
@@ -63,7 +59,7 @@ async def media_streamer(request: web.Request, db_id: str, speed: str):
             )
 
         # Set chunk size based on speed
-        chunk_size = 1024 * 1024 if speed == "FAST" else 512 * 1024
+        chunk_size =  1024 * 1024 if speed == "FAST" else 512 * 1024
 
         # Ensure we don't go past the file size
         until_bytes = min(until_bytes, file_size - 1)
@@ -72,7 +68,7 @@ async def media_streamer(request: web.Request, db_id: str, speed: str):
         offset, first_part_cut, last_part_cut, part_count = compute_offsets(from_bytes, until_bytes, chunk_size)
 
         # Request the file chunks
-        body = tg_connect["streamer"].yield_file(
+        body = tg_connect.yield_file(
             file_id, client['index'], offset, first_part_cut, last_part_cut, part_count, chunk_size
         )
 

FileStream/server/__init__.py

@@ -18,9 +18,10 @@ async def clear_inactive_clients():
         await asyncio.sleep(INACTIVITY_TIMEOUT)  # Check every INACTIVITY_TIMEOUT seconds
 
         now = asyncio.get_event_loop().time()  # Get current time
+        
         inactive_clients = [
-            client for client, data in ACTIVE_CLIENTS.items() 
-            if now - data["last_activity"] > INACTIVITY_TIMEOUT  # Compare with last_activity timestamp
+            client_id for client_id, tg_connect in ACTIVE_CLIENTS.items()
+            if now - tg_connect.last_activity > INACTIVITY_TIMEOUT  # Compare with last_activity timestamp
         ]
 
         for client in inactive_clients:

FileStream/utils/FileProcessors/custom_dl.py

@@ -1,26 +1,41 @@
+import os
+import time
+import math
 import asyncio
 import logging
-from typing import Dict, Union
-from pyrogram.types import Message
-from pyrogram import Client, utils, raw
-from pyrogram.session import Session, Auth
-from pyrogram.errors import AuthBytesInvalid
-from pyrogram.file_id import FileId, FileType, ThumbnailSource
+import traceback
+from aiohttp import web
+from pyrogram import raw
+from aiohttp.http_exceptions import BadStatusLine
 
-#--------------------Local Imports -------------------------------#
+#---------------------Local Upload---------------------#
 
-from .file_properties import get_file_ids
-from FileStream.bot import WORK_LOADS
+from FileStream.config import Telegram
+from FileStream.bot import req_client, FileStream
+from FileStream import utils, StartTime, __version__
+from FileStream.Tools import mime_identifier, Time_ISTKolNow
+from FileStream.utils.FileProcessors.custom_ul import TeleUploader
+from FileStream.server.exceptions import FIleNotFound, InvalidHash
+from FileStream.bot import MULTI_CLIENTS, WORK_LOADS, ACTIVE_CLIENTS
+from FileStream.server.render_template import render_page, render_upload
 
 
 class ByteStreamer:
-
     def __init__(self, client: Client):
         self.clean_timer = 30 * 60  # Cache cleanup timer set to 30 minutes
         self.client: Client = client
         self.cached_file_ids: Dict[str, FileId] = {}  # Cache to store file properties by db_id
+        self.last_activity: float = asyncio.get_event_loop().time()  # Track last activity time for the client
         asyncio.create_task(self.clean_cache())  # Start the cache cleanup task
 
+    def update_last_activity(self):
+        """Update the last activity time to the current time."""
+        self.last_activity = asyncio.get_event_loop().time()
+
+    def get_last_activity(self) -> float:
+        """Get the last activity time of this client."""
+        return self.last_activity
+
     async def get_file_properties(self, db_id: str, MULTI_CLIENTS) -> FileId:
         """
         Returns the properties of a media of a specific message in a FileId class.
@@ -206,6 +221,15 @@ class ByteStreamer:
         """
         while True:
             await asyncio.sleep(self.clean_timer)  # Wait for the cleanup interval
-            logging.debug("Cleaning cached file properties...")
+            logging.debug("*** Cleaning cached file properties...")
             self.cached_file_ids.clear()  # Clear the cache
             logging.debug("Cache cleaned.")
+
+    @staticmethod
+    def is_active(last_activity: float, timeout: float = 120) -> bool:
+        """
+        Determines if the client is active based on the last activity time.
+        If the client has been inactive for longer than the timeout period (default 30 minutes),
+        they are considered inactive.
+        """
+        return (asyncio.get_event_loop().time() - last_activity) < timeout