fix(musicseerr): enhance Tubidy search with query variations and censorship handling
Build and Push Docker Images / deploy (push) Blocked by required conditions Details
Build and Push Docker Images / build (api) (push) Successful in 1m10s Details
Build and Push Docker Images / build (nextgen) (push) Successful in 7m15s Details
Build and Push Docker Images / build (web) (push) Successful in 2m4s Details
Build and Push Docker Images / build (worker) (push) Has been cancelled Details
Build and Push Docker Images / build (musicseerr) (push) Has been cancelled Details

This commit is contained in:
fchinembiri 2026-07-10 20:57:29 +02:00
parent 0bf96f0213
commit f3779f00cd
1 changed files with 77 additions and 24 deletions

View File

@ -96,6 +96,7 @@ def scrape_tubidy_link(url: str, proxy_url: str) -> str:
def search_tubidy(query: str, proxy_url: str) -> list: def search_tubidy(query: str, proxy_url: str) -> list:
""" """
Searches Tubidy for the query and returns a list of results with ID, title, and link. Searches Tubidy for the query and returns a list of results with ID, title, and link.
Includes robust fallback variations if the initial search returns no results.
""" """
proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
headers = { headers = {
@ -115,31 +116,83 @@ def search_tubidy(query: str, proxy_url: str) -> list:
response = cf_requests.get(search_url, headers=headers, params=params, proxies=proxies, impersonate="chrome", timeout=15) response = cf_requests.get(search_url, headers=headers, params=params, proxies=proxies, impersonate="chrome", timeout=15)
response.raise_for_status() response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser") def parse_results(html: str) -> list:
results = [] soup = BeautifulSoup(html, "html.parser")
results_list = []
for media_body in soup.find_all("div", class_="media-body"):
a_tag = media_body.find("a")
if a_tag:
href = a_tag.get("href")
if href:
title = a_tag.get("aria-label") or a_tag.text.strip()
match = re.search(r'/watch/([^/]+)', href)
content_id = match.group(1) if match else None
link = href
if href.startswith("//"):
link = f"https:{href}"
elif href.startswith("/"):
link = f"{endpoint}{href}"
for media_body in soup.find_all("div", class_="media-body"): if content_id:
a_tag = media_body.find("a") results_list.append({
if a_tag: "id": content_id,
href = a_tag.get("href") "title": title,
if href: "link": link
title = a_tag.get("aria-label") or a_tag.text.strip() })
# Extract ID from /watch/content_id return results_list
match = re.search(r'/watch/([^/]+)', href)
content_id = match.group(1) if match else None
link = href
if href.startswith("//"):
link = f"https:{href}"
elif href.startswith("/"):
link = f"{endpoint}{href}"
if content_id: results = parse_results(response.text)
results.append({ if results:
"id": content_id, return results
"title": title,
"link": link # Fallback query variations if first search yielded no results
}) variations = []
return results
# Variation 1: Clean basic symbols and parentheticals
clean1 = re.sub(r'[\(\[][^\)\]]*[\)\]]', '', query)
clean1 = clean1.replace("-", " ")
clean1 = re.sub(r'\s+', ' ', clean1).strip()
variations.append(clean1)
# Variation 2: Clean and also remove feat/ft
clean2 = re.sub(r'(?i)\b(feat|ft)\.?\b.*$', '', clean1)
clean2 = re.sub(r'[^\w\s]', '', clean2)
clean2 = re.sub(r'\s+', ' ', clean2).strip()
variations.append(clean2)
# Variation 3: Handle censored words (e.g. fucked -> f*cked, fuck -> f*ck)
if "fucked" in clean2.lower() or "fuck" in clean2.lower():
clean3 = re.sub(r'(?i)\bfucked\b', 'f*cked', clean2)
clean3 = re.sub(r'(?i)\bfuck\b', 'f*ck', clean3)
variations.append(clean3)
# Variation 4: Try just the artist and the first 2 words of the title
parts = query.split(" - ", 1)
if len(parts) == 2:
artist = re.sub(r'[^\w\s]', '', parts[0]).strip()
title_words = re.sub(r'[^\w\s]', '', parts[1]).strip().split()
if len(title_words) > 1:
clean4 = f"{artist} {' '.join(title_words[:1])}"
variations.append(clean4)
# Try each variation in order
for var in variations:
var = var.strip()
if not var or var == query:
continue
logger.info(f"Retrying Tubidy search with variation: '{var}'")
params["q"] = var
try:
response = cf_requests.get(search_url, headers=headers, params=params, proxies=proxies, impersonate="chrome", timeout=15)
response.raise_for_status()
new_results = parse_results(response.text)
if new_results:
logger.info(f"Found {len(new_results)} results for variation '{var}'")
return new_results
except Exception as e:
logger.warning(f"Variation search for '{var}' failed: {e}")
return []
except Exception as e: except Exception as e:
logger.error(f"Failed to search Tubidy: {e}") logger.error(f"Failed to search Tubidy: {e}")
return [] return []