#1 __future__ 忘れ

.gitmodules

@@ -1,4 +1,6 @@
+[submodule "nizika_nico"]
+	path = nizika_nico
+	url = https://git.miteruzo.com/miteruzo/nizika_nico
 [submodule "ai"]
 	path = ai
 	url = https://git.miteruzo.com/miteruzo/nizika_broadcast
-	branch = main

atproto.pyi

@@ -0,0 +1,49 @@
+from datetime import datetime
+
+from atproto import models
+
+
+class Client:
+    app:    AppNamespace
+
+    def get_current_time_iso (self) -> datetime: ...
+
+    def get_post_thread (
+            self,
+            uri:            str,
+            parent_height:  int | None  = None
+    ) -> Response: ...
+
+    def follow (self, did: str) -> None: ...
+
+
+class AppNamespace:
+    bsky:   AppBskyNamespace
+
+
+class AppBskyNamespace:
+    notification:   AppBskyNotificationNamespace
+
+
+class AppBskyNotificationNamespace:
+    def list_notifications (self) -> Response: ...
+
+    def update_seen (self, seen: dict[str, datetime]) -> None: ...
+
+
+class Response:
+    notifications:  list[Notification]
+    thread:         (ThreadViewPost
+                     | models.AppBskyFeedDefs.NotFoundPost
+                     | models.AppBskyFeedDefs.BlockedPost)
+
+
+class Notification:
+    is_read:    bool
+    reason:     str
+    uri:        str
+    author:     ProfileView
+
+
+class ProfileView:
+    did:    str

db

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

eloquent.pyi

@@ -0,0 +1 @@
+db/eloquent.pyi

main.py

@@ -1,11 +1,25 @@
-from datetime import datetime, timedelta
-import time
+"""
+Bluesky のニジカがいろいろする．
+（近々機能ごとにファイル分けて systemd でイベント管理する予定）
+"""
+
+import io
+import random
 import sys
+import time
+from datetime import date, datetime
+from datetime import time as dt_time
+from datetime import timedelta
+from typing import cast
 
+import requests
 from atproto import Client, models
+from bs4 import BeautifulSoup
+from requests.exceptions import Timeout
 
-from ai.talk import Talk
 import account
+import nico
+from ai.talk import Talk
 
 
 def check_notifications (
@@ -16,7 +30,7 @@ def check_notifications (
     for notification in (client.app.bsky.notification.list_notifications ()
                          .notifications):
         if not notification.is_read:
-            if notification.reason in ['mention', 'reply']:
+            if notification.reason in ['mention', 'reply', 'quote']:
                 uris += [notification.uri]
             elif notification.reason == 'follow':
                 client.follow (notification.author.did)
@@ -41,26 +55,45 @@ def get_thread_contents (
                       'handle':     response.post.author.handle,
                       'name':       response.post.author.display_name,
                       'datetime':   response.post.record.created_at,
-                      'text':       response.post.record.text }]
+                      'text':       response.post.record.text,
+                      'embed':      response.post.record.embed }]
         response = response.parent
 
     return records
 
 
-def main () -> None:
+def main (
+) -> None:
+    time.sleep (60)
+
     client = Client (base_url = 'https://bsky.social')
 
     client.login (account.USER_ID, account.PASSWORD)
 
+    got_kiriban_at: date = datetime.now ().date () - timedelta (days = datetime.now ().hour < 15)
+    kiriban_list: list[tuple[int, nico.VideoInfo]] = nico.get_kiriban_list (got_kiriban_at)
+    kiriban_interval: timedelta = ((get_kiriban_dt_to_update () - datetime.now ())
+                                   / len (kiriban_list))
+    next_kiriban_at = datetime.now ()
+
     last_posted_at = datetime.now () - timedelta (hours = 6)
     has_got_snack_time = False
+    has_taken_hot_spring = False
+    watched_videos = []
     while True:
         now = datetime.now ()
 
         for uri in check_notifications (client):
             records = get_thread_contents (client, uri, 20)
             if len (records) > 0:
