fyrfli/interactive_bot
0
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

looks like i finally have this shit working

Browse source
This commit is contained in:
Camille Frantz 2025-10-30 16:59:50 +00:00
parent 62f474f3e9
commit b1a3309cc7
Signed by: fyrfli
SSH key fingerprint: SHA256:oxaZT6PRsoNy2x01TIrx3YE6H6p6gdmG2/Ir05d/bjc
1 changed files with 18 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

61
listen.py
View file

@ -5,19 +5,16 @@ from requests import get, post
from random import randint from random import randint
from os import getenv from os import getenv
from munch import munchify from munch import munchify
from lxml import html
from markdownify import markdownify as md from markdownify import markdownify as md
from pprint import pp from pprint import pp
import csv, io import csv, io
import sys import sys
import apprise import apprise
def strip_html(s):
return str(html.fromstring(s).text_content())
apprise_object = apprise.Apprise() apprise_object = apprise.Apprise()
apprise_object.add('pover://dP9LCGHZRMozXRn6K5PctGg6uZhaYc@af7xxr6ho94isdzc3uj92m3f82hdkb', tag='pover') apprise_object.add('pover://dP9LCGHZRMozXRn6K5PctGg6uZhaYc@af7xxr6ho94isdzc3uj92m3f82hdkb', tag='pover')
apprise_object.add('zulip://sandbox@chat.fyrfli.org/KiXbeBR7poxmUuuW3S6b9igX2ibpyfNC/sandbox', tag='zulip') apprise_object.add('zulip://sandbox@chat.fyrfli.org/KiXbeBR7poxmUuuW3S6b9igX2ibpyfNC/sandbox', tag='zulip')
# apprise_object.notify(body='is this even working from inside this program?', title='fyrfli at mastodon.social timeline', tag='zulip')
api_token = getenv('MASTODON_SOCIAL_FYRFLI_TOKEN') api_token = getenv('MASTODON_SOCIAL_FYRFLI_TOKEN')
@ -29,10 +26,6 @@ quotes_list = loads(dumps(list(csv.DictReader(io.StringIO(get(quotes_src).text))
instance_info = munchify(loads(get(f"https://{endpoint}/api/v1/instance").text)) instance_info = munchify(loads(get(f"https://{endpoint}/api/v1/instance").text))
streaming_api = f"{instance_info.urls.streaming_api}/api/v1/streaming?access_token={api_token}&stream=user" streaming_api = f"{instance_info.urls.streaming_api}/api/v1/streaming?access_token={api_token}&stream=user"
# print(streaming_api)
# sys.exit(0)
# streaming_api=f"wss://{endpoint}/api/v1/streaming?access_token={api_token}&stream=user"
post_uri = f"https://{endpoint}/api/v1/statuses" post_uri = f"https://{endpoint}/api/v1/statuses"
headers = { "Content-Type" : "application/json", headers = { "Content-Type" : "application/json",
"Authorization" : f"Bearer {api_token}", "Authorization" : f"Bearer {api_token}",
@ -62,7 +55,6 @@ def get_random_quote():
def process_request(the_post): def process_request(the_post):
print('processing ', the_post.url) print('processing ', the_post.url)
# pp(dumps(the_post), compact=True, indent=2)
hellos = ["hey" in the_post.content.lower(),"hey you" in the_post.content.lower(), "hi" in the_post.content.lower(), "hello" in the_post.content.lower()] hellos = ["hey" in the_post.content.lower(),"hey you" in the_post.content.lower(), "hi" in the_post.content.lower(), "hello" in the_post.content.lower()]
request_quote = ["random quote" in the_post.content.lower(), "quote please" in the_post.content.lower()] request_quote = ["random quote" in the_post.content.lower(), "quote please" in the_post.content.lower()]
if any(hellos): if any(hellos):
@ -82,59 +74,41 @@ def process_request(the_post):
async def hello(uri): async def hello(uri):
# async with websockets.connect(uri) as ws: async for ws in websockets.connect(streaming_api):
async for ws in websockets.connect(uri):
try: try:
message = await ws.recv() message = await ws.recv()
decoded = munchify(loads(message)) decoded = munchify(loads(message))
# print(decoded.event, decoded.payload[1:150])
# print(decoded.event, decoded.payload)
if decoded.event != "delete": if decoded.event != "delete":
the_post = munchify(loads(decoded.payload)) the_post = munchify(loads(decoded.payload))
# print(decoded.payload[1:250])
match decoded.event: match decoded.event:
case "delete": case "delete":
# message = f"a delete event for id: {decoded.payload}\\n===>\\n" message = f"a delete event for id: {decoded.payload}\n===>\n"
# print("a delete event for id ", decoded.payload, "\n===>\n")
message = 'a delete event for id: ', decoded.payload, '===>'
case "status.update": case "status.update":
# print("an update to an existing post: ", the_post.url, "\n===>\n") message = f"an update to an existing post: {the_post.url}\n===>\n"
# message = f"an update to an existing post: {the_post.url}\\n===>\\n"
message = 'an update to an existing post: ', the_post.url, '===>'
case "update": case "update":
if the_post.reblog: if the_post.reblog:
# print( "a reblog of ", the_post.reblog.url)
reblog_post_id = the_post.reblog.url.split('/')[4] reblog_post_id = the_post.reblog.url.split('/')[4]
reblogged_post = munchify(loads(get(f'{post_uri}/{reblog_post_id}').text)) reblogged_post = munchify(loads(get(f'{post_uri}/{reblog_post_id}').text))
# pp(md(reblogged_post.content)) message = f"a reblog of {the_post.reblog.url}"
# print("\n===>\n") if reblogged_post.content:
# message = f"a reblog of {the_post.reblog.url}\\n{pp(md(reblogged_post.content))}\\n===>\\n" message += "\n{(pp(md(reblogged_post.content)) or \"unknown content\")}\n===>\n"
message = 'a reblog of ', the_post.reblog.url, pp(md(reblogged_post.content)),'===>'
else: else:
# plain_text_content = strip_html(the_post.content)
post_content = md(the_post.content) post_content = md(the_post.content)
# print( "new status: ", the_post.url, "\n", post_content, "\n===>\n" ) message = f"new status: {the_post.url}\n{post_content}\n===>\n"
# message = f"new status: {the_post.url}\\n{post_content}\\n===>\\n"
message = 'new status: ', the_post.url, post_content, '===>'
if the_post.mentions: if the_post.mentions:
for mention in the_post.mentions: for mention in the_post.mentions:
if mention.username == "botsy": if mention.username == "botsy":
process_request(the_post) process_request(the_post)
# break
case "notification": case "notification":
process_request(the_post) process_request(the_post)
case _: case _:
# print('unknown event: ', decoded.event, "\n", message = f"unknown event: {decoded.event}\n{decoded.payload[1:200]}"
# " with payload beginning ", print(message)
# decoded.payload[1:150]) apprise_object.notify(body=message, title='fyrfli at mastodon.social timeline', tag='zulip')
# message = f"unknown event: {decoded.event}\\n{decoded.payload[1:200]}"
message = 'unknown event: ', decoded.event, decoded.payload[1:120]
pp(message)
apprise_object.notify(body=message, title='@fyrfli.mastodon.social timeline')
except asyncio.exceptions.CancelledError: except asyncio.exceptions.CancelledError:
print('keyboard interrupt received ... exiting') print('keyboard interrupt received ... exiting')
@ -145,15 +119,16 @@ async def hello(uri):
break break
except websockets.ConnectionClosed: except websockets.ConnectionClosed:
break print("connection closed ... reconnecting ...")
continue
except websockets.exceptions.ConnectionClosed:
print("connection closed ... reconnecting ...")
continue
async def main(): async def main():
async with websockets.serve(hello, streaming_api): async with websockets.connect(hello, streaming_api):
await asyncio.Future() await asyncio.Future()
# print(__name__)
# sys.exit(0)
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(hello(streaming_api)) asyncio.run(hello(streaming_api))