#17

#15 タグ関係なく追跡する動画リスト

Co-authored-by: miteruzo <miteruzo@naver.com>
Reviewed-on: #16

db/config.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import os
+from typing import TypedDict
+
+from eloquent import DatabaseManager, Model  # type: ignore
+
+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 DbConfig (TypedDict):
+    driver:     str
+    host:       str
+    database:   str
+    user:       str
+    password:   str
+    prefix:     str

db/models.py

@@ -58,6 +58,13 @@ class Tag (Model):
         return self.has_many (VideoTag)
 
 
+class TrackedVideo (Model):
+    id:     int
+    code:   str
+
+    __timestamps__ = False
+
+
 class User (Model):
     id:     int
     code:   str
@@ -74,6 +81,7 @@ class User (Model):
 class Video (Model):
     id:             int
     code:           str
+    user_id:        int | None
     title:          str
     description:    str
     uploaded_at:    datetime
@@ -81,6 +89,14 @@ class Video (Model):
 
     __timestamps__ = False
 
+    @property
+    def user (
+            self,
+    ) -> User | None:
+        if self.user_id is None:
+            return None
+        return self.belongs_to (User)
+
     @property
     def video_histories (
             self,

get_comments_by_video_code.py

@@ -0,0 +1,58 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+動画コードからコメントのリストを取得し，JSON 形式で出力する．
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import date, datetime
+from typing import TypedDict, cast
+
+from eloquent import DatabaseManager, Model
+
+from db.config import DB
+from db.models import Video
+
+DB
+
+
+def main (
+        video_code: str,
+) -> None:
+    video = Video.where ('code', video_code).first ()
+    if video:
+        comments: list[CommentDict] = []
+        for row in video.comments:
+            comment: CommentDict = {
+                    'id':           row.id,
+                    'video_id':     row.video_id,
+                    'comment_no':   row.comment_no,
+                    'user_id':      row.user_id,
+                    'content':      row.content,
+                    'posted_at':    row.posted_at,
+                    'nico_count':   row.nico_count,
+                    'vpos_ms':      row.vpos_ms }
+            comments.append (comment)
+        print (json.dumps (comments, default = str))
+    else:
+        print ('[]')
+
+
+class CommentDict (TypedDict):
+    id:         int
+    video_id:   int
+    comment_no: int
+    user_id:    int
+    content:    str
+    posted_at:  datetime
+    nico_count: int
+    vpos_ms:    int
+
+
+if __name__ == '__main__':
+    main (sys.argv[1])

get_kiriban_list.py

@@ -0,0 +1,67 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+動画履歴の情報を取得し，JSON 形式で出力する．
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import date, datetime
+from typing import TypedDict, cast
+
+from eloquent import DatabaseManager, Model
+
+from db.config import DB
+from db.models import Video, VideoHistory
+
+DB
+
+
+def main (
+        views_counts:   list[int],
+        base_date:      date,
+) -> None:
+    if not base_date:
+        base_date = datetime.now ().date ()
+
+    kiriban_list: list[tuple[int, str, str]] = []
+
+    latest_fetched_at = cast (date, (VideoHistory
+                                     .where ('fetched_at', '<=', base_date)
+                                     .max ('fetched_at')))
+
+    for views_count in views_counts:
+        targets = { vh.video.code for vh in (
+                VideoHistory
+                .where ('fetched_at', latest_fetched_at)
+                .where ('views_count', '>=', views_count)
+                .get ()) }
+
+        for code in targets:
+            if code in [kiriban[1] for kiriban in kiriban_list]:
+                continue
+
+            previous_views_count: int | None = (
+                    VideoHistory
+                    .where_has ('video', lambda q, code = code: q.where ('code', code))
+                    .where ('fetched_at', '<', latest_fetched_at)
+                    .max ('views_count'))
+            if previous_views_count is None:
+                previous_views_count = 0
+            if previous_views_count >= views_count:
+                continue
+
+            kiriban_list.append ((views_count, code,
+                                  (cast (Video, Video.where ('code', code).first ())
+                                   .uploaded_at)))
+
+    print (json.dumps (kiriban_list, default = str))
+
+
+if __name__ == '__main__':
+    main (map (int, sys.argv[2:]),
+          datetime.strptime (sys.argv[1], '%Y-%m-%d').date ())

get_videos.py

@@ -0,0 +1,65 @@
+# pylint: disable = missing-class-docstring
+# pylint: disable = missing-function-docstring
+
+"""
+全動画の情報を取得し，JSON 形式で出力する．
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import date, datetime
+from typing import TypedDict
+
+from eloquent import DatabaseManager, Model
+
+from db.config import DB
+from db.models import Video
+
+DB
+
+
+def main (
+) -> None:
+    videos: list[VideoDict] = []
+    for row in Video.all ():
+        deleted_at = row.deleted_at.date () if row.deleted_at else None
+        video: VideoDict = { 'id':          row.id,
+                             'code':        row.code,
+                             'user':        getattr (row.user, 'code', None),
+                             'title':       row.title,
+                             'description': row.description,
+                             'tags':        [],
+                             'uploaded_at': row.uploaded_at,
+                             'deleted_at':  deleted_at }
+        for video_tag in row.video_tags:
+            if video_tag.untagged_at is None:
+                video['tags'].append (video_tag.tag.name)
+        videos.append(video)
+
+    print (json.dumps (videos, default = str))
+
+
+class DbConfig (TypedDict):
+    driver:     str
+    host:       str
+    database:   str
+    user:       str
+    password:   str
+    prefix:     str
+
+
+class VideoDict (TypedDict):
+    id:             int
+    code:           str
+    user:           str | None
+    title:          str
+    description:    str
+    tags:           list[str]
+    uploaded_at:    datetime
+    deleted_at:     date | None
+
+
+if __name__ == '__main__':
+    main ()

migrations/2026_01_01_041600_create_tracked_videos.sql

@@ -0,0 +1,2 @@
+CREATE TABLE `nizika_nico`.`tracked_videos` (`id` BIGINT NOT NULL AUTO_INCREMENT , `code` VARCHAR(16) NOT NULL COMMENT '動画コード' , PRIMARY KEY (`id`)) ENGINE = InnoDB COMMENT = '追跡対象動画';
+ALTER TABLE `tracked_videos` ADD UNIQUE(`code`);

migrations/2026_03_05_122300_add_user_id_to_videos.sql

@@ -0,0 +1,3 @@
+ALTER TABLE `videos` ADD `user_id` BIGINT NULL DEFAULT NULL COMMENT 'ユーザ Id.' AFTER `code`;
+ALTER TABLE `videos` ADD INDEX(`user_id`);
+ALTER TABLE `videos` ADD FOREIGN KEY (`user_id`) REFERENCES `users`(`id`) ON DELETE RESTRICT ON UPDATE CASCADE;

update_db.py

@@ -20,24 +20,31 @@ import jaconv
 import requests
 from eloquent import DatabaseManager, Model
 
-from db.models import Comment, Tag, User, Video, VideoHistory, VideoTag
+from db.config import DB
+from db.models import (Comment,
+                       Tag,
+                       TrackedVideo,
+                       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 (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ'])
-    update_tables (api_data, now)
+    api_data = search_nico_by_tags (['伊地知ニジカ',
+                                     'ぼざろクリーチャーシリーズ',
+                                     'ぼざろクリーチャーシリーズ外伝'])
+
+    DB.begin_transaction ()
+    try:
+        update_tables (api_data, now)
+        DB.commit ()
+    except Exception:
+        DB.rollback ()
+        raise
 
 
 def update_tables (
@@ -48,8 +55,16 @@ def update_tables (
 
     for datum in api_data:
         tag_names: list[str] = datum['tags'].split ()
+        user: User | None = None
+        if datum['userId']:
+            user = User.where('code', str (datum['userId'])).first ()
+            if user is None:
+                user = User ()
+                user.code = str (datum['userId'])
+                user.save ()
         video = Video ()
         video.code = datum['contentId']
+        video.user_id = user.id if user else None
         video.title = datum['title']
         video.description = datum['description'] or ''
         video.uploaded_at = datetime.fromisoformat (datum['startTime'])
@@ -62,7 +77,7 @@ def update_tables (
         video_history.views_count = datum['viewCounter']
         video_history.upsert ()
         video_tags = [video_tag for video_tag in video.video_tags
-                                if video_tag.untagged_at is not None]
+                                if video_tag.untagged_at is None]
         tag: Tag | None
         video_tag: VideoTag | None
         for video_tag in video_tags:
@@ -114,9 +129,9 @@ def update_tables (
             video.save ()
 
 
-def fetch_comments (
+def fetch_video_data (
         video_code: str,
-) -> list[CommentResult]:
+) -> dict[str, Any]:
     time.sleep (1.2)
 
     headers = { 'X-Frontend-Id':      '6',
@@ -131,10 +146,14 @@ def fetch_comments (
     url = (f"https://www.nicovideo.jp/api/watch/v3_guest/{ video_code }"
             + f"?actionTrackId={ action_track_id }")
 
-    res = requests.post (url, headers = headers, timeout = 60).json ()
+    return requests.post (url, headers = headers, timeout = 60).json ()
 
+
+def fetch_comments (
+        video_code: str,
+) -> list[CommentResult]:
     try:
-        nv_comment = res['data']['comment']['nvComment']
+        nv_comment = fetch_video_data (video_code)['data']['comment']['nvComment']
     except KeyError:
         return []
     if nv_comment is None:
@@ -161,12 +180,6 @@ def fetch_comments (
         return []
 
 
-def search_nico_by_tag (
-        tag:    str,
-) -> list[VideoResult]:
-    return search_nico_by_tags ([tag])
-
-
 def search_nico_by_tags (
         tags:   list[str],
 ) -> list[VideoResult]:
@@ -194,6 +207,7 @@ def search_nico_by_tags (
                                      'targets':    'tagsExact',
                                      '_sort':      '-viewCounter',
                                      'fields':     ('contentId,'
+                                                    'userId,'
                                                     'title,'
                                                     'tags,'
                                                     'description,'
@@ -208,18 +222,26 @@ def search_nico_by_tags (
             pass
         to = until + timedelta (days = 1)
 
+    for video in TrackedVideo.get ():
+        if video.code in map (lambda v: v['contentId'], result_data):
+            continue
+        try:
+            video_data = fetch_video_data (video.code)['data']
+            result_data.append ({
+                    'contentId':    video.code,
+                    'userId':       video_data['video']['userId'],
+                    'title':        video_data['video']['title'],
+                    'tags':         ' '.join (map (lambda t: t['name'],
+                                                   video_data['tag']['items'])),
+                    'description':  video_data['video']['description'],
+                    'viewCounter':  video_data['video']['count']['view'],
+                    'startTime':    video_data['video']['registeredAt'] })
+        except Exception:
+            pass
+
     return result_data
 
 
-class DbConfig (TypedDict):
-    driver:     str
-    host:       str
-    database:   str
-    user:       str
-    password:   str
-    prefix:     str
-
-
 class VideoSearchParam (TypedDict):
     q:          str
     targets:    str
@@ -231,6 +253,7 @@ class VideoSearchParam (TypedDict):
 
 class VideoResult (TypedDict):
     contentId:      str
+    userId:         int | None
     title:          str
     tags:           str
     description:    str | None