nizika_bluesky/test.py

from __future__ import annotations

import asyncio
import time
from typing import TypedDict

import atproto
from atproto import Client
from atproto_client.models.app.bsky.feed.get_timeline import Response
from atproto_client.models.com.atproto.repo.strong_ref import Main

import account
from nizika_ai.consts import 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 (),
                          reply ())


async def like_posts (
) -> None:
    while True:
        for post in fetch_target_posts ():
            client.like (**post)

        await asyncio.sleep (60)


async def check_mentions (
) -> None:
    while True:
        for uri in check_notifications ():
            records = fetch_thread_contents (uri, 20)
            if records:
                record = records[0]
                content = record['text']
                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)

        await asyncio.sleep (60)


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:
        records.append ({ 'strong_ref': atproto.models.create_strong_ref (res.post),
                          'did':        res.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 })
        res = res.parent

    return records


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

    timeline: Response = client.get_timeline ()
    for feed in timeline.feed:
        if (feed.post.author.did != client.me.did
                and (feed.post.viewer.like is None)
                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


def _add_query (
        user:       User,
        content:    str,
        image_url:  str | 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.save ()
    # TODO: 履歴情報の追加


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


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


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


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