nizika_nico/update_db.py

# pylint: disable = missing-class-docstring
# pylint: disable = missing-function-docstring

"""
日次で実行し，ぼざクリ DB を最新に更新する．
"""

from __future__ import annotations

import json
import logging
import random
import string
import time
import unicodedata
from datetime import date, datetime, timedelta
from typing import Any, TypedDict, cast

import jaconv
import requests

from db.config import DB
from db.models import (Comment,
                       Tag,
                       TrackedVideo,
                       User,
                       Video,
                       VideoHistory,
                       VideoTag)

logger = logging.getLogger (__name__)
logging.basicConfig (
        level = logging.INFO,
        format = '%(asctime)s %(levelname)s %(message)s')


def main (
) -> None:
    now = datetime.now ()
    today = now.date ()

    search_result = search_nico_by_tags (['伊地知ニジカ',
                                          'ぼざろクリーチャーシリーズ',
                                          'ぼざろクリーチャーシリーズ外伝'])
    comments_by_video_code = fetch_comments_by_video_code (search_result['videos'])

    context: UpdateContext = { 'api_data':               search_result['videos'],
                               'comments_by_video_code': comments_by_video_code,
                               'deletable':              search_result['is_complete'] }

    connection = DB.connection ()
    connection.begin_transaction ()
    try:
        update_tables (context, now, today)
        connection.commit ()
    except Exception:
        connection.rollback ()
        raise


def update_tables (
        context:  UpdateContext,
        now:      datetime,
        today:    date,
) -> None:
    alive_video_codes: list[str] = []

    for datum in context['api_data']:
        tag_names = datum['tags'].split ()
        normalised_tag_names = {normalise (tag_name) for tag_name in tag_names}

        user: User | None = None
        if datum['userId'] is not None:
            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'])
        video.deleted_at = None
        video.upsert ()
        alive_video_codes.append (video.code)

        video_history = VideoHistory ()
        video_history.video_id = video.id
        video_history.fetched_at = today
        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 None]
        for video_tag in video_tags:
            tag = video_tag.tag
            if tag is None:
                continue
            if normalise (tag.name) in normalised_tag_names:
                continue
            video_tag.untagged_at = today
            video_tag.save ()

        for tag_name in tag_names:
            tag = Tag.where ('name', tag_name).first ()
            if tag is None:
                tag = Tag ()
                tag.name = tag_name
                tag.save ()

            video_tag = (VideoTag.where ('video_id', video.id)
                                 .where ('tag_id', tag.id)
                                 .first ())
            if video_tag is None:
                video_tag = VideoTag ()
                video_tag.video_id = video.id
                video_tag.tag_id = tag.id

            video_tag.tagged_at = getattr (video_tag, 'tagged_at', None) or today
            video_tag.untagged_at = None
            video_tag.upsert ()

        for com in context['comments_by_video_code'].get (video.code, []):
            user = User.where ('code', com['userId']).first ()
            if user is None:
                user = User ()
                user.code = com['userId']
                user.save ()

            comment = Comment ()
            comment.video_id = video.id
            comment.comment_no = com['no']
            comment.user_id = user.id
            comment.content = com['body']
            comment.posted_at = datetime.fromisoformat (com['postedAt'])
            comment.nico_count = com.get ('nicoruCount', 0)
            comment.vpos_ms = com.get ('vposMs', 0)
            comment.upsert ()

    if not context['deletable']:
        logger.warning ('skip soft-delete because the latest fetch was incomplete')
        return

    if not alive_video_codes:
        logger.warning ('skip soft-delete because no alive videos were fetched')
        return

    videos = (Video.where_not_in ('code', alive_video_codes)
                   .where_null ('deleted_at')
                   .get ())
    for video in videos:
        video.deleted_at = now
        video.save ()


def fetch_video_data (
        video_code: str,
) -> dict[str, Any]:
    time.sleep (1.2)

    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_code }'
           + f'?actionTrackId={ action_track_id }')

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


def fetch_comments_by_video_code (
        videos: list[VideoResult],
) -> dict[str, list[CommentResult]]:
    comments_by_video_code: dict[str, list[CommentResult]] = {}

    for video in videos:
        video_code = video['contentId']
        try:
            comments_by_video_code[video_code] = fetch_comments (video_code)
        except (KeyError,
                TypeError,
                ValueError,
                requests.RequestException) as exc:
            logger.warning ('failed to fetch comments: %s (%s)', video_code, exc)
            comments_by_video_code[video_code] = []

    return comments_by_video_code


def fetch_comments (
        video_code: str,
) -> list[CommentResult]:
    video_data = fetch_video_data (video_code)
    nv_comment = (video_data.get ('data', {})
                           .get ('comment', {})
                           .get ('nvComment'))
    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'

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

    return select_comments_from_threads (res)


