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

Reviewed-on: #4

.gitignore

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

.gitmodules

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

db

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

main.py

@@ -1,15 +1,22 @@
+"""
+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 nicolib
 import queries_to_answers as q2a
-from db.models import Comment, 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
 
@@ -20,21 +27,35 @@ 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:
         loop = asyncio.get_running_loop ()
         await loop.run_in_executor (None, q2a.main)
@@ -43,6 +64,10 @@ async def queries_to_answers (
 
 async def report_kiriban (
 ) -> None:
+    """
+    キリ番祝ひ
+    """
+
     while True:
         if not kiriban_list:
             await wait_until (time (15, 0))
@@ -53,29 +78,23 @@ async def report_kiriban (
             (views_count, video_info, uploaded_at) = (
                     kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
 
-        since_posted = datetime.now () - uploaded_at
-        _days = since_posted.days
-        _seconds = since_posted.seconds
-        (_hours, _seconds) = divmod (_seconds, 3600)
-        (_mins, _seconds) = divmod (_seconds, 60)
-
         video_code = video_info['contentId']
-        uri = f"https://www.nicovideo.jp/watch/{ video_code }"
-        (title, description, _) = nicolib.fetch_embed_info (uri)
         comments = fetch_comments (video_code)
         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]
-        prompt = f"{ _days }日{ _hours }時間{ _mins }分{ _seconds }秒前にニコニコに投稿された『{ video_info['title'] }』という動画が{ views_count }再生を突破しました。\n"
+        prompt = (f"{ _format_elapsed (uploaded_at) }前にニコニコに投稿された"
-        prompt += f"コメント数は{ len (comments) }件です。\n"
+                  f"『{ video_info['title'] }』という動画が{ views_count }再生を突破しました。\n"
+                  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
@@ -84,16 +103,7 @@ async def report_kiriban (
 このことについて、何かお祝いメッセージを下さい。
 ただし、そのメッセージ内には再生数の数値を添えてください。
 また、つけられたタグ、コメントからどのような動画か想像し、説明してください。"""
-        query = Query ()
+        _add_query (prompt, QueryType.KIRIBAN, { 'video_code': video_code })
-        query.user_id = None
-        query.target_character = Character.DEERJIKA.value
-        query.content = prompt
-        query.query_type = QueryType.KIRIBAN.value
-        query.model = GPTModel.GPT3_TURBO.value
-        query.sent_at = datetime.now ()
-        query.answered = False
-        query.transfer_data = { 'video_code': video_code }
-        query.save ()
 
         # 待ち時間計算
         dt = datetime.now ()
@@ -111,6 +121,10 @@ async def report_kiriban (
 
 async def update_kiriban_list (
 ) -> None:
+    """
+    キリ番リストの更新
+    """
+
     while True:
         await wait_until (time (15, 0))
 
@@ -129,54 +143,122 @@ async def update_kiriban_list (
 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
+    Parameters
-                                     .where ('fetched_at', '<=', base_date)
+    ----------
-                                     .max ('fetched_at')))
+    base_date:  date
+        基準日
 
-    for kiriban_views_count in KIRIBAN_VIEWS_COUNTS:
+    Return
-        targets = { vh.video.code for vh in (VideoHistory
+    ------
-                                             .where ('fetched_at', latest_fetched_at)
+    list[tuple[int, VideoInfo, datetime]]
-                                             .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 = nicolib.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
+    result = subprocess.run (
+            ['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:
+        kl = json.loads (result.stdout)
+    except Exception:
+        kl = []
+
+    return [(cast (int, k[0]), video_info, str_to_datetime (cast (str, k[2])))
+            for k in kl
+            if (video_info := nicolib.fetch_video_info (cast (str, k[1]))) is not None]
 
 
 def fetch_comments (
         video_code: str,
-) -> list[Comment]:
+) -> list[CommentDict]:
-    video = Video.where ('code', video_code).first ()
+    """
-    if video is None:
+    動画のコメント・リストを取得する．
-        return []
+
-    return video.comments
+    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:
@@ -184,9 +266,149 @@ async def wait_until (
     await asyncio.sleep ((datetime.combine (d, t) - dt).total_seconds ())
 
 
+async def report_snack_time (
+) -> None:
+    """
+    おやつタイムを報知する．
+    """
+
+    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:
             if history.query is not None:
                 histories.append ({ 'role': 'user', 'content': history.query.content })
             histories.append ({ 'role': 'assistant', 'content': history.content })
+
         for character in [Character.DEERJIKA, Character.GOATOH]:
             if query.target_character & character.value:
                 add_answer (query, character, user_name, histories)
+
         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