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

now responds to mentioned hellos, and requests for a random quote

Browse source
This commit is contained in:
Camille Frantz 2025-10-10 19:27:41 -05:00
parent 17d1dd48e4
commit 2d23297cb5
Signed by: fyrfli
SSH key fingerprint: SHA256:oxaZT6PRsoNy2x01TIrx3YE6H6p6gdmG2/Ir05d/bjc
1 changed files with 67 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

140
gts_socket_listen.py
View file

@ -3,26 +3,69 @@ import websockets
import asyncio
import json
import requests
import random
from os import getenv
from munch import DefaultMunch as dm
from munch import DefaultMunch as dm, Munch
from lxml import html
import pprint
import csv, io
def strip_html(s):
return str(html.fromstring(s).text_content())
api_token = getenv('LG_BOTSY_TOKEN')
fyrfli_token = getenv('LG_FYRFLI_TOKEN')
endpoint = getenv('LG_ENDPOINT')
quotes_src = getenv('QUOTES_SOURCE')
quotes_list = json.loads(json.dumps(list(csv.DictReader(io.StringIO(requests.get(quotes_src).text)))))
ws_uri=f"wss://{endpoint}/api/v1/streaming?access_token={api_token}&stream=user"
message = ""
post_uri = f"https://{endpoint}/api/v1/statuses"
note_to_post = ""
headers = { "Content-Type" : "application/json",
"Authorization" : f"Bearer {api_token}",
"User-Agent" : "fyrfliBOT" }
mention_id = ""
created_note = ""
status_dict = Munch({"status": "",
"in_reply_to_id": "",
"local_only": False,
"visibility": "mutuals_only",
"sensitive": False,
"content_type": "text/markdown"})
randomised_hellos = [
'hi back',
'hello to you too'
'aw hi hi',
'well hello there',
'mornin\' mornin\'',
'g\'day mate',
'allo'
]
def get_random_quote():
choise = quotes_list[random.randint(0, len(quotes_list) - 1)]
return choise['quote'] + '\n ~ ' + choise['author'] + ' ~'
def process_request(the_post):
print('processing ', the_post.url)
pprint.pprint(json.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()]
request_quote = ["random quote" in the_post.content.lower(), "quote please" in the_post.content.lower()]
if any(hellos):
respond_with = randomised_hellos[random.randint(0, len(randomised_hellos) - 1)]
status_dict.status = f"[@{the_post.account.acct}]({the_post.account.url}) {respond_with}!"
if any(request_quote):
status_dict.status = f"[@{the_post.account.acct}]({the_post.account.url}) here's a quote for you:\n\n" + get_random_quote()
if status_dict.status:
status_dict.in_reply_to_id = the_post.id
pprint.pprint(status_dict, compact=False, indent=2)
r = requests.post(
post_uri,
headers=headers,
data=json.dumps(status_dict)
)
print(r.reason, r.status_code, json.loads(r.text))
async def hello(uri):
# async with websockets.connect(uri) as ws:
@ -30,86 +73,35 @@ async def hello(uri):
try:
message = await ws.recv()
decoded = dm.fromDict(json.loads(message))
print(decoded.event, decoded.payload[1:49])
print(decoded.event, decoded.payload[1:150])
if decoded.event != "delete":
the_post = dm.fromDict(json.loads(decoded.payload))
match decoded.event:
case "delete":
print("a delete event for id ", decoded.payload, "\n===>\n")
case "update"|"status.update":
the_update = dm.fromDict(json.loads(decoded.payload))
print(json.dumps(the_update))
if the_update.reblog:
print(
"a reblog of ",
the_update.reblog.url,
"\n id ",
the_update.id,
"\n created at ",
the_update.created_at,
"\n===>\n"
)
case "status.update":
print("an update to an existing post: ", the_post.url, "\n===>\n")
case "update":
if the_post.reblog:
print( "a reblog of ", the_post.reblog.url, "\n===>\n" )
else:
plain_text_content = strip_html(the_update.content)
print(
"a status update (new or edit) ",
the_update.url,
"\n id ",
the_update.id,
"\n by ",
the_update.account.acct,
"\n created at ",
the_update.created_at,
"\n the content \n",
plain_text_content,
"\n===>\n"
)
if the_update.mentions:
for mention in the_update.mentions:
plain_text_content = strip_html(the_post.content)
print( "new status: ", the_post.url, "\n===>\n" )
if the_post.mentions:
for mention in the_post.mentions:
if mention.username == "botsy":
note_to_post = {
"status": f"[@{the_update.account.acct}](the_update.account.url) We come to love not by finding a perfect person, but by learning to see an imperfect person perfectly.\n ~ Sam Keen ~ ",
"in_reply_to_id": the_update.id,
"visibility": "mutuals_only",
"content_type": "text/markdown"
}
r = requests.post(
post_uri,
headers=headers,
json=note_to_post
)
print(r.reason, r.status_code)
process_request(the_post)
break
case "notification":
the_notif = dm.fromDict(json.loads(decoded.payload))
print(
"a ",
the_notif.type,
" notification\n",
" from ",
the_notif.account.acct,
" id ", the_notif.id, "\n",
" created at ", the_notif.created_at,
"\n===>\n"
)
note_to_post = {
"status": f"[@{the_notif.account.acct}](the_notif.account.url) We come to love not by finding a perfect person, but by learning to see an imperfect person perfectly.\n ~ Sam Keen ~ ",
"in_reply_to_id": the_notif.id,
"visibility": "mutuals_only",
"content_type": "text/markdown"
}
r = requests.post(
post_uri,
headers=headers,
json=note_to_post
)
print(r.reason, r.status_code)
process_request(the_post)
case _:
print('unknown event: ', decoded.event, "\n",
" with payload beginning ",
decoded.payload[1:100])
decoded.payload[1:150])
except asyncio.exceptions.CancelledError:
print('keyboard interrupt received ... exiting')
@ -129,3 +121,5 @@ async def main():
if __name__ == "__main__":
asyncio.run(hello(ws_uri))