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

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

update_db.py

@@ -21,7 +21,13 @@ import requests
 from eloquent import DatabaseManager, Model
 
 from db.config import DB
-from db.models import Comment, Tag, User, Video, VideoHistory, VideoTag
+from db.models import (Comment,
+                       Tag,
+                       TrackedVideo,
+                       User,
+                       Video,
+                       VideoHistory,
+                       VideoTag)
 
 
 def main (
@@ -115,9 +121,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',
@@ -132,10 +138,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:
@@ -162,12 +172,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]:
@@ -209,6 +213,22 @@ 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,
+                    '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