682 lines
24 KiB
Python
682 lines
24 KiB
Python
import base64
|
|
import requests
|
|
from .imgur.models.tag import Tag
|
|
from .imgur.models.album import Album
|
|
from .imgur.models.image import Image
|
|
from .imgur.models.account import Account
|
|
from .imgur.models.comment import Comment
|
|
from .imgur.models.tag_vote import TagVote
|
|
from .helpers.error import ImgurClientError
|
|
from .helpers.format import build_notification
|
|
from .helpers.format import format_comment_tree
|
|
from .helpers.format import build_notifications
|
|
from .imgur.models.conversation import Conversation
|
|
from .helpers.error import ImgurClientRateLimitError
|
|
from .helpers.format import build_gallery_images_and_albums
|
|
from .imgur.models.custom_gallery import CustomGallery
|
|
from .imgur.models.account_settings import AccountSettings
|
|
|
|
API_URL = 'https://api.imgur.com/'
|
|
MASHAPE_URL = 'https://imgur-apiv3.p.mashape.com/'
|
|
|
|
|
|
class AuthWrapper(object):
|
|
def __init__(self, access_token, refresh_token, client_id, client_secret):
|
|
self.current_access_token = access_token
|
|
|
|
if refresh_token is None:
|
|
raise TypeError('A refresh token must be provided')
|
|
|
|
self.refresh_token = refresh_token
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
|
|
def get_refresh_token(self):
|
|
return self.refresh_token
|
|
|
|
def get_current_access_token(self):
|
|
return self.current_access_token
|
|
|
|
def refresh(self):
|
|
data = {
|
|
'refresh_token': self.refresh_token,
|
|
'client_id': self.client_id,
|
|
'client_secret': self.client_secret,
|
|
'grant_type': 'refresh_token'
|
|
}
|
|
|
|
url = API_URL + 'oauth2/token'
|
|
|
|
response = requests.post(url, data=data)
|
|
|
|
if response.status_code != 200:
|
|
raise ImgurClientError('Error refreshing access token!', response.status_code)
|
|
|
|
response_data = response.json()
|
|
self.current_access_token = response_data['access_token']
|
|
|
|
|
|
class ImgurClient(object):
|
|
allowed_album_fields = {
|
|
'ids', 'title', 'description', 'privacy', 'layout', 'cover'
|
|
}
|
|
|
|
allowed_advanced_search_fields = {
|
|
'q_all', 'q_any', 'q_exactly', 'q_not', 'q_type', 'q_size_px'
|
|
}
|
|
|
|
allowed_account_fields = {
|
|
'bio', 'public_images', 'messaging_enabled', 'album_privacy', 'accepted_gallery_terms', 'username'
|
|
}
|
|
|
|
allowed_image_fields = {
|
|
'album', 'name', 'title', 'description'
|
|
}
|
|
|
|
def __init__(self, client_id, client_secret, access_token=None, refresh_token=None, mashape_key=None):
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
self.auth = None
|
|
self.mashape_key = mashape_key
|
|
|
|
if refresh_token is not None:
|
|
self.auth = AuthWrapper(access_token, refresh_token, client_id, client_secret)
|
|
|
|
self.credits = self.get_credits()
|
|
|
|
def set_user_auth(self, access_token, refresh_token):
|
|
self.auth = AuthWrapper(access_token, refresh_token, self.client_id, self.client_secret)
|
|
|
|
def get_client_id(self):
|
|
return self.client_id
|
|
|
|
def get_credits(self):
|
|
return self.make_request('GET', 'credits', None, True)
|
|
|
|
def get_auth_url(self, response_type='pin'):
|
|
return '%soauth2/authorize?client_id=%s&response_type=%s' % (API_URL, self.client_id, response_type)
|
|
|
|
def authorize(self, response, grant_type='pin'):
|
|
return self.make_request('POST', 'oauth2/token', {
|
|
'client_id': self.client_id,
|
|
'client_secret': self.client_secret,
|
|
'grant_type': grant_type,
|
|
'code' if grant_type == 'authorization_code' else grant_type: response
|
|
}, True)
|
|
|
|
def prepare_headers(self, force_anon=False):
|
|
headers = {}
|
|
if force_anon or self.auth is None:
|
|
if self.client_id is None:
|
|
raise ImgurClientError('Client credentials not found!')
|
|
else:
|
|
headers['Authorization'] = 'Client-ID %s' % self.get_client_id()
|
|
else:
|
|
headers['Authorization'] = 'Bearer %s' % self.auth.get_current_access_token()
|
|
|
|
if self.mashape_key is not None:
|
|
headers['X-Mashape-Key'] = self.mashape_key
|
|
|
|
return headers
|
|
|
|
|
|
def make_request(self, method, route, data=None, force_anon=False):
|
|
method = method.lower()
|
|
method_to_call = getattr(requests, method)
|
|
|
|
header = self.prepare_headers(force_anon)
|
|
url = (MASHAPE_URL if self.mashape_key is not None else API_URL) + ('3/%s' % route if 'oauth2' not in route else route)
|
|
|
|
if method in ('delete', 'get'):
|
|
response = method_to_call(url, headers=header, params=data, data=data)
|
|
else:
|
|
response = method_to_call(url, headers=header, data=data)
|
|
|
|
if response.status_code == 403 and self.auth is not None:
|
|
self.auth.refresh()
|
|
header = self.prepare_headers()
|
|
if method in ('delete', 'get'):
|
|
response = method_to_call(url, headers=header, params=data, data=data)
|
|
else:
|
|
response = method_to_call(url, headers=header, data=data)
|
|
|
|
self.credits = {
|
|
'UserLimit': response.headers.get('X-RateLimit-UserLimit'),
|
|
'UserRemaining': response.headers.get('X-RateLimit-UserRemaining'),
|
|
'UserReset': response.headers.get('X-RateLimit-UserReset'),
|
|
'ClientLimit': response.headers.get('X-RateLimit-ClientLimit'),
|
|
'ClientRemaining': response.headers.get('X-RateLimit-ClientRemaining')
|
|
}
|
|
|
|
# Rate-limit check
|
|
if response.status_code == 429:
|
|
raise ImgurClientRateLimitError()
|
|
|
|
try:
|
|
response_data = response.json()
|
|
except:
|
|
raise ImgurClientError('JSON decoding of response failed.')
|
|
|
|
if 'data' in response_data and isinstance(response_data['data'], dict) and 'error' in response_data['data']:
|
|
raise ImgurClientError(response_data['data']['error'], response.status_code)
|
|
|
|
return response_data['data'] if 'data' in response_data else response_data
|
|
|
|
def validate_user_context(self, username):
|
|
if username == 'me' and self.auth is None:
|
|
raise ImgurClientError('\'me\' can only be used in the authenticated context.')
|
|
|
|
def logged_in(self):
|
|
if self.auth is None:
|
|
raise ImgurClientError('Must be logged in to complete request.')
|
|
|
|
# Account-related endpoints
|
|
def get_account(self, username):
|
|
self.validate_user_context(username)
|
|
account_data = self.make_request('GET', 'account/%s' % username)
|
|
|
|
return Account(
|
|
account_data['id'],
|
|
account_data['url'],
|
|
account_data['bio'],
|
|
account_data['reputation'],
|
|
account_data['created'],
|
|
account_data['pro_expiration'],
|
|
)
|
|
|
|
def get_gallery_favorites(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
gallery_favorites = self.make_request('GET', 'account/%s/gallery_favorites/%d' % (username, page))
|
|
|
|
return build_gallery_images_and_albums(gallery_favorites)
|
|
|
|
def get_account_favorites(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
favorites = self.make_request('GET', 'account/%s/favorites/%d' % (username, page))
|
|
|
|
return build_gallery_images_and_albums(favorites)
|
|
|
|
def get_account_submissions(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
submissions = self.make_request('GET', 'account/%s/submissions/%d' % (username, page))
|
|
|
|
return build_gallery_images_and_albums(submissions)
|
|
|
|
def get_account_settings(self, username):
|
|
self.logged_in()
|
|
settings = self.make_request('GET', 'account/%s/settings' % username)
|
|
|
|
return AccountSettings(
|
|
settings['email'],
|
|
settings['high_quality'],
|
|
settings['public_images'],
|
|
settings['album_privacy'],
|
|
settings['pro_expiration'],
|
|
settings['accepted_gallery_terms'],
|
|
settings['active_emails'],
|
|
settings['messaging_enabled'],
|
|
settings['blocked_users']
|
|
)
|
|
|
|
def change_account_settings(self, username, fields):
|
|
post_data = {setting: fields[setting] for setting in set(self.allowed_account_fields).intersection(fields.keys())}
|
|
return self.make_request('POST', 'account/%s/settings' % username, post_data)
|
|
|
|
def get_email_verification_status(self, username):
|
|
self.logged_in()
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/verifyemail' % username)
|
|
|
|
def send_verification_email(self, username):
|
|
self.logged_in()
|
|
self.validate_user_context(username)
|
|
return self.make_request('POST', 'account/%s/verifyemail' % username)
|
|
|
|
def get_account_albums(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
|
|
albums = self.make_request('GET', 'account/%s/albums/%d' % (username, page))
|
|
return [Album(album) for album in albums]
|
|
|
|
def get_account_album_ids(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/albums/ids/%d' % (username, page))
|
|
|
|
def get_account_album_count(self, username):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/albums/count' % username)
|
|
|
|
def get_account_comments(self, username, sort='newest', page=0):
|
|
self.validate_user_context(username)
|
|
comments = self.make_request('GET', 'account/%s/comments/%s/%s' % (username, sort, page))
|
|
|
|
return [Comment(comment) for comment in comments]
|
|
|
|
def get_account_comment_ids(self, username, sort='newest', page=0):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/comments/ids/%s/%s' % (username, sort, page))
|
|
|
|
def get_account_comment_count(self, username):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/comments/count' % username)
|
|
|
|
def get_account_images(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
images = self.make_request('GET', 'account/%s/images/%d' % (username, page))
|
|
|
|
return [Image(image) for image in images]
|
|
|
|
def get_account_image_ids(self, username, page=0):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/images/ids/%d' % (username, page))
|
|
|
|
def get_account_images_count(self, username):
|
|
self.validate_user_context(username)
|
|
return self.make_request('GET', 'account/%s/images/count' % username)
|
|
|
|
# Album-related endpoints
|
|
def get_album(self, album_id):
|
|
album = self.make_request('GET', 'album/%s' % album_id)
|
|
return Album(album)
|
|
|
|
def get_album_images(self, album_id):
|
|
images = self.make_request('GET', 'album/%s/images' % album_id)
|
|
return [Image(image) for image in images]
|
|
|
|
def create_album(self, fields):
|
|
post_data = {field: fields[field] for field in set(self.allowed_album_fields).intersection(fields.keys())}
|
|
|
|
if 'ids' in post_data:
|
|
self.logged_in()
|
|
|
|
return self.make_request('POST', 'album', data=post_data)
|
|
|
|
def update_album(self, album_id, fields):
|
|
post_data = {field: fields[field] for field in set(self.allowed_album_fields).intersection(fields.keys())}
|
|
|
|
if isinstance(post_data['ids'], list):
|
|
post_data['ids'] = ','.join(post_data['ids'])
|
|
|
|
return self.make_request('POST', 'album/%s' % album_id, data=post_data)
|
|
|
|
def album_delete(self, album_id):
|
|
return self.make_request('DELETE', 'album/%s' % album_id)
|
|
|
|
def album_favorite(self, album_id):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'album/%s/favorite' % album_id)
|
|
|
|
def album_set_images(self, album_id, ids):
|
|
if isinstance(ids, list):
|
|
ids = ','.join(ids)
|
|
|
|
return self.make_request('POST', 'album/%s/' % album_id, {'ids': ids})
|
|
|
|
def album_add_images(self, album_id, ids):
|
|
if isinstance(ids, list):
|
|
ids = ','.join(ids)
|
|
|
|
return self.make_request('POST', 'album/%s/add' % album_id, {'ids': ids})
|
|
|
|
def album_remove_images(self, album_id, ids):
|
|
if isinstance(ids, list):
|
|
ids = ','.join(ids)
|
|
|
|
return self.make_request('DELETE', 'album/%s/remove_images' % album_id, {'ids': ids})
|
|
|
|
# Comment-related endpoints
|
|
def get_comment(self, comment_id):
|
|
comment = self.make_request('GET', 'comment/%d' % comment_id)
|
|
return Comment(comment)
|
|
|
|
def delete_comment(self, comment_id):
|
|
self.logged_in()
|
|
return self.make_request('DELETE', 'comment/%d' % comment_id)
|
|
|
|
def get_comment_replies(self, comment_id):
|
|
replies = self.make_request('GET', 'comment/%d/replies' % comment_id)
|
|
return format_comment_tree(replies)
|
|
|
|
def post_comment_reply(self, comment_id, image_id, comment):
|
|
self.logged_in()
|
|
data = {
|
|
'image_id': image_id,
|
|
'comment': comment
|
|
}
|
|
|
|
return self.make_request('POST', 'comment/%d' % comment_id, data)
|
|
|
|
def comment_vote(self, comment_id, vote='up'):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'comment/%d/vote/%s' % (comment_id, vote))
|
|
|
|
def comment_report(self, comment_id):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'comment/%d/report' % comment_id)
|
|
|
|
# Custom Gallery Endpoints
|
|
def get_custom_gallery(self, gallery_id, sort='viral', window='week', page=0):
|
|
gallery = self.make_request('GET', 'g/%s/%s/%s/%s' % (gallery_id, sort, window, page))
|
|
return CustomGallery(
|
|
gallery['id'],
|
|
gallery['name'],
|
|
gallery['datetime'],
|
|
gallery['account_url'],
|
|
gallery['link'],
|
|
gallery['tags'],
|
|
gallery['item_count'],
|
|
gallery['items']
|
|
)
|
|
|
|
def get_user_galleries(self):
|
|
self.logged_in()
|
|
galleries = self.make_request('GET', 'g')
|
|
|
|
return [CustomGallery(
|
|
gallery['id'],
|
|
gallery['name'],
|
|
gallery['datetime'],
|
|
gallery['account_url'],
|
|
gallery['link'],
|
|
gallery['tags']
|
|
) for gallery in galleries]
|
|
|
|
def create_custom_gallery(self, name, tags=None):
|
|
self.logged_in()
|
|
data = {'name': name}
|
|
|
|
if tags:
|
|
data['tags'] = ','.join(tags)
|
|
|
|
gallery = self.make_request('POST', 'g', data)
|
|
|
|
return CustomGallery(
|
|
gallery['id'],
|
|
gallery['name'],
|
|
gallery['datetime'],
|
|
gallery['account_url'],
|
|
gallery['link'],
|
|
gallery['tags']
|
|
)
|
|
|
|
def custom_gallery_update(self, gallery_id, name):
|
|
self.logged_in()
|
|
data = {
|
|
'id': gallery_id,
|
|
'name': name
|
|
}
|
|
|
|
gallery = self.make_request('POST', 'g/%s' % gallery_id, data)
|
|
|
|
return CustomGallery(
|
|
gallery['id'],
|
|
gallery['name'],
|
|
gallery['datetime'],
|
|
gallery['account_url'],
|
|
gallery['link'],
|
|
gallery['tags']
|
|
)
|
|
|
|
def custom_gallery_add_tags(self, gallery_id, tags):
|
|
self.logged_in()
|
|
|
|
if tags:
|
|
data = {'tags': ','.join(tags)}
|
|
else:
|
|
raise ImgurClientError('tags must not be empty!')
|
|
|
|
return self.make_request('PUT', 'g/%s/add_tags' % gallery_id, data)
|
|
|
|
def custom_gallery_remove_tags(self, gallery_id, tags):
|
|
self.logged_in()
|
|
|
|
if tags:
|
|
data = {'tags': ','.join(tags)}
|
|
else:
|
|
raise ImgurClientError('tags must not be empty!')
|
|
|
|
return self.make_request('DELETE', 'g/%s/remove_tags' % gallery_id, data)
|
|
|
|
def custom_gallery_delete(self, gallery_id):
|
|
self.logged_in()
|
|
return self.make_request('DELETE', 'g/%s' % gallery_id)
|
|
|
|
def filtered_out_tags(self):
|
|
self.logged_in()
|
|
return self.make_request('GET', 'g/filtered_out')
|
|
|
|
def block_tag(self, tag):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'g/block_tag', data={'tag': tag})
|
|
|
|
def unblock_tag(self, tag):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'g/unblock_tag', data={'tag': tag})
|
|
|
|
# Gallery-related endpoints
|
|
def gallery(self, section='hot', sort='viral', page=0, window='day', show_viral=True):
|
|
if section == 'top':
|
|
response = self.make_request('GET', 'gallery/%s/%s/%s/%d?showViral=%s'
|
|
% (section, sort, window, page, str(show_viral).lower()))
|
|
else:
|
|
response = self.make_request('GET', 'gallery/%s/%s/%d?showViral=%s'
|
|
% (section, sort, page, str(show_viral).lower()))
|
|
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def memes_subgallery(self, sort='viral', page=0, window='week'):
|
|
if sort == 'top':
|
|
response = self.make_request('GET', 'g/memes/%s/%s/%d' % (sort, window, page))
|
|
else:
|
|
response = self.make_request('GET', 'g/memes/%s/%d' % (sort, page))
|
|
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def memes_subgallery_image(self, item_id):
|
|
item = self.make_request('GET', 'g/memes/%s' % item_id)
|
|
return build_gallery_images_and_albums(item)
|
|
|
|
def subreddit_gallery(self, subreddit, sort='time', window='week', page=0):
|
|
if sort == 'top':
|
|
response = self.make_request('GET', 'gallery/r/%s/%s/%s/%d' % (subreddit, sort, window, page))
|
|
else:
|
|
response = self.make_request('GET', 'gallery/r/%s/%s/%d' % (subreddit, sort, page))
|
|
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def subreddit_image(self, subreddit, image_id):
|
|
item = self.make_request('GET', 'gallery/r/%s/%s' % (subreddit, image_id))
|
|
return build_gallery_images_and_albums(item)
|
|
|
|
def gallery_tag(self, tag, sort='viral', page=0, window='week'):
|
|
if sort == 'top':
|
|
response = self.make_request('GET', 'gallery/t/%s/%s/%s/%d' % (tag, sort, window, page))
|
|
else:
|
|
response = self.make_request('GET', 'gallery/t/%s/%s/%d' % (tag, sort, page))
|
|
|
|
return Tag(
|
|
response['name'],
|
|
response['followers'],
|
|
response['total_items'],
|
|
response['following'],
|
|
response['items']
|
|
)
|
|
|
|
def gallery_tag_image(self, tag, item_id):
|
|
item = self.make_request('GET', 'gallery/t/%s/%s' % (tag, item_id))
|
|
return build_gallery_images_and_albums(item)
|
|
|
|
def gallery_item_tags(self, item_id):
|
|
response = self.make_request('GET', 'gallery/%s/tags' % item_id)
|
|
|
|
return [TagVote(
|
|
item['ups'],
|
|
item['downs'],
|
|
item['name'],
|
|
item['author']
|
|
) for item in response['tags']]
|
|
|
|
def gallery_tag_vote(self, item_id, tag, vote):
|
|
self.logged_in()
|
|
response = self.make_request('POST', 'gallery/%s/vote/tag/%s/%s' % (item_id, tag, vote))
|
|
return response
|
|
|
|
def gallery_search(self, q, advanced=None, sort='time', window='all', page=0):
|
|
if advanced:
|
|
data = {field: advanced[field]
|
|
for field in set(self.allowed_advanced_search_fields).intersection(advanced.keys())}
|
|
else:
|
|
data = {'q': q}
|
|
|
|
response = self.make_request('GET', 'gallery/search/%s/%s/%s' % (sort, window, page), data)
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def gallery_random(self, page=0):
|
|
response = self.make_request('GET', 'gallery/random/random/%d' % page)
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def share_on_imgur(self, item_id, title, terms=0):
|
|
self.logged_in()
|
|
data = {
|
|
'title': title,
|
|
'terms': terms
|
|
}
|
|
|
|
return self.make_request('POST', 'gallery/%s' % item_id, data)
|
|
|
|
def remove_from_gallery(self, item_id):
|
|
self.logged_in()
|
|
return self.make_request('DELETE', 'gallery/%s' % item_id)
|
|
|
|
def gallery_item(self, item_id):
|
|
response = self.make_request('GET', 'gallery/%s' % item_id)
|
|
return build_gallery_images_and_albums(response)
|
|
|
|
def report_gallery_item(self, item_id):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'gallery/%s/report' % item_id)
|
|
|
|
def gallery_item_vote(self, item_id, vote='up'):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'gallery/%s/vote/%s' % (item_id, vote))
|
|
|
|
def gallery_item_comments(self, item_id, sort='best'):
|
|
response = self.make_request('GET', 'gallery/%s/comments/%s' % (item_id, sort))
|
|
return format_comment_tree(response)
|
|
|
|
def gallery_comment(self, item_id, comment):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'gallery/%s/comment' % item_id, {'comment': comment})
|
|
|
|
def gallery_comment_ids(self, item_id):
|
|
return self.make_request('GET', 'gallery/%s/comments/ids' % item_id)
|
|
|
|
def gallery_comment_count(self, item_id):
|
|
return self.make_request('GET', 'gallery/%s/comments/count' % item_id)
|
|
|
|
# Image-related endpoints
|
|
def get_image(self, image_id):
|
|
image = self.make_request('GET', 'image/%s' % image_id)
|
|
return Image(image)
|
|
|
|
def upload_from_path(self, path, config=None, anon=True):
|
|
if not config:
|
|
config = dict()
|
|
|
|
fd = open(path, 'rb')
|
|
contents = fd.read()
|
|
b64 = base64.b64encode(contents)
|
|
data = {
|
|
'image': b64,
|
|
'type': 'base64',
|
|
}
|
|
data.update({meta: config[meta] for meta in set(self.allowed_image_fields).intersection(config.keys())})
|
|
fd.close()
|
|
|
|
return self.make_request('POST', 'upload', data, anon)
|
|
|
|
def upload_from_url(self, url, config=None, anon=True):
|
|
if not config:
|
|
config = dict()
|
|
|
|
data = {
|
|
'image': url,
|
|
'type': 'url',
|
|
}
|
|
|
|
data.update({meta: config[meta] for meta in set(self.allowed_image_fields).intersection(config.keys())})
|
|
return self.make_request('POST', 'upload', data, anon)
|
|
|
|
def delete_image(self, image_id):
|
|
return self.make_request('DELETE', 'image/%s' % image_id)
|
|
|
|
def favorite_image(self, image_id):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'image/%s/favorite' % image_id)
|
|
|
|
# Conversation-related endpoints
|
|
def conversation_list(self):
|
|
self.logged_in()
|
|
|
|
conversations = self.make_request('GET', 'conversations')
|
|
return [Conversation(
|
|
conversation['id'],
|
|
conversation['last_message_preview'],
|
|
conversation['datetime'],
|
|
conversation['with_account_id'],
|
|
conversation['with_account'],
|
|
conversation['message_count'],
|
|
) for conversation in conversations]
|
|
|
|
def get_conversation(self, conversation_id, page=1, offset=0):
|
|
self.logged_in()
|
|
|
|
conversation = self.make_request('GET', 'conversations/%d/%d/%d' % (conversation_id, page, offset))
|
|
return Conversation(
|
|
conversation['id'],
|
|
conversation['last_message_preview'],
|
|
conversation['datetime'],
|
|
conversation['with_account_id'],
|
|
conversation['with_account'],
|
|
conversation['message_count'],
|
|
conversation['messages'],
|
|
conversation['done'],
|
|
conversation['page']
|
|
)
|
|
|
|
def create_message(self, recipient, body):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'conversations/%s' % recipient, {'body': body})
|
|
|
|
def delete_conversation(self, conversation_id):
|
|
self.logged_in()
|
|
return self.make_request('DELETE', 'conversations/%d' % conversation_id)
|
|
|
|
def report_sender(self, username):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'conversations/report/%s' % username)
|
|
|
|
def block_sender(self, username):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'conversations/block/%s' % username)
|
|
|
|
# Notification-related endpoints
|
|
def get_notifications(self, new=True):
|
|
self.logged_in()
|
|
response = self.make_request('GET', 'notification', {'new': str(new).lower()})
|
|
return build_notifications(response)
|
|
|
|
def get_notification(self, notification_id):
|
|
self.logged_in()
|
|
response = self.make_request('GET', 'notification/%d' % notification_id)
|
|
return build_notification(response)
|
|
|
|
def mark_notifications_as_read(self, notification_ids):
|
|
self.logged_in()
|
|
return self.make_request('POST', 'notification', ','.join(notification_ids))
|
|
|
|
# Memegen-related endpoints
|
|
def default_memes(self):
|
|
response = self.make_request('GET', 'memegen/defaults')
|
|
return [Image(meme) for meme in response]
|