Spaces:
Running
Running
| import re | |
| import time | |
| import pymongo | |
| import motor.motor_asyncio | |
| from bson.objectid import ObjectId | |
| from bson.errors import InvalidId | |
| from bson.json_util import dumps | |
| from pymongo.errors import PyMongoError | |
| class PUBLICFilesDB: | |
| def __init__(self, uri, database_name): | |
| self._client = motor.motor_asyncio.AsyncIOMotorClient(uri) | |
| self.db = self._client[database_name] | |
| self.files = self.db.Public_Files | |
| async def GetSeriesAllBy10(self, offset=None): | |
| try: | |
| filter = {"type": "TVSeries"} # Define the filter for movies | |
| total_results = await self.files.count_documents(filter) | |
| offset=int( offset if offset else 0) | |
| next_offset = offset + 10 | |
| if next_offset >= total_results: # Changed > to >= to handle the last page correctly | |
| next_offset = '' | |
| cursor = self.files.find(filter) # Apply the filter | |
| cursor.sort('$natural', pymongo.DESCENDING) # Use pymongo.DESCENDING | |
| cursor.skip(offset).limit(10) | |
| files = await cursor.to_list(length=10) | |
| return files, next_offset | |
| except PyMongoError as e: | |
| print(f"MongoDB error: {e}") | |
| return [], '' # Return empty list and '' on error | |
| except Exception as e: | |
| print(f"An unexpected error occurred: {e}") | |
| return [], '' | |
| async def GetMoviesAllBy10(self, offset=None): | |
| try: | |
| filter = {"type": "Movie"} # Define the filter for movies | |
| total_results = await self.files.count_documents(filter) | |
| offset=int( offset if offset else 0) | |
| next_offset = offset + 10 | |
| if next_offset >= total_results: # Changed > to >= to handle the last page correctly | |
| next_offset = '' | |
| cursor = self.files.find(filter) # Apply the filter | |
| cursor.sort('$natural', pymongo.DESCENDING) # Use pymongo.DESCENDING | |
| cursor.skip(offset).limit(10) | |
| files = await cursor.to_list(length=10) | |
| return files, next_offset | |
| except PyMongoError as e: | |
| print(f"MongoDB error: {e}") | |
| return [], '' # Return empty list and '' on error | |
| except Exception as e: | |
| print(f"An unexpected error occurred: {e}") | |
| return [], '' | |
| async def GetLatestAllBy10(self, offset=None): | |
| try: | |
| total_results = await self.files.count_documents({}) | |
| offset=int( offset if offset else 0) | |
| next_offset = offset + 10 | |
| if next_offset >= total_results: # Changed > to >= to handle the last page correctly | |
| next_offset = '' | |
| cursor = self.files.find({}) # Apply the filter | |
| cursor.sort('$natural', pymongo.DESCENDING ) # Use pymongo.DESCENDING | |
| cursor.skip(offset).limit(10) | |
| files = await cursor.to_list(length=10) | |
| return files, next_offset | |
| except PyMongoError as e: | |
| print(f"MongoDB error: {e}") | |
| return [], '' # Return empty list and '' on error | |
| except Exception as e: | |
| print(f"An unexpected error occurred: {e}") | |
| return [], '' | |