From: Christian Heller <c.heller@plomlompom.de>
Date: Wed, 27 Nov 2024 05:40:05 +0000 (+0100)
Subject: More broadly use .rel_path for ID'ing VideoFiles, base64-encode where necessary.
X-Git-Url: https://plomlompom.com/repos/%22https:/validator.w3.org/error?a=commitdiff_plain;h=576a6cec8b5e41b1819dbf76f90cca95f151c68f;p=ytplom
More broadly use .rel_path for ID'ing VideoFiles, base64-encode where necessary.
---
diff --git a/src/sync.py b/src/sync.py
index 98bdfcb..d0591f7 100755
--- a/src/sync.py
+++ b/src/sync.py
@@ -35,7 +35,8 @@ def sync_objects(host_names: tuple[str, str],
shared: tuple[Any, str]
) -> None:
"""Equalize both DB's object collections; prefer newer states to older."""
- cls, id_name = shared
+ cls, _ = shared
+ id_name = 'id_' if 'id' == cls.id_name else cls.id_name
obj_colls = cls.get_all(dbs[0]), cls.get_all(dbs[1])
for obj_0 in [obj for obj in obj_colls[0] if obj not in obj_colls[1]]:
id_ = getattr(obj_0, id_name)
diff --git a/src/templates/file_data.tmpl b/src/templates/file_data.tmpl
index 7ea70f9..d0e606d 100644
--- a/src/templates/file_data.tmpl
+++ b/src/templates/file_data.tmpl
@@ -8,7 +8,7 @@
<tr><th>YouTube ID:</th><td><a href="/{{page_names.yt_result}}/{{file.yt_id}}">{{file.yt_id}}</a></tr>
<tr><th>present:</th><td>{% if file.present %}<a href="/{{page_names.download}}/{{file.yt_id}}">yes</a>{% else %}no{% endif %}</td></tr>
</table>
-<form action="/{{page_names.file}}/{{file.yt_id}}" method="POST" />
+<form action="/{{page_names.file}}/{{file.rel_path_b64}}" method="POST" />
{% for flag_name in flag_names %}
{{ flag_name }}: <input type="checkbox" name="{{flag_name}}" {% if file.is_flag_set(flag_name) %}checked {% endif %} /><br />
{% endfor %}
diff --git a/src/templates/files.tmpl b/src/templates/files.tmpl
index 69543c8..9af8487 100644
--- a/src/templates/files.tmpl
+++ b/src/templates/files.tmpl
@@ -5,8 +5,8 @@
{{ macros.nav_head(page_names, "files") }}
<p>downloaded videos:</p>
<ul>
-{% for video_id, path in videos %}
-<li><a href="/{{page_names.file}}/{{video_id}}">{{ path }}</a>
+{% for rel_path_b64, full_path in files %}
+<li><a href="/{{page_names.file}}/{{rel_path_b64}}">{{ full_path }}</a>
{% endfor %}
</ul>
{% endblock %}
diff --git a/src/ytplom/misc.py b/src/ytplom/misc.py
index cd86352..06ab2a0 100644
--- a/src/ytplom/misc.py
+++ b/src/ytplom/misc.py
@@ -5,6 +5,7 @@ from typing import TypeAlias, Optional, NewType, Callable, Self, Any
from os import chdir, environ, getcwd, makedirs, scandir, remove as os_remove
from os.path import (dirname, isdir, isfile, exists as path_exists,
join as path_join, splitext, basename)
+from base64 import urlsafe_b64encode, urlsafe_b64decode
from random import shuffle
from time import time, sleep
from datetime import datetime, timedelta
@@ -43,13 +44,14 @@ FlagName = NewType('FlagName', str)
FlagsInt = NewType('FlagsInt', int)
AmountDownloads = NewType('AmountDownloads', int)
PlayerUpdateId = NewType('PlayerUpdateId', str)
+B64Str = NewType('B64Str', str)
PageNames: TypeAlias = dict[str, PathStr]
DownloadsIndex: TypeAlias = dict[YoutubeId, PathStr]
TemplateContext: TypeAlias = dict[
str, None | bool | PlayerUpdateId | Optional[PathStr] | YoutubeId
| QueryText | QuotaCost | list[FlagName] | 'VideoFile' | 'YoutubeVideo'
| PageNames | list['YoutubeVideo'] | list['YoutubeQuery']
- | list[tuple[YoutubeId, PathStr]] | list[tuple[PathStr, PathStr]]]
+ | list[tuple[B64Str, PathStr]] | list[tuple[PathStr, PathStr]]]
# major expected directories
PATH_HOME = PathStr(environ.get('HOME', ''))
@@ -73,7 +75,7 @@ NAME_TEMPLATE_YT_VIDEO = PathStr('yt_result.tmpl')
NAME_TEMPLATE_PLAYLIST = PathStr('playlist.tmpl')
# page names
-PAGE_NAMES: PageNames = {
+PAGE_NAMES = {
'download': PathStr('dl'),
'file': PathStr('file'),
'files': PathStr('files'),
@@ -202,6 +204,7 @@ class DbConnection:
class DbData:
"""Abstraction of common DB operation."""
+ id_name: str = 'id'
_table_name: str
_cols: tuple[str, ...]
@@ -223,7 +226,8 @@ class DbData:
@classmethod
def get_one(cls, conn: DbConnection, id_: str) -> Self:
"""Return single entry of id_ from DB."""
- sql = SqlText(f'SELECT * FROM {cls._table_name} WHERE id = ?')
+ sql = SqlText(f'SELECT * FROM {cls._table_name} '
+ f'WHERE {cls.id_name} = ?')
row = conn.exec(sql, (id_,)).fetchone()
if not row:
msg = f'no entry found for ID "{id_}" in table {cls._table_name}'
@@ -336,6 +340,7 @@ class YoutubeVideo(DbData):
class VideoFile(DbData):
"""Collects data about downloaded files."""
+ id_name = 'rel_path'
_table_name = 'files'
_cols = ('rel_path', 'yt_id', 'flags', 'last_update')
last_update: DatetimeStr
@@ -366,6 +371,16 @@ class VideoFile(DbData):
raise NotFoundException(f'no entry for file to Youtube ID {yt_id}')
return cls._from_table_row(row)
+ @classmethod
+ def get_by_b64(cls, conn: DbConnection, rel_path_b64: B64Str) -> Self:
+ """Retrieve by .rel_path provided as urlsafe_b64 encoding."""
+ return cls.get_one(conn, urlsafe_b64decode(rel_path_b64).decode())
+
+ @property
+ def rel_path_b64(self) -> B64Str:
+ """Return .rel_path as urlsafe_b64 e3ncoding."""
+ return B64Str(urlsafe_b64encode(self.rel_path.encode()).decode())
+
@property
def full_path(self) -> PathStr:
"""Return self.rel_path suffixed under PATH_DOWNLOADS."""
@@ -612,11 +627,17 @@ class DownloadsManager:
return [f.rel_path for f in self._files if f.missing]
@property
- def ids_to_paths(self) -> DownloadsIndex:
- """Return mapping YoutubeIds:paths of files downloaded to them."""
+ def ids_to_full_paths(self) -> DownloadsIndex:
+ """Return mapping YoutubeIds:full paths of files downloaded to them."""
self._sync_db()
return {f.yt_id: f.full_path for f in self._files}
+ @property
+ def rel_paths_b64_to_rel_paths(self) -> list[tuple[B64Str, PathStr]]:
+ """Return mapping .rel_paths b64-encoded:rel_paths of known files."""
+ self._sync_db()
+ return [(f.rel_path_b64, f.rel_path) for f in self._files]
+
@property
def ids_unfinished(self) -> set[YoutubeId]:
"""Return set of IDs of videos awaiting or currently in download."""
@@ -634,8 +655,9 @@ class DownloadsManager:
def queue_download(self, video_id: YoutubeId) -> None:
"""Add video_id to download queue *if* not already processed."""
- pre_existing = self.ids_unfinished | set(self._to_download
- + list(self.ids_to_paths))
+ pre_existing = (self.ids_unfinished
+ | set(self._to_download
+ + list(self.ids_to_full_paths)))
if video_id not in pre_existing:
self._to_download += [video_id]
@@ -695,7 +717,7 @@ class TaskHandler(BaseHTTPRequestHandler):
if PAGE_NAMES['playlist'] == page_name:
self._receive_player_command(list(postvars.keys())[0])
elif PAGE_NAMES['file'] == page_name:
- self._receive_video_flag(YoutubeId(toks_url[2]),
+ self._receive_video_flag(B64Str(toks_url[2]),
[FlagName(k) for k in postvars])
elif PAGE_NAMES['yt_queries'] == page_name:
self._receive_yt_query(QueryText(postvars['query'][0]))
@@ -715,11 +737,11 @@ class TaskHandler(BaseHTTPRequestHandler):
self._send_http(headers=[('Location', '/')], code=302)
def _receive_video_flag(self,
- yt_id: YoutubeId,
+ rel_path_b64: B64Str,
flag_names: list[FlagName]
) -> None:
conn = DbConnection()
- file = VideoFile.get_by_yt_id(conn, yt_id)
+ file = VideoFile.get_by_b64(conn, rel_path_b64)
flags = FlagsInt(0)
for flag_name in flag_names:
flags = FlagsInt(file.flags | FILE_FLAGS[flag_name])
@@ -728,7 +750,7 @@ class TaskHandler(BaseHTTPRequestHandler):
conn.commit_close()
file.ensure_absence_if_deleted()
self._send_http(headers=[('Location',
- f'/{PAGE_NAMES["file"]}/{yt_id}')],
+ f'/{PAGE_NAMES["file"]}/{rel_path_b64}')],
code=302)
def _receive_yt_query(self, query_txt: QueryText) -> None:
@@ -799,7 +821,7 @@ class TaskHandler(BaseHTTPRequestHandler):
elif PAGE_NAMES['files'] == page_name:
self._send_files_index()
elif PAGE_NAMES['file'] == page_name:
- self._send_file_data(YoutubeId(toks_url[2]))
+ self._send_file_data(B64Str(toks_url[2]))
elif PAGE_NAMES['yt_result'] == page_name:
self._send_yt_result(YoutubeId(toks_url[2]))
elif PAGE_NAMES['missing'] == page_name:
@@ -841,8 +863,8 @@ class TaskHandler(BaseHTTPRequestHandler):
self._send_http(img, [('Content-type', 'image/jpg')])
def _send_or_download_video(self, video_id: YoutubeId) -> None:
- if video_id in self.server.downloads.ids_to_paths:
- with open(self.server.downloads.ids_to_paths[video_id],
+ if video_id in self.server.downloads.ids_to_full_paths:
+ with open(self.server.downloads.ids_to_full_paths[video_id],
'rb') as video_file:
video = video_file.read()
self._send_http(content=video)
@@ -883,24 +905,23 @@ class TaskHandler(BaseHTTPRequestHandler):
NAME_TEMPLATE_YT_VIDEO,
{'video_data': video_data,
'is_temp': video_id in self.server.downloads.ids_unfinished,
- 'file_path': self.server.downloads.ids_to_paths.get(video_id,
- None),
+ 'file_path': self.server.downloads.ids_to_full_paths.get(
+ video_id, None),
'youtube_prefix': YOUTUBE_URL_PREFIX,
'queries': linked_queries})
- def _send_file_data(self, yt_id: YoutubeId) -> None:
+ def _send_file_data(self, rel_path_b64: B64Str) -> None:
conn = DbConnection()
- file = VideoFile.get_by_yt_id(conn, yt_id)
+ file = VideoFile.get_by_b64(conn, rel_path_b64)
conn.commit_close()
self._send_rendered_template(
NAME_TEMPLATE_FILE_DATA,
{'file': file, 'flag_names': list(FILE_FLAGS)})
def _send_files_index(self) -> None:
- videos = [(id_, PathStr(basename(path)))
- for id_, path in self.server.downloads.ids_to_paths.items()]
- videos.sort(key=lambda t: t[1])
- self._send_rendered_template(NAME_TEMPLATE_FILES, {'videos': videos})
+ files = self.server.downloads.rel_paths_b64_to_rel_paths
+ files.sort(key=lambda t: t[1])
+ self._send_rendered_template(NAME_TEMPLATE_FILES, {'files': files})
def _send_missing_json(self) -> None:
self._send_http(