mirror_maker/ebooks
1
0
Fork 0
mirror of https://git.kiwifarms.net/mint/ebooks.git synced 2024-10-22 23:29:06 +00:00
Code Issues Projects Releases Packages Wiki Activity

Replying to arbitrary posts by command (@bot reply <flakeid>)

Browse source
This commit is contained in:
Mint 2022-10-21 07:30:09 +03:00
parent 6fa9ec72d8
commit 8e87ef927d
2 changed files with 30 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pleroma.py
View file

@ -124,6 +124,11 @@ class Pleroma:
id = self._unpack_id(id) id = self._unpack_id(id)
return await self.request('POST', f'/api/v1/statuses/{id}/unpin') return await self.request('POST', f'/api/v1/statuses/{id}/unpin')
async def get_status(self, id):
id = self._unpack_id(id)
status = (await self.request('GET', f'/api/v1/statuses/{id}'))
return status
async def stream(self, stream_name, *, target_event_type=None): async def stream(self, stream_name, *, target_event_type=None):
async with self._session.ws_connect( async with self._session.ws_connect(
self.api_base_url + f'/api/v1/streaming?stream={stream_name}&access_token={self.access_token}' self.api_base_url + f'/api/v1/streaming?stream={stream_name}&access_token={self.access_token}'

36
reply.py
View file

@ -32,8 +32,11 @@ class ReplyBot:
return return
content = self.extract_toot(notification['status']['content'], self.cfg) content = self.extract_toot(notification['status']['content'], self.cfg)
if content in {'pin', 'unpin'}: if content.startswith('reply '):
await self.process_command(context, notification, content) command = content.split(' ')
await self.process_command(context, notification, command[0], command[1])
elif content in {'pin', 'unpin'}:
await self.process_command(context, notification, content, '')
else: else:
await self.reply(notification) await self.reply(notification)
@ -48,7 +51,7 @@ class ReplyBot:
return False return False
async def process_command(self, context, notification, command): async def process_command(self, context, notification, command, argument):
post_id = notification['status']['id'] post_id = notification['status']['id']
if notification['account']['id'] not in self.follows: # this user is unauthorized if notification['account']['id'] not in self.follows: # this user is unauthorized
await self.pleroma.react(post_id, '❌') await self.pleroma.react(post_id, '❌')
@ -60,7 +63,13 @@ class ReplyBot:
target_post_id = post['id'] target_post_id = post['id']
try: try:
await (self.pleroma.pin if command == 'pin' else self.pleroma.unpin)(target_post_id) if command in ('pin', 'unpin'):
await (self.pleroma.pin if command == 'pin' else self.pleroma.unpin)(target_post_id)
elif command == 'reply':
toot = await utils.make_post(self.cfg)
toot = self.cleanup_toot(toot, self.cfg)
status = await self.pleroma.get_status(argument)
await self.pleroma.reply(status, toot, cw=self.cfg['cw'])
except pleroma.BadRequest as exc: except pleroma.BadRequest as exc:
async with anyio.create_task_group() as tg: async with anyio.create_task_group() as tg:
tg.start_soon(self.pleroma.react, post_id, '❌') tg.start_soon(self.pleroma.react, post_id, '❌')
@ -69,19 +78,24 @@ class ReplyBot:
await self.pleroma.react(post_id, '✅') await self.pleroma.react(post_id, '✅')
async def reply(self, notification): async def reply(self, notification):
PAIRED_PUNCTUATION = re.compile(r"[{}]".format(re.escape('[](){}"‘’“”«»„'))) toot = await utils.make_post(self.cfg)
toot = await utils.make_post(self.cfg) # generate a toot toot = self.cleanup_toot(toot, self.cfg)
if self.cfg['strip_paired_punctuation']:
toot = PAIRED_PUNCTUATION.sub("", toot)
toot = toot.replace("@", "@\u200b") # sanitize mentions
toot = utils.remove_mentions(self.cfg, toot)
await self.pleroma.reply(notification['status'], toot, cw=self.cfg['cw']) await self.pleroma.reply(notification['status'], toot, cw=self.cfg['cw'])
@staticmethod
def cleanup_toot(text, cfg):
PAIRED_PUNCTUATION = re.compile(r"[{}]".format(re.escape('[](){}"‘’“”«»„')))
if cfg['strip_paired_punctuation']:
text = PAIRED_PUNCTUATION.sub("", text)
text = text.replace("@", "@\u200b") # sanitize mentions
text = utils.remove_mentions(cfg, text)
return text
@staticmethod @staticmethod
def extract_toot(toot, cfg): def extract_toot(toot, cfg):
text = utils.extract_post_content(toot) text = utils.extract_post_content(toot)
text = re.sub(r"^@\S+\s", r"", text) # remove the initial mention text = re.sub(r"^@\S+\s", r"", text) # remove the initial mention
text = text.lower() # treat text as lowercase for easier keyword matching (if this bot uses it) # text = text.lower() # treat text as lowercase for easier keyword matching (if this bot uses it)
return text return text
async def amain(): async def amain():