diff --git a/update_db.py b/update_db.py index 0a6ceae..3b4a8db 100644 --- a/update_db.py +++ b/update_db.py @@ -13,17 +13,18 @@ from dataclasses import dataclass from datetime import date, datetime, timedelta from typing import Any, Type, TypedDict, cast -import mysql.connector import requests -from mysql.connector.connection import MySQLConnectionAbstract +from eloquent import DatabaseManager, Model +from eloquent.orm.relations.dynamic_property import DynamicProperty - -class DbNull: - def __new__ ( - cls, - ): - delattr (cls, '__init__') -DbNullType = Type[DbNull] +config: dict[str, DbConfig] = { 'mysql': { 'driver': 'mysql', + 'host': 'localhost', + 'database': 'nizika_nico', + 'user': os.environ['MYSQL_USER'], + 'password': os.environ['MYSQL_PASS'], + 'prefix': '' } } +db = DatabaseManager (config) +Model.set_connection_resolver (db) class VideoSearchParam (TypedDict): @@ -105,51 +106,165 @@ class VideoTagRow (TypedDict): untagged_at: date | None -def main ( -) -> None: - conn = mysql.connector.connect (user = os.environ['MYSQL_USER'], - password = os.environ['MYSQL_PASS'], - database = 'nizika_nico') - if not isinstance (conn, MySQLConnectionAbstract): - raise TypeError - - now = datetime.now () +class DbConfig (TypedDict): + driver: str + host: str + database: str + user: str + password: str + prefix: str - video_dao = VideoDao (conn) - tag_dao = TagDao (conn) - video_tag_dao = VideoTagDao (conn) - video_history_dao = VideoHistoryDao (conn) - comment_dao = CommentDao (conn) - user_dao = UserDao (conn) +def main ( +) -> None: api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ']) - update_tables (video_dao, tag_dao, video_tag_dao, video_history_dao, comment_dao, user_dao, - api_data, now) + update_tables (api_data, now) conn.commit () conn.close () +class Comment (Model): + __timestamps__ = False + + @property + def video ( + self, + ) -> DynamicProperty: + return self.belongs_to (Video) + + @property + def user ( + self, + ) -> DynamicProperty: + return self.belongs_to (User) + + +class Tag (Model): + __timestamps__ = False + + @property + def video_tags ( + self, + ) -> DynamicProperty: + return self.has_many (VideoTag) + + @property + def videos ( + self, + ) -> DynamicProperty: + return self.belongs_to_many (Video) + + +class User (Model): + __timestamps__ = False + + @property + del comments ( + self, + ) -> DynamicProperty: + return self.has_many (Comment) + + +class Video (Model): + __timestamps__ = False + + @property + def video_histories ( + self, + ) -> DynamicProperty: + return self.has_many (VideoHistory) + + @property + def video_tags ( + self, + ) -> DynamicProperty: + return self.has_many (VideoTag) + + @property + def tags ( + self, + ) -> DynamicProperty: + return self.belongs_to_many (Tag) + + @property + def comments ( + self, + ) -> DynamicProperty: + return self.has_many (Comment) + + def upsert ( + self, + ) -> None: + row = Video.where ('code', self.code).first () + if row is not None: + self.id = row.id + self.save () + + +class VideoHistory (Model): + __timestamps__ = False + + @property + def video ( + self, + ) -> DynamicProperty: + return self.belongs_to (Video) + + def upsert ( + self, + ) -> None: + row = (Video + .where ('video_id', self.video_id) + .where ('fetched_at', self.fetched_at) + .first ()) + if row is not None: + self.id = row.id + self.save () + + +class VideoTag (Model): + __timestamps__ = False + + @property + def video ( + self, + ) -> DynamicProperty: + return self.belongs_to (Video) + + @property + def tag ( + self, + ) -> DynamicProperty: + return self.belongs_to (Tag) + + def upsert ( + self, + ) -> None: + row = (Video + .where ('video_id', self.video_id) + .where ('tag_id', self.tag_id) + .first ()) + if row is not None: + self.id = row.id + self.save () + + def update_tables ( - video_dao: VideoDao, - tag_dao: TagDao, - video_tag_dao: VideoTagDao, - video_history_dao: VideoHistoryDao, - comment_dao: CommentDao, - user_dao: UserDao, - api_data: list[VideoResult], - now: datetime, + api_data: list[VideoResult], + now: datetime, ) -> None: video_ids: list[int] = [] for datum in api_data: tag_names: list[str] = datum['tags'].split () - video = VideoDto (code = datum['contentId'], - title = datum['title'], - description = datum['description'] or '', - uploaded_at = datetime.fromisoformat (datum['startTime'])) - video_dao.upsert (video, False) - if video.id_ is not None: + video = Video () + video.code = datum['contentId'], + video.title = datum['title'], + video.description = datum['description'] or '', + video.uploaded_at = datetime.fromisoformat (datum['startTime']) + video.upsert () + if video.id is not None: video_ids.append (video.id_) video_history = VideoHistoryDto (video_id = video.id_, fetched_at = now, @@ -298,12 +413,6 @@ def search_nico_by_tags ( class VideoDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def find ( self, video_id: int, @@ -473,26 +582,7 @@ class VideoDao: return video -@dataclass (slots = True) -class VideoDto: - code: str - title: str - description: str - uploaded_at: datetime - id_: int | None = None - deleted_at: datetime | DbNullType = DbNull - video_tags: list[VideoTagDto] | None = None - comments: list[CommentDto] | None = None - video_histories: list[VideoHistoryDto] | None = None - - class VideoTagDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def fetch_by_video_id ( self, video_id: int, @@ -690,24 +780,7 @@ class VideoTagDao: return video_tag -@dataclass (slots = True) -class VideoTagDto: - video_id: int - tag_id: int - tagged_at: date - id_: int | None = None - untagged_at: date | DbNullType = DbNull - video: VideoDto | None = None - tag: TagDto | None = None - - class TagDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def find ( self, tag_id: int, @@ -783,19 +856,7 @@ class TagDao: name = row['name']) -@dataclass (slots = True) -class TagDto: - name: str - id_: int | None = None - - class VideoHistoryDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def fetch_by_video_id ( self, video_id: int, @@ -884,22 +945,7 @@ class VideoHistoryDao: return video_history -@dataclass (slots = True) -class VideoHistoryDto: - video_id: int - fetched_at: date - views_count: int - id_: int | None = None - video: VideoDto | None = None - - class CommentDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def fetch_by_video_id ( self, video_id: int, @@ -1001,27 +1047,7 @@ class CommentDao: return comment -@dataclass (slots = True) -class CommentDto: - video_id: int - comment_no: int - user_id: int - content: str - posted_at: datetime - id_: int | None = None - nico_count: int = 0 - vpos_ms: int | DbNullType = DbNull - video: VideoDto | None = None - user: UserDto | None = None - - class UserDao: - def __init__ ( - self, - conn: MySQLConnectionAbstract, - ): - self.conn = conn - def fetch_by_code ( self, user_code: str @@ -1062,11 +1088,5 @@ class UserDao: code = row['code']) -@dataclass (slots = True) -class UserDto: - code: str - id_: int | None = None - - if __name__ == '__main__': main ()