Ich schreibe ein Programm was von der ilovemusic Website die Streams abspielen soll. Das funktioniert auch. Nur das Problem ist wenn ich die Scraper Klasse in der main.py aufrufe über scraper_task = asyncio.create_task(self.scraper.get_genresong_info(self.player.stream))
Dann wird die Methode init_browser
aufgerufen wo geprüft wird ob eine Browser Instanz bereits Aktiv ist oder nicht. Wenn nicht soll dieser Code ausgeführt werden:
async def init_browser(self):
logger.info("Initializing browser...")
if self.browser is None:
logger.warn("Browser not initialized. Initializing...")
try:
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch()
self.page = await self.browser.new_page()
logger.info(f"Navigating to {self.config.get_ilovemusic_provider()}...")
await self.page.goto(self.config.get_ilovemusic_provider())
logger.info("Browser initialized and page loaded.")
except Exception as e:
logger.error(f"Error during browser initialization: {e}", exc_info=True)
else:
logger.warn("Browser already initialized.")
return self.page, self.browser
Die Log naricht wird auch ausgegeben jedoch was im Try/exept steht wird ohne jegliche Fehlermeldung nicht ausgeführt. Wenn ich im Try Block print("Test")
schreibe wird der Code Normal ausgeführt. Könnte Jemand mir Bitte Weiter helfen ?
main.py
import asyncio
from base import Scraper, Player
class Main:
def __init__(self):
self.scraper = Scraper()
self.player = Player()
self.player.stream = "iloveradio"
async def run(self):
# player_task = asyncio.create_task(self.player.play())
scraper_task = asyncio.create_task(self.scraper.get_genresong_info(self.player.stream))
if __name__ == "__main__":
main = Main()
asyncio.run(main.run())
scraper.py
import asyncio
from base import Config
from base.logger import LoggingHandler
from playwright.async_api import async_playwright
logger = LoggingHandler("scraper").get_logger()
class Scraper:
def __init__(self):
self.config = Config()
self.browser = None
self.page = None
self.playwright = None
async def init_browser(self):
logger.info("Initializing browser...")
if self.browser is None:
logger.warn("Browser not initialized. Initializing...")
try:
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch()
self.page = await self.browser.new_page()
logger.info(f"Navigating to {self.config.get_ilovemusic_provider()}...")
await self.page.goto(self.config.get_ilovemusic_provider())
logger.info("Browser initialized and page loaded.")
except Exception as e:
logger.error(f"Error during browser initialization: {e}", exc_info=True)
else:
logger.warn("Browser already initialized.")
return self.page, self.browser
async def get_channels(self):
page, _ = await self.init_browser()
if page:
logger.info("Fetching channels...")
return await page.query_selector_all('section[id^="channel-"]')
else:
logger.error("Page not initialized.")
return []
async def get_genre_img(self, genre):
channels = await self.get_channels()
for section in channels:
data_channel_name = await section.get_attribute("data-channel-name")
if data_channel_name == genre:
img_tag = await section.query_selector("img.hidden")
if img_tag:
img_url = await img_tag.get_attribute("src")
logger.info("Image found. Returning image URL")
return self.config.get_ilovemusic_provider() + img_url
return "Image not found"
async def get_genresong_info(self, genre):
channels = await self.get_channels()
for section in channels:
data_channel_name = await section.get_attribute("data-channel-name")
if data_channel_name == genre:
title_elem = await section.query_selector('h3')
artist_elem = await section.query_selector('h2')
title = await title_elem.inner_text() if title_elem else "Song not found"
artist = await artist_elem.inner_text() if artist_elem else "Artist not found"
return {"title": title, "artist": artist}
return {"title": "Song not found", "artist": "Artist not found"}
async def get_current_song_info(self, genre):
previous_info = await self.get_genresong_info(genre)
img_url = await self.get_genre_img(genre)
while True:
try:
current_info = await self.get_genresong_info(genre)
if current_info != previous_info:
previous_info = current_info
logger.info(f"Title: {current_info['title']} Artist: {current_info['artist']}")
logger.info(f"Image URL: {img_url}")
await asyncio.sleep(5)
except Exception as e:
logger.error(f"Error: {e}")
break
async def close_browser(self):
if self.browser:
logger.warn("Closing browser...")
await self.browser.close()
self.browser = None
else:
logger.warn("Browser already closed.")