aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorPinapelz <yukais@pinapelz.com>2026-03-12 13:56:30 -0700
committerPinapelz <yukais@pinapelz.com>2026-03-12 13:56:50 -0700
commitcaa3cf245186ab0f6fb33e63a7dd838d834da12e (patch)
treebc5742a134ecabf0b9d35cc12b1d6f67defd5da7
parent5658441ab9b703c95a48e654d41e45cc3a55ffd3 (diff)
refactor: move to common NewsSource interface
cleanup imports by defining initaliazers modules and decorator remove legacy scrapers remove single factory for sega games (sites don't change that much)
-rw-r--r--bandai_namco/__init__.py19
-rw-r--r--bandai_namco/wmmt.py5
-rw-r--r--bemani/__init__.py7
-rw-r--r--bemani/ddr.py63
-rw-r--r--bemani/iidx.py68
-rw-r--r--community/__init__.py12
-rw-r--r--community/disc.py4
-rw-r--r--community/wacca_plus/__init__.py0
-rw-r--r--community/wacca_plus/wacca_plus.py2
-rw-r--r--constants.py4
-rw-r--r--generate.py44
-rw-r--r--konami/__init__.py6
-rw-r--r--news_feed.py354
-rw-r--r--scrapers/__init__.py0
-rw-r--r--scrapers/base.py8
-rw-r--r--scrapers/registry.py22
-rw-r--r--sega/__init__.py23
-rw-r--r--sega/chuni_intl.py166
-rw-r--r--sega/chuni_jp.py169
-rw-r--r--sega/maimaidx_intl.py49
-rw-r--r--sega/maimaidx_jp.py89
-rw-r--r--sega/ongeki_jp.py102
-rw-r--r--taito/__init__.py7
-rw-r--r--taito/music_diver.py3
-rw-r--r--taito/street_fighter.py142
25 files changed, 655 insertions, 713 deletions
diff --git a/bandai_namco/__init__.py b/bandai_namco/__init__.py
new file mode 100644
index 0000000..0e0ce82
--- /dev/null
+++ b/bandai_namco/__init__.py
@@ -0,0 +1,19 @@
+from bandai_namco.taiko import parse_taiko_blog_site
+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,
+)
+
+__all__ = [
+ "parse_taiko_blog_site",
+ "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",
+] \ No newline at end of file
diff --git a/bandai_namco/wmmt.py b/bandai_namco/wmmt.py
index b7ea927..1a6bbbe 100644
--- a/bandai_namco/wmmt.py
+++ b/bandai_namco/wmmt.py
@@ -2,11 +2,8 @@ import re
from datetime import datetime, timedelta, timezone
from enum import Enum
from urllib.parse import urljoin
-import sys
-import os
import pytz
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../")))
-import constants
+from .. import constants
from bs4 import BeautifulSoup
BASE_URL = "https://wanganmaxi-official.com"
diff --git a/bemani/__init__.py b/bemani/__init__.py
new file mode 100644
index 0000000..f16ed0a
--- /dev/null
+++ b/bemani/__init__.py
@@ -0,0 +1,7 @@
+from bemani.sdvx import parse_exceed_gear_news_site
+from bemani.polaris_chord import parse_polaris_chord_news_site
+
+__all__ = [
+ "parse_exceed_gear_news_site",
+ "parse_polaris_chord_news_site",
+]
diff --git a/bemani/ddr.py b/bemani/ddr.py
deleted file mode 100644
index b5ae93c..0000000
--- a/bemani/ddr.py
+++ /dev/null
@@ -1,63 +0,0 @@
-"""
-Currently unused as e-eamusement app feed is favored. Here for archival purposes
-"""
-from bs4 import BeautifulSoup
-from datetime import datetime
-from urllib.parse import urljoin
-import time
-
-def parse_ddr_world_news_site(html: str):
- base_url = "https://p.eagate.573.jp"
- soup = BeautifulSoup(html, 'html.parser')
- news_entries = []
-
- for div in soup.select("div#info > div.news_one"):
- style = div.get('style', '')
- if 'none' in style:
- continue
-
- title_tag = div.select_one("div.news_title > div.title")
- date_tag = div.select_one("div.news_title > div.date")
- headline = title_tag.get_text(strip=True) if title_tag else None
- date_str = date_tag.get_text(strip=True) if date_tag else None
-
- try:
- dt = datetime.strptime(date_str, "%Y/%m/%d")
- date_iso = dt.strftime("%Y-%m-%d")
- timestamp = int(time.mktime(dt.timetuple()))
- except Exception:
- date_iso, timestamp = None, None
- paras = [p.get_text(strip=True, separator="\n\n")
- for p in div.find_all("p", recursive=False)]
- if not paras:
- for child in div.find_all(recursive=False):
- cls = child.get("class", [])
- if "news_title" in cls or "img_news_center" in cls:
- continue
- if child.name == "div":
- paras.append(child.get_text(strip=True, separator="\n\n"))
-
- content = "\n\n\n".join(paras) if paras else None
- if content:
- content = f"\n{content}\n"
-
- images = []
- for img in div.select("div.img_news_center img"):
- raw_src = img.get("data-src") or img.get("src")
- if raw_src:
- full_url = urljoin(base_url, raw_src)
- images.append({"image": full_url, "link": None})
-
- news_entries.append({
- "date": date_iso,
- "identifier": "DDR",
- "type": None,
- "timestamp": timestamp,
- "headline": headline,
- "content": content,
- "url": None,
- "images": images,
- 'is_ai_summary': False
- })
-
- return news_entries
diff --git a/bemani/iidx.py b/bemani/iidx.py
deleted file mode 100644
index de7f34c..0000000
--- a/bemani/iidx.py
+++ /dev/null
@@ -1,68 +0,0 @@
-from bs4 import BeautifulSoup
-from datetime import datetime
-from urllib.parse import urljoin
-import re
-
-KEY_TERMS_TL = [
- ("クプロ", "QPro")
-]
-
-# Legacy code. e-amuse feed provides better data
-def parse_pinky_crush_news_site(html: str):
- base_url = "https://p.eagate.573.jp"
- type_map = {
- "i_01": "NEWSONG",
- "i_02": "RANKING",
- "i_03": "EVENT",
- "i_04": "SHOP",
- "i_05": "OTHER"
- }
- soup = BeautifulSoup(html, "html.parser")
- news_items = []
-
- for li in soup.select("#info-news > li"):
- date_elem = li.select_one(".news-main > li:nth-of-type(1)")
- headline_elem = li.select_one(".news-main > li:nth-of-type(2)")
- content_elem = li.select_one(".news-main > li:nth-of-type(3)")
- type_class = li.get("class", [None])[0]
- if not (date_elem and content_elem):
- continue
- date_str = date_elem.text.strip()
- try:
- dt = datetime.strptime(date_str, "%Y/%m/%d")
- timestamp = int(dt.timestamp())
- except ValueError:
- timestamp = None
-
- headline = headline_elem.a.text.strip() if headline_elem.a else headline_elem.text.strip()
-
- for a in content_elem.select("a[href]"):
- href = urljoin(base_url, a["href"])
- text = a.get_text(strip=True)
- a.replace_with(f"[{text}]({href})")
-
- for br in content_elem.find_all("br"):
- br.replace_with("\n")
-
- content = content_elem.get_text().strip()
-
- content = content.replace(
- " e-amusement ベーシックコース ",
- " e-amusement ベーシックコース "
- )
- content = content.replace("※", "\n※")
- content = re.sub(r"\n[ \t]+", "\n", content)
- content = re.sub(r'\s*/\s*', '/', content)
- news_items.append({
- "date": date_str,
- "identifier": "IIDX",
- "type": type_map[type_class],
- "timestamp": timestamp,
- "headline": headline,
- "content": content,
- "url": None,
- "images": [],
- 'is_ai_summary': False
- })
-
- return news_items
diff --git a/community/__init__.py b/community/__init__.py
new file mode 100644
index 0000000..835b7e6
--- /dev/null
+++ b/community/__init__.py
@@ -0,0 +1,12 @@
+from community.disc import fetch_messages
+from community.museca_plus import parse_museca_plus_news_site
+from community.rbdx import get_carousel_posts
+from community.wacca_plus.wacca_plus import parse_announcement_messages, check_is_generation_possible
+
+__all__ = [
+ "fetch_messages",
+ "parse_museca_plus_news_site",
+ "get_carousel_posts",
+ "parse_announcement_messages",
+ "check_is_generation_possible",
+] \ No newline at end of file
diff --git a/community/disc.py b/community/disc.py
index b6b7815..90e32f6 100644
--- a/community/disc.py
+++ b/community/disc.py
@@ -1,8 +1,6 @@
import os
import requests
-from dotenv import load_dotenv
-load_dotenv()
def fetch_messages(channel_id: str):
url = f"https://discord.com/api/v9/channels/{channel_id}/messages?limit=50"
@@ -23,4 +21,4 @@ def fetch_messages(channel_id: str):
"x-discord-timezone": "America/Vancouver",
}
response = requests.get(url, headers=headers)
- return response.json()
+ return response.json() \ No newline at end of file
diff --git a/community/wacca_plus/__init__.py b/community/wacca_plus/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/community/wacca_plus/__init__.py
diff --git a/community/wacca_plus/wacca_plus.py b/community/wacca_plus/wacca_plus.py
index 666a243..c15bbf7 100644
--- a/community/wacca_plus/wacca_plus.py
+++ b/community/wacca_plus/wacca_plus.py
@@ -6,8 +6,6 @@ import os
import time
import openai
import json
-import sys
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../")))
from summarizer import generate_headline_and_content_from_images
diff --git a/constants.py b/constants.py
index 6125147..dc1c680 100644
--- a/constants.py
+++ b/constants.py
@@ -3,8 +3,6 @@ from enum import Enum
DAYS_LIMIT=14
SOUND_VOLTEX_EXCEED_GEAR_NEWS_SITE ="https://p.eagate.573.jp/game/sdvx/vi/news/index.html"
-IIDX_PINKY_CRUSH_NEWS_SITE="https://p.eagate.573.jp/game/2dx/32/info/index.html" # legacy should not be used, eamuse feed is more verbose
-DDR_WORLD_NEWS_SITE="https://p.eagate.573.jp/game/ddr/ddrworld/info/index.html"
POLARIS_CHORD_NEWS_SITE="https://p.eagate.573.jp/game/polarischord/pc/news/news.html"
POLARIS_CHORD_RECENT_NEWS_LIMIT=15
@@ -54,11 +52,13 @@ class CHUNITHM_VERSION(Enum):
LUMINOUS_PLUS = 1
VERSE = 2
X_VERSE = 3
+ X_VERSE_X = 4
class MAIMAIDX_VERSION(Enum):
PRISM = 1
PRISM_PLUS = 2
CIRCLE = 3
+ CIRCLE_PLUS = 4
class ONGEKI_VERSION(Enum):
REFRESH = 1
diff --git a/generate.py b/generate.py
index ab7d968..819ce08 100644
--- a/generate.py
+++ b/generate.py
@@ -156,11 +156,8 @@ def generate_news_file(filename, url, version=None, formatted_name: str = None):
# For e-amusement games you can choose to pull from a specific implementation of the scraper or the generic feed provided
# by the e-amusement app. Information is different
-def generate_iidx_news_file(eamuse_feed: bool=False):
- if eamuse_feed:
- news = generate_news_file("iidx_news", constants.EAMUSE_APP_API_ROUTE, constants.IIDX_EAMUSE_APP_ID)
- else: # legacy should not be used, use eamuse app feed above
- news = generate_news_file("iidx_news", constants.IIDX_PINKY_CRUSH_NEWS_SITE)
+def generate_iidx_news_file():
+ news = generate_news_file("iidx_news", constants.EAMUSE_APP_API_ROUTE, constants.IIDX_EAMUSE_APP_ID)
attempt_broadcast_notifications(news, "New information for beatmania IIDX", "iidx")
return news
@@ -169,11 +166,8 @@ def generate_sdvx_news_file():
attempt_broadcast_notifications(news, "New Information for SOUND VOLTEX","sdvx")
return news
-def generate_ddr_news_file(eamuse_feed: bool=False):
- if eamuse_feed:
- news = generate_news_file("ddr_news", constants.EAMUSE_APP_FEED, constants.DDR_EAMUSE_APP_ID)
- else:
- news = generate_news_file("ddr_news", constants.DDR_WORLD_NEWS_SITE)
+def generate_ddr_news_file():
+ news = generate_news_file("ddr_news", constants.EAMUSE_APP_FEED, constants.DDR_EAMUSE_APP_ID)
attempt_broadcast_notifications(news, "New information for DanceDanceRevolution", "ddr")
return news
@@ -213,12 +207,12 @@ def generate_gitadora_news_file():
return news
def generate_chunithm_jp_news_file():
- news = generate_news_file("chunithm_jp_news", constants.CHUNITHM_JP_NEWS_SITE, constants.CHUNITHM_VERSION.X_VERSE)
+ news = generate_news_file("chunithm_jp_news", constants.CHUNITHM_JP_NEWS_SITE, constants.CHUNITHM_VERSION.X_VERSE_X)
attempt_broadcast_notifications(news, "New information for CHUNITHM (Japan ver.)", "chunithm_jp")
return news
def generate_maimaidx_jp_news_file():
- news = generate_news_file("maimaidx_jp_news", constants.MAIMAIDX_JP_NEWS_SITE, constants.MAIMAIDX_VERSION.CIRCLE)
+ news = generate_news_file("maimaidx_jp_news", constants.MAIMAIDX_JP_NEWS_SITE, constants.MAIMAIDX_VERSION.CIRCLE_PLUS)
attempt_broadcast_notifications(news, "New information for maimai DX (Japan ver.)", "maimaidx_jp")
return news
@@ -286,8 +280,8 @@ if __name__ == "__main__":
os.makedirs(OUTPUT_DIR)
sdvx_news_data = generate_sdvx_news_file()
polaris_news_data = generate_polaris_chord_news_file()
- iidx_news_data = generate_iidx_news_file(eamuse_feed=True)
- ddr_news_data = generate_ddr_news_file(eamuse_feed=True)
+ iidx_news_data = generate_iidx_news_file()
+ ddr_news_data = generate_ddr_news_file()
dance_rush_news_data = generate_dance_rush_news_file()
dance_around_news_data = generate_dance_around_news_file()
gitadora_news_data = generate_gitadora_news_file()
@@ -334,14 +328,14 @@ if __name__ == "__main__":
dance_around_news_data,
wmmt_news
)
- log_output("Creating merged news.json file for all news that are within " + str(constants.DAYS_LIMIT) + " days old")
- log_output("Computing and Attaching Archived IDs for merged feed")
- for item in news:
- if 'archive_hash' not in item:
- hash_value = compute_json_hash(json.dumps(item, sort_keys=True))
- item['archive_hash'] = hash_value
- if ARCHIVE_NEWS:
- save_news_to_db(news)
- with open(OUTPUT_DIR+'/news.json', 'w') as json_file:
- json.dump(attach_news_meta_data(news), json_file)
- log_output("JOB DONE", "TASK")
+ # log_output("Creating merged news.json file for all news that are within " + str(constants.DAYS_LIMIT) + " days old")
+ # log_output("Computing and Attaching Archived IDs for merged feed")
+ # for item in news:
+ # if 'archive_hash' not in item:
+ # hash_value = compute_json_hash(json.dumps(item, sort_keys=True))
+ # item['archive_hash'] = hash_value
+ # if ARCHIVE_NEWS:
+ # save_news_to_db(news)
+ # with open(OUTPUT_DIR+'/news.json', 'w') as json_file:
+ # json.dump(attach_news_meta_data(news), json_file)
+ # log_output("JOB DONE", "TASK")
diff --git a/konami/__init__.py b/konami/__init__.py
new file mode 100644
index 0000000..f7f4d66
--- /dev/null
+++ b/konami/__init__.py
@@ -0,0 +1,6 @@
+from konami.eamuse_app import parse_news_api_route, parse_news_page
+
+__all__ = [
+ "parse_news_api_route",
+ "parse_news_page",
+] \ No newline at end of file
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)
diff --git a/scrapers/__init__.py b/scrapers/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/scrapers/__init__.py
diff --git a/scrapers/base.py b/scrapers/base.py
new file mode 100644
index 0000000..8d9839f
--- /dev/null
+++ b/scrapers/base.py
@@ -0,0 +1,8 @@
+from abc import ABC, abstractmethod
+
+
+class NewsSource(ABC):
+
+ @abstractmethod
+ def fetch(self, version=None) -> list[dict]:
+ pass
diff --git a/scrapers/registry.py b/scrapers/registry.py
new file mode 100644
index 0000000..6d9bde6
--- /dev/null
+++ b/scrapers/registry.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from scrapers.base import NewsSource
+
+_registry: dict[str, type["NewsSource"]] = {}
+
+
+def register(url_key: str):
+ def decorator(cls):
+ _registry[url_key] = cls
+ return cls
+ return decorator
+
+
+def get_source(url_key: str) -> Optional[type["NewsSource"]]:
+ return _registry.get(url_key)
+
+
+def get_all() -> dict[str, type["NewsSource"]]:
+ return dict(_registry)
diff --git a/sega/__init__.py b/sega/__init__.py
new file mode 100644
index 0000000..242ab52
--- /dev/null
+++ b/sega/__init__.py
@@ -0,0 +1,23 @@
+from sega.chuni_jp import parse_chuni_jp_news_site, parse_chuni_jp_post_images
+from sega.chuni_intl import (
+ parse_chuni_intl_api_route,
+ parse_chuni_intl_news_site,
+ parse_chuni_intl_post_images,
+)
+from sega.maimaidx_jp import parse_maimaidx_jp_news_site
+from sega.maimaidx_intl import parse_maimaidx_intl_api_route
+from sega.ongeki_jp import parse_ongeki_news_site
+from sega.idac import parse_idac_news_site, get_promo_image
+
+__all__ = [
+ "parse_chuni_jp_news_site",
+ "parse_chuni_jp_post_images",
+ "parse_chuni_intl_api_route",
+ "parse_chuni_intl_news_site",
+ "parse_chuni_intl_post_images",
+ "parse_maimaidx_jp_news_site",
+ "parse_maimaidx_intl_api_route",
+ "parse_ongeki_news_site",
+ "parse_idac_news_site",
+ "get_promo_image",
+] \ No newline at end of file
diff --git a/sega/chuni_intl.py b/sega/chuni_intl.py
index 64d279c..816b857 100644
--- a/sega/chuni_intl.py
+++ b/sega/chuni_intl.py
@@ -1,100 +1,11 @@
import re
from datetime import datetime, timedelta, timezone
-from enum import Enum
import json
from urllib.parse import urljoin
from bs4 import BeautifulSoup
-class ParserVersion(Enum):
- ALPHA = 1
-
-
-def make_chuni_intl_parser(identifier: str, parser: ParserVersion):
- def alpha_parser(html: str):
- """
- Confirmed on:
- LUMINOUS PLUS
- """
- soup = BeautifulSoup(html, "html.parser")
- base_url = "https://info-chunithm.sega.com/"
- items = soup.select("li.news--list__item")
- results = []
-
- for item in items:
- a_tag = item.select_one("a.news--list__post")
- if not a_tag:
- continue
-
- url = urljoin(base_url, a_tag["href"])
- date_text = item.select_one("div.news--date").text.strip()
- headline = item.select_one("p.news--title").text.strip()
- img_tag = item.select_one("div.news--thumbnail img")
- image_url = urljoin(base_url, img_tag["src"]) if img_tag else None
-
- date_match = re.match(r"(\d{4})\.(\d{1,2})\.(\d{1,2})", date_text)
- if not date_match:
- continue
- year, month, day = map(int, date_match.groups())
- jst = timezone(timedelta(hours=9))
- dt = datetime(year, month, day, tzinfo=jst)
- timestamp = int(dt.timestamp())
-
- results.append(
- {
- "date": dt.strftime("%Y-%m-%d"),
- "identifier": identifier,
- "type": None,
- "timestamp": timestamp,
- "headline": None,
- "content": headline,
- "url": url,
- "images": [{"image": image_url, "link": url}] if image_url else [],
- 'is_ai_summary': False
- }
- )
-
- return results
-
- if parser == ParserVersion.ALPHA:
- return alpha_parser
-
-
-def make_image_extractor(version: ParserVersion):
- """
- Gets all the images from a full post page as CHUNITHM intl has more relevant images
- hidden in the actual posts
- """
-
- def image_extractor_alpha(html: str):
- base_url = "https://info-chunithm.sega.com/"
- soup = BeautifulSoup(html, "html.parser")
- images = []
- news_post = soup.select_one(".news--post")
- if not news_post:
- return images
-
- for img in news_post.find_all("img"):
- src = img.get("src") or img.get("data-src")
- if not src:
- continue
-
- full_url = urljoin(base_url, src)
- parent = img.find_parent("a")
- link = parent.get("href") if parent and parent.name == "a" else None
-
- images.append(
- {"image": full_url, "link": urljoin(base_url, link) if link else None}
- )
-
- return images
-
- if version == ParserVersion.ALPHA:
- return image_extractor_alpha
- else:
- raise ValueError("Unknown Parser Version")
-
def parse_chuni_intl_api_route(raw_api_data: str, identifier: str, limit: int):
route_data = json.loads(raw_api_data)
route_data = route_data[:limit]
@@ -126,7 +37,76 @@ def parse_chuni_intl_api_route(raw_api_data: str, identifier: str, limit: int):
return entries
-parse_chuni_intl_news_site = make_chuni_intl_parser(
- "CHUNITHM_INTL", ParserVersion.ALPHA
-)
-parse_chuni_intl_post_images = make_image_extractor(ParserVersion.ALPHA)
+def parse_chuni_intl_post_images(html: str):
+ """
+ Gets all the images from a full post page as CHUNITHM intl has more relevant images
+ hidden in the actual posts.
+ """
+ base_url = "https://info-chunithm.sega.com/"
+ soup = BeautifulSoup(html, "html.parser")
+ images = []
+ news_post = soup.select_one(".news--post")
+ if not news_post:
+ return images
+
+ for img in news_post.find_all("img"):
+ src = img.get("src") or img.get("data-src")
+ if not src:
+ continue
+
+ full_url = urljoin(base_url, src)
+ parent = img.find_parent("a")
+ link = parent.get("href") if parent and parent.name == "a" else None
+
+ images.append(
+ {"image": full_url, "link": urljoin(base_url, link) if link else None}
+ )
+
+ return images
+
+
+def parse_chuni_intl_news_site(html: str):
+ """
+ Confirmed on:
+ LUMINOUS PLUS
+ """
+ identifier = "CHUNITHM_INTL"
+ soup = BeautifulSoup(html, "html.parser")
+ base_url = "https://info-chunithm.sega.com/"
+ items = soup.select("li.news--list__item")
+ results = []
+
+ for item in items:
+ a_tag = item.select_one("a.news--list__post")
+ if not a_tag:
+ continue
+
+ url = urljoin(base_url, a_tag["href"])
+ date_text = item.select_one("div.news--date").text.strip()
+ headline = item.select_one("p.news--title").text.strip()
+ img_tag = item.select_one("div.news--thumbnail img")
+ image_url = urljoin(base_url, img_tag["src"]) if img_tag else None
+
+ date_match = re.match(r"(\d{4})\.(\d{1,2})\.(\d{1,2})", date_text)
+ if not date_match:
+ continue
+ year, month, day = map(int, date_match.groups())
+ jst = timezone(timedelta(hours=9))
+ dt = datetime(year, month, day, tzinfo=jst)
+ timestamp = int(dt.timestamp())
+
+ results.append(
+ {
+ "date": dt.strftime("%Y-%m-%d"),
+ "identifier": identifier,
+ "type": None,
+ "timestamp": timestamp,
+ "headline": None,
+ "content": headline,
+ "url": url,
+ "images": [{"image": image_url, "link": url}] if image_url else [],
+ "is_ai_summary": False,
+ }
+ )
+
+ return results \ No newline at end of file
diff --git a/sega/chuni_jp.py b/sega/chuni_jp.py
index 452e153..a914270 100644
--- a/sega/chuni_jp.py
+++ b/sega/chuni_jp.py
@@ -1,114 +1,93 @@
import re
from datetime import datetime, timedelta, timezone
-from enum import Enum
from urllib.parse import urljoin
from bs4 import BeautifulSoup
-class ParserVersion(Enum):
- ALPHA = 1
-
-
-def make_chuni_jp_parser(identifier: str, parser: ParserVersion):
- def alpha_parser(html: str):
- """
- Confirmed on:
- VERSE
- """
- soup = BeautifulSoup(html, "html.parser")
- news_entries = []
- news_wrapper = soup.find("div", class_="newsMainWrapper-left")
- if not news_wrapper:
- return news_entries
- for a_tag in news_wrapper.find_all("a", href=True):
- if not a_tag.find("div", class_="chuniCommonBox-inner"):
- continue
- news_dict = {}
- news_url = a_tag.get("href")
- news_dict["url"] = news_url
+def parse_chuni_jp_news_site(html: str):
+ """
+ Confirmed on:
+ VERSE
+ """
+ identifier = "CHUNITHM_JP"
+ soup = BeautifulSoup(html, "html.parser")
+ news_entries = []
+ news_wrapper = soup.find("div", class_="newsMainWrapper-left")
+ if not news_wrapper:
+ return news_entries
+ for a_tag in news_wrapper.find_all("a", href=True):
+ if not a_tag.find("div", class_="chuniCommonBox-inner"):
+ continue
+ news_dict = {}
+ news_url = a_tag.get("href")
+ news_dict["url"] = news_url
- date_container = a_tag.find("div", class_="chuniCommonBox-inner-title")
- date_str = None
- if date_container:
- title_span = date_container.find("span", class_="title")
- if title_span:
- text = title_span.get_text(strip=True)
- date_match = re.search(r"(\d{4}\.\d{2}\.\d{2})", text)
- if date_match:
- date_str = date_match.group(1)
- news_dict["date"] = date_str
- news_dict["type"] = None
- timestamp = None
- if date_str:
- try:
- dt = datetime.strptime(date_str, "%Y.%m.%d")
- dt = dt.replace(tzinfo=timezone(timedelta(hours=9)))
- timestamp = int(dt.timestamp())
- except Exception:
- timestamp = None
- news_dict["timestamp"] = timestamp
+ date_container = a_tag.find("div", class_="chuniCommonBox-inner-title")
+ date_str = None
+ if date_container:
+ title_span = date_container.find("span", class_="title")
+ if title_span:
+ text = title_span.get_text(strip=True)
+ date_match = re.search(r"(\d{4}\.\d{2}\.\d{2})", text)
+ if date_match:
+ date_str = date_match.group(1)
+ news_dict["date"] = date_str
+ news_dict["type"] = None
+ timestamp = None
+ if date_str:
+ try:
+ dt = datetime.strptime(date_str, "%Y.%m.%d")
+ dt = dt.replace(tzinfo=timezone(timedelta(hours=9)))
+ timestamp = int(dt.timestamp())
+ except Exception:
+ timestamp = None
+ news_dict["timestamp"] = timestamp
- main_content = a_tag.find("div", class_="chuniCommonBox-inner-main")
- content_text = ""
- if main_content:
- content_text = main_content.get_text(separator=" ", strip=True)
- news_dict["content"] = content_text
+ main_content = a_tag.find("div", class_="chuniCommonBox-inner-main")
+ content_text = ""
+ if main_content:
+ content_text = main_content.get_text(separator=" ", strip=True)
+ news_dict["content"] = content_text
- images = {"image": None, "link": None}
- if main_content:
- img_tag = main_content.find("img")
- if img_tag:
- images["image"] = img_tag.get("src")
- images["link"] = news_url
- news_dict["images"] = [images]
- news_dict["identifier"] = identifier
- news_dict["is_ai_summary"] = False
+ images = {"image": None, "link": None}
+ if main_content:
+ img_tag = main_content.find("img")
+ if img_tag:
+ images["image"] = img_tag.get("src")
+ images["link"] = news_url
+ news_dict["images"] = [images]
+ news_dict["identifier"] = identifier
+ news_dict["is_ai_summary"] = False
- news_entries.append(news_dict)
+ news_entries.append(news_dict)
- return news_entries
+ return news_entries
- if parser == ParserVersion.ALPHA:
- return alpha_parser
-
-def make_image_extractor(version: ParserVersion):
+def parse_chuni_jp_post_images(html: str):
"""
- Gets all the images from a full post page as CHUNITHM intl has more relevant images
- hidden in the actual posts
+ Gets all the images from a full post page as CHUNITHM JP has more relevant images
+ hidden in the actual posts.
"""
+ base_url = "https://info-chunithm.sega.jp/"
+ soup = BeautifulSoup(html, "html.parser")
+ images = []
- def image_extractor_alpha(html: str):
- base_url = "https://info-chunithm.sega.jp/"
- soup = BeautifulSoup(html, "html.parser")
- images = []
-
- container = soup.select_one(".chuniCommonBox-inner-main")
- if not container:
- return images
- for img in container.find_all("img"):
- if img.find_parent("p") and "©" in img.find_parent("p").text:
- continue
-
- src = img.get("src") or img.get("data-src")
- if not src:
- continue
- full_url = urljoin(base_url, src)
- parent = img.find_parent("a")
- link = parent.get("href") if parent and parent.name == "a" else None
- images.append(
- {"image": full_url, "link": urljoin(base_url, link) if link else None}
- )
+ container = soup.select_one(".chuniCommonBox-inner-main")
+ if not container:
return images
+ for img in container.find_all("img"):
+ if img.find_parent("p") and "©" in img.find_parent("p").text:
+ continue
- if version == ParserVersion.ALPHA:
- return image_extractor_alpha
- else:
- raise ValueError("Unknown Parser Version")
-
-
-parse_chuni_jp_news_site = make_chuni_jp_parser(
- "CHUNITHM_JP", ParserVersion.ALPHA
-)
-parse_chuni_jp_post_images = make_image_extractor(ParserVersion.ALPHA)
+ src = img.get("src") or img.get("data-src")
+ if not src:
+ continue
+ full_url = urljoin(base_url, src)
+ parent = img.find_parent("a")
+ link = parent.get("href") if parent and parent.name == "a" else None
+ images.append(
+ {"image": full_url, "link": urljoin(base_url, link) if link else None}
+ )
+ return images
diff --git a/sega/maimaidx_intl.py b/sega/maimaidx_intl.py
index 3e26a37..8182117 100644
--- a/sega/maimaidx_intl.py
+++ b/sega/maimaidx_intl.py
@@ -1,53 +1,7 @@
from bs4 import BeautifulSoup
from datetime import datetime, timezone, timedelta
-from enum import Enum
import json
-class ParserVersion(Enum):
- ALPHA=1
-
-def make_maimaidx_intl_parser(identifier: str, parser: ParserVersion):
- """
- Parses the download page of maimai dx intl site. API route method below is preferred as information is the same
- """
- def alpha_parser(html: str):
- """
- Confirmed on:
- PRISM
- """
- soup = BeautifulSoup(html, "html.parser")
- items = soup.select(".dl--pop__item")
-
- entries = []
- for item in items:
- date_text = item.select_one(".dl--pop__head").text.strip().replace(" UP", "")
- dt = datetime.strptime(date_text, "%Y.%m.%d").replace(tzinfo=timezone(timedelta(hours=9)))
- timestamp = int(dt.timestamp())
-
- img_tag = item.select_one("a.dl--pop__thumb img")
- image_url = img_tag["srcset"] if img_tag else None
- full_image_url = image_url.replace("../", "https://maimai.sega.com/") if image_url else None
-
- entry = {
- "date": date_text,
- "identifier": identifier,
- "type": None,
- "timestamp": timestamp,
- "headline": None,
- "content": f"New maimai DX International News / maimai DX International の新しいお知らせ\n\n{full_image_url}",
- "url": None,
- "images": [
- {
- "image": full_image_url,
- "link": None
- }
- ],
- 'is_ai_summary': False
- }
- entries.append(entry)
- return entries
- if parser == ParserVersion.ALPHA:
- return alpha_parser
def parse_maimaidx_intl_api_route(raw_api_data: str, identifier: str, limit: int):
route_data = json.loads(raw_api_data)
@@ -84,6 +38,3 @@ def parse_maimaidx_intl_api_route(raw_api_data: str, identifier: str, limit: int
}
entries.append(entry)
return entries
-
-
-parse_maimaidx_intl_news_site = make_maimaidx_intl_parser("MAIMAIDX_INTL", ParserVersion.ALPHA)
diff --git a/sega/maimaidx_jp.py b/sega/maimaidx_jp.py
index 1314325..2b61c9a 100644
--- a/sega/maimaidx_jp.py
+++ b/sega/maimaidx_jp.py
@@ -1,60 +1,53 @@
from bs4 import BeautifulSoup
from datetime import datetime, timezone, timedelta
from urllib.parse import urljoin
-from enum import Enum
-class ParserVersion(Enum):
- ALPHA=1
-def make_maimaidx_jpn_parser(identifier: str, parser: ParserVersion):
- def alpha_parser(html: str):
- """
- Confirmed on:
- PRISM PLUS
- """
- soup = BeautifulSoup(html, "html.parser")
- base_url = "https://info-maimai.sega.jp/"
- news_items = []
+def parse_maimaidx_jp_news_site(html: str):
+ """
+ Confirmed on:
+ PRISM PLUS
+ """
+ identifier = "MAIMAIDX_JP"
+ soup = BeautifulSoup(html, "html.parser")
+ base_url = "https://info-maimai.sega.jp/"
+ news_items = []
- news_boxes = soup.select(".maiPager-content .newsBox")
- for box in news_boxes:
- a_tag = box.select_one("a")
- url = urljoin(base_url, a_tag["href"]) if a_tag and a_tag.get("href") else None
+ news_boxes = soup.select(".maiPager-content .newsBox")
+ for box in news_boxes:
+ a_tag = box.select_one("a")
+ url = urljoin(base_url, a_tag["href"]) if a_tag and a_tag.get("href") else None
- img_tag = box.select_one("img")
- image_url = urljoin(base_url, img_tag["src"]) if img_tag else None
+ img_tag = box.select_one("img")
+ image_url = urljoin(base_url, img_tag["src"]) if img_tag else None
- date_tag = box.select_one(".newsDate")
- raw_date = date_tag.get_text(strip=True) if date_tag else None
+ date_tag = box.select_one(".newsDate")
+ raw_date = date_tag.get_text(strip=True) if date_tag else None
- jst = timezone(timedelta(hours=9))
- try:
- dt = datetime.strptime(raw_date.split(" ")[0], "%Y.%m.%d").replace(tzinfo=jst)
- timestamp = int(dt.timestamp())
- except:
- dt = None
- timestamp = 0
+ jst = timezone(timedelta(hours=9))
+ try:
+ dt = datetime.strptime(raw_date.split(" ")[0], "%Y.%m.%d").replace(tzinfo=jst)
+ timestamp = int(dt.timestamp())
+ except Exception:
+ dt = None
+ timestamp = 0
- content_tag = box.select_one(".newsLink")
- content = content_tag.get_text(strip=True) if content_tag else None
+ content_tag = box.select_one(".newsLink")
+ content = content_tag.get_text(strip=True) if content_tag else None
- news_items.append({
- "date": raw_date,
- "identifier": identifier,
- "type": None,
- "timestamp": timestamp,
- "headline": None,
- "content": content,
- "url": url,
- 'is_ai_summary': False,
- "images": [{
- "image": image_url,
- "link": url
- }] if image_url else []
- })
+ news_items.append({
+ "date": raw_date,
+ "identifier": identifier,
+ "type": None,
+ "timestamp": timestamp,
+ "headline": None,
+ "content": content,
+ "url": url,
+ "is_ai_summary": False,
+ "images": [{
+ "image": image_url,
+ "link": url
+ }] if image_url else []
+ })
- return news_items
- if parser == ParserVersion.ALPHA:
- return alpha_parser
-
-parse_maimaidx_jp_news_site = make_maimaidx_jpn_parser("MAIMAIDX_JP", ParserVersion.ALPHA)
+ return news_items \ No newline at end of file
diff --git a/sega/ongeki_jp.py b/sega/ongeki_jp.py
index f9c2dc4..c173189 100644
--- a/sega/ongeki_jp.py
+++ b/sega/ongeki_jp.py
@@ -1,68 +1,58 @@
-import time
-from datetime import datetime
-from enum import Enum
+from datetime import datetime, timezone, timedelta
from bs4 import BeautifulSoup
+JST = timezone(timedelta(hours=9))
-class ParserVersion(Enum):
- ALPHA = 1
+def parse_ongeki_news_site(html: str):
+ identifier = "ONGEKI_JPN"
+ soup = BeautifulSoup(html, "html.parser")
+ items = []
-def make_ongeki_parser(identifier: str, parser: ParserVersion):
- def alpha_parser(html: str):
- soup = BeautifulSoup(html, "html.parser")
- items = []
+ for li in soup.select("li.p-news__listChild"):
+ a_tag = li.select_one("a.p-news__listLink")
+ url = a_tag["href"] if a_tag else None
- for li in soup.select("li.p-news__listChild"):
- a_tag = li.select_one("a.p-news__listLink")
- url = a_tag["href"] if a_tag else None
+ img_tag = li.select_one(".p-news__listThumb img")
+ image_url = img_tag["src"] if img_tag else None
+ image_alt = img_tag["alt"] if img_tag else ""
+ image_link = url if image_url else None
- img_tag = li.select_one(".p-news__listThumb img")
- image_url = img_tag["src"] if img_tag else None
- image_alt = img_tag["alt"] if img_tag else ""
- image_link = url if image_url else None
+ date_type_text = li.select_one(".p-news__listTextUpper")
+ date_text = (
+ date_type_text.text.strip().split("/")[0].strip()
+ if date_type_text
+ else None
+ )
+ type_text = (
+ date_type_text.text.strip().split("/")[-1].strip()
+ if date_type_text and "/" in date_type_text.text
+ else None
+ )
- date_type_text = li.select_one(".p-news__listTextUpper")
- date_text = (
- date_type_text.text.strip().split("/")[0].strip()
- if date_type_text
- else None
- )
- type_text = (
- date_type_text.text.strip().split("/")[-1].strip()
- if "/" in date_type_text.text
- else None
- )
+ timestamp = None
+ if date_text:
+ try:
+ dt = datetime.strptime(date_text, "%Y.%m.%d %a").replace(tzinfo=JST)
+ timestamp = int(dt.timestamp())
+ except Exception:
+ timestamp = None
- timestamp = None
- if date_text:
- try:
- dt = datetime.strptime(date_text, "%Y.%m.%d %a")
- timestamp = int(time.mktime(dt.timetuple()))
- except:
- timestamp = None
+ entry = {
+ "date": date_text,
+ "identifier": identifier,
+ "type": type_text if type_text not in ["GAME", "CARDMAKER"] else None,
+ "timestamp": timestamp,
+ "headline": None,
+ "content": image_alt,
+ "url": url,
+ "is_ai_summary": False,
+ "images": [{"image": image_url, "link": image_link}]
+ if image_url
+ else [],
+ }
- entry = {
- "date": date_text,
- "identifier": identifier,
- "type": type_text if type_text not in ["GAME", "CARDMAKER"] else None,
- "timestamp": timestamp,
- "headline": None,
- "content": image_alt,
- "url": url,
- "is_ai_summary": False,
- "images": [{"image": image_url, "link": image_link}]
- if image_url
- else [],
- }
+ items.append(entry)
- items.append(entry)
-
- return items
-
- if parser == ParserVersion.ALPHA:
- return alpha_parser
-
-
-parse_ongeki_news_site = make_ongeki_parser("ONGEKI_JPN", ParserVersion.ALPHA)
+ return items \ No newline at end of file
diff --git a/taito/__init__.py b/taito/__init__.py
new file mode 100644
index 0000000..bc55d25
--- /dev/null
+++ b/taito/__init__.py
@@ -0,0 +1,7 @@
+from taito.music_diver import parse_music_diver_news_json
+from taito.street_fighter import parse_sf_news_site
+
+__all__ = [
+ "parse_music_diver_news_json",
+ "parse_sf_news_site",
+] \ No newline at end of file
diff --git a/taito/music_diver.py b/taito/music_diver.py
index 5469ad5..efab0b0 100644
--- a/taito/music_diver.py
+++ b/taito/music_diver.py
@@ -52,6 +52,7 @@ def parse_music_diver_news_json(data_str: str):
"headline": post["title"],
"content": content,
"url": None,
- "images": images
+ "images": images,
+ "is_ai_summary": False
})
return news_posts
diff --git a/taito/street_fighter.py b/taito/street_fighter.py
index 987b72b..bf58090 100644
--- a/taito/street_fighter.py
+++ b/taito/street_fighter.py
@@ -3,15 +3,12 @@ from bs4 import BeautifulSoup
import re
from datetime import datetime
from urllib.parse import urljoin
-from enum import Enum
from constants import STREET_FIGHTER_NEWS_SITE
import requests
import base64
IMAGE_LIMIT = 10 # only allow 10 images to be processed as b64 is expensive to store
-class ParserVersion(Enum):
- ALPHA = 1
def _convert_image_to_base64(img_url: str):
headers = {
@@ -26,81 +23,72 @@ def _convert_image_to_base64(img_url: str):
else:
raise Exception(f"Failed to fetch image from URL: {img_url}, status code: {response.status_code}")
-def make_sf_parser(identifier: str, parser: ParserVersion):
- def alpha_parser(html: str):
- soup = BeautifulSoup(html, "html.parser")
- news_entries = []
- img_processed = 0
- news_links = soup.find_all('a', class_='btn_latestnews')
- for link in news_links:
- try:
- url = link.get('href', '')
- if url.startswith('/'):
- url = urljoin(STREET_FIGHTER_NEWS_SITE, url)
- info_p = link.find('p', class_='info_list_event')
- if not info_p:
- continue
- date_span = info_p.find('span', class_='latestnews_date')
- if not date_span:
- continue
- date_text = date_span.get_text(strip=True)
- date_match = re.match(r'(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2})\s* (.+)', date_text)
- if not date_match:
- continue
- date_str = date_match.group(1)
- time_str = date_match.group(2)
- datetime_str = f"{date_str} {time_str}"
- try:
- post_date = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M")
- timestamp = int(post_date.timestamp())
- except ValueError:
- continue
- headline_span = info_p.find('span', class_='info_list_txt')
- headline = headline_span.get_text(strip=True) if headline_span else ""
- headline = re.sub(r'<br\s*/?>', ' ', headline)
- headline = re.sub(r'\s+', ' ', headline).strip()
- images = []
- img_div = link.find('div', class_='image')
- if img_div:
- img_tag = img_div.find('img')
- if img_tag:
- img_src = img_tag.get('src', '')
- if img_src.startswith('/'):
- img_src = urljoin('https://sf6ta.jp', img_src)
- if img_processed <= IMAGE_LIMIT:
- try:
- img_b64 = _convert_image_to_base64(img_src)
- images.append({
- 'image': img_b64,
- 'link': url
- })
- except Exception:
- pass # Failed likely due to 403. Just show no images in that case
- img_processed += 1
- news_entry = {
- 'date': post_date.strftime("%Y-%m-%d %H:%M"),
- 'identifier': identifier,
- 'type': None,
- 'timestamp': timestamp,
- 'headline': None,
- 'content': headline, # content should be prio-ed over headline
- 'url': url,
- 'images': images,
- 'is_ai_summary': False
- }
- news_entries.append(news_entry)
- except Exception as e:
+def parse_sf_news_site(html: str):
+ identifier = "STREET_FIGHTER"
+ soup = BeautifulSoup(html, "html.parser")
+ news_entries = []
+ img_processed = 0
+ news_links = soup.find_all('a', class_='btn_latestnews')
+ for link in news_links:
+ try:
+ url = link.get('href', '')
+ if url.startswith('/'):
+ url = urljoin(STREET_FIGHTER_NEWS_SITE, url)
+ info_p = link.find('p', class_='info_list_event')
+ if not info_p:
continue
+ date_span = info_p.find('span', class_='latestnews_date')
+ if not date_span:
+ continue
+ date_text = date_span.get_text(strip=True)
+ date_match = re.match(r'(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2})\s* (.+)', date_text)
+ if not date_match:
+ continue
+ date_str = date_match.group(1)
+ time_str = date_match.group(2)
+ datetime_str = f"{date_str} {time_str}"
+ try:
+ post_date = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M")
+ timestamp = int(post_date.timestamp())
+ except ValueError:
+ continue
+ headline_span = info_p.find('span', class_='info_list_txt')
+ headline = headline_span.get_text(strip=True) if headline_span else ""
+ headline = re.sub(r'<br\s*/?>', ' ', headline)
+ headline = re.sub(r'\s+', ' ', headline).strip()
+ images = []
+ img_div = link.find('div', class_='image')
+ if img_div:
+ img_tag = img_div.find('img')
+ if img_tag:
+ img_src = img_tag.get('src', '')
+ if img_src.startswith('/'):
+ img_src = urljoin('https://sf6ta.jp', img_src)
+ if img_processed <= IMAGE_LIMIT:
+ try:
+ img_b64 = _convert_image_to_base64(img_src)
+ images.append({
+ 'image': img_b64,
+ 'link': url
+ })
+ except Exception:
+ pass # Failed likely due to 403. Just show no images in that case
+ img_processed += 1
+ news_entry = {
+ 'date': post_date.strftime("%Y-%m-%d %H:%M"),
+ 'identifier': identifier,
+ 'type': None,
+ 'timestamp': timestamp,
+ 'headline': None,
+ 'content': headline, # content should be prio-ed over headline
+ 'url': url,
+ 'images': images,
+ 'is_ai_summary': False
+ }
+ news_entries.append(news_entry)
- return news_entries
-
- if parser == ParserVersion.ALPHA:
- return alpha_parser
- else:
- raise ValueError("Unknown Parser Version")
-
+ except Exception:
+ continue
-parse_sf_news_site = make_sf_parser(
- "STREET_FIGHTER", ParserVersion.ALPHA
-)
+ return news_entries \ No newline at end of file
send patches to the email below
yukais@pinapelz.com
include the subject [PATCH repo_name]
pinapelz.com
homepage