Work on a sane API (ca8b02d0) · Commits · git-mirror / Youtubevideofinder

.pylintrc

+1 −1

Original line number	Diff line number	Diff line
		@@ -35,7 +35,7 @@ confidence=
		# --enable=similarities". If you want to run only the classes checker, but have
		# no Warning level messages displayed, use"--disable=all --enable=classes
		# --disable=W"
		disable=redefined-builtin,line-too-long
		disable=redefined-builtin,line-too-long,invalid-name,too-few-public-methods
		enable=

app.py

+5 −2

Original line number	Diff line number	Diff line
		@@ -17,13 +17,13 @@ async def youtubev2(id):
		"""
		Provides backwards compatibility for the old endpoint.
		"""
		return (await lostmediafinder.YouTubeResponse.generateAsync(id)).coerce_to_api_version(2).json()
		return (await lostmediafinder.YouTubeResponse.generate(id)).coerce_to_api_version(2).json()

		async def wrapperYT(id):
		"""
		Wrapper for generateAsync
		"""
		return await lostmediafinder.YouTubeResponse.generateAsync(id)
		return await lostmediafinder.YouTubeResponse.generate(id)

		@app.route("/api/v<int:v>/<site>/<id>")
		@app.route("/api/v<int:v>/<id>")
		@@ -59,6 +59,9 @@ def parse_changelog(changelog):
		return parsed

		class E:
		"""
		What does this do? Good question
		"""
		name: str
		type: str

lostmediafinder/finder.py

+8 −8

Original line number	Diff line number	Diff line
		@@ -19,7 +19,7 @@ class WaybackMachine(YouTubeService):
		name = "Wayback Machine"

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		async 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 +53,7 @@ class InternetArchive(YouTubeService):
		]

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		async def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		responses = []
		is_dark = False
		for template in cls.items_tried:
		@@ -84,7 +84,7 @@ class GhostArchive(YouTubeService):
		Queries GhostArchive for the video you requested.
		"""
		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		async 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
		@@ -116,7 +116,7 @@ class Ya(YouTubeService):
		)

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False):
		async def _run(cls, id, includeRaw=True, asynchronous=False):
		vid = id
		assert cls._getFromConfig("ya", "enabled"), "#youtubearchive API access is not enabled"
		auth = HTTPBasicAuth(cls._getFromConfig("ya", "username"), cls._getFromConfig("ya", "password"))
		@@ -144,7 +144,7 @@ class Filmot(YouTubeService):
		cooldown: int = 2

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		async def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		enabled = cls._getFromConfig("filmot", "enabled")
		assert enabled, "Filmot API access is not enabled."
		key = cls._getFromConfig("filmot", "key")
		@@ -177,7 +177,7 @@ class Playboard(YouTubeService):
		note = "The Playboard scraper is unreliable; please verify values yourself."

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False):
		async def _run(cls, id, includeRaw=True, asynchronous=False):
		user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.%s.0.0 Safari/537.36"
		user_agent = user_agent % random.randint(0, 100)
		url = f"https://playboard.co/en/video/{id}"
		@@ -185,7 +185,7 @@ class Playboard(YouTubeService):
		rawraw = {"status_code": code, "ua_used": user_agent}
		lastupdated = time.time()
		available = None
		if code == 200 or code == 429:
		if code in {200, 429}:
		archived = True
		available = url
		elif code == 404:
		@@ -207,7 +207,7 @@ class NoxinfluencerService(YouTubeService):
		endpoint = "https://www.noxinfluencer.com/youtube/video-analytics/"

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False):
		async def _run(cls, id, includeRaw=True, asynchronous=False):
		user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.%s.0.0 Safari/537.36"
		user_agent = user_agent % random.randint(0, 100)
		url = cls.endpoint

lostmediafinder/types.py

+6 −31

Original line number	Diff line number	Diff line
		"""
		The classes that are used to store the response data.
		"""

		import asyncio as aio
		import copy
		import dataclasses
		import time
		import typing
		import re

		import cachetools.func
		import nest_asyncio

		from snscrape.base import _JSONDataclass as JSONDataclass

		@@ -25,8 +23,6 @@ def update_cnfig(ya, filmot, version):
		config.filmot = filmot
		config.config_version = version

		nest_asyncio.apply()

		T = typing.TypeVar("T", bound="YouTubeService") # pylint: disable=invalid-name
		# (this name is fine)

		@@ -72,7 +68,7 @@ class Service(JSONDataclass):
		return val

		@classmethod
		def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		async def _run(cls, id, includeRaw=True, asynchronous=False) -> T:
		raise NotImplementedError("Subclass Service and impl the _run function")

		@classmethod
		@@ -81,7 +77,7 @@ 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, **kwargs):
		async def run(cls, id: str, includeRaw=True, **kwargs):
		"""
		Retrieves the data from the service.
		Arguments:
		@@ -89,7 +85,7 @@ class Service(JSONDataclass):
		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.
		"""
		try:
		return cls._run(id, includeRaw=includeRaw, **kwargs)
		return await 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)
		@@ -101,14 +97,6 @@ class Service(JSONDataclass):
		available=None
		)

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

		@classmethod
		def getName(cls) -> str:
		"""
		@@ -166,7 +154,6 @@ class YouTubeResponse(JSONDataclass):
		return self

		def _convert_v3_to_v2(selfNEW):
		import copy
		self = copy.deepcopy(selfNEW)
		assert self.api_version == 3
		self.api_version = 2
		@@ -191,12 +178,11 @@ class YouTubeResponse(JSONDataclass):
		return bool(re.match(r"^[A-Za-z0-9_-]{10}[AEIMQUYcgkosw048]$", id))

		@classmethod
		def generate(cls, id, asyncio=False):
		async def generate(cls, id, asyncio=False):
		"""
		Runs all the Services.
		Arguments:
		id: The video ID
		asyncio: Whether or not to use asyncio.run_until_complete; this is implied if you use generateAsync
		"""
		if not cls.verifyId(id):
		return cls(status="bad.id", id=id, keys=[])
		@@ -204,21 +190,10 @@ class YouTubeResponse(JSONDataclass):
		services = cls._get_services()
		for subclass in services:
		result = None
		if asyncio:
		result = aio.get_event_loop().run_until_complete(subclass.runAsync(id))
		else:
		result = subclass.run(id)
		result = await subclass.run(id)
		keys.append(result)
		return cls(id=id, status="ok", keys=keys)

		@classmethod
		async def generateAsync(cls, args, *kwargs):
		"""
		Runs all the Services asynchronously.
		"""
		kwargs['asyncio'] = True
		return cls.generate(args, *kwargs)

		def __str__(self):
		services = "Services:\n"
		for i in self.keys: