from config import instance import time import json import requests import logging logger = logging.getLogger('fedi_api') 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"] } r = requests.get(instance_point + "/notifications", json=params, headers=headers) 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) if r.status_code == 200: return r.json() else: http_code = r.status_code logger.error(f'Ошибка получения контекста треда {status_id}: {http_code}') logger.error(str(r.headers)) 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() def get_status(status_id): r = requests.get(instance_point + f"/statuses/{status_id}", headers=headers) return r.json() def post_status(text, reply_to_status_id=None, poll_options=None, poll_expires=345600, attachments=None): 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", "language": "ru", "poll": poll } if attachments: params['media_ids'] = attachments r = requests.post(instance_point + "/statuses", json=params, headers=headers) return r.json() 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}')