diff options
Diffstat (limited to 'news_feed.py')
| -rw-r--r-- | news_feed.py | 354 |
1 files changed, 227 insertions, 127 deletions
diff --git a/news_feed.py b/news_feed.py index c962e82..9bae903 100644 --- a/news_feed.py +++ b/news_feed.py @@ -20,31 +20,15 @@ Generic format for a news entry. All keys are considered to be nullable """ from site_scraper import SiteScraper, download_site_as_html -import konami.eamuse_app as eamuse_app -import bemani.sdvx as sound_voltex -import bemani.iidx as iidx -import bemani.ddr as ddr -import sega.chuni_jp as chunithm_jp -import bemani.polaris_chord as polaris_chord -import sega.chuni_intl as chuni_intl -import sega.maimaidx_jp as maimaidx_jp -import sega.maimaidx_intl as maimaidx_intl -import sega.ongeki_jp as ongeki_jp -import sega.idac as idac -import taito.music_diver as music_diver -import taito.street_fighter as street_fighter -import bandai_namco.taiko as taiko -import bandai_namco.wmmt as wmmt -import community.disc as disc -import community.wacca_plus.wacca_plus as wac_plus -import community.museca_plus as mus_plus -import community.rbdx as rbdx +from scrapers.base import NewsSource +import scrapers.registry as registry import constants import translate import summarizer from datetime import datetime + def _attach_llm_summaries(news_posts: list, game_name: str): for post in news_posts: image_urls = [img["image"] for img in post.get("images", []) if "image" in img] @@ -60,176 +44,292 @@ def _attach_llm_summaries(news_posts: list, game_name: str): post["is_ai_summary"] = True -def get_news(news_url: str, version=None) -> list: - if news_url == constants.SOUND_VOLTEX_EXCEED_GEAR_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(sound_voltex.parse_exceed_gear_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts, overrides=[("ボルテ", "SDVX")]) +# --------------------------------------------------------------------------- +# BEMANI (Specific feeds because these provide better information) +# --------------------------------------------------------------------------- + +@registry.register(constants.SOUND_VOLTEX_EXCEED_GEAR_NEWS_SITE) +class SoundVoltexSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from bemani.sdvx import parse_exceed_gear_news_site + site_data = download_site_as_html(constants.SOUND_VOLTEX_EXCEED_GEAR_NEWS_SITE) + news_posts = sorted(parse_exceed_gear_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts, overrides=[("ボルテ", "SDVX")]) + +# Can't find a Polaris feed on EAM app so this is here instead +@registry.register(constants.POLARIS_CHORD_NEWS_SITE) +class PolarisChordSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from bemani.polaris_chord import parse_polaris_chord_news_site + from bemani.iidx import KEY_TERMS_TL + site_data = download_site_as_html(constants.POLARIS_CHORD_NEWS_SITE) + news_posts = sorted( + parse_polaris_chord_news_site(site_data, constants.POLARIS_CHORD_RECENT_NEWS_LIMIT), + key=lambda x: x['timestamp'], + reverse=True, + ) + return translate.add_translate_text_to_en(news_posts, KEY_TERMS_TL) - elif news_url == constants.IIDX_PINKY_CRUSH_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(iidx.parse_pinky_crush_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts, iidx.KEY_TERMS_TL) - elif news_url == constants.POLARIS_CHORD_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(polaris_chord.parse_polaris_chord_news_site(site_data, constants.POLARIS_CHORD_RECENT_NEWS_LIMIT), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts, iidx.KEY_TERMS_TL) +# --------------------------------------------------------------------------- +# E-AMUSEMENT APP FEEDS (General Konami/BEMANI) +# --------------------------------------------------------------------------- - elif news_url == constants.EAMUSE_APP_API_ROUTE: - site_data = download_site_as_html(news_url+"/?uuid_to="+version+"&format=json") +@registry.register(constants.EAMUSE_APP_API_ROUTE) +class EamuseAppSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from konami.eamuse_app import parse_news_api_route + from bemani.iidx import KEY_TERMS_TL + site_data = download_site_as_html( + constants.EAMUSE_APP_API_ROUTE + "/?uuid_to=" + version + "&format=json" + ) match version: case constants.IIDX_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "IIDX_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts, iidx.KEY_TERMS_TL) + news_posts = sorted(parse_news_api_route(site_data, "IIDX_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts, KEY_TERMS_TL) case constants.DDR_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "DDR_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "DDR_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.SDVX_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "SOUND_VOLTEX_EAMUSEMENT", constants.EAMUSE_POST_SITE ), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "SOUND_VOLTEX_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.JUBEAT_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "JUBEAT_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "JUBEAT_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.POPN_MUSIC_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "POPN_MUSIC_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "POPN_MUSIC_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.GITADORA_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "GITADORA_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "GITADORA_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.NOSTALGIA_EAMUSE_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "NOSTALGIA_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "NOSTALGIA_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.DANCE_RUSH_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "DANCE_RUSH_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "DANCE_RUSH_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case constants.DANCE_AROUND_APP_ID: - news_posts= sorted(eamuse_app.parse_news_api_route(site_data, "DANCE_AROUND_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) + news_posts = sorted(parse_news_api_route(site_data, "DANCE_AROUND_EAMUSEMENT", constants.EAMUSE_POST_SITE), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) case _: raise ValueError("Cannot find provided e-amuse app gameId", version) - elif news_url == constants.DDR_WORLD_NEWS_SITE: - scraper = SiteScraper(headless=True) - site_data = scraper.get_page_source(news_url) - scraper.close() - news_posts = sorted(ddr.parse_ddr_world_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) +# --------------------------------------------------------------------------- +# SEGA +# --------------------------------------------------------------------------- + +@registry.register(constants.CHUNITHM_JP_NEWS_SITE) +class ChunithmJPSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.chuni_jp import parse_chuni_jp_news_site, parse_chuni_jp_post_images + site_data = download_site_as_html(constants.CHUNITHM_JP_NEWS_SITE) + if version not in [constants.CHUNITHM_VERSION.VERSE, constants.CHUNITHM_VERSION.X_VERSE]: + return [] + news_posts = sorted(parse_chuni_jp_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) news_posts = translate.add_translate_text_to_en(news_posts) + if constants.CHUNI_RECURSIVE_IMAGE: + for i in range(len(news_posts)): + if not news_posts[i]["url"]: + continue + post_site_data = download_site_as_html(news_posts[i]["url"]) + post_images = parse_chuni_jp_post_images(post_site_data) + news_posts[i]["images"].extend([ + image for image in post_images + if not any(existing["image"] == image["image"] for existing in news_posts[i]["images"]) + ]) + return news_posts - elif news_url == constants.CHUNITHM_JP_NEWS_SITE: - site_data = download_site_as_html(news_url) - if version in [ constants.CHUNITHM_VERSION.VERSE, constants.CHUNITHM_VERSION.X_VERSE ]: - news_posts = sorted(chunithm_jp.parse_chuni_jp_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) - if constants.CHUNI_RECURSIVE_IMAGE: - for i in range(len(news_posts)): - if not news_posts[i]["url"]: - continue - post_site_data = download_site_as_html(news_posts[i]["url"]) - post_images = chunithm_jp.parse_chuni_jp_post_images(post_site_data) - news_posts[i]["images"].extend([image for image in post_images if not any(existing_image['image'] == image['image'] for existing_image in news_posts[i]["images"])]) - elif news_url == constants.CHUNITHM_INTL_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(chuni_intl.parse_chuni_intl_api_route(site_data, "CHUNITHM_INTL", constants.CHUNITHM_INTL_RECENT_NEWS_LIMIT), key=lambda x: x['timestamp'], reverse=True) +@registry.register(constants.CHUNITHM_INTL_NEWS_SITE) +class ChunithmIntlSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.chuni_intl import parse_chuni_intl_api_route, parse_chuni_intl_post_images + site_data = download_site_as_html(constants.CHUNITHM_INTL_NEWS_SITE) + news_posts = sorted( + parse_chuni_intl_api_route(site_data, "CHUNITHM_INTL", constants.CHUNITHM_INTL_RECENT_NEWS_LIMIT), + key=lambda x: x['timestamp'], + reverse=True, + ) if constants.CHUNI_RECURSIVE_IMAGE: for i in range(len(news_posts)): if not news_posts[i]["url"]: continue post_site_data = download_site_as_html(news_posts[i]["url"]) - post_images = chuni_intl.parse_chuni_intl_post_images(post_site_data) - news_posts[i]["images"].extend([image for image in post_images if not any(existing_image['image'] == image['image'] for existing_image in news_posts[i]["images"])]) + post_images = parse_chuni_intl_post_images(post_site_data) + news_posts[i]["images"].extend([ + image for image in post_images + if not any(existing["image"] == image["image"] for existing in news_posts[i]["images"]) + ]) + return news_posts - elif news_url == constants.MAIMAIDX_JP_NEWS_SITE: - site_data = download_site_as_html(news_url) - if version in [ constants.MAIMAIDX_VERSION.PRISM_PLUS, constants.MAIMAIDX_VERSION.CIRCLE ]: - news_posts = sorted(maimaidx_jp.parse_maimaidx_jp_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) +@registry.register(constants.MAIMAIDX_JP_NEWS_SITE) +class MaimaiDXJPSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.maimaidx_jp import parse_maimaidx_jp_news_site + site_data = download_site_as_html(constants.MAIMAIDX_JP_NEWS_SITE) + if version not in [ + constants.MAIMAIDX_VERSION.PRISM_PLUS, + constants.MAIMAIDX_VERSION.CIRCLE, + constants.MAIMAIDX_VERSION.CIRCLE_PLUS, + ]: + return [] + news_posts = sorted(parse_maimaidx_jp_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) - elif news_url == constants.MAIMAIDX_INTL_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(maimaidx_intl.parse_maimaidx_intl_api_route(site_data, "MAIMAIDX_INTL", constants.MAIMAIDX_INTL_RECENT_NEWS_LIMIT), key=lambda x: x['timestamp'], reverse=True) + +@registry.register(constants.MAIMAIDX_INTL_NEWS_SITE) +class MaimaiDXIntlSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.maimaidx_intl import parse_maimaidx_intl_api_route + site_data = download_site_as_html(constants.MAIMAIDX_INTL_NEWS_SITE) + news_posts = sorted( + parse_maimaidx_intl_api_route(site_data, "MAIMAIDX_INTL", constants.MAIMAIDX_INTL_RECENT_NEWS_LIMIT), + key=lambda x: x['timestamp'], + reverse=True, + ) _attach_llm_summaries(news_posts, "maimai DX International") + return news_posts - elif news_url == constants.ONGEKI_JP_NEWS_SITE: - site_data = download_site_as_html(news_url) - if version == constants.ONGEKI_VERSION.REFRESH: - news_posts = sorted(ongeki_jp.parse_ongeki_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) - elif news_url == constants.IDAC_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(idac.parse_idac_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) +@registry.register(constants.ONGEKI_JP_NEWS_SITE) +class OngekiJPSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.ongeki_jp import parse_ongeki_news_site + site_data = download_site_as_html(constants.ONGEKI_JP_NEWS_SITE) + if version != constants.ONGEKI_VERSION.REFRESH: + return [] + news_posts = sorted(parse_ongeki_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) + + +@registry.register(constants.IDAC_NEWS_SITE) +class IDACSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from sega.idac import parse_idac_news_site, get_promo_image + site_data = download_site_as_html(constants.IDAC_NEWS_SITE) + news_posts = sorted(parse_idac_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) for news in news_posts: - promo_image_url = idac.get_promo_image(download_site_as_html(news["url"])) + promo_image_url = get_promo_image(download_site_as_html(news["url"])) if promo_image_url.endswith("png") or promo_image_url.endswith("jpg"): news["images"] = [{'image': promo_image_url, 'link': None}] else: news["images"] = [] - news_posts = translate.add_translate_text_to_en(news_posts) + return translate.add_translate_text_to_en(news_posts) - elif news_url == constants.MUSIC_DIVER_NEWS: - api_data = download_site_as_html(news_url) - news_posts = sorted(music_diver.parse_music_diver_news_json(api_data), key=lambda x: x['timestamp'], reverse=True) - elif news_url == constants.STREET_FIGHTER_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(street_fighter.parse_sf_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) +# --------------------------------------------------------------------------- +# Taito +# --------------------------------------------------------------------------- +@registry.register(constants.MUSIC_DIVER_NEWS) +class MusicDiverSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from taito.music_diver import parse_music_diver_news_json + api_data = download_site_as_html(constants.MUSIC_DIVER_NEWS) + return sorted(parse_music_diver_news_json(api_data), key=lambda x: x['timestamp'], reverse=True) - elif news_url == constants.TAIKO_BLOG_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(taiko.parse_taiko_blog_site(site_data), key=lambda x: x['timestamp'], reverse=True) - news_posts = translate.add_translate_text_to_en(news_posts) - elif news_url == constants.WANGAN_MAXI_GENERIC: +@registry.register(constants.STREET_FIGHTER_NEWS_SITE) +class StreetFighterSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from taito.street_fighter import parse_sf_news_site + site_data = download_site_as_html(constants.STREET_FIGHTER_NEWS_SITE) + news_posts = sorted(parse_sf_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) + + +# --------------------------------------------------------------------------- +# BANDAI NAMCO +# --------------------------------------------------------------------------- + +@registry.register(constants.TAIKO_BLOG_SITE) +class TaikoBlogSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from bandai_namco.taiko import parse_taiko_blog_site + site_data = download_site_as_html(constants.TAIKO_BLOG_SITE) + news_posts = sorted(parse_taiko_blog_site(site_data), key=lambda x: x['timestamp'], reverse=True) + return translate.add_translate_text_to_en(news_posts) + + +@registry.register(constants.WANGAN_MAXI_GENERIC) +class WanganMaxiSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from bandai_namco.wmmt import ( + get_wmmt_na_news_post_links, + get_wmmt_asia_oce_news_post_links, + get_wmmt_jp_news_post_links, + parse_wmmt_na_news, + parse_wmmt_asia_oce_news, + parse_wmmt_jp_news, + ) news_posts = [] + na_site_data = download_site_as_html(constants.WANGAN_MAXI_NA_NEWS_SITE, response_encoding="utf-8") - prelim_na_news_data = wmmt.get_wmmt_na_news_post_links(na_site_data) + prelim_na_news_data = get_wmmt_na_news_post_links(na_site_data) for data in prelim_na_news_data: post_site_data = download_site_as_html(data["url"]) - news = wmmt.parse_wmmt_na_news(post_site_data, data) + news = parse_wmmt_na_news(post_site_data, data) if news is not None: news_posts.append(news) + asia_oce_site_data = download_site_as_html(constants.WANGAN_MAXI_ASIA_OCE_NEWS_SITE, response_encoding="utf-8") - prelim_asia_oce_news_data = wmmt.get_wmmt_asia_oce_news_post_links(asia_oce_site_data) + prelim_asia_oce_news_data = get_wmmt_asia_oce_news_post_links(asia_oce_site_data) for data in prelim_asia_oce_news_data: post_site_data = download_site_as_html(data["url"]) - news = wmmt.parse_wmmt_asia_oce_news(post_site_data, data) + news = parse_wmmt_asia_oce_news(post_site_data, data) if news is not None: news_posts.append(news) + jp_site_data = download_site_as_html(constants.WANGAN_MAXI_JP_NEWS_SITE, response_encoding="utf-8") - prelim_jp_news_data = wmmt.get_wmmt_jp_news_post_links(jp_site_data) + prelim_jp_news_data = get_wmmt_jp_news_post_links(jp_site_data) jp_news = [] for data in prelim_jp_news_data: post_site_data = download_site_as_html(data["url"], response_encoding="utf-8") - news = wmmt.parse_wmmt_jp_news(post_site_data, data) + news = parse_wmmt_jp_news(post_site_data, data) if news is not None: jp_news.append(news) jp_news = translate.add_translate_text_to_en(jp_news) news_posts.extend(jp_news) - news_posts = sorted(news_posts, key=lambda x: x['timestamp'], reverse=True) - return news_posts + return sorted(news_posts, key=lambda x: x['timestamp'], reverse=True) + + +# --------------------------------------------------------------------------- +# Community +# --------------------------------------------------------------------------- + +@registry.register(constants.WACCA_PLUS_MAGIC_STRING) +class WaccaPlusSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from community.wacca_plus.wacca_plus import parse_announcement_messages, check_is_generation_possible + from community.disc import fetch_messages + if not check_is_generation_possible(): + return [] + messages = fetch_messages(constants.WACCA_PLUS_MAGIC_STRING) + return sorted(parse_announcement_messages(messages), key=lambda x: x['timestamp'], reverse=True) - elif news_url == constants.WACCA_PLUS_MAGIC_STRING: - if not wac_plus.check_is_generation_possible(): - news_posts = [] - else: - messages = disc.fetch_messages(constants.WACCA_PLUS_MAGIC_STRING) - news_posts = sorted(wac_plus.parse_announcement_messages(messages), key=lambda x: x['timestamp'], reverse=True) - elif news_url == constants.MUSECA_PLUS_NEWS_SITE: - site_data = download_site_as_html(news_url) - news_posts = sorted(mus_plus.parse_museca_plus_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) +@registry.register(constants.MUSECA_PLUS_NEWS_SITE) +class MusecaPlusSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from community.museca_plus import parse_museca_plus_news_site + site_data = download_site_as_html(constants.MUSECA_PLUS_NEWS_SITE) + return sorted(parse_museca_plus_news_site(site_data), key=lambda x: x['timestamp'], reverse=True) - elif news_url == constants.RB_DELUXE_PLUS_NEWS: - site_data = download_site_as_html(news_url) - news_posts = rbdx.get_carousel_posts(site_data) + +@registry.register(constants.RB_DELUXE_PLUS_NEWS) +class RBDeluxePlusSource(NewsSource): + def fetch(self, version=None) -> list[dict]: + from community.rbdx import get_carousel_posts + site_data = download_site_as_html(constants.RB_DELUXE_PLUS_NEWS) + news_posts = get_carousel_posts(site_data) _attach_llm_summaries(news_posts, "REFLEC BEAT PLUS DELUXE") + return news_posts - else: - news_posts = [] - return news_posts + +def get_news(news_url: str, version=None) -> list[dict]: + source_cls = registry.get_source(news_url) + if source_cls is None: + return [] + return source_cls().fetch(version) |