ホロジュール収集プログラム（Python）のアップデート

## はじめに

以前に、ホロライブの配信予定や動画情報をホロジュールと Youtube から収集するプログラムを作成しましたが、ホロライブの所属メンバーが増えてグループ構成など複雑になってきたので、収集する情報を増やすために、ホロジュール収集プログラムをあらためて作成しました。

ちなみに、ホロジュールの API は公開されている（コラボ情報も含めて！）のですが、とりあえず前回と同様に Web スクレイピングを利用して配信予定の情報を収集しています。

なお、同時に開発している FastAPI 版の WebAPI のための準備でもあります。

## Selenium のバージョンアップ

Selenium 4.6 から、Selenium 自体に ChromeDriver の自動更新機能「Selenium Manager」が搭載されたため、環境構築の手順を簡略化でき、ドライバーの初期化もシンプルになりました。

```python
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 を利用したモデルの定義

```python
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とモデルの相互変換（例としてオブジェクトからドキュメントへの変換）

```python
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)
```

### アプリケーション設定の読み込み

```python
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 を利用）](https://qiita.com/kerobot/items/8b371ba51501ddc806a7)

```powershell
> poetry --version
Poetry version 1.3.2

> pyenv --version
pyenv 3.1.1
```

### MongoDB の確認

```powershell
> mongosh --version
1.6.0

> mongosh localhost:27017/admin -u admin -p
```

### データベースの作成とロール（今回は dbOwner ）の設定

```powershell
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](https://console.developers.google.com/?hl=JA)

* Google Developer Console にログイン
* ダッシュボードでプロジェクトを作成
* ライブラリで YouTube Data API v3 を有効化
* 認証情報で認証情報を作成して APIキー を取得

## プロジェクト作成

### プロジェクトを作成してディレクトリに入る

```powershell
> poetry new holocollect --name app
Created package app in holocollect
> cd holocollect
```

### プロジェクトで利用する Python を pyenv でインストール

```powershell
> pyenv install 3.11.1
```

### プロジェクトで利用するローカルの Python のバージョンを変更

```powershell
> pyenv local 3.11.1
> python -V
Python 3.11.1
```

### バージョンを指定して Python 仮想環境を作成（pyenv で管理している Python のパスを指定）

```powershell
> python -c "import sys; print(sys.executable)"
> poetry env use C:\Users\[UserName]\.pyenv\pyenv-win\versions\3.11.1\python.exe
```

### パッケージの追加

```powershell
> 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 を読み込みます。

```python
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 と ホロジュール の設定を読み込みます。

```python
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 をシングルトンで管理して利用します。

```python
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）

配信情報を保持するモデルクラスです。

```python
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）

配信情報のコレクションを保持する反復可能クラスです。

```python
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）

配信者情報を保持するモデルクラスです。

```python
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）

配信者情報のコレクションを保持するクラスです。

```python
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の動画情報とあわせて配信情報のコレクションを取得します。

```python
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）

このプログラムのエントリポイントです。

```python
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キーを設定

```text
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>"
```

### ログの設定

```json
{
    "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 の設定（デバッグ実行用）

```json
{
    "version": "1.0.0",
    "configurations": [
        {
            "name": "ホロコレクトのモジュール実行",
            "type": "python",
            "request": "launch",
            "module": "app",
            "justMyCode": true,
            "args": ["--csvpath","c:\\temp\\holodule.csv"]
        }
    ]
}
```

### プログラムの実行

```powershell
> poetry run python -m app --csvpath c:\temp\holodule.csv
```

## 実行結果

ホロジュールと Youtube から取得した情報が MongoDB へ登録されます。

```powershell
[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 でも確認

![img01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/283306/3b2d9928-1c47-2ace-04a9-291458aa3e92.png)

## おわりに

前回との変更点は少ないですが、MongoDB へ登録する情報を増やすことができました。
コラボ情報も欲しいところですが、それはまた次回にします。

引き続き、登録した情報を利用する Web API を開発します。