Fix a few bugs, allow modifying config on the fly (610a1e9a) · Commits · git-mirror / Youtubevideofinder

config.example.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -6,3 +6,4 @@ class ya:
		class filmot:
		key = "fill this in"

		config_version = 2
		No newline at end of file

lostmediafinder/main.py

+7 −6

Original line number	Diff line number	Diff line
		@@ -9,10 +9,11 @@ from switch import Switch

		from . import Response

		@click.command()
		@click.command
		@click.option("--format", default="text", help="Selects which format to output to stdout.", type=click.Choice(["json", "text"]))
		@click.argument("id")
		def main(id: str, format: str) -> int:
		@click.pass_context
		def main(ctx, id: str, format: str) -> int:
		"""
		Parses CLI arguments and returns the Response for the video ID <IDENT>.

		@@ -31,8 +32,8 @@ def main(id: str, format: str) -> int:
		click.echo(str(response).strip())
		else:
		raise AssertionError("This should never occur!")
		errors = [item for item in response.keys if item.error]
		errors = [service for service in response.keys if service.error]
		code = 2 if errors else 0
		return code
		ctx.exit(code)

		sys.exit(main())
		main() # pylint: disable=no-value-for-parameter

lostmediafinder/finder.py

+12 −16

Original line number	Diff line number	Diff line
		@@ -9,7 +9,6 @@ import requests
		from requests.auth import HTTPBasicAuth
		from switch import Switch

		import config
		from .types import Service, T

		class WaybackMachine(Service):
		@@ -19,7 +18,7 @@ class WaybackMachine(Service):
		name = "Wayback Machine"

		@classmethod
		def _run(cls, id, includeRaw=True) -> T:
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		ismeta = False
		lien = f"https://web.archive.org/web/2oe_/http://wayback-fakeurl.archive.org/yt/{id}"
		response = requests.get(lien, allow_redirects=False, timeout=15)
		@@ -53,7 +52,7 @@ class InternetArchive(Service):
		]

		@classmethod
		def _run(cls, id, includeRaw=True) -> T:
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		responses = []
		is_dark = False
		for template in cls.items_tried:
		@@ -84,7 +83,7 @@ class GhostArchive(Service):
		Queries GhostArchive for the video you requested.
		"""
		@classmethod
		def _run(cls, id, includeRaw=True) -> T:
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		link = f"https://ghostarchive.org/varchive/{id}"
		code = requests.get(link).status_code
		rawraw = code if includeRaw else None
		@@ -114,15 +113,12 @@ class Ya(Service):
		note = ("To retrieve a video from #youtubearchive, join #youtubearchive on hackint IRC and ask for help. "
		"Remember <a href='https://wiki.archiveteam.org/index.php/Archiveteam:IRC#How_do_I_chat_on_IRC?'>IRC etiquette</a>!"
		)
		enabled = config.ya.enabled
		username = config.ya.username
		password = config.ya.password

		@classmethod
		def _run(cls, id, includeRaw=True):
		def _run(cls, id, includeRaw=True, asynchronous=False):
		vid = id
		assert cls.enabled, "#youtubearchive API access is not enabled"
		auth = HTTPBasicAuth(cls.username, cls.password)
		assert cls._getFromConfig("enabled"), "#youtubearchive API access is not enabled"
		auth = HTTPBasicAuth(cls._getFromConfig("username"), cls._getFromConfig("password"))
		comments = False
		count = requests.get("https://ya.borg.xyz/cgi-bin/capture-count?v=" + vid, auth=auth, timeout=5).text
		if not count:
		@@ -141,20 +137,20 @@ class Filmot(Service):
		"""
		Queries Filmot for the video you requested.
		"""
		key = config.filmot.key
		enabled = getattr(config.filmot, "enabled", False)

		lastretrieved: int = 0
		cooldown: int = 2

		@classmethod
		def _run(cls, id, includeRaw=True) -> T:
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		enabled = cls._getFromConfig("enabled")
		assert enabled, "Filmot API access is not enabled."
		key = cls._getFromConfig("key")
		while time.time() - cls.lastretrieved < cls.cooldown:
		time.sleep(0.1)
		lastupdated = time.time()
		cls.lastretrieved = time.time()
		lastupdated = time.time()
		assert cls.enabled, "Filmot API access is not enabled."
		metadata = requests.get(f"https://filmot.com/api/getvideos?key={cls.key}&id={id}&flags=1").json()
		metadata = requests.get(f"https://filmot.com/api/getvideos?key={key}&id={id}&flags=1").json()
		rawraw = metadata if includeRaw else None
		if len(metadata) > 0: # pylint: disable=simplifiable-if-statement
		archived = True

