miteruzo
/
nizika_bluesky
1
0
Fork 0
Code Issues 10 Pull Requests 0 Releases 0 Wiki Activity
Browse Source

#11

ai-migration
みてるぞ 1 week ago
parent
3e8c2c5fe0
commit
7ff729b256
5 changed files with 125 additions and 176 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +0
    -58
      atproto.pyi
  2. +0
    -5
      atproto/models/AppBskyFeedDefs.pyi
  3. +0
    -5
      atproto_client/models/app/bsky/feed/get_timeline.pyi
  4. +124
    -107
      main.py
  5. +1
    -1
      nizika_ai

+ 0
- 58
atproto.pyi View File

@@ -1,58 +0,0 @@
from datetime import datetime

from atproto.models.AppBskyFeedDefs import BlockedPost, NotFoundPost
from atproto_client.models.app.bsky.feed import get_timeline


class Client:
app: AppNamespace

def get_current_time_iso (self) -> datetime: ...

def get_post_thread (
self,
uri: str,
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


class AppBskyNamespace:
notification: AppBskyNotificationNamespace


class AppBskyNotificationNamespace:
def list_notifications (self) -> Response: ...

def update_seen (self, seen: dict[str, datetime]) -> None: ...


class Response:
notifications: list[Notification]
thread: (ThreadViewPost
| NotFoundPost
| BlockedPost)


class ThreadViewPost:
pass


class Notification:
is_read: bool
reason: str
uri: str
author: ProfileView


class ProfileView:
did: str

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

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

class BlockedPost:
pass

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

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


class Response:
feed: list[FeedViewPost]

+ 124
- 107
main.py View File

@@ -5,16 +5,14 @@ import os
import time import time
from datetime import datetime from datetime import datetime
from io import BytesIO from io import BytesIO
from typing import TypedDict
from typing import Any, TypedDict


import atproto
import atproto # type: ignore
import requests import requests
from atproto import Client
from atproto.models import AppBskyEmbedExternal, AppBskyEmbedImages
from atproto.models.AppBskyFeedPost import ReplyRef
from atproto.models.app.bsky.embed import images
from atproto.models.com.atproto.repo import strong_ref
from atproto_client.models.app.bsky.feed.get_timeline import Response
from atproto import Client # type: ignore
from atproto.models import AppBskyEmbedExternal, AppBskyEmbedImages # type: ignore
from atproto.models.AppBskyFeedPost import ReplyRef # type: ignore
from atproto_client.models.app.bsky.feed.get_timeline import Response # type: ignore
from requests.exceptions import Timeout from requests.exceptions import Timeout


import account import account
@@ -68,7 +66,6 @@ async def check_mentions (
records = fetch_thread_contents (uri, 20) records = fetch_thread_contents (uri, 20)
if records: if records:
record = records[0] record = records[0]
content = record['text']
image_url: str | None = None image_url: str | None = None
if record['embed'] and hasattr (record['embed'], 'images'): if record['embed'] and hasattr (record['embed'], 'images'):
image_url = ('https://cdn.bsky.app/img/feed_fullsize/plain' image_url = ('https://cdn.bsky.app/img/feed_fullsize/plain'
@@ -86,95 +83,110 @@ async def check_mentions (


async def answer ( async def answer (
) -> None: ) -> None:
answered_flags = (
AnsweredFlag
.where ('platform', Platform.BLUESKY.value)
.where ('answered', False)
.get ())
for answered_flag in answered_flags:
answer = answered_flag.answer
match QueryType (answer.query.query_type):
case QueryType.BLUESKY_COMMENT:
td = answer.query.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)
reply_ref = ReplyRef (root = strong_ref, parent = strong_ref)
client.post (answer.content, reply_to = reply_ref)
flag.answered = True
flag.save ()
case QueryType.KIRIBAN | QueryType.NICO_REPORT:
td = answer.query.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:
upload = client.com.atproto.repo.upload_blob (
BytesIO (requests.get (thumbnail, timeout = 60).content))
thumb = upload.blob
except Timeout:
thumb = None

external = AppBskyEmbedExternal.External (
title = title,
description = description,
thumb = thumb,
uri = uri)
embed_external = AppBskyEmbedExternal.Main (external = external)
client.post (answer.content, embed = embed_external)
flag.answered = True
flag.save ()
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,
embed = AppBskyEmbedImages.Main (images = [image]))
flag.answered = True
flag.save ()
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,
embed = AppBskyEmbedImages.Main (images = [image]))
flag.answered = True
flag.save ()
except Exception:
pass
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
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, reply_to = reply_ref)
except Exception as e:
print (f"[answer/reply] { type (e).__name__ }: { e }")
continue
answered_flag.answered = True
answered_flag.save ()
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, embed = embed_external)
except Exception as e:
print (f"[answer/nico-post] { type (e).__name__ }: { e }")
continue
answered_flag.answered = True
answered_flag.save ()
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,
embed = AppBskyEmbedImages.Main (images = [image]))
answered_flag.answered = True
answered_flag.save ()
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,
embed = AppBskyEmbedImages.Main (images = [image]))
answered_flag.answered = True
answered_flag.save ()
except Exception:
pass
await asyncio.sleep (10)




def check_notifications ( def check_notifications (
@@ -208,14 +220,19 @@ def fetch_thread_contents (


records: list[Record] = [] records: list[Record] = []
while res: while res:
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,
'datetime': res.post.record.created_at,
'text': res.post.record.text,
'embed': res.post.record.embed })
res = res.parent
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 return records


@@ -241,7 +258,7 @@ def _add_query (
user: User, user: User,
content: str, content: str,
image_url: str | None = None, image_url: str | None = None,
transfer_data: dict | None = None,
transfer_data: dict[str, Any] | None = None,
) -> None: ) -> None:
query = Query () query = Query ()
query.user_id = user.id query.user_id = user.id


+ 1
- 1
nizika_ai

@@ -1 +1 @@
Subproject commit 4e5bd13ab45ec024bda6746db4959d32fced56da
Subproject commit 3be6d9063c987deaceee24a1d16296d21319778c

Write Preview
Loading…
Cancel
Save