FMN_bot/src/fedi_api.py

148 lines
4.1 KiB
Python
Raw Normal View History

from config import instance
2022-09-05 13:49:48 +00:00
import time
2022-08-31 10:20:49 +00:00
import json
import requests
2022-09-06 23:55:12 +00:00
from loguru import logger
2022-08-31 10:20:49 +00:00
instance_point = f"https://{instance}/api/v1"
with open(".auth", mode='rt') as auth:
tkn = auth.read().replace('\n', '')
2022-11-16 01:42:54 +00:00
s = requests.Session()
s.headers.update({
"Authorization": "Bearer " + tkn,
"Accept-encoding": 'gzip'
})
2022-08-31 10:20:49 +00:00
def get_notifications():
params = {
"limit": 15,
"types": ["mention"]
2022-08-31 10:20:49 +00:00
}
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.get(instance_point + "/notifications", json=params)
r.raise_for_status()
success = 1
return r.json()
except:
logger.exception('Error get notificatios')
time.sleep(30)
logger.info('Retrying get notificatios...')
2022-08-31 10:20:49 +00:00
def mark_as_read_notification(id_notification):
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.post(instance_point + f"/notifications/{id_notification}/dismiss")
r.raise_for_status()
success = 1
return r.json()
except:
logger.exception(f'Error read notification {id_notification}')
time.sleep(30)
logger.info(f'Retrying read notification {id_notification}...')
2022-08-31 10:20:49 +00:00
def get_status_context(status_id):
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.get(instance_point + f"/statuses/{status_id}/context")
r.raise_for_status()
success = 1
return r.json()
except:
logger.exception(f'Ошибка получения контекста треда {status_id}')
2022-09-05 13:44:39 +00:00
time.sleep(30)
logger.info('Повторный запрос треда...')
2022-08-31 10:20:49 +00:00
def get_status(status_id):
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.get(instance_point + f"/statuses/{status_id}")
r.raise_for_status()
success = 1
return r.json()
except:
logger.exception(f'Error get status {status_id}')
time.sleep(30)
logger.info(f'Retrying get status {status_id}')
2022-08-31 10:20:49 +00:00
2022-08-31 23:06:06 +00:00
def post_status(text, reply_to_status_id=None, poll_options=None, poll_expires=345600, attachments=None):
2022-08-31 10:20:49 +00:00
poll = None
if poll_options is not None:
poll = {
"options": poll_options,
"expires_in": poll_expires,
"multiple": True
}
params = {
"status": text,
"in_reply_to_id": reply_to_status_id,
"visibility": "unlisted",
"content_type": "text/plain",
2022-08-31 23:06:06 +00:00
"language": "ru",
2022-08-31 10:20:49 +00:00
"poll": poll
}
2022-08-31 23:06:06 +00:00
if attachments:
params['media_ids'] = attachments
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.post(instance_point + "/statuses", json=params)
r.raise_for_status()
success = 1
return r.json()
except:
logger.exception('Error send status, retrying...')
time.sleep(5)
2022-08-31 10:20:49 +00:00
2022-08-31 23:06:06 +00:00
def upload_attachment(file_path):
file = {
"file": open(file_path, mode='rb')
}
params = {
"description": "Fediverse Movie Night\nВоскресенье, 21:00\nLIVE ON XXIV Production",
}
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.post(instance_point + "/media", params, files=file)
r.raise_for_status()
success = 1
return r.json()['id']
except:
logger.exception(f'Error uploading {file_path} attachment')
time.sleep(5)
logger.info(f'Retrying upload {file_path}...')
def mute_user(acct_id=str, acct=str, duration=None):
params = {
"duration": duration
}
success = 0
while success == 0:
try:
2022-11-16 01:42:54 +00:00
r = s.post(instance_point + '/accounts' + f"/{acct_id}/mute", params)
r.raise_for_status()
logger.info(f'Пользователь {acct} был заглушен на {duration} secs')
success = 1
except:
logger.exception(f'Ошибка глушения {acct}')
time.sleep(5)
logger.info(f'Повторное глушение {acct}...')