nizika_nico/get_kiriban_list.py

# 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]['contentId'] for kiriban in kiriban_list]:
                continue

            previous_views_count: int | None = (
                    VideoHistory
                    .where_has ('videos', 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 ())