File size: 10,373 Bytes
7e58923 4bc7387 510a8f2 4bc7387 63e3048 6146a19 ae3ede2 58138cb e1c913c 63c1006 4bc7387 7e58923 6a45bf5 5ce8d0d 7e58923 f94628b ded5ac7 0f09a72 f94628b 6b74354 f94628b c07416b 0f09a72 ad3ab65 0f09a72 e843d84 bae3b41 ad3ab65 0f09a72 bb5ced0 160098c 0f09a72 cb07cc1 0f09a72 160098c 0f09a72 160098c 0f09a72 160098c 7191a0b 0f09a72 179ee8b 160098c 179ee8b dbc7bc9 0f09a72 dbc7bc9 c07416b d69267d 8a6f020 2e809a6 8a6f020 c07416b e1c913c c07416b 6146a19 c07416b 4bc7387 ddfd898 6146a19 e1c913c 6146a19 2dbf28a c07416b 4bc7387 5ce8d0d c07416b 4bc7387 f19d74e c07416b 63c1006 e1c913c e6bda95 5ce8d0d 44f1011 e1c913c 44f1011 e1c913c 63c1006 f19d74e e1c913c 44f1011 e1c913c 44f1011 e1c913c 44f1011 e1c913c 44f1011 e1c913c c07416b 5ce8d0d 4bc7387 7e58923 74078ba 7e58923 c07416b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 |
from flask import Flask, render_template, jsonify, request
from ytmusicapi import YTMusic
import os
import logging
import requests
from datetime import datetime, timedelta
import time
import asyncio
import cloudscraper
from pydantic import BaseModel
from urllib.parse import urlparse, parse_qs
from collections import defaultdict
import threading
import aiohttp
app = Flask(__name__)
ytmusic = YTMusic()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@app.route('/')
def index():
return render_template('index.html')
@app.route('/search', methods=['POST'])
def search():
query = request.json.get('query', '')
search_results = ytmusic.search(query, filter="songs")
return jsonify(search_results)
@app.route('/searcht', methods=['POST'])
def searcht():
query = request.json.get('query', '')
logger.info(f"serch query: {query}")
search_results = ytmusic.search(query, filter="songs")
first_song = next((song for song in search_results if 'videoId' in song and song['videoId']), {}) if search_results else {}
return jsonify(first_song)
# Function to extract track ID from Amazon Music URL
def extract_amazon_track_id(url: str):
if "music.amazon.com" in url:
# Case 1: URL contains trackAsin (e.g., https://music.amazon.com/albums/B01N48U32A?trackAsin=B01NAE38YO&do=play)
parsed_url = urlparse(url)
query_params = parse_qs(parsed_url.query)
if "trackAsin" in query_params:
return query_params["trackAsin"][0]
# Case 2: URL is a direct track link (e.g., https://music.amazon.com/tracks/B0DNTPYT5S)
if "/tracks/" in url:
return url.split("/tracks/")[-1].split("?")[0]
return None
# Function to get track info from Song.link API
def get_song_link_info(url: str):
# Check if the URL is from Amazon Music
if "music.amazon.com" in url:
track_id = extract_amazon_track_id(url)
if track_id:
# Use the working format for Amazon Music tracks
api_url = f"https://api.song.link/v1-alpha.1/links?type=song&platform=amazonMusic&id={track_id}&userCountry=US"
else:
# If no track ID is found, use the original URL
api_url = f"https://api.song.link/v1-alpha.1/links?url={url}&userCountry=US"
else:
# For non-Amazon Music URLs, use the standard format
api_url = f"https://api.song.link/v1-alpha.1/links?url={url}&userCountry=US"
# Make the API call
response = requests.get(api_url)
if response.status_code == 200:
return response.json()
else:
return None
# Function to extract Tidal or YouTube URL
def extract_url(links_by_platform: dict, platform: str):
if platform in links_by_platform:
return links_by_platform[platform]["url"]
return None
# Function to extract track title and artist from entities
def extract_track_info(entities_by_unique_id: dict, platform: str):
for entity in entities_by_unique_id.values():
if entity["apiProvider"] == platform:
return entity["title"], entity["artistName"]
return None, None
@app.route('/match', methods=['POST'])
async def match():
data = request.json
track_url = data.get('url')
if not track_url:
raise HTTPException(status_code=400, detail="No URL provided")
track_info = get_song_link_info(track_url)
if not track_info:
raise HTTPException(status_code=404, detail="Could not fetch track info")
youtube_url = extract_url(track_info["linksByPlatform"], "youtube")
entityUniqueId = track_info["entityUniqueId"]
logger.info(f"songlink info: {entityUniqueId}")
title = track_info["entitiesByUniqueId"][entityUniqueId]["title"]
artist = track_info["entitiesByUniqueId"][entityUniqueId]["artistName"]
if youtube_url:
video_id = youtube_url.split("v=")[1] if "v=" in youtube_url else None
if title and artist:
filename = f"{title} - {artist}"
return {"url": youtube_url, "filename": filename, "track_id": video_id}
else:
return {"url": youtube_url, "filename": "Unknown Track - Unknown Artist", "track_id": video_id}
else:
search_query = f'{title}+{artist}'
search_results = ytmusic.search(search_query, filter="songs")
first_song = next((song for song in search_results if 'videoId' in song and song['videoId']), {}) if search_results else {}
if 'videoId' in first_song:
videoId = first_song["videoId"]
ym_url = f'https://www.youtube.com/watch?v={videoId}'
return {"filename": search_query, "url": ym_url, "track_id": videoId}
else:
raise HTTPException(status_code=404, detail="Video ID not found")
# If no URLs found, return an error
raise HTTPException(status_code=404, detail="No matching URL found")
class ApiRotator:
def __init__(self, apis):
self.apis = apis
self.last_successful_index = None
def get_prioritized_apis(self):
if self.last_successful_index is not None:
# Move the last successful API to the front
rotated_apis = (
[self.apis[self.last_successful_index]] +
self.apis[:self.last_successful_index] +
self.apis[self.last_successful_index+1:]
)
return rotated_apis
return self.apis
def update_last_successful(self, index):
self.last_successful_index = index
# In your function:
api_rotator = ApiRotator([
"https://cobalt-api.ayo.tf",
"https://dwnld.nichind.dev",
"https://cobalt-api.kwiatekmiki.com",
"http://34.107.254.11"
])
async def get_track_download_url(track_id: str, quality: str) -> str:
apis = api_rotator.get_prioritized_apis()
session = cloudscraper.create_scraper() # Requires cloudscraper package
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
for i, api_url in enumerate(apis):
try:
logger.info(f"Attempting to get download URL from: {api_url}")
y_url = f"https://youtu.be/{track_id}"
response = session.post(
api_url,
timeout=20,
json={"url": y_url, "audioFormat": "mp3", "downloadMode": "audio", "audioBitrate": quality},
headers=headers
)
logger.info(f"Response status: {response.status_code}")
logger.info(f"Response content: {response.content}")
if response.headers.get('content-type', '').startswith('application/json'):
json_response = response.json()
error_code = json_response.get("error", {}).get("code", "")
if error_code == "error.api.content.video.unavailable":
logger.warning(f"Video unavailable error from {api_url}")
break # Only break for specific error
if "url" in json_response:
api_rotator.update_last_successful(i)
return json_response["url"]
except Exception as e:
logger.error(f"Failed with {api_url}: {str(e)}")
continue
logger.error(f"No download URL found")
return {"error": "Download URL not found"}
async def get_audio_download_url(track_id: str, quality: str) -> str:
youtube_url = f'https://www.youtube.com/watch?v={track_id}'
donwnload_url = f'https://chrunos-shadl.hf.space/yt/dl?url={youtube_url}&type=audio'
return donwnload_url
async def get_download_url(track_id, quality):
url = "https://chrunos-ytdl2.hf.space/download"
youtube_url = f'https://www.youtube.com/watch?v={track_id}'
async with aiohttp.ClientSession() as session:
data = {"url": youtube_url}
try:
async with session.post(url, json=data) as response:
if response.status == 200:
result = await response.json()
return result.get('download_url')
else:
print(f"请求失败,状态码: {response.status}")
return None
except aiohttp.ClientError as e:
print(f"发生客户端错误: {e}")
return None
@app.route('/track_dl', methods=['POST'])
async def track_dl():
data = request.get_json()
track_id = data.get('track_id')
quality = data.get('quality', '128')
try:
quality_num = int(quality)
if quality_num > 128 or quality.upper() == 'FLAC':
return jsonify({
"error": "Quality above 128 or FLAC is for Premium users Only.",
"premium": "https://chrunos.com/premium-shortcuts/"
}), 400
dl_url = await get_download_url(track_id, quality)
logger.info(dl_url)
if dl_url and "http" in dl_url:
result = {
"url": dl_url,
"premium": "https://chrunos.com/premium-shortcuts/"
}
return jsonify(result)
else:
return jsonify({
"error": "Failed to Fetch the Track.",
"premium": "https://chrunos.com/premium-shortcuts/"
}), 400
except ValueError:
return jsonify({
"error": "Invalid quality value provided. It should be a valid integer or FLAC.",
"premium": "https://chrunos.com/premium-shortcuts/"
}), 400
@app.route('/get_artist', methods=['GET'])
def get_artist():
artist_id = request.args.get('id')
artist_info = ytmusic.get_artist(artist_id)
return jsonify(artist_info)
@app.route('/get_album', methods=['GET'])
def get_album():
album_id = request.args.get('id')
album_info = ytmusic.get_album(album_id)
return jsonify(album_info)
@app.route('/get_song', methods=['GET'])
def get_song():
song_id = request.args.get('id')
song_info = ytmusic.get_song(song_id)
return jsonify(song_info)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=7860)
|