nizika_bluesky/main.py

from __future__ import annotations

import asyncio
import os
import time
from datetime import datetime
from io import BytesIO
from typing import Any, TypedDict

import atproto  # type: ignore
import requests
from atproto import Client  # type: ignore
from atproto_client.models import AppBskyEmbedExternal, AppBskyEmbedImages  # type: ignore
from atproto_client.models.app.bsky.feed.get_timeline import Response  # type: ignore
from atproto_client.models.app.bsky.feed.post import ReplyRef  # type: ignore
from requests.exceptions import Timeout

import account
import nicolib
from nizika_ai.consts import Character, GPTModel, Platform, QueryType
from nizika_ai.models import Answer, AnsweredFlag, Query, User

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

time.sleep (60)

client = Client (base_url = 'https://bsky.social')
client.login (account.USER_ID, account.PASSWORD)


async def main (
) -> None:
    """
    メーン処理
    """

    await asyncio.gather (like_posts (),
                          check_mentions (),
                          answer ())


async def like_posts (
) -> None:
    while True:
        try:
            for post in fetch_target_posts ():
                client.like (**post)
        except Exception as e:
            print (f"[like_posts] { type (e).__name__ }: { e }")

        await asyncio.sleep (60)


async def check_mentions (
) -> None:
    while True:
        try:
            for uri in check_notifications ():
                records = fetch_thread_contents (uri, 20)
                if records:
                    record = records[0]
                    image_url: str | None = None
                    if record['embed'] and hasattr (record['embed'], 'images'):
                        image_url = ('https://cdn.bsky.app/img/feed_fullsize/plain'
                                     f"/{ record['did'] }"
                                     f"/{ record['embed'].images[0].image.ref.link }")
                    user = _fetch_user (record['did'], record['name'])
                    _add_query (user, record['text'], image_url, {
                            'uri':  record['strong_ref']['uri'],
                            'cid':  record['strong_ref']['cid'] })
        except Exception as e:
            print (f"[check_mentions] { type (e).__name__ }: { e }")

        await asyncio.sleep (60)


async def answer (
) -> None:
    while True:
        answered_flags = (
                AnsweredFlag
                .where ('platform', Platform.BLUESKY.value)
                .where ('answered', False)
                .get ())
        for answered_flag in answered_flags:
            td: dict[str, Any]
            answer = answered_flag.answer
            answered_flag.answered = True
            answered_flag.save ()
            match QueryType (answer.query_rel.query_type):
                case QueryType.BLUESKY_COMMENT:
                    td = answer.query_rel.transfer_data or { }
                    uri: str | None = td.get ('uri')
                    cid: str | None = td.get ('cid')
                    if (not uri) or (not cid):
                        continue

                    sref = { 'uri': uri, 'cid': cid }
                    strong_ref = atproto.models.create_strong_ref (sref)  # type: ignore
                    reply_ref = ReplyRef (root = strong_ref, parent = strong_ref)
                    try:
                        client.post (answer.content[:250], reply_to = reply_ref)
                    except Exception as e:
                        print (f"[answer/reply] { type (e).__name__ }: { e }")
                        continue
                case QueryType.KIRIBAN | QueryType.NICO_REPORT:
                    td = answer.query_rel.transfer_data or { }
                    video_code: str | None = td.get ('video_code')
                    if not video_code:
                        continue

                    uri = f"https://www.nicovideo.jp/watch/{ video_code }"
                    (title, description, thumbnail) = nicolib.fetch_embed_info (uri)
                    try:
                        resp = requests.get (thumbnail, timeout = 60)
                        resp.raise_for_status ()
                        upload = client.com.atproto.repo.upload_blob (BytesIO (resp.content))
                        thumb = upload.blob
                    except Timeout:
                        thumb = None
                    except Exception as e:
                        print (f"[answer/nico-thumb] { type (e).__name__ }: { e }")
                        thumb = None

                    external = AppBskyEmbedExternal.External (
                            title       = title,
                            description = description,
                            thumb       = thumb,
                            uri         = uri)
                    embed_external = AppBskyEmbedExternal.Main (external = external)
                    try:
                        client.post (answer.content[:250], embed = embed_external)
                    except Exception as e:
                        print (f"[answer/nico-post] { type (e).__name__ }: { e }")
                        continue
                case QueryType.SNACK_TIME:
                    try:
                        with open ('./assets/snack-time.jpg', 'rb') as f:
                            image = AppBskyEmbedImages.Image (
                                    alt = (
                                        '左に喜多ちゃん、右に人面鹿のニジカが'
                                        'V字に並んでいる。'
                                        '喜多ちゃんは右手でピースサインをして'
                                        '片目をウインクしている。'
                                        'ニジカは両手を広げ、'
                                        '右手にスプーンを持って'
                                        'ポーズを取っている。'
                                        '背景には'
                                        '赤と黄色の放射線状の模様が広がり、'
                                        '下部に「おやつタイムだ!!!!」という'
                                        '日本語のテキストが表示されている。'),
                                    image = client.com.atproto.repo.upload_blob (f).blob)
                        client.post (answer.content[:250],
                                     embed = AppBskyEmbedImages.Main (images = [image]))
                    except Exception:
                        pass
                case QueryType.HOT_SPRING:
                    try:
                        with open ('./assets/hot-spring.jpg', 'rb') as f:
                            image = AppBskyEmbedImages.Image (
                                    alt = ('左に喜多ちゃん、右にわさび県産滋賀県が'
                                           'V字に並んでいる。'
                                           '喜多ちゃんは右手でピースサインをして'
                                           '片目をウインクしている。'
                                           'わさび県産滋賀県はただ茫然と'
                                           '立ち尽くしている。'
                                           '背景には'
                                           '血と空の色をした放射線状の模様が広がり、'
                                           '下部に「温泉に入ろう!!!」という'
                                           '日本語のテキストが表示されている。'),
                                    image = client.com.atproto.repo.upload_blob (f).blob)
                        client.post (answer.content[:250],
                                     embed = AppBskyEmbedImages.Main (images = [image]))
                    except Exception:
                        pass
        await asyncio.sleep (10)


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

    notifications = client.app.bsky.notification.list_notifications ()
    for notification in notifications.notifications:
        if not notification.is_read:
            match notification.reason:
                case 'mention' | 'reply' | 'quote':
                    uris.append (notification.uri)
                case 'follow':
                    client.follow (notification.author.did)

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

    return uris


