nizika_nico/update_db.py

import json
import os
import random
import string
import time
from dataclasses import dataclass
from datetime import datetime

import mysql.connector
import requests


def main (
) -> None:
    conn = mysql.connector.connect (host     = os.environ['MYSQL_HOST'],
                                    user     = os.environ['MYSQL_USER'],
                                    password = os.environ['MYSQL_PASS'])

    video_dao = VideoDao (conn)

    api_data = search_nico_by_tags (['伊地知ニジカ', 'ぼざろクリーチャーシリーズ'])

    update_video_table (video_dao, api_data)

    # TODO: 書くこと


def update_video_table (
        video_dao,
        api_data:   list[dict],
) -> None:
    # TODO: 書くこと

    video_dao.upsert_all (videos)

    video_dao.delete_nonexistent_data (video_id_list)

    # TODO: 書くこと


def fetch_comments (
        video_id:   str,
) -> list:
    headers = { 'X-Frontend-Id':      '6',
                'X-Frontend-Version': '0' }

    action_track_id = (
            ''.join (random.choice (string.ascii_letters + string.digits)
                     for _ in range (10)) 
            + '_'
            + str (random.randrange (10 ** 12, 10 ** 13)))

    url = (f"https://www.nicovideo.jp/api/watch/v3_guest/{ video_id }"
            + f"?actionTrackId={ action_track_id }")

    res = requests.post (url, headers = headers, timeout = 60).json ()

    try:
        nv_comment = res['data']['comment']['nvComment']
    except KeyError:
        return []
    if nv_comment is None:
        return []

    headers = { 'X-Frontend-Id':        '6',
                'X-Frontend-Version':   '0',
                'Content-Type':         'application/json' }

    params = { 'params':        nv_comment['params'],
               'additionals':   { },
               'threadKey':     nv_comment['threadKey'] }

    url = nv_comment['server'] + '/v1/threads'

    res = (requests.post (url, json.dumps (params),
                          headers = headers,
                          timeout = 60)
            .json ())

    try:
        return res['data']['threads'][1]['comments']
    except (IndexError, KeyError):
        return []


def search_nico_by_tag (
        tag:    str,
) -> list[dict]:
    return search_nico_by_tags ([tag])


def search_nico_by_tags (
        tags:   list[str],
) -> list[dict]:
    url = ('https://snapshot.search.nicovideo.jp'
            + '/api/v2/snapshot/video/contents/search')

    # TODO: 年月日の設定ができてゐなぃのと，100 件までしか取得できなぃので何とかすること

    query_filter = json.dumps ({ 'type':    'or',
                                 'filters': [ 
            { 'type':          'range',
              'field':         'startTime',
              'from':          f"{year}-{start}T00:00:00+09:00",
              'to':            f"{year}-{end}T23:59:59+09:00",
              'include_lower': True }] })

    params: dict[str, int | str]
    params = { 'q':          ' OR '.join (tags),
               'targets':    'tagsExact',
               '_sort':      '-viewCounter',
               'fields':     'contentId,title,tags,viewCounter,startTime',
               '_limit':     100,
               'jsonFilter': query_filter }

    res = requests.get (url, params = params, timeout = 60).json ()

    return res['data']


class VideoDao:
    def __init__ (
            self,
            conn
    ):
        self.conn = conn

    def fetch (
            self,
            video_id:               int,
            with_relation_tables:   bool    = True,
    ) -> VideoDto | None:
        with self.conn.cursor () as c:
            c.execute ("""
                    SELECT
                      id,
                      code,
                      title,
                      description,
                      uploaded_at,
                      deleted_at
                    FROM
                      videos
                    WHERE
                      id = %s
                    ORDER BY
                      id""", video_id)
            row = c.fetchone ()
            if row is None:
                return None
            return self._create_dto_from_row (row, with_relation_tables)

    def fetch_all (
            self,
            with_relation_tables: bool  = True,
    ) -> list[VideoDto]:
        with self.conn.cursor () as c:
            c.execute ("""
                    SELECT
                      id,
                      code,
                      title,
                      description,
                      uploaded_at,
                      deleted_at
                    FROM
                      videos
                    ORDER BY
                      id""")
            videos: list[VideoDto] = []
            for row in c.fetchall ():
                videos.append (self._create_dto_from_row (row, with_relation_tables))
            return videos

    def _create_dto_from_row (
            self,
            row,
            with_relation_tables:   bool,
    ) -> VideoDto:
        video = VideoDto (id_             = row['id'],
                          code            = row['code'],
                          title           = row['title'],
                          description     = row['description'],
                          uploaded_at     = row['uploaded_at'],
                          deleted_at      = row['deleted_at'],
                          video_tags      = None,
                          comments        = None,
                          video_histories = None)
        if with_relation_tables:
            video.video_tags = VideoTagDao (self.conn).fetch_by_video_id (video.id_, False)
            for i in range (len (video.video_tags)):
                video.video_tags[i].video = video
            video.comments = CommentDao (self.conn).fetch_by_video_id (video.id_, False)
            for i in range (len (video.comments)):
                video.comments[i].video = video
            video.video_histories = VideoHistoryDao (self.conn).fetch_by_video_id (video.id_, False)
            for i in range (len (video.video_histories)):
                video.video_histories[i].video = video
        return video


@dataclass (slots = True)
class VideoDto:
    id_:                int
    code:               str
    title:              str
    description:        str
    uploaded_at:        datetime
    deleted_at:         datetime | None
    video_tags:         list[VideoTagDto] | None
    comments:           list[CommentDto] | None
    video_histories:    list[VideoHistoryDto] | None


class VideoTagDao:
    def __init__ (
            self,
            conn,
    ):
        self.conn = conn

    def fetch_by_video_id (
            self,
            video_id:               int,
            with_relation_tables:   bool    = True,
    ) -> list[VideoTagDto]:
        with self.conn.cursor () as c:
            c.execute ("""
                    SELECT
                      id,
                      video_id,
                      tag_id,
                      tagged_at,
                      untagged_at
                    FROM
                      video_tags
                    WHERE
                      video_id = %s
                    ORDER BY
                      id""", video_id)
            video_tags: list[VideoTagDto] = []
            for row in c.fetchall ():
                video_tags.append (self._create_dto_from_row (row, with_relation_tables))
            return video_tags

    def _create_dto_from_row (
            self,
            row,
            with_relation_tables:   bool,
    ) -> VideoTagDto:
        video_tag = VideoTagDto (id_            = row['id'],
                                 video_id       = row['video_id'],
                                 tag_id         = row['tag_id'],
                                 tagged_at      = row['tagged_at'],
                                 untagged_at    = row['untagged_at'],
                                 video          = None,
                                 tag            = None)
        if with_relation_tables:
            video_tag.video = VideoDao (self.conn).fetch (video_tag.video_id, True)
            video_tag.tag = TagDao (self.conn).fetch (video_tag.tag_id, True)
        return video_tag


@dataclass (slots = True)
class VideoTagDto:
    id_:            int
    video_id:       int
    tag_id:         int
    tagged_at:      datetime
    untagged_at:    datetime
    video:          VideoDto
    tag:            TagDto


if __name__ == '__main__':
    main ()