64d3082a8a
* Bump pylint from 2.13.5 to 3.1.0 Bumps [pylint](https://github.com/pylint-dev/pylint) from 2.13.5 to 3.1.0. - [Release notes](https://github.com/pylint-dev/pylint/releases) - [Commits](https://github.com/pylint-dev/pylint/compare/v2.13.5...v3.1.0) --- updated-dependencies: - dependency-name: pylint dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> * bend knee to pylint --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Jesse Bannon <jbann1994@gmail.com>
240 lines
8.2 KiB
Python
240 lines
8.2 KiB
Python
import math
|
|
from typing import Any
|
|
from typing import Dict
|
|
from typing import List
|
|
from typing import Optional
|
|
from typing import Set
|
|
from urllib.parse import urlparse
|
|
|
|
from ytdl_sub.entries.base_entry import BaseEntry
|
|
from ytdl_sub.entries.base_entry import BaseEntryT
|
|
from ytdl_sub.entries.entry import Entry
|
|
from ytdl_sub.entries.script.variable_definitions import VARIABLES
|
|
from ytdl_sub.entries.script.variable_definitions import VariableDefinitions
|
|
from ytdl_sub.entries.script.variable_types import MetadataVariable
|
|
|
|
v: VariableDefinitions = VARIABLES
|
|
|
|
|
|
# pylint: disable=protected-access
|
|
|
|
|
|
class EntryParent(BaseEntry):
|
|
@classmethod
|
|
def _sort_entries(cls, entries: List[BaseEntryT]) -> List[BaseEntryT]:
|
|
"""Try sorting by playlist_id first, then fall back to uid"""
|
|
return sorted(
|
|
entries,
|
|
key=lambda ent: (ent._kwargs_get(v.playlist_index.metadata_key, math.inf), ent.uid),
|
|
)
|
|
|
|
def __init__(self, entry_dict: Dict, working_directory: str):
|
|
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
|
|
self._parent_children: List["EntryParent"] = []
|
|
self._entry_children: List[Entry] = []
|
|
|
|
def parent_children(self) -> List["EntryParent"]:
|
|
"""This parent's children that are also parents"""
|
|
return self._parent_children
|
|
|
|
def entry_children(self) -> List[Entry]:
|
|
"""This parent's children that are entries"""
|
|
return self._entry_children
|
|
|
|
def num_children(self) -> int:
|
|
"""
|
|
Returns
|
|
-------
|
|
Total number of (nested) entry children
|
|
"""
|
|
return sum(parent.num_children() for parent in self.parent_children()) + len(
|
|
self.entry_children()
|
|
)
|
|
|
|
def _sibling_entry_metadata(self) -> List[Dict[str, Any]]:
|
|
sibling_entry_metadata: List[Dict[str, Any]] = []
|
|
variable_filter: Set[MetadataVariable] = (
|
|
v.required_entry_variables() | v.default_entry_variables()
|
|
)
|
|
for entry in self.entry_children():
|
|
sibling_entry_metadata.append(
|
|
{var.metadata_key: entry._kwargs_get(var.metadata_key) for var in variable_filter}
|
|
)
|
|
return sibling_entry_metadata
|
|
|
|
def _set_child_variables(
|
|
self, include_sibling_metadata: bool, parents: Optional[List["EntryParent"]] = None
|
|
) -> "EntryParent":
|
|
if parents is None:
|
|
parents = [self]
|
|
|
|
kwargs_to_add: Dict[str, Any] = {}
|
|
if include_sibling_metadata:
|
|
kwargs_to_add[v.sibling_metadata.metadata_key] = self._sibling_entry_metadata()
|
|
|
|
if len(parents) >= 1:
|
|
kwargs_to_add[v.playlist_metadata.metadata_key] = parents[-1]._kwargs
|
|
if len(parents) >= 2:
|
|
kwargs_to_add[v.source_metadata.metadata_key] = parents[-2]._kwargs
|
|
if len(parents) >= 3:
|
|
raise ValueError(
|
|
"ytdl-sub currently does support more than 3 layers of playlists/entries. "
|
|
"If you encounter this error, please file a ticket with the URLs used."
|
|
)
|
|
|
|
for entry_child in self.entry_children():
|
|
entry_child._kwargs = dict(entry_child._kwargs, **kwargs_to_add)
|
|
|
|
for parent_child in self.parent_children():
|
|
parent_child._set_child_variables(
|
|
include_sibling_metadata=include_sibling_metadata, parents=parents + [parent_child]
|
|
)
|
|
|
|
return self
|
|
|
|
def _read_children_from_entry_dicts(self, entry_dicts: List[Dict]) -> "EntryParent":
|
|
"""
|
|
Populates a tree of EntryParents that belong to this instance
|
|
"""
|
|
entries = [
|
|
EntryParent(
|
|
entry_dict=entry_dict,
|
|
working_directory=self.working_directory(),
|
|
)._read_children_from_entry_dicts(entry_dicts)
|
|
for entry_dict in entry_dicts
|
|
if entry_dict in self
|
|
]
|
|
|
|
self._parent_children = self._sort_entries(
|
|
[ent for ent in entries if self.is_entry_parent(ent)]
|
|
)
|
|
self._entry_children = self._sort_entries(
|
|
[ent.to_type(Entry) for ent in entries if self.is_entry(ent)]
|
|
)
|
|
|
|
return self
|
|
|
|
def get_thumbnail_url(self, thumbnail_id: str) -> Optional[str]:
|
|
"""
|
|
Downloads a specific thumbnail from a YTDL entry's thumbnail list
|
|
|
|
Parameters
|
|
----------
|
|
thumbnail_id:
|
|
Id of the thumbnail defined in the parent's thumbnail
|
|
|
|
Returns
|
|
-------
|
|
Desired thumbnail url if it exists. None if it does not.
|
|
"""
|
|
for thumbnail in self._kwargs_get("thumbnails", []):
|
|
if thumbnail["id"] == thumbnail_id:
|
|
return thumbnail["url"]
|
|
return None
|
|
|
|
def __contains__(self, item: Dict | BaseEntry) -> bool:
|
|
playlist_id: Optional[str] = None
|
|
if isinstance(item, dict):
|
|
playlist_id = item.get("playlist_id")
|
|
elif isinstance(item, BaseEntry):
|
|
playlist_id = item._kwargs_get("playlist_id")
|
|
|
|
if not playlist_id:
|
|
return False
|
|
|
|
return self.uid == playlist_id or any(item in child for child in self.parent_children())
|
|
|
|
@classmethod
|
|
def _get_disconnected_root_parent(
|
|
cls, url: str, parents: List["EntryParent"]
|
|
) -> Optional["EntryParent"]:
|
|
"""
|
|
Sometimes the root-level parent is disconnected via playlist_ids Find it if it exists.
|
|
"""
|
|
|
|
def _url_matches(parent: "EntryParent"):
|
|
url_parsed = urlparse(url)
|
|
parent_parsed = urlparse(parent.webpage_url)
|
|
return (
|
|
url_parsed.hostname == parent_parsed.hostname
|
|
and url_parsed.path == parent_parsed.path
|
|
)
|
|
|
|
def _uid_is_uploader_id(parent: "EntryParent"):
|
|
return parent.uid == parent.uploader_id
|
|
|
|
top_level_parents = [parent for parent in parents if parent.num_children() == 0]
|
|
|
|
# If more than 1 parent exists, assume the uploader_id is the root parent
|
|
if len(top_level_parents) > 1:
|
|
top_level_parents = [
|
|
parent for parent in top_level_parents if _uid_is_uploader_id(parent)
|
|
]
|
|
|
|
if len(top_level_parents) > 1:
|
|
top_level_parents = [parent for parent in top_level_parents if _url_matches(parent)]
|
|
|
|
if len(top_level_parents) == 0:
|
|
return None
|
|
if len(top_level_parents) == 1:
|
|
return top_level_parents[0]
|
|
raise ValueError(
|
|
"Detected multiple top-level parents. "
|
|
"Please file an issue on GitHub with the URLs used to produce this error"
|
|
)
|
|
|
|
@classmethod
|
|
def from_entry_dicts(
|
|
cls,
|
|
url: str,
|
|
entry_dicts: List[Dict],
|
|
working_directory: str,
|
|
include_sibling_metadata: bool,
|
|
) -> List["EntryParent"]:
|
|
"""
|
|
Reads all entry dicts and builds a tree of EntryParents
|
|
"""
|
|
parents = [
|
|
EntryParent(
|
|
entry_dict=entry_dict, working_directory=working_directory
|
|
)._read_children_from_entry_dicts(entry_dicts)
|
|
for entry_dict in entry_dicts
|
|
if cls.is_entry_parent(entry_dict)
|
|
]
|
|
|
|
if not parents:
|
|
return []
|
|
|
|
# If a disconnected root parent exists, connect it here
|
|
if (root_parent := cls._get_disconnected_root_parent(url, parents)) is not None:
|
|
parents.remove(root_parent)
|
|
root_parent._parent_children = parents
|
|
parents = [root_parent]
|
|
|
|
for parent in parents:
|
|
parent._set_child_variables(include_sibling_metadata=include_sibling_metadata)
|
|
|
|
return parents
|
|
|
|
@classmethod
|
|
def from_entry_dicts_with_no_parents(
|
|
cls,
|
|
parents: List["EntryParent"],
|
|
entry_dicts: List[Dict],
|
|
working_directory: str,
|
|
) -> List[Entry]:
|
|
"""
|
|
Reads all entries that do not have any parents
|
|
"""
|
|
|
|
def _in_any_parents(entry_dict: Dict):
|
|
return any(entry_dict in parent for parent in parents)
|
|
|
|
return [
|
|
Entry(
|
|
entry_dict=entry_dict,
|
|
working_directory=working_directory,
|
|
)
|
|
for entry_dict in entry_dicts
|
|
if cls.is_entry(entry_dict) and not _in_any_parents(entry_dict)
|
|
]
|