From 06b9d015d04153aa6d3eaee56396d187e28c830f Mon Sep 17 00:00:00 2001 From: miteruzo Date: Sat, 9 Nov 2024 06:02:57 +0900 Subject: [PATCH] =?UTF-8?q?#8=20=E5=9E=8B=E5=AE=89=E5=85=A8=E6=80=A7?= =?UTF-8?q?=E4=B8=8D=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- atproto.pyi | 15 ++- atproto/models.pyi | 0 atproto/models/AppBskyFeedDefs.pyi | 5 + .../models/app/bsky/feed/get_timeline.pyi | 5 + main.py | 119 ++++++++++++------ 5 files changed, 101 insertions(+), 43 deletions(-) delete mode 100644 atproto/models.pyi create mode 100644 atproto/models/AppBskyFeedDefs.pyi create mode 100644 atproto_client/models/app/bsky/feed/get_timeline.pyi diff --git a/atproto.pyi b/atproto.pyi index ed46197..41abdcf 100644 --- a/atproto.pyi +++ b/atproto.pyi @@ -1,6 +1,7 @@ from datetime import datetime -from atproto import models +from atproto.models.AppBskyFeedDefs import BlockedPost, NotFoundPost +from atproto_client.models.app.bsky.feed import get_timeline class Client: @@ -14,8 +15,12 @@ class Client: parent_height: int | None = None ) -> Response: ... + def get_timeline (self) -> get_timeline.Response: ... + def follow (self, did: str) -> None: ... + def like (self, uri: str, cid: str) -> None: ... + class AppNamespace: bsky: AppBskyNamespace @@ -34,8 +39,12 @@ class AppBskyNotificationNamespace: class Response: notifications: list[Notification] thread: (ThreadViewPost - | models.AppBskyFeedDefs.NotFoundPost - | models.AppBskyFeedDefs.BlockedPost) + | NotFoundPost + | BlockedPost) + + +class ThreadViewPost: + pass class Notification: diff --git a/atproto/models.pyi b/atproto/models.pyi deleted file mode 100644 index e69de29..0000000 diff --git a/atproto/models/AppBskyFeedDefs.pyi b/atproto/models/AppBskyFeedDefs.pyi new file mode 100644 index 0000000..6a80cd7 --- /dev/null +++ b/atproto/models/AppBskyFeedDefs.pyi @@ -0,0 +1,5 @@ +class NotFoundPost: + pass + +class BlockedPost: + pass diff --git a/atproto_client/models/app/bsky/feed/get_timeline.pyi b/atproto_client/models/app/bsky/feed/get_timeline.pyi new file mode 100644 index 0000000..a701328 --- /dev/null +++ b/atproto_client/models/app/bsky/feed/get_timeline.pyi @@ -0,0 +1,5 @@ +from atproto.models.AppBskyFeedDefs import FeedViewPost + + +class Response: + feed: list[FeedViewPost] diff --git a/main.py b/main.py index e882306..6feeda1 100644 --- a/main.py +++ b/main.py @@ -3,6 +3,8 @@ Bluesky のニジカがいろいろする． （近々機能ごとにファイル分けて systemd でイベント管理する予定） """ +from __future__ import annotations + import io import random import sys @@ -10,10 +12,11 @@ import time from datetime import date, datetime from datetime import time as dt_time from datetime import timedelta -from typing import cast +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 @@ -21,45 +24,14 @@ import account import nico from ai.talk import Talk - -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 +TARGET_WORDS = ['deerjika', 'ニジカ', 'ぼっち', '虹夏', '郁代', 'バーカ', + 'kfif', 'kita-flatten-ikuyo-flatten', 'ラマ田', 'ゴートう', + 'ぼざクリ', 'オオミソカ', '伊地知', '喜多ちゃん', + '喜タイ', '洗澡鹿', 'シーザオ', '今日は本当に', + 'ダイソーで', '変なチンチン', 'daisoで', 'だね～(笑)', + 'おやつタイム', 'わさしが', 'わさび県', 'たぬマ', 'にくまる', + 'ルイズマリー', '餅', 'ニジゴ', 'ゴニジ', 'ニジニジ', + '新年だよね', 'うんこじゃん', 'ほくほくのジャガイモ'] def main ( @@ -109,6 +81,8 @@ def main ( 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))) @@ -257,6 +231,46 @@ def main ( 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]: @@ -305,5 +319,30 @@ def get_kiriban_dt_to_update ( 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 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:])