Merge pull request 'SM 最新化' (#4) from chore/submodule-bump into main

Reviewed-on: #4

.gitignore

@@ -0,0 +1 @@
+__pycache__

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "nizika_ai"]
 	path = nizika_ai
 	url = https://git.miteruzo.com/miteruzo/nizika_ai
-[submodule "nizika_nico"]
+[submodule "nicolib"]
-	path = nizika_nico
+	path = nicolib
-	url = https://git.miteruzo.com/miteruzo/nizika_nico.git
+	url = https://git.miteruzo.com/miteruzo/nicolib.git

db

@@ -1 +0,0 @@
-./nizika_nico/db

main.py

@@ -1,15 +1,24 @@
+"""
+AI ニジカ常時稼動バッチ
+"""
+
 from __future__ import annotations
 
 import asyncio
+import json
+import os
+import random
+import subprocess
+from asyncio import Lock
 from datetime import date, datetime, time, timedelta
-from typing import TypedDict, cast
+from typing import Any, Callable, TypedDict, cast
-
-import requests
-from bs4 import BeautifulSoup
-from requests.exceptions import Timeout
 
+import nicolib
 import queries_to_answers as q2a
-from db.models import Video, VideoHistory
+from nicolib import VideoInfo
+from nizika_ai.config import DB
+from nizika_ai.consts import Character, GPTModel, QueryType
+from nizika_ai.models import Query
 
 KIRIBAN_VIEWS_COUNTS: list[int] = sorted ({ *range (1_000, 10_000, 1_000),
                                             *range (10_000, 1_000_001, 10_000),
@@ -18,191 +27,238 @@ KIRIBAN_VIEWS_COUNTS: list[int] = sorted ({ *range (1_000, 10_000, 1_000),
                                             4_545, 194_245, 245_194, 510_245 },
                                           reverse = True)
 
-kiriban_list: list[tuple[int, VideoInfo, datetime]]
+kiriban_list: list[tuple[int, VideoInfo, datetime]] = []
+
+watched_videos: set[str] = set ()
+
+lock = Lock ()
 
 
 async def main (
 ) -> None:
+    """
+    メーン処理
+    """
+
     await asyncio.gather (
             queries_to_answers (),
             report_kiriban (),
             report_nico (),
-            update_kiriban_list ())
+            update_kiriban_list (),
+            report_snack_time (),
+            report_hot_spring_time (),
+            reconnect_db ())
 
 
 async def queries_to_answers (
 ) -> None:
+    """
+    クエリ処理
+    """
+
     while True:
-        q2a.main ()
+        loop = asyncio.get_running_loop ()
+        await loop.run_in_executor (None, q2a.main)
         await asyncio.sleep (10)
 
 
 async def report_kiriban (
 ) -> None:
+    """
+    キリ番祝ひ
+    """
+
     while True:
+        if not kiriban_list:
+            await wait_until (time (15, 0))
+            continue
+
         # キリ番祝ひ
-        (views_count, video_info, uploaded_at) = (
+        async with lock:
-                kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
+            (views_count, video_info, uploaded_at) = (
-        since_posted = datetime.now () - uploaded_at
+                    kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
-        uri = f"https://www.nicovideo.jp/watch/{ video_info['contentId'] }"
+
-        (title, description, thumbnail) = fetch_embed_info (uri)
+        video_code = video_info['contentId']
-        try:
+        comments = fetch_comments (video_code)
-            upload = client.com.atproto.repo.upload_blob (
-                    io.BytesIO (requests.get (thumbnail,
-                                              timeout = 60).content))
-            thumb = upload.blob
-        except Timeout:
-            thumb = None
-        comments = nico.get_comments (video_info['contentId'])
         popular_comments = sorted (comments,
-                                   key      = lambda c: c.nico_count,
+                                   key      = lambda c: c['nico_count'],
                                    reverse  = True)[:10]
         latest_comments = sorted (comments,
-                                  key       = lambda c: c.posted_at,
+                                  key       = lambda c: c['posted_at'],
                                   reverse   = True)[:10]
-        embed_external = models.AppBskyEmbedExternal.Main (
+        prompt = (f"{ _format_elapsed (uploaded_at) }前にニコニコに投稿された"
-                external = models.AppBskyEmbedExternal.External (
+                  f"『{ video_info['title'] }』という動画が{ views_count }再生を突破しました。\n"
-                    title       = title,
+                  f"コメント数は{ len (comments) }件です。\n")
-                    description = description,
-                    thumb       = thumb,
-                    uri         = uri))
-        prompt = f"{ since_posted.days }日と{ since_posted.seconds }秒前にニコニコに投稿された『{ video_info['title'] }』という動画が{ views_count }再生を突破しました。\n"
-        prompt += f"コメント数は{ len (comments) }件です。\n"
         if video_info['tags']:
             prompt += f"つけられたタグは「{ '」、「'.join (video_info['tags']) }」です。\n"
         if comments:
-            prompt += f"人気のコメントは次の通りです：「{ '」、「'.join (c.content for c in popular_comments) }」\n"
+            prompt += f"人気のコメントは次の通りです：「{ '」、「'.join (c['content'] for c in popular_comments) }」\n"
-            prompt += f"最新のコメントは次の通りです：「{ '」、「'.join (c.content for c in latest_comments) }」\n"
+            if latest_comments != popular_comments:
+                prompt += f"最新のコメントは次の通りです：「{ '」、「'.join (c['content'] for c in latest_comments) }」\n"
         prompt += f"""
 概要には次のように書かれています：
 ```html
 { video_info['description'] }
 ```
-このことについて、ニジカちゃんからのお祝いメッセージを下さい。
+このことについて、何かお祝いメッセージを下さい。
-ただし、そのメッセージ内には再生数の数値とその多さに応じたリアクションを添えてください。
+ただし、そのメッセージ内には再生数の数値を添えてください。
-また、ぜひ投稿からこの再生数に至るまでにかかった時間や、つけられたタグ、コメントに対して思いを馳せてください。
+また、つけられたタグ、コメントからどのような動画か想像し、説明してください。"""
-好きなコメントがあったら教えてね。"""
+        _add_query (prompt, QueryType.KIRIBAN, { 'video_code': video_code })
+
         # 待ち時間計算
         dt = datetime.now ()
         d = dt.date ()
         if dt.hour >= 15:
             d += timedelta (days = 1)
-        td = datetime.combine (d, time (15, 0)) - dt
+        remain = max (len (kiriban_list), 1)
-        if kiriban_list:
+        td = (datetime.combine (d, time (15, 0)) - dt) / remain
-            td /= len (kiriban_list)
+        # まれに時刻跨ぎでマイナスになるため
+        if td.total_seconds () < 0:
+            td = timedelta (seconds = 0)
+
         await asyncio.sleep (td.total_seconds ())
 
 
 async def update_kiriban_list (
 ) -> None:
+    """
+    キリ番リストの更新
+    """
+
     while True:
         await wait_until (time (15, 0))
-        kiriban_list += fetch_kiriban_list (datetime.now ().date ())
+
+        new_list = fetch_kiriban_list (datetime.now ().date ())
+        if not new_list:
+            continue
+
+        async with lock:
+            have = { k[1]['contentId'] for k in kiriban_list }
+            for item in new_list:
+                if item[1]['contentId'] not in have:
+                    kiriban_list.append (item)
+                    have.add (item[1]['contentId'])
 
 
 def fetch_kiriban_list (
         base_date:  date,
 ) -> list[tuple[int, VideoInfo, datetime]]:
-    _kiriban_list: list[tuple[int, VideoInfo, datetime]] = []
-
-    latest_fetched_at = cast (date, (VideoHistory
-                                     .where ('fetched_at', '<=', base_date)
-                                     .max ('fetched_at')))
-
-    for kiriban_views_count in KIRIBAN_VIEWS_COUNTS:
-        targets = { vh.video.code for vh in (VideoHistory
-                                             .where ('fetched_at', latest_fetched_at)
-                                             .where ('views_count', '>=', kiriban_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 ('video', lambda q: 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 >= kiriban_views_count:
-                continue
-            video_info = fetch_video_info (code)
-            if video_info is not None:
-                _kiriban_list.append ((kiriban_views_count, video_info,
-                                      cast (Video, Video.where ('code', code).first ()).uploaded_at))
-
-    return _kiriban_list
-
-
-def fetch_video_info (
-        video_code: str,
-) -> VideoInfo | None:
-    video_info: dict[str, str | list[str]] = { 'contentId': video_code }
-
-    bs = create_bs_from_url (f"https://www.nicovideo.jp/watch/{ video_code }")
-    if bs is None:
-        return None
-
-    try:
-        title = bs.find ('title')
-        if title is None:
-            return None
-        video_info['title'] = '-'.join (title.text.split ('-')[:(-1)])[:(-1)]
-
-        tags: str = bs.find ('meta', attrs = { 'name': 'keywords' }).get ('content') # type: ignore
-        video_info['tags'] = tags.split (',')
-
-        video_info['description'] = bs.find ('meta', attrs = { 'name': 'description' }).get ('content') # type: ignore
-    except Exception:
-        return None
-
-    return cast (VideoInfo, video_info)
-
-
-def create_bs_from_url (
-        url:    str,
-        params: dict | None = None,
-) -> BeautifulSoup | None:
     """
-    URL から BeautifulSoup インスタンス生成
+    キリ番を迎へた動画のリストを取得する．
 
     Parameters
     ----------
-    url:    str
+    base_date:  date
-        捜査する URL
+        基準日
-    params: dict
-        パラメータ
 
     Return
     ------
-    BeautifulSoup | None
+    list[tuple[int, VideoInfo, datetime]]
-        BeautifulSoup オブゼクト（失敗したら None）
+        動画リスト（キリ番基準再生数、対象動画情報、投稿日時のタプル）
     """
 
-    if params is None:
+    result = subprocess.run (
-        params = { }
+            ['python3', '/root/nizika_nico/get_kiriban_list.py',
-
+             str (base_date), *map (str, KIRIBAN_VIEWS_COUNTS)],
+            cwd             = '/root/nizika_nico',
+            env             = os.environ,
+            capture_output  = True,
+            text            = True)
+    kl: list[list[int | str]]
     try:
-        req = requests.get (url, params = params, timeout = 60)
+        kl = json.loads (result.stdout)
-    except Timeout:
+    except Exception:
-        return None
+        kl = []
 
-    if req.status_code != 200:
+    return [(cast (int, k[0]), video_info, str_to_datetime (cast (str, k[2])))
-        return None
+            for k in kl
+            if (video_info := nicolib.fetch_video_info (cast (str, k[1]))) is not None]
 
-    req.encoding = req.apparent_encoding
 
-    return BeautifulSoup (req.text, 'hecoml.parser')
+def fetch_comments (
+        video_code: str,
+) -> list[CommentDict]:
+    """
+    動画のコメント・リストを取得する．
+
+    Parameters
+    ----------
+    video_code: str
+        ニコニコの動画コード
+
+    Return
+    ------
+    list[CommentDict]
+        コメント・リスト
+    """
+
+    result = subprocess.run (
+            ['python3', 'get_comments_by_video_code.py', video_code],
+            cwd             = '/root/nizika_nico',
+            env             = os.environ,
+            capture_output  = True,
+            text            = True)
+    rows: list[dict[str, Any]] = json.loads (result.stdout)
+    comments: list[CommentDict] = []
+    for row in rows:
+        row['posted_at'] = str_to_datetime (row['posted_at'])
+        comments.append (cast (CommentDict, row))
+
+    return comments
+
+
+def fetch_latest_deerjika (
+) -> VideoInfo | None:
+    """
+    最新のぼざクリ動画を取得する．
+
+    Return
+    ------
+    VideoInfo | None
+        動画情報
+    """
+
+    return nicolib.fetch_latest_video (['伊地知ニジカ',
+                                        'ぼざろクリーチャーシリーズ',
+                                        'ぼざろクリーチャーシリーズ外伝'])
 
 
 async def report_nico (
 ) -> None:
-    ...
+    """
+    ニコニコから最新のぼざクリを取得し，まだ報知してゐなかったら報知する．
+    """
+
+    while True:
+        latest_deerjika = fetch_latest_deerjika ()
+        if latest_deerjika and latest_deerjika['contentId'] not in watched_videos:
+            video = latest_deerjika
+            watched_videos.add (video['contentId'])
+
+            prompt = f"""ニコニコに『{ video['title'] }』という動画がアップされました。
+つけられたタグは「{ '」、「'.join (video['tags']) }」です。
+概要には次のように書かれています：
+```html
+{ video['description'] }
+```
+このことについて、みんなに告知するとともに、ニジカちゃんの感想を教えてください。"""
+            _add_query (prompt, QueryType.NICO_REPORT, { 'video_code': video['contentId'] })
+
+        await asyncio.sleep (60)
 
 
 async def wait_until (
         t:  time,
-):
+) -> None:
+    """
+    指定した時刻まで待つ．
+
+    Parameters
+    ----------
+    t:  time
+        次に実行を続行するまでの時刻
+    """
+
     dt = datetime.now ()
     d = dt.date ()
     if dt.time () >= t:
@@ -210,16 +266,149 @@ async def wait_until (
     await asyncio.sleep ((datetime.combine (d, t) - dt).total_seconds ())
 
 
-class VideoInfo (TypedDict):
+async def report_snack_time (
-    contentId:      str
+) -> None:
-    title:          str
+    """
-    tags:           list[str]
+    おやつタイムを報知する．
-    description:    str
+    """
+
+    while True:
+        await wait_until (time (15, 0))
+        _add_query ('おやつタイムだ!!!!', QueryType.SNACK_TIME)
+
+
+async def report_hot_spring_time (
+) -> None:
+    """
+    温泉タイムを報知する．
+    """
+
+    while True:
+        await wait_until (time (21, 0))
+        _add_query ('温泉に入ろう!!!', QueryType.HOT_SPRING)
+
+
+async def reconnect_db (
+) -> None:
+    while True:
+        await asyncio.sleep (600)
+        try:
+            ensure_mysql_alive ()
+        except Exception as ex:
+            if getattr (ex, 'args', [None])[0] not in (2006, 2013):
+                raise
+            print (f"[reconnect_db] { type (ex).__name__ }: { ex }")
+            safe_reconnect ()
+
+
+def ensure_mysql_alive (
+) -> None:
+    conn = DB.connection ('mysql').get_connection ()
+    conn.ping ()
+
+
+def safe_reconnect (
+) -> None:
+    try:
+        DB.reconnect ('mysql')
+    except Exception as ex:
+        if getattr (ex, 'args', [None])[0] not in (2006, 2013):
+            raise
+        print (f"[safe_reconnect] { type (ex).__name__ }: { ex }")
+
+
+def run_with_mysql_retry (
+        fn:         Callable[..., Any],
+        *args,
+        **kwargs,
+) -> Any:
+    last = None
+    for _ in range (2):
+        try:
+            ensure_mysql_alive ()
+            return fn (*args, **kwargs)
+        except Exception as ex:
+            if getattr (ex, 'args', [None])[0] not in (2006, 2013):
+                raise
+            last = ex
+            print (f"[run_with_mysql_retry] { type (ex).__name__ }: { ex }")
+            safe_reconnect ()
+    if last:
+        raise last
+
+
+def _add_query (
+        content:        str,
+        query_type:     QueryType,
+        transfer_data:  dict | None = None,
+) -> None:
+    query = Query ()
+    query.user_id = None
+    query.target_character = Character.DEERJIKA.value
+    query.content = content
+    query.query_type = query_type.value
+    query.model = GPTModel.GPT4_O.value
+    query.sent_at = datetime.now ()
+    query.answered = False
+    if transfer_data is not None:
+        query.transfer_data = transfer_data
+    run_with_mysql_retry (query.save)
+
+
+def _format_elapsed (
+        uploaded_at:    datetime,
+) -> str:
+    """
+    指定した時刻から現在までの時間を見やすぃ文字列に変換する．
+
+    Parameters
+    ----------
+    uploaded_at:    datetime
+        基準日時
+
+    Return
+    ------
+    str
+        変換後文字列
+    """
+
+    delta = datetime.now () - uploaded_at
+    days = delta.days
+    seconds = delta.seconds
+    (hours, seconds) = divmod (seconds, 3600)
+    (mins, seconds) = divmod (seconds, 60)
+
+    return f"{ days }日{ hours }時間{ mins }分{ seconds }秒"
+
+
+def str_to_datetime (
+        s:  str,
+) -> datetime:
+    formats: list[str] = [
+            '%Y-%m-%d %H:%M:%S.%f',
+            '%Y-%m-%d %H:%M:%S']
+    for f in formats:
+        try:
+            return datetime.strptime (s, f)
+        except ValueError:
+            pass
+    raise ValueError ('うんち!w')
+
+
+class CommentDict (TypedDict):
+    id:         int
+    video_id:   int
+    comment_no: int
+    user_id:    int
+    content:    str
+    posted_at:  datetime
+    nico_count: int
+    vpos_ms:    int
 
 
 kiriban_list = (
-        fetch_kiriban_list ((d := datetime.now ()).date ()
+        fetch_kiriban_list ((now := datetime.now ()).date ()
-                            - timedelta (days = d.hour < 15)))
+                            - timedelta (days = 1 if now.hour < 15 else 0)))
 
 if __name__ == '__main__':
     asyncio.run (main ())

queries_to_answers.py

@@ -1,3 +1,7 @@
+"""
+DB の queries テーブルにたまってゐるクエリを AI に処理させ answers テーブルに流す．
+"""
+
 from __future__ import annotations
 
 import random
@@ -5,33 +9,45 @@ from datetime import datetime
 from typing import TypedDict
 
 from nizika_ai.config import DB
-from nizika_ai.consts import AnswerType, Character, Platform
+from nizika_ai.consts import Character, Platform, QueryType
-from nizika_ai.models import Answer, AnsweredFlag, Query, User
+from nizika_ai.models import Answer, AnsweredFlag, Query
 from nizika_ai.talk import Talk
 
 
 def main (
 ) -> None:
-    DB.begin_transaction ()
+    """
+    メーン処理
+    """
+
     queries: list[Query] = Query.where ('answered', False).get ()
     if not queries:
         return
+
     query: Query = random.choice (queries)
-    user: User = query.user
+
-    user_name: str | None = None
+    DB.begin_transaction ()
-    if query.user_id is not None:
+    try:
-        user_name = user.name
+        user_name = query.user.name if query.user_id else None
-    histories: list[History] = []
+
-    for history in query.answer_histories:
+        histories: list[History] = []
-        if history.query is not None:
+        for history in query.answer_histories:
-            histories.append ({ 'role': 'user', 'content': history.query.content })
+            if history.query is not None:
-        histories.append ({ 'role': 'assistant', 'content': history.content })
+                histories.append ({ 'role': 'user', 'content': history.query.content })
-    for character in [Character.DEERJIKA, Character.GOATOH]:
+            histories.append ({ 'role': 'assistant', 'content': history.content })
-        if query.target_character & character.value:
+
-            add_answer (query, character, user_name, histories)
+        for character in [Character.DEERJIKA, Character.GOATOH]:
-    query.answered = True
+            if query.target_character & character.value:
-    query.save ()
+                add_answer (query, character, user_name, histories)
-    DB.commit ()
+
+        query.answered = True
+
+        query.save ()
+
+        DB.commit ()
+    except Exception:
+        DB.rollback ()
+        raise
 
 
 def add_answer (
@@ -40,19 +56,35 @@ def add_answer (
         user_name:  str | None,
         histories:  list[History],
 ) -> None:
+    """
+    AI の返答を DB に積む．
+
+    Parameters
+    ----------
+    query:      Query
+        クエリ
+    character:  Character
+        返答するキャラクタ
+    user_name:  str | None
+        クエリの主
+    histories:  list[History]
+        履歴
+    """
+
     message: str | list[dict[str, str | dict[str, str]]]
-    if query.image_url is None:
+    if query.image_url:
-        message = query.content
-    else:
         message = [{ 'type': 'text', 'text': query.content },
                    { 'type': 'image_url', 'image_url': query.image_url }]
+    else:
+        message = query.content
+
     answer = Answer ()
     answer.query_id = query.id
     answer.character = character.value
     answer.content = Talk.main (message, user_name, histories,
                                 goatoh_mode = character == Character.GOATOH)
-    answer.answer_type = query.query_type
     answer.sent_at = datetime.now ()
+    answer.answer_type = query.query_type
     answer.save ()
     add_answered_flags (answer)
 
@@ -60,14 +92,35 @@ def add_answer (
 def add_answered_flags (
         answer: Answer,
 ) -> None:
-    answer_type:    AnswerType
+    """
+    返答済フラグを付与する．
+
+    Parameters
+    ----------
+    answer: Answer
+        返答モデル
+    """
+
+    answer_type: QueryType
     try:
-        answer_type = AnswerType (answer.answer_type)
+        answer_type = QueryType (answer.query_rel.query_type)
-    except Exception:
+    except (TypeError, ValueError):
         return
-    if answer_type in [AnswerType.YOUTUBE_REPLY]:
+
+    if answer_type in (QueryType.YOUTUBE_COMMENT,
+                       QueryType.YOUTUBE_SYSTEM,
+                       QueryType.KIRIBAN,
+                       QueryType.NICO_REPORT,
+                       QueryType.SNACK_TIME,
+                       QueryType.HOT_SPRING):
         add_answered_flag (answer, Platform.YOUTUBE)
-    if answer_type in [AnswerType.BLUESKY_REPLY]:
+
+    if answer_type in (QueryType.BLUESKY_COMMENT,
+                       QueryType.BLUESKY_SYSTEM,
+                       QueryType.KIRIBAN,
+                       QueryType.NICO_REPORT,
+                       QueryType.SNACK_TIME,
+                       QueryType.HOT_SPRING):
         add_answered_flag (answer, Platform.BLUESKY)
 
 
@@ -75,6 +128,17 @@ def add_answered_flag (
         answer:     Answer,
         platform:   Platform,
 ) -> None:
+    """
+    返答済フラグを付与する．
+
+    Parameters
+    ----------
+    answer:     Answer
+        返答モデル
+    platform:   Platform
+        プラットフォーム
+    """
+
     answered_flag = AnsweredFlag ()
     answered_flag.answer_id = answer.id
     answered_flag.platform = platform.value
@@ -83,6 +147,10 @@ def add_answered_flag (
 
 
 class History (TypedDict):
+    """
+    会話履歴の 1 要素；ユーザや AI の発話を簡易に保持する型
+    """
+
     role:       str
     content:    str