-                answer = Talk.main (records[0]['text'],
+                answer = Talk.main ((records[0]['text']
+                                     if (records[0]['embed'] is None
+                                         or not hasattr (records[0]['embed'],
+                                                         'images'))
+                                     else [
+                        { 'type': 'text', 'text': records[0]['text'] },
+                        { 'type': 'image_url', 'image_url': {
+                            'url': f"https://cdn.bsky.app/img/feed_fullsize/plain/{ records[0]['did'] }/{ records[0]['embed'].images[0].image.ref.link }" } }]),
                                     records[0]['name'],
                                     [*map (lambda record: {
                                                'role': ('assistant'
@@ -70,26 +103,189 @@ def main () -> None:
                                                'content':
                                                    record['text']},
                                            reversed (records[1:]))])
-                client.send_post (answer,
+                client.post (answer,
                              reply_to = models.AppBskyFeedPost.ReplyRef (
                                  parent = records[0]['strong_ref'],
                                  root   = records[-1]['strong_ref']))
 
+        if kiriban_list and datetime.now () >= next_kiriban_at:
+            (views_count, video_info) = (
+                    kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
+            uri = f"https://www.nicovideo.jp/watch/{ video_info['contentId'] }"
+            (title, description, thumbnail) = get_embed_info (uri)
+            try:
+                upload = client.com.atproto.repo.upload_blob (
+                        io.BytesIO (requests.get (thumbnail,
+                                                  timeout = 60).content))
+                thumb = upload.blob
+            except Timeout:
+                thumb = None
+            embed_external = models.AppBskyEmbedExternal.Main (
+                    external = models.AppBskyEmbedExternal.External (
+                        title       = title,
+                        description = description,
+                        thumb       = thumb,
+                        uri         = uri))
+            client.post (Talk.main (f"""
+ニコニコの『{ video_info['title'] }』という動画が{ views_count }再生を突破しました。
+つけられたタグは「{ '」、「'.join (video_info['tags']) }」です。
+概要には次のように書かれています：
+```html
+{ video_info['description'] }
+```
+このことについて、ニジカちゃんからのお祝いメッセージを下さい。"""),
+                         embed = embed_external)
+            next_kiriban_at += kiriban_interval
+            last_posted_at = now
+
+        latest_deerjika = nico.get_latest_deerjika ()
+        if latest_deerjika is not None:
+            for datum in [e for e in [latest_deerjika]
+                            if e['contentId'] not in watched_videos]:
+                watched_videos += [datum['contentId']]
+
+                uri = f"https://www.nicovideo.jp/watch/{ datum['contentId'] }"
+                (title, description, thumbnail) = get_embed_info (uri)
+                try:
+                    upload = client.com.atproto.repo.upload_blob (
+                            io.BytesIO (requests.get (thumbnail,
+                                                      timeout = 60).content))
+                    thumb = upload.blob
+                except Timeout:
+                    thumb = None
+
+                embed_external = models.AppBskyEmbedExternal.Main (
+                        external = models.AppBskyEmbedExternal.External (
+                            title       = title,
+                            description = description,
+                            thumb       = thumb,
+                            uri         = uri))
+                client.post (Talk.main (f"""
+ニコニコに『{ datum['title'] }』という動画がアップされました。
+つけられたタグは「{ '」、「'.join (datum['tags']) }」です。
+概要には次のように書かれています：
+```html
+{ datum['description'] }
+```
+このことについて、みんなに告知するとともに、ニジカちゃんの感想を教えてください。 """),
+                             embed = embed_external)
+                last_posted_at = now
+
         if now.hour == 14 and has_got_snack_time:
             has_got_snack_time = False
 
-        if now.hour == 15 and not has_got_snack_time:
-            client.send_post (Talk.main ('おやつタイムだ!!!!'))
-            last_posted_at = now
+        if now.hour == 15:
+            if got_kiriban_at < datetime.now ().date ():
+                kiriban_list = nico.get_kiriban_list (datetime.now ().date ())
+                got_kiriban_at = datetime.now ().date ()
+                kiriban_interval = ((get_kiriban_dt_to_update () - datetime.now ())
+                                    / len (kiriban_list))
+                next_kiriban_at = datetime.now ()
 
+            if not has_got_snack_time:
+                try:
+                    with open ('./assets/snack-time.jpg', 'rb') as f:
+                        image = models.AppBskyEmbedImages.Image (
+                                alt   = ('左に喜多ちゃん、右に人面鹿のニジカが'
+                                         'V字に並んでいる。'
+                                         '喜多ちゃんは右手でピースサインをして'
+                                         '片目をウインクしている。'
+                                         'ニジカは両手を広げ、'
+                                         '右手にスプーンを持って'
+                                         'ポーズを取っている。'
+                                         '背景には'
+                                         '赤と黄色の放射線状の模様が広がり、'
+                                         '下部に「おやつタイムだ!!!!」という'
+                                         '日本語のテキストが表示されている。'),
+                                image = client.com.atproto.repo.upload_blob (f).blob)
+                    client.post (Talk.main ('おやつタイムだ!!!!'),
+                                 embed = models.app.bsky.embed.images.Main (
+                                     images = [image]))
+                    last_posted_at = now
+                except Exception:
+                    pass
                 has_got_snack_time = True
 
+        if now.hour == 20 and has_taken_hot_spring:
+            has_taken_hot_spring = False
+
+        if now.hour == 21 and not has_taken_hot_spring:
+            try:
+                with open ('./assets/hot-spring.jpg', 'rb') as f:
+                    image = models.AppBskyEmbedImages.Image (
+                            alt     = ('左に喜多ちゃん、右にわさび県産滋賀県が'
+                                       'V字に並んでいる。'
+                                       '喜多ちゃんは右手でピースサインをして'
+                                       '片目をウインクしている。'
+                                       'わさび県産滋賀県はただ茫然と'
+                                       '立ち尽くしている。'
+                                       '背景には'
+                                       '血と空の色をした放射線状の模様が広がり、'
+                                       '下部に「温泉に入ろう!!!」という'
+                                       '日本語のテキストが表示されている。'),
+                            image   = client.com.atproto.repo.upload_blob (f).blob)
+                client.post (Talk.main ('温泉に入ろう!!!'),
+                             embed = models.app.bsky.embed.images.Main (
+                                 images = [image]))
+                last_posted_at = now
+            except Exception:
+                pass
+            has_taken_hot_spring = True
+
         if now - last_posted_at >= timedelta (hours = 6):
-            client.send_post (Talk.main (''))
+            client.post (Talk.main ('今どうしてる?'))
             last_posted_at = now
 
         time.sleep (60)
 
 
+def get_embed_info (
+        url:    str
+) -> tuple[str, str, str]:
+    title:          str = ''
+    description:    str = ''
+    thumbnail:      str = ''
+
+    try:
+        res = requests.get (url, timeout = 60)
+    except Timeout:
+        return ('', '', '')
+
+    if res.status_code != 200:
+        return ('', '', '')
+
+    soup = BeautifulSoup (res.text, 'html.parser')
+
+    tmp = soup.find ('title')
+    if tmp is not None:
+        title = tmp.text
+
+    tmp = soup.find ('meta', attrs = { 'name': 'description' })
+    if tmp is not None and hasattr (tmp, 'get'):
+        try:
+            description = cast (str, tmp.get ('content'))
+        except Exception:
+            pass
+
+    tmp = soup.find ('meta', attrs = { 'name': 'thumbnail' })
+    if tmp is not None and hasattr (tmp, 'get'):
+        try:
+            thumbnail = cast (str, tmp.get ('content'))
+        except Exception:
+            pass
+
+    return (title, description, thumbnail)
+
+
+def get_kiriban_dt_to_update (
+) -> datetime:
+    now = datetime.now ()
+    today = now.date ()
+    dt = datetime.combine (today, dt_time (15, 0))
+    if dt <= now:
+        dt += timedelta (days = 1)
+    return dt
+
+
 if __name__ == '__main__':
     main (*sys.argv[1:])

nico.py

@@ -0,0 +1,158 @@
+"""
+ニコニコのニジカ動画取得モヂュール
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import date, timedelta
+from typing import TypedDict, cast
+
+import requests
+from bs4 import BeautifulSoup
+from requests.exceptions import Timeout
+from eloquent import DatabaseManager, Model
+
+from db.models import Tag, Video, VideoHistory, VideoTag
+
+CONFIG: dict[str, DbConfig] = { 'mysql': { 'driver':    'mysql',
+                                           'host':      'localhost',
+                                           'database':  'nizika_nico',
+                                           'user':      os.environ['MYSQL_USER'],
+                                           'password':  os.environ['MYSQL_PASS'],
+                                           'prefix':    '' } }
+
+DB = DatabaseManager (CONFIG)
+Model.set_connection_resolver (DB)
+
+KIRIBAN_VIEWS_COUNTS: set[int] = { *range (100, 1_000, 100),
+                                   *range (1_000, 10_000, 1_000),
+                                   *range (10_000, 1_000_001, 10_000),
+                                   194, 245, 510, 114_514, 1_940, 2_450, 5_100, 24_500,
+                                   51_000, 2_424 }
+
+
+class VideoInfo (TypedDict):
+    contentId:      str
+    title:          str
+    tags:           list[str]
+    description:    str
+
+
+def get_latest_deerjika (
+) -> VideoInfo | None:
+    tag = '伊地知ニジカ OR ぼざろクリーチャーシリーズ'
+    url = f"https://www.nicovideo.jp/tag/{ tag }"
+
+    params = { 'sort':  'f',
+               'order': 'd' }
+
+    video_info = { }
+
+    bs = get_bs_from_url (url, params)
+    if bs is None:
+        return None
+
+    try:
+        video = (bs.find_all ('ul', class_ = 'videoListInner')[1]
+                 .find ('li', class_ = 'item'))
+
+        video_info['contentId'] = video['data-video-id']
+    except Exception:
+        return None
+
+    return get_video_info (video_info['contentId'])
+
+
+def get_bs_from_url (
+        url:    str,
+        params: dict    = { },
+) -> BeautifulSoup | None:
+    """
+    URL から BeautifulSoup インスタンス生成
+
+    Parameters
+    ----------
+    url:    str
+        捜査する URL
+    params: dict
+        パラメータ
+
+    Return
+    ------
+    BeautifulSoup | None
+        BeautifulSoup オブゼクト（失敗したら None）
+    """
+
+    try:
+        req = requests.get (url, params = params, timeout = 60)
+    except Timeout:
+        return None
+
+    if req.status_code != 200:
+        return None
+
+    req.encoding = req.apparent_encoding
+
+    return BeautifulSoup (req.text, 'html.parser')
+
+
+def get_video_info (
+        video_code: str,
+) -> VideoInfo | None:
+    video_info: dict[str, str | list[str]] = { 'contentId': video_code }
+
+    bs = get_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 get_kiriban_list (
+        base_date:  date,
+) -> list[tuple[int, VideoInfo]]:
+    kiriban_list: list[tuple[int, VideoInfo]] = []
+
+    latest_fetched_at = cast (date, VideoHistory.max ('fetched_at'))
+    previous_fetched_at = cast (date, (VideoHistory
+                                       .where ('fetched_at', '<', latest_fetched_at)
+                                       .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 ()) }
+                   - { vh.video.code for vh in (VideoHistory
+                                                .where ('fetched_at', previous_fetched_at)
+                                                .where ('views_count', '>=', kiriban_views_count)
+                                                .get ()) })
+        for code in targets:
+            video_info = get_video_info (code)
+            if video_info is not None:
+                kiriban_list.append ((kiriban_views_count, video_info))
+
+    return kiriban_list
+
+
+class DbConfig (TypedDict):
+    driver:     str
+    host:       str
+    database:   str
+    user:       str
+    password:   str
+    prefix:     str