FMN_bot/src/fedi_api.py

97 lines
2.8 KiB
Python
Raw Normal View History

from config import instance
2022-09-05 16:49:48 +03:00
import time
2022-08-31 13:20:49 +03:00
import json
import requests
import logging
logger = logging.getLogger('fedi_api')
2022-08-31 13:20:49 +03:00
instance_point = f"https://{instance}/api/v1"
with open(".auth", mode='rt') as auth:
tkn = auth.read().replace('\n', '')
headers= {
"Authorization": "Bearer " + tkn
}
def get_notifications():
params = {
"limit": 15,
"types": ["mention"]
2022-08-31 13:20:49 +03:00
}
r = requests.get(instance_point + "/notifications", json=params, headers=headers)
2022-08-31 13:20:49 +03:00
return r.json()
def mark_as_read_notification(id_notification):
r = requests.post(instance_point + f"/notifications/{id_notification}/dismiss", headers=headers)
return r.json()
def get_status_context(status_id):
r = requests.get(instance_point + f"/statuses/{status_id}/context", headers=headers)
2022-09-05 16:44:39 +03:00
if r.status_code == 200:
return r.json()
else:
http_code = r.status_code
logger.error(f'Ошибка получения контекста треда {status_id}: {http_code}')
2022-09-05 17:03:57 +03:00
logger.error(str(r.headers))
2022-09-05 16:44:39 +03:00
while r.status_code != 200:
time.sleep(30)
logger.info('Повторный запрос треда...')
r = requests.get(instance_point + f"/statuses/{status_id}/context", headers=headers)
return r.json()
2022-08-31 13:20:49 +03:00
def get_status(status_id):
r = requests.get(instance_point + f"/statuses/{status_id}", headers=headers)
return r.json()
2022-09-01 02:06:06 +03:00
def post_status(text, reply_to_status_id=None, poll_options=None, poll_expires=345600, attachments=None):
2022-08-31 13:20:49 +03: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-09-01 02:06:06 +03:00
"language": "ru",
2022-08-31 13:20:49 +03:00
"poll": poll
}
2022-09-01 02:06:06 +03:00
if attachments:
params['media_ids'] = attachments
2022-08-31 13:20:49 +03:00
r = requests.post(instance_point + "/statuses", json=params, headers=headers)
return r.json()
2022-09-01 02:06:06 +03: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",
}
r = requests.post(instance_point + "/media", params, files=file, headers=headers)
return r.json()['id']
def mute_user(acct_id=str, acct=str, duration=None):
params = {
"duration": duration
}
r = requests.post(instance_point + '/accounts' + f"/{acct_id}/mute", params, headers=headers)
if r.status_code == 200:
logger.info(f'Пользователь {acct} был заглушен на {duration} secs')
else:
logger.error(f'Ошибка глушения {r.status_code} - {acct}')