| import re | |
| from typing import Union | |
| import aiohttp | |
| from bs4 import BeautifulSoup | |
| from youtubesearchpython.__future__ import VideosSearch | |
| class RessoAPI: | |
| def __init__(self): | |
| self.regex = r"^(https:\/\/m.resso.com\/)(.*)$" | |
| self.base = "https://m.resso.com/" | |
| async def valid(self, link: str): | |
| """Check if the link is a valid Resso mobile URL.""" | |
| return bool(re.search(self.regex, link)) | |
| async def track(self, url, playid: Union[bool, str] = None): | |
| """Extract track info from a Resso link and find a matching YouTube video.""" | |
| if playid: | |
| url = self.base + url | |
| async with aiohttp.ClientSession() as session: | |
| async with session.get(url) as response: | |
| if response.status != 200: | |
| return False | |
| html = await response.text() | |
| soup = BeautifulSoup(html, "html.parser") | |
| title, des = None, None | |
| for tag in soup.find_all("meta"): | |
| if tag.get("property", None) == "og:title": | |
| title = tag.get("content", None) | |
| if tag.get("property", None) == "og:description": | |
| des = tag.get("content", None) | |
| try: | |
| des = des.split("·")[0] | |
| except Exception: | |
| pass | |
| if not title or not des: | |
| return None | |
| results = VideosSearch(title, limit=1) | |
| yt_results = (await results.next()).get("result", []) | |
| if not yt_results: | |
| return None | |
| result = yt_results[0] | |
| track_details = { | |
| "title": result["title"], | |
| "link": result["link"], | |
| "vidid": result["id"], | |
| "duration_min": result["duration"], | |
| "thumb": result["thumbnails"][0]["url"].split("?")[0], | |
| } | |
| return track_details, result["id"] | |