NAME_TEMPLATE_PLAYLIST = PathStr('playlist.tmpl')
PATH_DIR_TEMP = PathStr(path_join(PATH_DIR_DOWNLOADS, NAME_DIR_TEMP))
-EXPECTED_DIRS = [PATH_DIR_DOWNLOADS, PATH_DIR_TEMP, PATH_DIR_THUMBNAILS]
PATH_TEMPLATE_QUERIES = PathStr(path_join(PATH_DIR_TEMPLATES,
NAME_TEMPLATE_QUERIES))
TIMESTAMP_FMT = '%Y-%m-%d %H:%M:%S.%f'
'''
+def _ensure_expected_dirs(expected_dirs: list[PathStr]) -> None:
+ """Ensure existance of expected_dirs _as_ directories."""
+ for dir_name in expected_dirs:
+ if not path_exists(dir_name):
+ print(f'creating expected directory: {dir_name}')
+ makedirs(dir_name)
+ elif not isdir(dir_name):
+ msg = f'at expected directory path {dir_name} found non-directory'
+ raise Exception(msg)
+
+
class DatabaseConnection:
"""Wrapped sqlite3.Connection."""
def __init__(self) -> None:
self._to_download: list[VideoId] = []
+ _ensure_expected_dirs([PATH_DIR_DOWNLOADS, PATH_DIR_TEMP])
@staticmethod
def _id_from_filename(path: PathStr,
self.downloads = downloads_db
-def ensure_expected_dirs_and_files() -> None:
- """Ensure existance of all dirs and files we need for proper operation."""
- for dir_name in EXPECTED_DIRS:
- if not path_exists(dir_name):
- print(f'creating expected directory: {dir_name}')
- makedirs(dir_name)
- elif not isdir(dir_name):
- msg = f'at expected directory path {dir_name} found non-directory'
- raise Exception(msg)
-
-
class TaskHandler(BaseHTTPRequestHandler):
"""Handler for GET and POST requests to our server."""
server: Server
self._send_http(bytes(html, 'utf8'))
def _send_thumbnail(self, filename: PathStr) -> None:
+ _ensure_expected_dirs([PATH_DIR_THUMBNAILS])
path_thumbnail = path_join(PATH_DIR_THUMBNAILS, filename)
if not path_exists(path_thumbnail):
video_id = splitext(filename)[0]
"""Create DownloadsDb, Player, run server loop."""
downloads_db = DownloadsDb()
downloads_db.clean_unfinished()
- ensure_expected_dirs_and_files()
Thread(target=downloads_db.download_loop, daemon=False).start()
server = Server(Player(), downloads_db, ('localhost', HTTP_PORT),
TaskHandler)