diff --git a/eloquent.pyi b/eloquent.pyi index 9925d9a..9d2ce86 100644 --- a/eloquent.pyi +++ b/eloquent.pyi @@ -51,6 +51,10 @@ class DatabaseManager: class Model: + id: int + + _Model__exists: bool + def has_one ( self, related_model: Type[_ModelT], @@ -82,6 +86,11 @@ class Model: @classmethod def find (cls, id_: int) -> Self | None: ... + @classmethod + def query ( + cls, + ) -> QueryBuilder[Self]: ... + @overload @classmethod def where ( diff --git a/models.py b/models.py new file mode 100644 index 0000000..ced158e --- /dev/null +++ b/models.py @@ -0,0 +1,169 @@ +# pylint: disable = missing-class-docstring +# pylint: disable = missing-function-docstring + +""" +ぼざクリ DB の構成 +""" + +from __future__ import annotations + +from datetime import date, datetime + +from eloquent import Model + + +class Comment (Model): + # pylint: disable = too-many-instance-attributes + + id: int + video_id: int + comment_no: int + user_id: int + content: str + posted_at: datetime + nico_count: int + vpos_ms: int + + __timestamps__ = False + + @property + def video ( + self, + ) -> Video: + return self.belongs_to (Video) + + @property + def user ( + self, + ) -> User: + return self.belongs_to (User) + + def upsert ( + self, + ) -> None: + upsert (self, 'video_id', 'comment_no') + + +class Tag (Model): + id: int + name: str + + __timestamps__ = False + + @property + def video_tags ( + self, + ) -> VideoTag: + return self.has_many (VideoTag) + + +class User (Model): + id: int + code: str + + __timestamps__ = False + + @property + def comments ( + self, + ) -> Comment: + return self.has_many (Comment) + + +class Video (Model): + id: int + code: str + title: str + description: str + uploaded_at: datetime + deleted_at: datetime | None + + __timestamps__ = False + + @property + def video_histories ( + self, + ) -> VideoHistory: + return self.has_many (VideoHistory) + + @property + def video_tags ( + self, + ) -> VideoTag: + return self.has_many (VideoTag) + + @property + def comments ( + self, + ) -> Comment: + return self.has_many (Comment) + + def upsert ( + self, + ) -> None: + upsert (self, 'code') + + +class VideoHistory (Model): + id: int + video_id: int + fetched_at: date + views_count: int + + __timestamps__ = False + + @property + def video ( + self, + ) -> Video: + return self.belongs_to (Video) + + def upsert ( + self, + ) -> None: + upsert (self, 'video_id', 'fetched_at') + + +class VideoTag (Model): + id: int + video_id: int + tag_id: int + tagged_at: date + untagged_at: date | None + + __timestamps__ = False + + @property + def video ( + self, + ) -> Video: + return self.belongs_to (Video) + + @property + def tag ( + self, + ) -> Tag: + return self.belongs_to (Tag) + + def upsert ( + self, + ) -> None: + upsert (self, 'video_id', 'tag_id') + + +def upsert ( + model: Model, + *args: str, +) -> None: + q = model.query () + + for arg in args: + q = q.where (arg, getattr (model, arg)) + + row = q.first () + + if row is not None: + model.id = row.id + model._Model__exists = True # pylint: disable = protected-access + + model.save () diff --git a/update_db.py b/update_db.py index af258ad..5ff33ca 100644 --- a/update_db.py +++ b/update_db.py @@ -13,24 +13,26 @@ import random import string import time import unicodedata -from datetime import date, datetime, timedelta +from datetime import datetime, timedelta from typing import Any, TypedDict, cast import requests from eloquent import DatabaseManager, Model -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) +from models import Comment, Tag, User, Video, VideoHistory, VideoTag def main ( ) -> None: + 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) + now = datetime.now () api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ']) @@ -207,167 +209,6 @@ def search_nico_by_tags ( return result_data -class Comment (Model): - # pylint: disable = too-many-instance-attributes - - id: int - video_id: int - comment_no: int - user_id: int - content: str - posted_at: datetime - nico_count: int - vpos_ms: int - - __timestamps__ = False - - @property - def video ( - self, - ) -> Video: - return self.belongs_to (Video) - - @property - def user ( - self, - ) -> User: - return self.belongs_to (User) - - def upsert ( - self, - ) -> None: - row = (Comment.where ('video_id', self.video_id) - .where ('comment_no', self.comment_no) - .first ()) - if row is not None: - self.id = row.id - self.__exists = True # pylint: disable = unused-private-member - self.save () - - -class Tag (Model): - id: int - name: str - - __timestamps__ = False - - @property - def video_tags ( - self, - ) -> VideoTag: - return self.has_many (VideoTag) - - -class User (Model): - id: int - code: str - - __timestamps__ = False - - @property - def comments ( - self, - ) -> Comment: - return self.has_many (Comment) - - -class Video (Model): - id: int - code: str - title: str - description: str - uploaded_at: datetime - deleted_at: datetime | None - - __timestamps__ = False - - @property - def video_histories ( - self, - ) -> VideoHistory: - return self.has_many (VideoHistory) - - @property - def video_tags ( - self, - ) -> VideoTag: - return self.has_many (VideoTag) - - @property - def comments ( - self, - ) -> Comment: - 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.__exists = True # pylint: disable = unused-private-member - self.save () - - -class VideoHistory (Model): - id: int - video_id: int - fetched_at: date - views_count: int - - __timestamps__ = False - - @property - def video ( - self, - ) -> Video: - return self.belongs_to (Video) - - def upsert ( - self, - ) -> None: - row = (VideoHistory.where ('video_id', self.video_id) - .where ('fetched_at', self.fetched_at) - .first ()) - if row is not None: - self.id = row.id - self.__exists = True # pylint: disable = unused-private-member - self.save () - - -class VideoTag (Model): - id: int - video_id: int - tag_id: int - tagged_at: date - untagged_at: date | None - - __timestamps__ = False - - @property - def video ( - self, - ) -> Video: - return self.belongs_to (Video) - - @property - def tag ( - self, - ) -> Tag: - return self.belongs_to (Tag) - - def upsert ( - self, - ) -> None: - row = (VideoTag.where ('video_id', self.video_id) - .where ('tag_id', self.tag_id) - .first ()) - if row is not None: - self.id = row.id - self.__exists = True # pylint: disable = unused-private-member - self.save () - - class DbConfig (TypedDict): driver: str host: str