はじめに
以前に、ホロライブの配信予定や動画情報をホロジュールと Youtube から収集するプログラムを作成しましたが、ホロライブの所属メンバーが増えてグループ構成など複雑になってきたので、収集する情報を増やすために、ホロジュール収集プログラムをあらためて作成しました。
ちなみに、ホロジュールの API は公開されている（コラボ情報も含めて！）のですが、とりあえず前回と同様に Web スクレイピングを利用して配信予定の情報を収集しています。
なお、同時に開発している FastAPI 版の WebAPI のための準備でもあります。
Selenium のバージョンアップ
Selenium 4.6 から、Selenium 自体に ChromeDriver の自動更新機能「Selenium Manager」が搭載されたため、環境構築の手順を簡略化でき、ドライバーの初期化もシンプルになりました。
def __setup_options(self) -> webdriver.ChromeOptions:
options = webdriver.ChromeOptions()
# ヘッドレスモードとする
options.add_argument('--headless=new')
return options
...
# オプションのセットアップ
options = self.__setup_options()
# ドライバの初期化（オプション（ヘッドレスモード）とプロファイルを指定）
self.__driver = webdriver.Chrome(options=options)
# 指定したドライバに対して最大で10秒間待つように設定する
self.__wait = WebDriverWait(self.__driver, 10)
Pydantic の利用
モデルの定義に Pydantic を利用することで、バリデーションや型ヒントによる安全性が向上し、JSONとモデルの相互変換や、アプリケーション設定の読み込みも楽になりました。
Pydantic はデータ検証ライブラリです。スキーマの検証やシリアライズが型アノテーションによって制御されています。カスタムバリデータやカスタムシリアライザも使用できます。
Pydantic を利用したモデルの定義
from datetime import datetime, timezone, timedelta
from typing_extensions import Annotated
from pydantic import BaseModel, ConfigDict, Field, computed_field
from pydantic.functional_validators import BeforeValidator
class ScheduleModel(BaseModel):
# フィールドに別名や規定値、説明などを指定できる
id: PyObjectId | None = Field(alias="_id", default=None, description="スケジュールID")
code: str | None = Field(default=None, description="配信者コード")
video_id: str | None = Field(default=None, description="動画ID")
# default_factory で動的な規定値を指定することもできる
streaming_at: datetime | None = Field(default_factory=lambda: datetime.now(tz=JST), description="配信日時")
name: str | None = Field(default=None, description="配信者名")
title: str | None = Field(default=None, description="タイトル")
url: str | None = Field(default=None, description="Youtube URL")
description: str | None = Field(default=None, description="概要")
published_at: datetime | None = Field(default_factory=lambda: datetime.now(tz=JST), description="投稿日時")
channel_id: str | None = Field(default=None, description="チャンネルID")
channel_title: str | None = Field(default=None, description="チャンネル名")
tags: list[str] = Field(default_factory=list, description="タグ")
# モデルの各種設定
model_config = ConfigDict(
# 別名でのアクセスを許可するか（例えば id と _id）
populate_by_name=True,
# 任意の型を許可するか
arbitrary_types_allowed=True,
# サンプルデータの宣言
json_schema_extra={
"example": {
"code": "HL0000",
"video_id": "動画ID",
"streaming_at": "2023-12-01T12:00:00Z",
"name": "配信者名",
"title": "タイトル",
"url": "Youtube URL",
"description": "概要",
"published_at": "2023-12-01T12:00:00Z",
"channel_id": "チャンネルID",
"channel_title": "チャンネル名",
"tags": []
}
},
)
@computed_field # フィールド同士の計算によりセットされるフィールド
@property
def key(self) -> str:
return self.code + "_" + self.streaming_at.strftime("%Y%m%d_%H%M%S") if (self.code is not None and self.streaming_at is not None) else ""
JSONとモデルの相互変換（例としてオブジェクトからドキュメントへの変換）
db = MongoDB.getInstance().holoduledb
collection = db.schedules
# video_id が一致するドキュメントを削除
video_ids = [schedule.video_id for schedule in self.schedules]
collection.delete_many({"video_id": {"$in": video_ids}})
# ScheduleModelオブジェクトをドキュメントに変換して一括登録
dumps = [schedule.model_dump(by_alias=True, exclude=["id"]) for schedule in self.schedules]
collection.insert_many(dumps)
アプリケーション設定の読み込み
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class MongoSettings(BaseSettings):
uri: str
database: str
# .env ファイルから mongo_～ となっている設定情報を取得
model_config = SettingsConfigDict(env_file=".env", env_prefix='mongo_')
@lru_cache
def get_mongo_settings() -> MongoSettings:
# キャッシュしたMongoDBの設定を取得
return MongoSettings()
開発環境
- Windows 11
- PowerShell 7.3.8
- Visual Studio Code 1.83
- Python 3.11 + Poetry + pyenv
- MongoDB 6.0
開発準備
Poetry と pyenv の確認
参考：Windows で Python の開発環境を構築する（Poetry と pyenv を利用）
> poetry --version
Poetry version 1.3.2
> pyenv --version
pyenv 3.1.1
MongoDB の確認
> mongosh --version
1.6.0
> mongosh localhost:27017/admin -u admin -p
データベースの作成とロール（今回は dbOwner ）の設定
MongoDB > use holoduledb
MongoDB > db.createUser( { user:"owner", pwd:"password", roles:[{ "role" : "dbOwner", "db" : "holoduledb" }] } );
Web スクレイピングで利用する google-chrome の導入
Windows の場合は、最新の Google Chrome をインストールしておくだけ
YouTube の動画情報を取得するための YouTube Data API v3 の有効化と API キーの取得
- Google Developer Console にログイン
- ダッシュボードでプロジェクトを作成
- ライブラリで YouTube Data API v3 を有効化
- 認証情報で認証情報を作成して APIキー を取得
プロジェクト作成
プロジェクトを作成してディレクトリに入る
> poetry new holocollect --name app
Created package app in holocollect
> cd holocollect
プロジェクトで利用する Python を pyenv でインストール
> pyenv install 3.11.1
プロジェクトで利用するローカルの Python のバージョンを変更
> pyenv local 3.11.1
> python -V
Python 3.11.1
バージョンを指定して Python 仮想環境を作成（pyenv で管理している Python のパスを指定）
> python -c "import sys; print(sys.executable)"
> poetry env use C:\Users\[UserName]\.pyenv\pyenv-win\versions\3.11.1\python.exe
パッケージの追加
> poetry add pylint
> poetry add pymongo
> poetry add beautifulsoup4
> poetry add requests
> poetry add selenium
> poetry add oauth2client
> poetry add google-api-python-client
> poetry add lxml
> poetry add pydantic-settings
プログラム実装
ログ（app/logger.py）
ログ設定として logger.json を読み込みます。
import json
import datetime
from os.path import join
from logging import Logger, config, getLogger
def get_logger(log_dir: str, json_path: str, verbose: bool=False) -> Logger | None:
try:
with open(json_path, "r", encoding="utf-8") as f:
log_config = json.load(f)
except FileNotFoundError:
print(f"設定ファイル {json_path} が見つかりません。")
raise
except json.JSONDecodeError:
print(f"設定ファイル {json_path} の形式が正しくありません。")
raise
# ログファイル名を日付とする
log_path = join(log_dir, f"{datetime.datetime.now().strftime('%Y%m%d')}.log")
log_config["handlers"]["rotateFileHandler"]["filename"] = log_path
# verbose引数が True の場合、レベルをINFOからDEBUGに置換
if verbose:
log_config["root"]["level"] = "DEBUG"
log_config["handlers"]["consoleHandler"]["level"] = "DEBUG"
log_config["handlers"]["rotateFileHandler"]["level"] = "DEBUG"
try:
# ロギングの設定を適用
config.dictConfig(log_config)
except ValueError as e:
print(f"ログ設定の適用に失敗しました：{e}")
raise
# ロガーを取得
logger = getLogger(__name__)
return logger
設定（app/settings.py）
.env から MongoDB と YOUTUBE と ホロジュール の設定を読み込みます。
import urllib.request
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class MongoSettings(BaseSettings):
uri: str
database: str
model_config = SettingsConfigDict(env_file=".env", env_prefix='mongo_')
class YoutubeSettings(BaseSettings):
api_key: str
api_service_name: str
api_version: str
url_pattern: str
model_config = SettingsConfigDict(env_file=".env", env_prefix='youtube_')
class HoloduleSettings(BaseSettings):
url: str
model_config = SettingsConfigDict(env_file=".env", env_prefix='holodule_')
async def check_holodule_url(self) -> bool:
return await self.__check_url(self.url)
async def __check_url(self, url: str) -> bool:
try:
async with urllib.request.urlopen(url) as response:
return True
except Exception:
return False
@lru_cache
def get_mongo_settings() -> MongoSettings:
# キャッシュしている設定情報を返却
return MongoSettings()
@lru_cache
def get_youtube_settings() -> YoutubeSettings:
# キャッシュしている設定情報を返却
return YoutubeSettings()
@lru_cache
def get_holodule_settings() -> HoloduleSettings:
# キャッシュしている設定情報を返却
return HoloduleSettings()
MongoDB（app/mongodb.py）
MongoClient をシングルトンで管理して利用します。
import atexit
from pymongo import MongoClient
from app.settings import get_mongo_settings
mongo_settings = get_mongo_settings()
class MongoDB:
_instance = None
@staticmethod
def getInstance() -> MongoClient:
if MongoDB._instance is None:
MongoDB()
return MongoDB._instance
def __init__(self):
if MongoDB._instance is not None:
raise Exception("このクラスはシングルトンです。")
else:
MongoDB._instance = MongoClient(mongo_settings.uri)
# クリーンアップ関数の登録
atexit.register(self.close)
def close(self):
if MongoDB._instance is not None:
MongoDB._instance.close()
スケジュールモデル（app/models/schedule.py）
配信情報を保持するモデルクラスです。
import re
from datetime import datetime, timezone, timedelta
from typing_extensions import Annotated
from pydantic import BaseModel, ConfigDict, Field, computed_field
from pydantic.functional_validators import BeforeValidator
PyObjectId = Annotated[str, BeforeValidator(str)]
UTC = timezone.utc
JST = timezone(timedelta(hours=+9), "JST")
class ScheduleModel(BaseModel):
id: PyObjectId | None = Field(alias="_id", default=None, description="スケジュールID")
code: str | None = Field(default=None, description="配信者コード")
video_id: str | None = Field(default=None, description="動画ID")
streaming_at: datetime | None = Field(default_factory=lambda: datetime.now(tz=JST), description="配信日時")
name: str | None = Field(default=None, description="配信者名")
title: str | None = Field(default=None, description="タイトル")
url: str | None = Field(default=None, description="Youtube URL")
description: str | None = Field(default=None, description="概要")
published_at: datetime | None = Field(default_factory=lambda: datetime.now(tz=JST), description="投稿日時")
channel_id: str | None = Field(default=None, description="チャンネルID")
channel_title: str | None = Field(default=None, description="チャンネル名")
tags: list[str] = Field(default_factory=list, description="タグ")
model_config = ConfigDict(
populate_by_name=True, # エイリアス名でのアクセスを許可するか（例えば id と _id）
arbitrary_types_allowed=True, # 任意の型を許可するか
json_schema_extra={
"example": {
"code": "HL0000",
"video_id": "動画ID",
"streaming_at": "2023-12-01T12:00:00Z",
"name": "配信者名",
"title": "タイトル",
"url": "Youtube URL",
"description": "概要",
"published_at": "2023-12-01T12:00:00Z",
"channel_id": "チャンネルID",
"channel_title": "チャンネル名",
"tags": []
}
},
)
@computed_field
@property
def key(self) -> str:
return self.code + "_" + self.streaming_at.strftime("%Y%m%d_%H%M%S") if (self.code is not None and self.streaming_at is not None) else ""
def set_video_info(self, video_id: str, title: str, description: str, published_at: datetime,
channel_id: str, channel_title: str, tags: list[str]):
self.video_id = video_id
self.title = title
self.description = re.sub(r'[\r\n\"\']', '', description)[:1000]
self.published_at = published_at
self.channel_id = channel_id
self.channel_title = channel_title
self.tags = tags
スケジュールコレクション（app/models/schedules.py）
配信情報のコレクションを保持する反復可能クラスです。
from pydantic import BaseModel
import pymongo
import csv
from logging import getLogger
from app.models.schedule import ScheduleModel
from app.mongodb import MongoDB
logger = getLogger(__name__)
class ScheduleCollection(BaseModel):
schedules: list[ScheduleModel] = []
index: int = 0
def __iter__(self) -> 'ScheduleCollection':
self.index = 0
return self
def __next__(self) -> ScheduleModel:
if self.index < len(self.schedules):
result = self.schedules[self.index]
self.index += 1
return result
else:
raise StopIteration
def __len__(self) -> int:
return len(self.schedules)
def __getitem__(self, index: int) -> ScheduleModel:
return self.schedules[index]
def append(self, schedule: ScheduleModel) -> None:
self.schedules.append(schedule)
def remove_at(self, index: int) -> None:
self.schedules.pop(index)
def output_to_csv(self, filepath: str) -> None:
try:
with open(filepath, "w", newline="", encoding="utf_8_sig") as csvfile:
csvwriter = csv.writer(csvfile, delimiter=",")
csvwriter.writerow([attr for attr in vars(ScheduleModel())])
for schedule in self.schedules:
csvwriter.writerow([value for value in vars(schedule).values()])
except (FileNotFoundError, PermissionError) as e:
self._logger.error("CSV エラーが発生しました。%s", e, exc_info=True)
raise
def save_to_mongodb(self) -> None:
try:
db = MongoDB.getInstance().holoduledb
collection = db.schedules
# video_id が一致するドキュメントを削除
video_ids = [schedule.video_id for schedule in self.schedules]
collection.delete_many({"video_id": {"$in": video_ids}})
# ScheduleModelオブジェクトをドキュメントに変換して一括登録
dumps = [schedule.model_dump(by_alias=True, exclude=["id"]) for schedule in self.schedules]
collection.insert_many(dumps)
except pymongo.errors.ConnectionError as e:
self.logger.error("MongoDB 接続に失敗しました。%s", e, exc_info=True)
raise
except pymongo.errors.OperationFailure as e:
self.logger.error("MongoDB 操作に失敗しました。%s", e, exc_info=True)
raise
except pymongo.errors.PyMongoError as e:
self.logger.error("MongoDB エラーが発生しました。%s", e, exc_info=True)
raise
配信者モデル（app/models/streamer.py）
配信者情報を保持するモデルクラスです。
from typing_extensions import Annotated
from pydantic import BaseModel, ConfigDict, Field
from pydantic.functional_validators import BeforeValidator
PyObjectId = Annotated[str, BeforeValidator(str)]
class StreamerModel(BaseModel):
id: PyObjectId | None = Field(alias="_id", default=None, description="ストリーマーID")
code: str | None = Field(default=None, description="ストリーマーコード")
name: str | None = Field(default=None, description="ストリーマー名")
group: str | None = Field(default=None, description="グループ")
affiliations: list[str] | None = Field(default=None, description="所属")
image_name: str | None = Field(default=None, description="画像名")
channel_id: str | None = Field(default=None, description="チャンネルID")
is_retired: bool | None = Field(default=False, description="引退済み")
model_config = ConfigDict(
populate_by_name=True, # エイリアス名でのアクセスを許可するか（例えば id と _id）
arbitrary_types_allowed=True, # 任意の型を許可するか
json_schema_extra={
"example": {
"code": "HL0000",
"name": "ホロライブ",
"group": "hololive",
"affiliations": ['bland', 'jp'],
"image_name": "hololive.jpg",
"channel_id": "@hololive",
"is_retired": False
}
},
)
配信者コレクション（app/models/streamers.py）
配信者情報のコレクションを保持するクラスです。
from typing import ClassVar
from pydantic import BaseModel, ConfigDict
import pymongo
from logging import getLogger
from app.models.streamer import StreamerModel
from app.mongodb import MongoDB
logger = getLogger(__name__)
class StreamerCollection(BaseModel):
streamers: ClassVar[dict[str, StreamerModel]] = {
# 配信者の情報はとりあえずクラス変数で固定で管理（グループや所属の情報を追加）
"ホロライブ" : StreamerModel(code="HL0000", name="ホロライブ", group="hololive", affiliations=["bland","jp"], image_name="hololive.jpg", channel_id="@hololive"),
"ときのそら" : StreamerModel(code="HL0001", name="ときのそら", group="hololive", affiliations=["gen0","jp"], image_name="tokino_sora.jpg", channel_id="@TokinoSora"),
"ロボ子さん" : StreamerModel(code="HL0002", name="ロボ子さん", group="hololive", affiliations=["gen0","jp"], image_name="robokosan.jpg", channel_id="@Robocosan"),
"さくらみこ" : StreamerModel(code="HL0003", name="さくらみこ", group="hololive", affiliations=["gen0","jp"], image_name="sakura_miko.jpg", channel_id="@SakuraMiko"),
"星街すいせい" : StreamerModel(code="HL0004", name="星街すいせい", group="hololive", affiliations=["gen0","jp"], image_name="hoshimachi_suisei.jpg", channel_id="@HoshimachiSuisei"),
"AZKi" : StreamerModel(code="HL0005", name="AZKi", group="hololive", affiliations=["gen0","jp"], image_name="azki.jpg", channel_id="@AZKi"),
"夜空メル" : StreamerModel(code="HL0101", name="夜空メル", group="hololive", affiliations=["gen1","jp"], image_name="yozora_mel.jpg", channel_id="@YozoraMel"),
"アキ・ローゼンタール" : StreamerModel(code="HL0102", name="アキ・ローゼンタール", group="hololive", affiliations=["gen1","jp"], image_name="aki_rosenthal.jpg", channel_id="@AkiRosenthal"),
"赤井はあと" : StreamerModel(code="HL0103", name="赤井はあと", group="hololive", affiliations=["gen1","jp"], image_name="haachama.jpg", channel_id="@AkaiHaato"),
"白上フブキ" : StreamerModel(code="HL0104", name="白上フブキ", group="hololive", affiliations=["gen1","gamers","jp"], image_name="shirakami_fubuki.jpg", channel_id="@ShirakamiFubuki"),
"夏色まつり" : StreamerModel(code="HL0105", name="夏色まつり", group="hololive", affiliations=["gen1","jp"], image_name="natsuiro_matsuri.jpg", channel_id="@NatsuiroMatsuri"),
"湊あくあ" : StreamerModel(code="HL0201", name="湊あくあ", group="hololive", affiliations=["gen2","jp"], image_name="minato_aqua.jpg", channel_id="@MinatoAqua"),
"紫咲シオン" : StreamerModel(code="HL0202", name="紫咲シオン", group="hololive", affiliations=["gen2","jp"], image_name="murasaki_shion.jpg", channel_id="@MurasakiShion"),
"百鬼あやめ" : StreamerModel(code="HL0203", name="百鬼あやめ", group="hololive", affiliations=["gen2","jp"], image_name="nakiri_ayame.jpg", channel_id="@NakiriAyame"),
"癒月ちょこ" : StreamerModel(code="HL0204", name="癒月ちょこ", group="hololive", affiliations=["gen2","jp"], image_name="yuzuki_choco.jpg", channel_id="@YuzukiChoco"),
"大空スバル" : StreamerModel(code="HL0205", name="大空スバル", group="hololive", affiliations=["gen2","jp"], image_name="oozora_subaru.jpg", channel_id="@OozoraSubaru"),
"大神ミオ" : StreamerModel(code="HL0G02", name="大神ミオ", group="hololive", affiliations=["gamers","jp"], image_name="ookami_mio.jpg", channel_id="@OokamiMio"),
"猫又おかゆ" : StreamerModel(code="HL0G03", name="猫又おかゆ", group="hololive", affiliations=["gamers","jp"], image_name="nekomata_okayu.jpg", channel_id="@NekomataOkayu"),
"戌神ころね" : StreamerModel(code="HL0G04", name="戌神ころね", group="hololive", affiliations=["gamers","jp"], image_name="inugami_korone.jpg", channel_id="@InugamiKorone"),
"兎田ぺこら" : StreamerModel(code="HL0301", name="兎田ぺこら", group="hololive", affiliations=["gen3","jp"], image_name="usada_pekora.jpg", channel_id="@usadapekora"),
"潤羽るしあ" : StreamerModel(code="HL0302", name="潤羽るしあ", group="hololive", affiliations=["gen3","jp"], image_name="uruha_rushia.jpg", channel_id="@UruhaRushia", is_retired=True),
"不知火フレア" : StreamerModel(code="HL0303", name="不知火フレア", group="hololive", affiliations=["gen3","jp"], image_name="shiranui_flare.jpg", channel_id="@ShiranuiFlare"),
"白銀ノエル" : StreamerModel(code="HL0304", name="白銀ノエル", group="hololive", affiliations=["gen3","jp"], image_name="shirogane_noel.jpg", channel_id="@ShiroganeNoel"),
"宝鐘マリン" : StreamerModel(code="HL0305", name="宝鐘マリン", group="hololive", affiliations=["gen3","jp"], image_name="housyou_marine.jpg", channel_id="@HoushouMarine"),
"天音かなた" : StreamerModel(code="HL0401", name="天音かなた", group="hololive", affiliations=["gen4","jp"], image_name="amane_kanata.jpg", channel_id="@AmaneKanata"),
"桐生ココ" : StreamerModel(code="HL0402", name="桐生ココ", group="hololive", affiliations=["gen4","jp"], image_name="kiryu_coco.jpg", channel_id="@KiryuCoco", is_retired=True),
"角巻わため" : StreamerModel(code="HL0403", name="角巻わため", group="hololive", affiliations=["gen4","jp"], image_name="tsunomaki_watame.jpg", channel_id="@TsunomakiWatame"),
"常闇トワ" : StreamerModel(code="HL0404", name="常闇トワ", group="hololive", affiliations=["gen4","jp"], image_name="tokoyami_towa.jpg", channel_id="@TokoyamiTowa"),
"姫森ルーナ" : StreamerModel(code="HL0405", name="姫森ルーナ", group="hololive", affiliations=["gen4","jp"], image_name="himemori_luna.jpg", channel_id="@HimemoriLuna"),
"獅白ぼたん" : StreamerModel(code="HL0501", name="獅白ぼたん", group="hololive", affiliations=["gen5","jp"], image_name="shishiro_botan.jpg", channel_id="@ShishiroBotan"),
"雪花ラミィ" : StreamerModel(code="HL0502", name="雪花ラミィ", group="hololive", affiliations=["gen5","jp"], image_name="yukihana_lamy.jpg", channel_id="@YukihanaLamy"),
"尾丸ポルカ" : StreamerModel(code="HL0503", name="尾丸ポルカ", group="hololive", affiliations=["gen5","jp"], image_name="omaru_polka.jpg", channel_id="@OmaruPolka"),
"桃鈴ねね" : StreamerModel(code="HL0504", name="桃鈴ねね", group="hololive", affiliations=["gen5","jp"], image_name="momosuzu_nene.jpg", channel_id="@MomosuzuNene"),
"魔乃アロエ" : StreamerModel(code="HL0505", name="魔乃アロエ", group="hololive", affiliations=["gen5","jp"], image_name="mano_aloe.jpg", channel_id="@ManoAloe", is_retired=True),
"ラプラス" : StreamerModel(code="HL0601", name="ラプラス・ダークネス", group="hololive", affiliations=["gen6","jp"], image_name="laplus_darknesss.jpg", channel_id="@LaplusDarknesss"),
"鷹嶺ルイ" : StreamerModel(code="HL0602", name="鷹嶺ルイ", group="hololive", affiliations=["gen6","jp"], image_name="takane_lui.jpg", channel_id="@TakaneLui"),
"博衣こより" : StreamerModel(code="HL0603", name="博衣こより", group="hololive", affiliations=["gen6","jp"], image_name="hakui_koyori.jpg", channel_id="@HakuiKoyori"),
"沙花叉クロヱ" : StreamerModel(code="HL0604", name="沙花叉クロヱ", group="hololive", affiliations=["gen6","jp"], image_name="sakamata_chloe.jpg", channel_id="@SakamataChloe"),
"風真いろは" : StreamerModel(code="HL0605", name="風真いろは", group="hololive", affiliations=["gen6","jp"], image_name="kazama_iroha.jpg", channel_id="@kazamairoha"),
"hololive DEV_IS" : StreamerModel(code="HLDI00", name="hololive DEV_IS", group="hololive_DEV_IS)", affiliations=["bland","jp"], image_name="hololive_dev_is.jpg", channel_id="@hololiveDEV_IS"),
"火威青" : StreamerModel(code="HLDI01", name="火威青", group="hololive_DEV_IS", affiliations=["dev_is","jp"], image_name="hiodoshi_ao.jpg", channel_id="@HiodoshiAo"),
"儒烏風亭らでん" : StreamerModel(code="HLDI02", name="儒烏風亭らでん", group="hololive_DEV_IS", affiliations=["dev_is","jp"], image_name="juufuutei_raden.jpg", channel_id="@JuufuuteiRaden"),
"一条莉々華" : StreamerModel(code="HLDI03", name="一条莉々華", group="hololive_DEV_IS", affiliations=["dev_is","jp"], image_name="otonose_kanade.jpg", channel_id="@OtonoseKanade"),
"音乃瀬奏" : StreamerModel(code="HLDI04", name="音乃瀬奏", group="hololive_DEV_IS", affiliations=["dev_is","jp"], image_name="ichijou_ririka.jpg", channel_id="@IchijouRirika"),
"轟はじめ" : StreamerModel(code="HLDI05", name="轟はじめ", group="hololive_DEV_IS", affiliations=["dev_is","jp"], image_name="todoroki_hajime.jpg", channel_id="@TodorokiHajime"),
"hololive ID" : StreamerModel(code="HLID00", name="hololive Indonesia", group="hololive_id)", affiliations=["bland","id"], image_name="hololive_id.jpg", channel_id="@hololiveIndonesia"),
"Risu" : StreamerModel(code="HLID01", name="Ayunda Risu", group="hololive_id", affiliations=["gen1","id"], image_name="ayunda_risu.jpg", channel_id="@AyundaRisu"),
"Moona" : StreamerModel(code="HLID02", name="Moona Hoshinova", group="hololive_id",affiliations=["gen1","id"], image_name="moona_hoshinova.jpg", channel_id="@MoonaHoshinova"),
"Iofi" : StreamerModel(code="HLID03", name="Airani Iofifteen", group="hololive_id", affiliations=["gen1","id"], image_name="airani_iofifteen.jpg", channel_id="@AiraniIofifteen"),
"Ollie" : StreamerModel(code="HLID04", name="Kureiji Ollie", group="hololive_id", affiliations=["gen2","id"], image_name="kureiji_ollie.jpg", channel_id="@KureijiOllie"),
"Anya" : StreamerModel(code="HLID05", name="Anya Melfissa", group="hololive_id", affiliations=["gen2","id"], image_name="anya_melfissa.jpg", channel_id="@AnyaMelfissa"),
"Reine" : StreamerModel(code="HLID06", name="Pavolia Reine", group="hololive_id", affiliations=["gen2","id"], image_name="pavolia_reine.jpg", channel_id="@PavoliaReine"),
"Zeta" : StreamerModel(code="HLID07", name="Vestia Zeta", group="hololive_id", affiliations=["gen3","id"], image_name="vestia_zeta.jpg", channel_id="@VestiaZeta"),
"Kaela" : StreamerModel(code="HLID08", name="Kaela Kovalskia",group="hololive_id", affiliations=["gen3","id"], image_name="kaela_kovalskia.jpg", channel_id="@KaelaKovalskia"),
"Kobo" : StreamerModel(code="HLID09", name="Kobo Kanaeru", group="hololive_id", affiliations=["gen3","id"], image_name="kobo_kanaeru.jpg", channel_id="@KoboKanaeru"),
"hololive EN" : StreamerModel(code="HLEN00", name="hololive English", group="hololive_en)", affiliations=["bland","en"], image_name="hololive_en.jpg", channel_id="@hololiveEnglish"),
"Calli" : StreamerModel(code="HLEN01", name="Mori Calliope", group="hololive_en", affiliations=["gen1","en"], image_name="mori_calliope.jpg", channel_id="@MoriCalliope"),
"Kiara" : StreamerModel(code="HLEN02", name="Takanashi Kiara", group="hololive_en", affiliations=["gen1","en"], image_name="takanashi_kiara.jpg", channel_id="@TakanashiKiara"),
"Ina" : StreamerModel(code="HLEN03", name="Ninomae Ina'nis", group="hololive_en", affiliations=["gen1","en"], image_name="ninomae_ina'nis.jpg", channel_id="@NinomaeInanis"),
"Gura" : StreamerModel(code="HLEN04", name="Gawr Gura", group="hololive_en", affiliations=["gen1","en"], image_name="gawr_gura.jpg", channel_id="@GawrGura"),
"Amelia" : StreamerModel(code="HLEN05", name="Watson Amelia", group="hololive_en", affiliations=["gen1","en"], image_name="watson_amelia.jpg", channel_id="@WatsonAmelia"),
"IRyS" : StreamerModel(code="HLEN06", name="IRyS", group="hololive_en", affiliations=["hope","gen2","en"], image_name="irys.jpg", channel_id="@IRyS"),
"Fauna" : StreamerModel(code="HLEN07", name="Ceres Fauna", group="hololive_en", affiliations=["gen2","en"], image_name="ceres_fauna.jpg", channel_id="@CeresFauna"),
"Kronii" : StreamerModel(code="HLEN08", name="Ouro Kronii", group="hololive_en", affiliations=["gen2","en"], image_name="ouro_kronii.jpg", channel_id="@OuroKronii"),
"Mumei" : StreamerModel(code="HLEN09", name="Nanashi Mumei", group="hololive_en", affiliations=["gen2","en"], image_name="nanashi_mumei.jpg", channel_id="@NanashiMumei"),
"Baelz" : StreamerModel(code="HLEN10", name="Hakos Baelz", group="hololive_en", affiliations=["gen2","en"], image_name="hakos_baelz.jpg", channel_id="@HakosBaelz"),
"Sana" : StreamerModel(code="HLEN11", name="Tsukumo Sana", group="hololive_en", affiliations=["gen2","en"], image_name="tsukumo_sana.jpg", channel_id="@TsukumoSana", is_retired=True),
"Shiori" : StreamerModel(code="HLEN12", name="Shiori Novella", group="hololive_en", affiliations=["gen3","en"], image_name="shiori_novella.jpg", channel_id="@ShioriNovella"),
"Bijou" : StreamerModel(code="HLEN13", name="Koseki Bijou", group="hololive_en", affiliations=["gen3","en"],image_name="koseki_bijou.jpg", channel_id="@KosekiBijou"),
"Nerissa" : StreamerModel(code="HLEN14", name="Nerissa Ravencroft", group="hololive_en", affiliations=["gen3","en"], image_name="nerissa_ravencroft.jpg", channel_id="@NerissaRavencroft"),
"FUWAMOCO" : StreamerModel(code="HLEN15", name="FUWAMOCO", group="hololive_en", affiliations=["gen3","en"], image_name="fuwamoco.jpg", channel_id="@FUWAMOCOch"),
}
def get_streamer_by_name(self, name: str) -> StreamerModel | None:
return StreamerCollection.streamers.get(name, None)
def save_to_mongodb(self) -> None:
try:
db = MongoDB.getInstance().holoduledb
collection = db.streamers
for streamer in StreamerCollection.streamers.values():
# codeをキーにして更新
query = {"code": streamer.code}
dump = streamer.model_dump(by_alias=True, exclude=["id"])
collection.replace_one(query, dump, upsert=True)
except pymongo.errors.ConnectionError as e:
logger.error("MongoDB 接続に失敗しました。%s", e, exc_info=True)
raise
except pymongo.errors.OperationFailure as e:
logger.error("MongoDB 操作に失敗しました。%s", e, exc_info=True)
raise
except pymongo.errors.PyMongoError as e:
logger.error("MongoDB エラーが発生しました。%s", e, exc_info=True)
raise
収集処理（app/collector.py）
ホロジュールから配信情報を収集し、YOUTUBEの動画情報とあわせて配信情報のコレクションを取得します。
import re
from datetime import datetime, timezone, timedelta, date
from logging import getLogger
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from app.settings import get_youtube_settings, get_holodule_settings
from app.models.schedule import ScheduleModel
from app.models.schedules import ScheduleCollection
from app.models.streamers import StreamerCollection
logger = getLogger(__name__)
holodule_settings = get_holodule_settings()
youtube_settings = get_youtube_settings()
class Collector:
def __init__(self):
# WebDriver 関連
self.__driver = None
self.__wait = None
# Model 関連
self.__streamers = StreamerCollection()
self.__schedules = ScheduleCollection()
# YouTube Data API v3 を利用するための準備
self.__youtube = build(youtube_settings.api_service_name, youtube_settings.api_version, developerKey=youtube_settings.api_key, cache_discovery=False)
def __setup_options(self) -> webdriver.ChromeOptions:
options = webdriver.ChromeOptions()
# ヘッドレスモードとする
options.add_argument('--headless=new')
return options
def __get_schedules(self) -> ScheduleCollection:
# 取得対象の URL に遷移
self.__driver.get(holodule_settings.url)
# <div class="holodule" style="margin-top:10px;">が表示されるまで待機する
self.__wait.until(EC.presence_of_element_located((By.CLASS_NAME, "holodule")))
# ページソースの取得
html = self.__driver.page_source.encode("utf-8")
# ページソースの解析（パーサとして lxml を指定）
soup = BeautifulSoup(html, "lxml")
# タイトルの取得（確認用）
head = soup.find("head")
title = head.find("title").text
logger.info('TITLE : %s', title)
# TODO : ここからはページの構成に合わせて決め打ち = ページの構成が変わったら動かない
# スケジュールの取得
schedules = ScheduleCollection()
date_string = ""
today = date.today()
tab_pane = soup.find("div", class_="tab-pane show active")
containers = tab_pane.find_all("div", class_="container")
for container in containers:
# 日付のみ取得
div_date = container.find("div", class_="holodule navbar-text")
if div_date is not None:
date_text = div_date.text.strip()
match_date = re.search(r"[0-9]{1,2}/[0-9]{1,2}", date_text)
dates = match_date.group(0).split("/")
month = int(dates[0])
day = int(dates[1])
year = today.year
if month == 12 and today.month == 1:
year = year - 1
elif month == 1 and today.month == 12:
year = year + 1
date_string = f"{year}/{month}/{day}"
# 配信者毎のスケジュール
thumbnails = container.find_all("a", class_="thumbnail")
if thumbnails is not None:
for thumbnail in thumbnails:
# Youtube URL
stream_url = thumbnail.get("href")
if stream_url is None or re.match(youtube_settings.url_pattern, stream_url) is None:
continue
# 時刻（先に取得しておいた日付と合体）
div_time = thumbnail.find("div", class_="col-4 col-sm-4 col-md-4 text-left datetime")
if div_time is None:
continue
time_text = div_time.text.strip()
match_time = re.search(r"[0-9]{1,2}:[0-9]{1,2}", time_text)
times = match_time.group(0).split(":")
hour = int(times[0])
minute = int(times[1])
datetime_string = f"{date_string} {hour}:{minute}"
stream_datetime = datetime.strptime(datetime_string, "%Y/%m/%d %H:%M")
# 配信者の名前
div_name = thumbnail.find("div", class_="col text-right name")
if div_name is None:
continue
stream_name = div_name.text.strip()
# リストに追加
streamer = self.__streamers.get_streamer_by_name(stream_name)
if streamer is None:
continue
schedule = ScheduleModel(code=streamer.code, url=stream_url, streaming_at=stream_datetime, name=streamer.name)
schedules.append(schedule)
return schedules
def __get_youtube_video_info(self, youtube_url: str) -> tuple:
try:
logger.info('YOUTUBE_URL : %s', youtube_url)
# Youtube の URL から ID を取得
match_video = re.search(r"^[^v]+v=(.{11}).*", youtube_url)
if not match_video:
logger.error("YouTube URL が不正です。")
return None
video_id = match_video.group(1)
# Youtube はスクレイピングを禁止しているので YouTube Data API (v3) で情報を取得
search_response = self.__youtube.videos().list(
# 結果として snippet のみを取得
part="snippet",
# 検索条件は id
id=video_id,
# 1件のみ取得
maxResults=1
).execute()
# 検索結果から情報を取得
for search_result in search_response.get("items", []):
# id
video_id = search_result["id"]
# タイトル
title = search_result["snippet"]["title"]
# 説明
description = search_result["snippet"]["description"]
# 投稿日
datetime_string = search_result["snippet"]["publishedAt"]
published_at = datetime.fromisoformat(datetime_string).astimezone(tz=timezone(timedelta(hours=+9)))
# チャンネルID
channel_id = search_result["snippet"]["channelId"]
# チャンネルタイトル
channel_title = search_result["snippet"]["channelTitle"]
# タグ（設定されていない＝キーが存在しない場合あり）
tags = search_result["snippet"].setdefault("tags", [])
# 取得した情報を返却
return (video_id, title, description, published_at, channel_id, channel_title, tags)
logger.error("指定したIDに一致する動画がありません。")
return None
except HttpError as e:
logger.error("HTTP エラー %d が発生しました。%s" % (e.resp.status, e.content))
raise
except Exception as e:
logger.error("エラーが発生しました。%s" % e)
raise
def get_holodules(self) -> ScheduleCollection:
try:
# オプションのセットアップ
options = self.__setup_options()
# ドライバの初期化（オプション（ヘッドレスモード）とプロファイルを指定）
self.__driver = webdriver.Chrome(options=options)
# 指定したドライバに対して最大で10秒間待つように設定する
self.__wait = WebDriverWait(self.__driver, 10)
# ホロジュール情報の取得
self.__schedules = self.__get_schedules()
# Youtube情報の取得
for schedule in self.__schedules:
try:
# ホロジュール情報に動画情報を付与
logger.info('SCHEDULE_NAME : %s', schedule.name)
logger.info('SCHEDULE_AT : %s', schedule.streaming_at)
video_info = self.__get_youtube_video_info(schedule.url)
if video_info == None:
continue
schedule.set_video_info(*video_info)
logger.info('SCHEDULE_TITLE : %s', schedule.title)
except Exception as e:
logger.error("エラーが発生しました。", exc_info=True)
raise e
except Exception as e:
logger.error("エラーが発生しました。", exc_info=True)
raise e
finally:
# ドライバを閉じる
if self.__driver is not None and len(self.__driver.window_handles) > 0:
self.__driver.close()
return self.__schedules
def save_to_mongodb(self):
# 配信者情報のDB登録
self.__streamers.save_to_mongodb()
# ホロジュール情報のDB登録
self.__schedules.save_to_mongodb()
def output_to_csv(self, filepath: str):
# ホロジュール情報のCSV出力
self.__schedules.output_to_csv(filepath)
メイン（app/main.py）
このプログラムのエントリポイントです。
import sys
import os
import argparse
from app.collector import Collector
from app.logger import get_logger
RETURN_SUCCESS = 0
RETURN_FAILURE = -1
def main():
# Logger 関連
logger = get_logger("logs", "logger.json", False)
# parser を作る（説明を指定できる）
parser = argparse.ArgumentParser(description="ホロジュールのHTMLをSelenium + BeautifulSoup4 + Youtube API で解析して MongoDB へ登録")
# コマンドライン引数を設定する（説明を指定できる）
parser.add_argument("--csvpath", nargs="?", help="出力するCSVファイルのパス")
# コマンドライン引数を解析する
args = parser.parse_args()
# ファイルパスの取得
is_output = False
csvpath = args.csvpath
if csvpath is not None:
# ディレクトリパスの取得と存在確認
dirpath = os.path.dirname(csvpath)
logger.info("出力ディレクトリパス : %s", dirpath)
if os.path.exists(dirpath) == False:
logger.error("出力するCSVファイルのディレクトリパスが存在しません。 : %s", dirpath)
return RETURN_FAILURE
is_output = True
try:
# Collectorオブジェクトの生成
collector = Collector()
logger.info("ホロジュールの取得を開始します。")
# ホロジュールの取得
schedules = collector.get_holodules()
logger.info("ホロジュールを取得しました。 : %s件", len(schedules))
# ホロジュールの登録
collector.save_to_mongodb()
logger.info("ホロジュールを登録しました。")
# ホロジュールの出力
if is_output == True:
collector.output_to_csv(csvpath)
logger.info("ホロジュールを出力しました。 : %s件", len(schedules))
return RETURN_SUCCESS
except:
logger.error("エラーが発生しました。", exc_info=True)
return RETURN_FAILURE
if __name__ == "__main__":
sys.exit(main())
プログラム実行
.env ファイルを作成して、URLやAPIキーを設定
MONGO_URI = "mongodb://[user]:[password]@127.0.0.1:27017/[db]"
MONGO_DATABASE = "holoduledb"
YOUTUBE_API_KEY = "<Youtube Data API Key>"
YOUTUBE_API_SERVICE_NAME = "youtube"
YOUTUBE_API_VERSION = "v3"
YOUTUBE_URL_PATTERN = "<Youtube URL Pattern>"
HOLODULE_URL = "<Holodule URL>"
ログの設定
{
"version": 1,
"disable_existing_loggers": false,
"root": {
"level": "INFO",
"handlers": [
"consoleHandler",
"rotateFileHandler"
]
},
"handlers": {
"consoleHandler": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "consoleFormatter",
"stream": "ext://sys.stdout"
},
"rotateFileHandler": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "INFO",
"formatter": "rotateFileFormatter",
"filename": "./logs/yyyymmdd.log",
"encoding": "utf-8",
"when": "MIDNIGHT",
"backupCount": "31"
}
},
"formatters": {
"consoleFormatter": {
"format": "[%(levelname)-8s]%(funcName)s - %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S"
},
"rotateFileFormatter": {
"format": "%(asctime)s|%(levelname)-8s|%(name)s|%(funcName)s|%(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S"
}
}
}
lounch.json の設定（デバッグ実行用）
{
"version": "1.0.0",
"configurations": [
{
"name": "ホロコレクトのモジュール実行",
"type": "python",
"request": "launch",
"module": "app",
"justMyCode": true,
"args": ["--csvpath","c:\\temp\\holodule.csv"]
}
]
}
プログラムの実行
> poetry run python -m app --csvpath c:\temp\holodule.csv
実行結果
ホロジュールと Youtube から取得した情報が MongoDB へ登録されます。
[INFO ]main - 出力ディレクトリパス : c:\temp
[INFO ]main - ホロジュールの取得を開始します。
DevTools listening on ws://127.0.0.1:62900/devtools/browser/87ed5929-f99e-451d-988d-995cf41c0099
[INFO ]__get_schedules - TITLE : ホロライブプロダクション配信予定スケジュール『ホロジュール』
[INFO ]get_holodules - SCHEDULE_NAME : Zeta
[INFO ]get_holodules - SCHEDULE_AT : 2024-01-07 00:01:00
[INFO ]__get_youtube_video_info - YOUTUBE_URL : https://www.youtube.com/watch?v=dCZEeq9T9NY
[INFO ]get_holodules - SCHEDULE_TITLE : 【Identity V】uweeeee
[INFO ]get_holodules - SCHEDULE_NAME : 星街すいせい
[INFO ]get_holodules - SCHEDULE_AT : 2024-01-07 00:29:00
[INFO ]__get_youtube_video_info - YOUTUBE_URL : https://www.youtube.com/watch?v=g7mVm0ySTw8
[INFO ]get_holodules - SCHEDULE_TITLE : 【マリオカート8DX】緊急配信⚠れんしゅう進捗やばい【ホロライブ / 星街すいせい 】
...
[INFO ]main - ホロジュールを取得しました。 : 81件
[INFO ]main - ホロジュールを登録しました。
[INFO ]main - ホロジュールを出力しました。 : 81件
MongoDB Compass でも確認
おわりに
前回との変更点は少ないですが、MongoDB へ登録する情報を増やすことができました。
コラボ情報も欲しいところですが、それはまた次回にします。
引き続き、登録した情報を利用する Web API を開発します。