nizika_bluesky/main.py

"""
Bluesky のニジカがいろいろする．
（近々機能ごとにファイル分けて systemd でイベント管理する予定）
"""

from __future__ import annotations

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 TypedDict, cast

import requests
from atproto import Client, models
from atproto_client.models.app.bsky.feed.get_timeline import Response
from bs4 import BeautifulSoup
from requests.exceptions import Timeout

import account
import nico
from ai.talk import Talk

TARGET_WORDS = ['deerjika', 'ニジカ', 'ぼっち', '虹夏', '郁代', 'バーカ',
                'kfif', 'kita-flatten-ikuyo-flatten', 'ラマ田', 'ゴートう',
                'ぼざクリ', 'オオミソカ', '伊地知', '喜多ちゃん',
                '喜タイ', '洗澡鹿', 'シーザオ', '今日は本当に',
                'ダイソーで', '変なチンチン', 'daisoで', 'だね～(笑)',
                'おやつタイム', 'わさしが', 'わさび県', 'たぬマ', 'にくまる',
                'ルイズマリー', '餅', 'ニジゴ', 'ゴニジ', 'ニジニジ',
                '新年だよね', 'うんこじゃん', 'ほくほくのジャガイモ']


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, datetime]] = (
            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']
                                     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'
                                                        if (record['handle']
                                                            == account.USER_ID)
                                                        else 'user'),
                                               'content':
                                                   record['text']},
                                           reversed (records[1:]))])
                client.post (answer,
                             reply_to = models.AppBskyFeedPost.ReplyRef (
                                 parent = records[0]['strong_ref'],
                                 root   = records[-1]['strong_ref']))

        like_posts (client)

        if kiriban_list and datetime.now () >= next_kiriban_at:
            (views_count, video_info, uploaded_at) = (
                    kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
            since_posted = datetime.now () - uploaded_at
            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
            comments = nico.get_comments (video_info['contentId'])
            popular_comments = sorted (comments,
                                       key      = lambda c: c.nico_count,
                                       reverse  = True)[:10]
            latest_comments = sorted (comments,
                                      key       = lambda c: c.posted_at,
                                      reverse   = True)[:10]
            embed_external = models.AppBskyEmbedExternal.Main (
                    external = models.AppBskyEmbedExternal.External (
                        title       = title,
                        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 latest_comments) }」\n"
            prompt += f"""
概要には次のように書かれています：
```html
{ video_info['description'] }
```
このことについて、ニジカちゃんからのお祝いメッセージを下さい。
ただし、そのメッセージ内には再生数の数値とその多さに応じたリアクションを添えてください。
ちなみに1000再生以下はしょぼいです。
また、ぜひ投稿からこの再生数に至るまでにかかった時間や、つけられたタグ、コメントに対して思いを馳せてください。
好きなコメントがあったら教えてね。"""
            client.post (Talk.main (prompt), 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:
            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.post (Talk.main ('今どうしてる?'))
            last_posted_at = now

        time.sleep (60)


def check_notifications (
        client: Client,
) -> list:
    (uris, last_seen_at) = ([], client.get_current_time_iso ())

    for notification in (client.app.bsky.notification.list_notifications ()
                         .notifications):
        if not notification.is_read:
            if notification.reason in ['mention', 'reply', 'quote']:
                uris += [notification.uri]
            elif notification.reason == 'follow':
                client.follow (notification.author.did)

    client.app.bsky.notification.update_seen ({ 'seen_at': last_seen_at })

    return uris


def get_thread_contents (
        client:         Client,
        uri:            str,
        parent_height:  int,
) -> list:
    response = (client.get_post_thread (uri           = uri,
                                        parent_height = parent_height)
                .thread)
    records = []
    while response is not None:
        records += [{ 'strong_ref': models.create_strong_ref (response.post),
                      'did':        response.post.author.did,
                      'handle':     response.post.author.handle,
                      'name':       response.post.author.display_name,
                      'datetime':   response.post.record.created_at,
                      'text':       response.post.record.text,
                      'embed':      response.post.record.embed }]
        response = response.parent

    return records


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


def like_posts (
        client: Client,
) -> None:
    for post in get_target_posts (client):
        client.like (**post)


def get_target_posts (
        client: Client,
) -> list[LikeParams]:
    posts = []

    timeline: Response = client.get_timeline ()
    for feed in timeline.feed:
        if (feed.post.author.did != client.me.did
                and any (target_word in feed.post.record.text.lower () for target_word in TARGET_WORDS)):
            posts.append (LikeParams({ 'uri': feed.post.uri, 'cid': feed.post.cid }))

    return posts


class LikeParams (TypedDict):
    uri:    str
    cid:    str


if __name__ == '__main__':
    main (*sys.argv[1:])