キリ番追加

get_kiriban_list.py

@@ -0,0 +1,66 @@
+# 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
+
+from eloquent import DatabaseManager, Model
+
+from db.config import DB
+from db.models import VideoHistory
+
+DB
+
+
+def main (
+        views_counts:   list[int],
+        base_date:      date,
+) -> None:
+    if not base_date:
+        base_date = date.now ()
+
+    kiriban_list: list[list[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]['contentId'] for kiriban in kiriban_list]:
+                continue
+
+            previous_views_count: int | None = (
+                    VideoHistory
+                    .where_has ('viedo', 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 (sys.argv[2:], datetime.strptime (sys.argv[1], '%Y-%m-%d').date ())

get_videos.py

@@ -14,20 +14,14 @@ from typing import TypedDict
 
 from eloquent import DatabaseManager, Model
 
+from db.config import DB
 from db.models import Video
 
+DB
+
 
 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)
-
     videos: list[VideoDict] = []
     for row in Video.all ():
         deleted_at = row.deleted_at.date () if row.deleted_at else None