2758be8680
* Fix pylint warnings across all 24 Python files in web_server - Add module, class, and function docstrings (C0114, C0115, C0116) - Fix import ordering: stdlib before third-party before local (C0411) - Replace wildcard imports with explicit named imports (W0401) - Remove trailing whitespace and add missing final newlines (C0303, C0304) - Replace dict() with dict literals (R1735) - Remove unused imports and variables (W0611, W0612) - Narrow broad Exception catches to specific exceptions (W0718) - Replace f-string logging with lazy % formatting (W1203) - Fix variable naming: UPPER_CASE for constants, snake_case for locals (C0103) - Add pylint disable comments for necessary global statements (W0603) - Fix no-else-return, simplifiable-if-expression, singleton-comparison - Fix bad indentation in stripe.py (W0311) - Add encoding="utf-8" to open() calls (W1514) - Add check=True to subprocess.run() calls (W1510) - Register Celery task modules via conf.include * Update `package-lock.json` add peer dependencies
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""File system path management for user streams, VODs, and profile pictures."""
|
|
|
|
import os
|
|
|
|
|
|
class PathManager():
|
|
"""Manages paths for user stream data, VODs, and profile pictures."""
|
|
|
|
def __init__(self) -> None:
|
|
self.root_path = "user_data"
|
|
self.vod_directory_name = "vods"
|
|
self.stream_directory_name = "stream"
|
|
self.profile_picture_name = "index.png"
|
|
self.stream_index_name = "index.m3u8"
|
|
self.stream_thumbnail_name = "index.png"
|
|
|
|
def _create_directory(self, path):
|
|
"""
|
|
Create a directory if it does not exist
|
|
"""
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
os.chmod(path, 0o777)
|
|
|
|
def create_user(self, username):
|
|
"""
|
|
Create directories for user stream data if they do not exist
|
|
"""
|
|
self._create_directory(os.path.join(self.root_path, username))
|
|
vods_path = self.get_vods_path(username)
|
|
stream_path = self.get_stream_path(username)
|
|
|
|
self._create_directory(vods_path)
|
|
self._create_directory(stream_path)
|
|
|
|
def delete_user(self, username):
|
|
"""
|
|
Delete directories for user stream data
|
|
"""
|
|
user_path = self.get_user_path(username)
|
|
if os.path.exists(user_path):
|
|
os.rmdir(user_path)
|
|
|
|
def get_user_path(self, username):
|
|
"""Returns the base path for a user's data directory."""
|
|
return os.path.join(self.root_path, username)
|
|
|
|
def get_vods_path(self, username):
|
|
"""Returns the path to a user's VODs directory."""
|
|
return os.path.join(self.root_path, username, self.vod_directory_name)
|
|
|
|
def get_stream_path(self, username):
|
|
"""Returns the path to a user's stream directory."""
|
|
return os.path.join(self.root_path, username, self.stream_directory_name)
|
|
|
|
def get_stream_file_path(self, username):
|
|
"""Returns the path to a user's stream index file."""
|
|
return os.path.join(self.get_stream_path(username), "index.m3u8")
|
|
|
|
def get_current_stream_thumbnail_file_path(self, username):
|
|
"""Returns the path to a user's current stream thumbnail."""
|
|
return os.path.join(self.get_stream_path(username), "index.png")
|
|
|
|
def get_vod_file_path(self, username, vod_id):
|
|
"""Returns the path to a specific VOD file."""
|
|
return os.path.join(self.get_vods_path(username), f"{vod_id}.mp4")
|
|
|
|
def get_vod_thumbnail_file_path(self, username, vod_id):
|
|
"""Returns the path to a specific VOD thumbnail."""
|
|
return os.path.join(self.get_vods_path(username), f"{vod_id}.png")
|
|
|
|
def get_profile_picture_file_path(self, username):
|
|
"""Returns the path to a user's profile picture."""
|
|
return os.path.join(self.root_path, username, self.profile_picture_name)
|