Browse Source

#8 型安全性不明

main
みてるぞ 5 days ago
parent
commit
06b9d015d0
5 changed files with 101 additions and 43 deletions
  1. +12
    -3
      atproto.pyi
  2. +0
    -0
      atproto/models.pyi
  3. +5
    -0
      atproto/models/AppBskyFeedDefs.pyi
  4. +5
    -0
      atproto_client/models/app/bsky/feed/get_timeline.pyi
  5. +79
    -40
      main.py

+ 12
- 3
atproto.pyi View File

@@ -1,6 +1,7 @@
from datetime import datetime 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: class Client:
@@ -14,8 +15,12 @@ class Client:
parent_height: int | None = None parent_height: int | None = None
) -> Response: ... ) -> Response: ...


def get_timeline (self) -> get_timeline.Response: ...

def follow (self, did: str) -> None: ... def follow (self, did: str) -> None: ...


def like (self, uri: str, cid: str) -> None: ...



class AppNamespace: class AppNamespace:
bsky: AppBskyNamespace bsky: AppBskyNamespace
@@ -34,8 +39,12 @@ class AppBskyNotificationNamespace:
class Response: class Response:
notifications: list[Notification] notifications: list[Notification]
thread: (ThreadViewPost thread: (ThreadViewPost
| models.AppBskyFeedDefs.NotFoundPost
| models.AppBskyFeedDefs.BlockedPost)
| NotFoundPost
| BlockedPost)


class ThreadViewPost:
pass




class Notification: class Notification:


+ 0
- 0
atproto/models.pyi View File


+ 5
- 0
atproto/models/AppBskyFeedDefs.pyi View File

@@ -0,0 +1,5 @@
class NotFoundPost:
pass

class BlockedPost:
pass

+ 5
- 0
atproto_client/models/app/bsky/feed/get_timeline.pyi View File

@@ -0,0 +1,5 @@
from atproto.models.AppBskyFeedDefs import FeedViewPost


class Response:
feed: list[FeedViewPost]

+ 79
- 40
main.py View File

@@ -3,6 +3,8 @@ Bluesky のニジカがいろいろする.
(近々機能ごとにファイル分けて systemd でイベント管理する予定) (近々機能ごとにファイル分けて systemd でイベント管理する予定)
""" """


from __future__ import annotations

import io import io
import random import random
import sys import sys
@@ -10,10 +12,11 @@ import time
from datetime import date, datetime from datetime import date, datetime
from datetime import time as dt_time from datetime import time as dt_time
from datetime import timedelta from datetime import timedelta
from typing import cast
from typing import TypedDict, cast


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


@@ -21,45 +24,14 @@ import account
import nico import nico
from ai.talk import Talk 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 ( def main (
@@ -109,6 +81,8 @@ def main (
parent = records[0]['strong_ref'], parent = records[0]['strong_ref'],
root = records[-1]['strong_ref'])) root = records[-1]['strong_ref']))


like_posts (client)

if kiriban_list and datetime.now () >= next_kiriban_at: if kiriban_list and datetime.now () >= next_kiriban_at:
(views_count, video_info, uploaded_at) = ( (views_count, video_info, uploaded_at) = (
kiriban_list.pop (random.randint (0, len (kiriban_list) - 1))) kiriban_list.pop (random.randint (0, len (kiriban_list) - 1)))
@@ -257,6 +231,46 @@ def main (
time.sleep (60) 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 ( def get_embed_info (
url: str url: str
) -> tuple[str, str, str]: ) -> tuple[str, str, str]:
@@ -305,5 +319,30 @@ def get_kiriban_dt_to_update (
return dt 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__': if __name__ == '__main__':
main (*sys.argv[1:]) main (*sys.argv[1:])

Loading…
Cancel
Save