def select_comments_from_threads (
        response: dict[str, Any],
) -> list[CommentResult]:
    threads = response.get ('data', {}).get ('threads', [])
    if not isinstance (threads, list):
        return []

    main_comments: list[CommentResult] = []
    fallback_comments: list[CommentResult] = []

    for thread in threads:
        comments = thread.get ('comments') if isinstance (thread, dict) else None
        if not isinstance (comments, list):
            continue

        casted_comments = cast (list[CommentResult], comments)
        if len (casted_comments) > len (fallback_comments):
            fallback_comments = casted_comments

        fork = str (thread.get ('fork', '')).lower ()
        label = str (thread.get ('label', '')).lower ()
        thread_id = str (thread.get ('id', '')).lower ()
        if fork == 'main' or 'main' in label or 'main' in thread_id:
            main_comments = casted_comments

    selected_comments = main_comments or fallback_comments
    deduped_comments: dict[int, CommentResult] = {}
    for comment in selected_comments:
        comment_no = comment.get ('no')
        if not isinstance (comment_no, int):
            continue
        deduped_comments[comment_no] = comment

    return [deduped_comments[comment_no]
            for comment_no in sorted (deduped_comments)]


def search_nico_by_tags (
        tags:   list[str],
) -> SearchNicoResult:
    today = datetime.now ()

    url = ('https://snapshot.search.nicovideo.jp'
           + '/api/v2/snapshot/video/contents/search')

    result_by_video_code: dict[str, VideoResult] = {}
    is_complete = True
    to = datetime (2022, 12, 3)
    while to <= today:
        time.sleep (1.2)
        until = to + timedelta (days = 14)
        # pylint: disable = consider-using-f-string
        query_filter = json.dumps ({ 'type':    'or',
                                     'filters': [
                { 'type':          'range',
                  'field':         'startTime',
                  'from':          ('%04d-%02d-%02dT00:00:00+09:00'
                                    % (to.year, to.month, to.day)),
                  'to':            ('%04d-%02d-%02dT23:59:59+09:00'
                                    % (until.year, until.month, until.day)),
                  'include_lower': True }] })
        params: VideoSearchParam = { 'q':          ' OR '.join (tags),
                                     'targets':    'tagsExact',
                                     '_sort':      '-viewCounter',
                                     'fields':     ('contentId,'
                                                    'userId,'
                                                    'title,'
                                                    'tags,'
                                                    'description,'
                                                    'viewCounter,'
                                                    'startTime'),
                                     '_limit':     100,
                                     'jsonFilter': query_filter }
        try:
            response = requests.get (
                    url,
                    params = cast (dict[str, int | str], params),
                    timeout = 60)
            response.raise_for_status ()
            res = response.json ()
            for datum in cast (list[VideoResult], res.get ('data', [])):
                result_by_video_code[datum['contentId']] = datum
        except (ValueError, requests.RequestException) as exc:
            logger.warning ('snapshot fetch failed: %s - %s (%s)',
                            to.date (),
                            until.date (),
                            exc)
            is_complete = False
        to = until + timedelta (days = 1)

    for video in TrackedVideo.get ():
        if video.code in result_by_video_code:
            continue
        try:
            tracked_video = video
            video_data = fetch_video_data (tracked_video.code)['data']
            owner = video_data.get ('owner') or {}
            video_info = video_data['video']
            result_by_video_code[tracked_video.code] = {
                    'contentId':    tracked_video.code,
                    'userId':       owner.get ('id'),
                    'title':        video_info['title'],
                    'tags':         ' '.join (map (lambda t: t['name'],
                                                   video_data['tag']['items'])),
                    'description':  video_info['description'],
                    'viewCounter':  video_info['count']['view'],
                    'startTime':    video_info['registeredAt'] }
        except (KeyError,
                TypeError,
                ValueError,
                requests.RequestException) as exc:
            logger.warning ('tracked video fetch failed: %s (%s)', video.code, exc)
            is_complete = False

    return { 'videos':      list (result_by_video_code.values ()),
             'is_complete': is_complete }


def normalise (
        text: str,
) -> str:
    return jaconv.hira2kata (
            unicodedata.normalize ('NFKC', text.strip ())).lower ()


class SearchNicoResult (TypedDict):
    videos:      list['VideoResult']
    is_complete: bool


class UpdateContext (TypedDict):
    api_data:                list['VideoResult']
    comments_by_video_code:  dict[str, list['CommentResult']]
    deletable:               bool


class VideoSearchParam (TypedDict):
    q:          str
    targets:    str
    _sort:      str
    fields:     str
    _limit:     int
    jsonFilter: str


class VideoResult (TypedDict):
    contentId:    str
    userId:       int | None
    title:        str
    tags:         str
    description:  str | None
    viewCounter:  int
    startTime:    str


class CommentResult (TypedDict):
    no:           int
    userId:       str
    body:         str
    postedAt:     str
    nicoruCount:  int
    vposMs:       int


if __name__ == '__main__':
    main ()