import asyncio import logging import aiohttp import aiofiles import os import json from typing import List, Dict, Optional from config import Config class ShazamService: """ Real Shazam HTTP API bilan ishlash uchun service klass Faqat real API, demo ma'lumotlar yo'q! """ def __init__(self): # Real Shazam API endpoints self.base_url = "https://shazam.p.rapidapi.com" self.headers = { "X-RapidAPI-Key": "your-real-api-key-here", # Bu yerda real API key bo'lishi kerak "X-RapidAPI-Host": "shazam.p.rapidapi.com" } logging.info("✅ ShazamService real HTTP API bilan ishga tushdi!") async def search_songs(self, query: str, limit: int = 10) -> List[Dict]: """ Qo'shiqlarni qidiruv - faqat real HTTP API """ if not query or not query.strip(): logging.warning("Bo'sh qidiruv so'rovi") return [] try: logging.info(f"🔍 Real HTTP API qidiruv: '{query}' (limit: {limit})") # Real Shazam API endpoint url = f"{self.base_url}/search" params = { 'term': query.strip(), 'locale': 'en-US', 'offset': '0', 'limit': str(limit) } async with aiohttp.ClientSession() as session: async with session.get(url, headers=self.headers, params=params, timeout=10) as response: if response.status == 200: data = await response.json() songs = [] if 'tracks' in data and 'hits' in data['tracks']: for hit in data['tracks']['hits']: if hit.get('type') != 'MUSIC': continue # API response formatini to'g'ri parse qilish heading = hit.get('heading', {}) title = heading.get('title', '').strip() artist = heading.get('subtitle', '').strip() if not title or not artist: continue song_info = { 'id': hit.get('key', f'track_{len(songs)}_{query}'), 'title': title, 'artist': artist, 'album': '', 'duration': 0, 'cover_url': '', 'preview_url': '', 'download_url': '', 'file_size': 0, 'genre': '', 'year': '' } # Rasm URL images = hit.get('images', {}) song_info['cover_url'] = ( images.get('default', '') or images.get('play', '') or images.get('blurred', '') ) # Preview URL stores = hit.get('stores', {}) if 'apple' in stores: apple_data = stores['apple'] song_info['preview_url'] = apple_data.get('previewurl', '') songs.append(song_info) if not songs: logging.warning(f"Real HTTP API dan '{query}' uchun natija topilmadi") return [] logging.info(f"✅ '{query}' uchun {len(songs)} ta real natija topildi") return songs elif response.status == 401: logging.error("❌ Real API key noto'g'ri yoki muddati tugagan") return [] elif response.status == 429: logging.error("❌ Real API limit tugagan") return [] else: logging.error(f"❌ Real HTTP API xatosi: {response.status}") return [] except asyncio.TimeoutError: logging.error("❌ Real HTTP API timeout") return [] except Exception as e: logging.error(f"❌ Real HTTP API qidiruv xatosi: {e}") return [] async def recognize_audio(self, audio_file_path: str) -> Optional[Dict]: """ Audio faylni tanib olish - faqat real HTTP API """ if not os.path.exists(audio_file_path): logging.error(f"Audio fayl topilmadi: {audio_file_path}") return None try: logging.info(f"🎵 Real HTTP API audio tanib olish: {audio_file_path}") # Real Shazam API endpoint url = f"{self.base_url}/songs/v2/detect" # Audio faylni yuklash async with aiofiles.open(audio_file_path, 'rb') as f: audio_data = await f.read() # Multipart form data data = aiohttp.FormData() data.add_field('upload_file', audio_data, filename=os.path.basename(audio_file_path)) async with aiohttp.ClientSession() as session: async with session.post(url, headers=self.headers, data=data, timeout=30) as response: if response.status == 200: result = await response.json() if 'track' in result: track_data = result['track'] song_info = { 'id': track_data.get('key', f'recognized_{os.path.basename(audio_file_path)}'), 'title': track_data.get('title', '').strip(), 'artist': track_data.get('subtitle', '').strip(), 'album': '', 'duration': 0, 'cover_url': '', 'preview_url': '', 'download_url': '', 'genre': '', 'year': '' } # Rasm URL if 'images' in track_data and track_data['images']: song_info['cover_url'] = ( track_data['images'].get('coverart', '') or track_data['images'].get('background', '') ) # Preview URL if 'hub' in track_data and 'actions' in track_data['hub']: for action in track_data['hub']['actions']: if action.get('type') == 'uri' and 'uri' in action: song_info['preview_url'] = action['uri'] break if song_info['title'] and song_info['artist']: logging.info(f"✅ Real HTTP API audio tanib olindi: {song_info['title']} - {song_info['artist']}") return song_info logging.warning("Real HTTP API audio tanib olmadi") return None elif response.status == 401: logging.error("❌ Real API key noto'g'ri") return None else: logging.error(f"❌ Real HTTP API audio tanib olish xatosi: {response.status}") return None except Exception as e: logging.error(f"❌ Real HTTP API audio tanib olish xatosi: {e}") return None async def get_song_details(self, song_id: str) -> Optional[Dict]: """ Qo'shiq tafsilotlarini olish - faqat real HTTP API """ if not song_id: return None try: logging.info(f"📋 Real HTTP API song details: {song_id}") # Real Shazam API endpoint url = f"{self.base_url}/songs/get-details" params = {'key': song_id, 'locale': 'en-US'} async with aiohttp.ClientSession() as session: async with session.get(url, headers=self.headers, params=params, timeout=10) as response: if response.status == 200: track_info = await response.json() if track_info: song_info = { 'id': song_id, 'title': track_info.get('title', '').strip(), 'artist': track_info.get('subtitle', '').strip(), 'album': '', 'duration': 0, 'cover_url': '', 'preview_url': '', 'download_url': '', 'lyrics': '', 'genre': '', 'year': '', 'label': '' } # Rasm URL if 'images' in track_info and track_info['images']: song_info['cover_url'] = ( track_info['images'].get('coverart', '') or track_info['images'].get('background', '') ) # Preview URL if 'hub' in track_info and 'actions' in track_info['hub']: for action in track_info['hub']['actions']: if action.get('type') == 'uri' and 'uri' in action: song_info['preview_url'] = action['uri'] break # Qo'shiq matnini olish try: lyrics_url = f"{self.base_url}/songs/get-lyrics" lyrics_params = {'key': song_id} async with session.get(lyrics_url, headers=self.headers, params=lyrics_params, timeout=10) as lyrics_response: if lyrics_response.status == 200: lyrics_data = await lyrics_response.json() if 'lyrics' in lyrics_data: song_info['lyrics'] = lyrics_data['lyrics'] except Exception as e: logging.debug(f"Lyrics olishda xato: {e}") logging.info(f"✅ Real HTTP API song details olindi: {song_info['title']}") return song_info logging.warning(f"Real HTTP API song details topilmadi: {song_id}") return None elif response.status == 401: logging.error("❌ Real API key noto'g'ri") return None else: logging.error(f"❌ Real HTTP API song details xatosi: {response.status}") return None except Exception as e: logging.error(f"❌ Real HTTP API song details xatosi: {e}") return None async def get_top_tracks(self, genre: str = 'POP', limit: int = 10) -> List[Dict]: """ Top qo'shiqlarni olish - faqat real HTTP API """ try: logging.info(f"🏆 Real HTTP API top tracks: {genre} (limit: {limit})") # Turli mamlakatlarni sinab ko'ramiz countries = ['US', 'GB', 'UZ', 'RU', 'DE', 'FR'] for country in countries: try: logging.debug(f"Trying country: {country}") # Real Shazam API endpoint url = f"{self.base_url}/charts/track" params = { 'locale': 'en-US', 'pageSize': str(limit), 'startFrom': '0' } async with aiohttp.ClientSession() as session: async with session.get(url, headers=self.headers, params=params, timeout=10) as response: if response.status == 200: data = await response.json() songs = [] if 'tracks' in data: for track in data['tracks']: if not track.get('title') or not track.get('subtitle'): continue song_info = { 'id': track.get('key', f'top_{len(songs)}_{country}'), 'title': track.get('title', '').strip(), 'artist': track.get('subtitle', '').strip(), 'album': '', 'duration': 0, 'cover_url': '', 'preview_url': '', 'download_url': '', 'genre': genre, 'country': country } # Rasm URL if 'images' in track and track['images']: song_info['cover_url'] = ( track['images'].get('coverart', '') or track['images'].get('background', '') ) # Preview URL if 'hub' in track and 'actions' in track['hub']: for action in track['hub']['actions']: if action.get('type') == 'uri' and 'uri' in action: song_info['preview_url'] = action['uri'] break songs.append(song_info) if songs: logging.info(f"✅ {len(songs)} ta real top track olindi ({country})") return songs elif response.status == 401: logging.error("❌ Real API key noto'g'ri") return [] else: logging.debug(f"Country {country} failed: HTTP {response.status}") continue except Exception as e: logging.debug(f"Country {country} failed: {e}") continue logging.error("Hech bir mamlakatdan real top tracks olinmadi") return [] except Exception as e: logging.error(f"❌ Real HTTP API top tracks xatosi: {e}") return [] async def download_song(self, song_info: Dict, quality: str = '192') -> Optional[str]: """ Qo'shiqni yuklash - real API orqali """ try: logging.info(f"⬇️ Real HTTP API qo'shiq yuklash: {song_info.get('title', 'Unknown')} - {quality}kbps") # Real download endpoint (bu yerda real download service bo'lishi kerak) download_url = f"{self.base_url}/songs/download" params = { 'key': song_info['id'], 'quality': quality } async with aiohttp.ClientSession() as session: async with session.get(download_url, headers=self.headers, params=params, timeout=60) as response: if response.status == 200: # Faylni saqlash filename = f"{song_info['artist']} - {song_info['title']}.mp3" filename = self._clean_filename(filename) filepath = f"{Config.TEMP_DIR}{filename}" # Temp papkani yaratish os.makedirs(Config.TEMP_DIR, exist_ok=True) async with aiofiles.open(filepath, 'wb') as f: async for chunk in response.content.iter_chunked(8192): await f.write(chunk) # Fayl hajmini tekshirish file_size = os.path.getsize(filepath) if file_size < 1000: # 1KB dan kichik bo'lsa logging.error(f"Yuklangan fayl juda kichik: {file_size} bytes") os.remove(filepath) return None logging.info(f"✅ Real HTTP API qo'shiq yuklandi: {filename} ({file_size} bytes)") return filepath elif response.status == 401: logging.error("❌ Real API key noto'g'ri") return None elif response.status == 404: logging.error("❌ Qo'shiq yuklash uchun mavjud emas") return None else: logging.error(f"❌ Real HTTP API yuklash xatosi: HTTP {response.status}") return None except Exception as e: logging.error(f"❌ Real HTTP API qo'shiq yuklash xatosi: {e}") return None def _clean_filename(self, filename: str) -> str: """ Fayl nomini tozalash """ import re # Maxsus belgilarni olib tashlash filename = re.sub(r'[<>:"/\\|?*]', '', filename) # Unicode belgilarni normallash filename = filename.encode('ascii', 'ignore').decode('ascii') # Bo'sh joylarni _ bilan almashtirish filename = re.sub(r'\s+', '_', filename) # Uzunlikni cheklash if len(filename) > 100: name, ext = os.path.splitext(filename) filename = name[:96] + ext return filename or "unknown_song.mp3"