lostmediafinder/types.py

+31 −19

Original line number	Diff line number	Diff line
		@@ -13,6 +13,18 @@ import nest_asyncio

		from snscrape.base import _JSONDataclass as JSONDataclass

		import config
		def update_cnfig(ya, filmot, version):
		"""
		Updates the configuration for the #youtubearchive and Filmot services.

		Check lostmediafinder.config for documentation.
		"""

		config.ya = ya
		config.filmot = filmot
		config.config_version = version

		nest_asyncio.apply()

		T = typing.TypeVar("T", bound="Service") # pylint: disable=invalid-name
		@@ -48,24 +60,13 @@ class Service(JSONDataclass):
		suppl: str = ""
		error: bool = False

		@classmethod
		def _run(cls, id, includeRaw=True) -> T:
		raise NotImplementedError("Subclass Service and impl the _run function")
		@staticmethod
		def _getFromConfig(key):
		return getattr(config, key)

		@classmethod
		def __run(cls, id, includeRaw=True) -> T:
		try:
		return cls._run(id, includeRaw=includeRaw)
		except Exception as ename: # pylint: disable=broad-except
		note = f"An error occured while retrieving data from {cls.getName()}."
		print(ename)
		rawraw = f"{type(ename)}{repr(ename)}" if includeRaw else None
		return cls(
		archived=False, capcount=0, error=True,
		lastupdated=time.time(), name=cls.getName(), note=note,
		rawraw=rawraw, metaonly=False, comments=False,
		available=None
		)
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		raise NotImplementedError("Subclass Service and impl the _run function")

		@classmethod
		# cache has a max of 128 items; items are cached for 600 seconds (10min)
		@@ -73,14 +74,25 @@ class Service(JSONDataclass):
		# maxsize=128, ttl=600
		# might add this to config.py later
		@cachetools.func.ttl_cache
		def run(cls, id: str, includeRaw=True):
		def run(cls, id: str, includeRaw=True, **kwargs):
		"""
		Retrieves the data from the service.
		Arguments:
		id (str): The video ID.
		includeRaw (bool): Whether or not to include the raw data as sent from the service. If you don't need this data, turn this off; it's only the default for compatibility.
		"""
		return cls.__run(id, includeRaw)
		try:
		return cls._run(id, includeRaw=includeRaw, **kwargs)
		except Exception as ename: # pylint: disable=broad-except
		note = f"An error occured while retrieving data from {cls.getName()}."
		print(ename)
		rawraw = f"{type(ename)}{repr(ename)}" if includeRaw else None
		return cls(
		archived=False, capcount=0, error=True,
		lastupdated=time.time(), name=cls.getName(), note=note,
		rawraw=rawraw, metaonly=False, comments=False,
		available=None
		)

		@classmethod
		async def runAsync(cls, id, includeRaw=True):
		@@ -88,7 +100,7 @@ class Service(JSONDataclass):
		Runs cls.run(...) but it's async.
		This currently still uses blocking networking (requests)!
		"""
		return cls.run(id, includeRaw)
		return cls.run(id, includeRaw, asynchronous=True)

		@classmethod
		def getName(cls) -> str: