"""Reddit source code and datastore model classes."""
from granary import reddit as gr_reddit
from granary import source as gr_source
from oauth_dropins import reddit as oauth_reddit
from webutil.util import json_dumps, json_loads
from prawcore.exceptions import NotFound
from flask_app import app
import models
import util
[docs]
class Reddit(models.Source):
"""A Reddit account.
The key name is the username.
"""
GR_CLASS = gr_reddit.Reddit
OAUTH_START = oauth_reddit.Start
SHORT_NAME = 'reddit'
TYPE_LABELS = {
'post': 'submission',
'comment': 'comment',
}
CAN_PUBLISH = False
DISABLE_HTTP_CODES = ('401', '403')
USERNAME_KEY_ID = True
URL_CANONICALIZER = util.UrlCanonicalizer(domain=GR_CLASS.DOMAIN)
[docs]
@staticmethod
def new(auth_entity=None, **kwargs):
"""Creates and returns a :class:`Reddit` entity.
Args:
auth_entity (oauth_dropins.reddit.RedditAuth):
kwargs: property values
"""
assert 'username' not in kwargs
assert 'id' not in kwargs
user = json_loads(auth_entity.user_json)
gr_source = gr_reddit.Reddit(auth_entity.refresh_token)
return Reddit(username=user.get('name'),
auth_entity=auth_entity.key,
url=gr_source.user_url(user.get('name')),
name=user.get('name'),
picture=user.get('icon_img'),
**kwargs)
[docs]
def silo_url(self):
"""Returns the Reddit account URL, e.g. https://reddit.com/user/foo."""
return self.gr_source.user_url(self.username)
[docs]
def label_name(self):
"""Returns the username."""
return self.username
[docs]
def get_activities_response(self, *args, **kwargs):
"""Set user_id manually.
...since Reddit sometimes (always?) 400s our calls to
https://oauth.reddit.com/api/v1/me (via PRAW's Reddit.user.me() ).
"""
kwargs.setdefault('user_id', self.username)
if kwargs.get('count'):
kwargs['count'] = min(kwargs['count'], 10)
try:
return super().get_activities_response(*args, **kwargs)
except NotFound:
# this user was deleted or banned
raise models.DisableSource()
[docs]
def search_for_links(self):
"""Searches for activities with links to any of this source's web sites.
Returns:
list of dict: ActivityStreams activities
"""
urls = {util.schemeless(util.fragmentless(url), slashes=False)
for url in self.domain_urls
if not util.in_webmention_blocklist(util.domain_from_link(url))}
if not urls:
return []
# Search syntax: https://www.reddit.com/wiki/search
url_query = ' OR '.join(f'site:"{u}" OR selftext:"{u}"' for u in urls)
return self.get_activities(
search_query=url_query, group_id=gr_source.SEARCH, etag=self.last_activities_etag,
fetch_replies=False, fetch_likes=False, fetch_shares=False, count=50)
class Callback(oauth_reddit.Callback):
def finish(self, auth_entity, state=None):
util.maybe_add_or_delete_source(Reddit, auth_entity, state)
app.add_url_rule('/reddit/start',
view_func=util.oauth_starter(oauth_reddit.Start).as_view('reddit_start', '/reddit/callback'), methods=['POST'])
app.add_url_rule('/reddit/callback',
view_func=Callback.as_view('reddit_callback', 'unused to_path'))