def fetch_thread_contents (
        uri:            str,
        parent_height:  int,
) -> list[Record]:
    post_thread = client.get_post_thread (uri = uri, parent_height = parent_height)
    if not post_thread:
        return []

    res = post_thread.thread

    records: list[Record] = []
    while res:
        if hasattr (res, 'post'):
            records.append ({ 'strong_ref': { 'uri':    res.post.uri,
                                              'cid':    res.post.cid },
                              'did':        res.post.author.did,
                              'handle':     res.post.author.handle,
                              'name':       (res.post.author.display_name
                                             or res.post.author.handle),
                              'datetime':   res.post.record.created_at,
                              'text':       getattr (res.post.record, 'text', None) or '',
                              'embed':      getattr (res.post.record, 'embed', None) })
            res = res.parent
        else:
            break

    return records


def fetch_target_posts (
) -> list[LikeParams]:
    posts: list[LikeParams] = []

    timeline: Response = client.get_timeline ()
    for feed in timeline.feed:
        me = getattr (client, 'me', None)
        my_did = me.did if me else ''
        if (feed.post.author.did != my_did
                and (feed.post.viewer.like is None)
                and any (target_word in (feed.post.record.text or '').casefold ()
                         for target_word in TARGET_WORDS)):
            posts.append (LikeParams ({ 'uri': feed.post.uri, 'cid': feed.post.cid }))

    return posts


def _add_query (
        user:           User,
        content:        str,
        image_url:      str | None  = None,
        transfer_data:  dict[str, Any] | None = None,
) -> None:
    query = Query ()
    query.user_id = user.id
    query.target_character = Character.DEERJIKA.value
    query.content = content
    query.image_url = image_url or None
    query.query_type = QueryType.BLUESKY_COMMENT.value
    query.model = GPTModel.GPT3_TURBO.value
    query.sent_at = datetime.now ()
    query.answered = False
    query.transfer_data = transfer_data
    query.save ()
    # TODO: 履歴情報の追加


def _fetch_user (
        did:    str,
        name:   str,
) -> User:
    user = User.where ('platform', Platform.BLUESKY.value).where ('code', did).first ()
    if user is None:
        user = User ()
        user.platform = Platform.BLUESKY.value
        user.code = did
    user.name = name
    user.save ()
    return user


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


class Record (TypedDict):
    strong_ref: StrongRef
    did:        str
    handle:     str
    name:       str
    datetime:   str
    text:       str
    embed:      object


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


if __name__ == '__main__':
    asyncio.run (main ())