tubearchivist
/
tubearchivist
mirror of https://github.com/tubearchivist/tubearchivist.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

add type hints to filesystem scanner

This commit is contained in:
Simon 2023-08-09 09:45:51 +07:00
parent 313bbe8b49
commit 8411889db7
No known key found for this signature in database
GPG Key ID: 2C15AA5E89985DD4
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
tubearchivist/home/src/index/filesystem.py
View File

@ -15,26 +15,26 @@ from home.src.ta.helper import ignore_filelist
class Scanner: class Scanner:
"""scan index and filesystem""" """scan index and filesystem"""
VIDEOS = AppConfig().config["application"]["videos"] VIDEOS: str = AppConfig().config["application"]["videos"]
def __init__(self, task=False): def __init__(self, task=False) -> None:
self.task = task self.task = task
self.to_delete = False self.to_delete: set[str] = set()
self.to_index = False self.to_index: set[str] = set()
def scan(self): def scan(self) -> None:
"""scan the filesystem""" """scan the filesystem"""
downloaded = self._get_downloaded() downloaded: set[str] = self._get_downloaded()
indexed = self._get_indexed() indexed: set[str] = self._get_indexed()
self.to_index = downloaded - indexed self.to_index = downloaded - indexed
self.to_delete = indexed - downloaded self.to_delete = indexed - downloaded
def _get_downloaded(self): def _get_downloaded(self) -> set[str]:
"""get downloaded ids""" """get downloaded ids"""
if self.task: if self.task:
self.task.send_progress(["Scan your filesystem for videos."]) self.task.send_progress(["Scan your filesystem for videos."])
downloaded = set() downloaded: set = set()
channels = ignore_filelist(os.listdir(self.VIDEOS)) channels = ignore_filelist(os.listdir(self.VIDEOS))
for channel in channels: for channel in channels:
folder = os.path.join(self.VIDEOS, channel) folder = os.path.join(self.VIDEOS, channel)
@ -43,7 +43,7 @@ class Scanner:
return downloaded return downloaded
def _get_indexed(self): def _get_indexed(self) -> set:
"""get all indexed ids""" """get all indexed ids"""
if self.task: if self.task:
self.task.send_progress(["Get all videos indexed."]) self.task.send_progress(["Get all videos indexed."])
@ -52,13 +52,13 @@ class Scanner:
response = IndexPaginate("ta_video", data).get_results() response = IndexPaginate("ta_video", data).get_results()
return {i["youtube_id"] for i in response} return {i["youtube_id"] for i in response}
def apply(self): def apply(self) -> None:
"""apply all changes""" """apply all changes"""
self.delete() self.delete()
self.index() self.index()
self.url_fix() self.url_fix()
def delete(self): def delete(self) -> None:
"""delete videos from index""" """delete videos from index"""
if not self.to_delete: if not self.to_delete:
print("nothing to delete") print("nothing to delete")
@ -72,7 +72,7 @@ class Scanner:
for youtube_id in self.to_delete: for youtube_id in self.to_delete:
YoutubeVideo(youtube_id).delete_media_file() YoutubeVideo(youtube_id).delete_media_file()
def index(self): def index(self) -> None:
"""index new""" """index new"""
if not self.to_index: if not self.to_index:
print("nothing to index") print("nothing to index")
@ -91,7 +91,7 @@ class Scanner:
CommentList(self.to_index, task=self.task).index() CommentList(self.to_index, task=self.task).index()
def url_fix(self): def url_fix(self) -> None:
""" """
update path v0.3.6 to v0.3.7 update path v0.3.6 to v0.3.7
fix url not matching channel-videoid pattern fix url not matching channel-videoid pattern