Source code for twitter

"""Twitter source code and datastore model classes.

The Twitter API is dead, and so is this code.
"""
import logging

from flask import request
from granary import twitter as gr_twitter
from granary import source as gr_source
from oauth_dropins import twitter as oauth_twitter
from webutil.flask_util import error
from webutil.util import json_dumps, json_loads

from flask_app import app
import models
import util

logger = logging.getLogger(__name__)




[docs]
class Twitter(models.Source):
  """A Twitter account.

  The key name is the username.
  """
  GR_CLASS = gr_twitter.Twitter
  OAUTH_START = oauth_twitter.Start
  SHORT_NAME = 'twitter'
  TYPE_LABELS = {
    'post': 'tweet',
    'comment': '@-reply',
    'repost': 'retweet',
    'like': 'favorite',
  }
  TRANSIENT_ERROR_HTTP_CODES = ('404',)
  CAN_LISTEN = False
  CAN_PUBLISH = False
  AUTH_MODEL = oauth_twitter.TwitterAuth
  MICROPUB_TOKEN_PROPERTY = 'token_secret'
  HAS_BLOCKS = True
  URL_CANONICALIZER = gr_twitter.Twitter.URL_CANONICALIZER
  USERNAME_KEY_ID = True



[docs]
  @staticmethod
  def new(auth_entity=None, **kwargs):
    """Creates and returns a :class:`Twitter` entity.

    Args:
      auth_entity (oauth_dropins.twitter.TwitterAuth)
      kwargs: property values
    """
    assert 'username' not in kwargs
    assert 'id' not in kwargs
    user = json_loads(auth_entity.user_json)
    gr_source = gr_twitter.Twitter(*auth_entity.access_token())
    actor = gr_source.user_to_actor(user)
    return Twitter(username=user['screen_name'],
                   auth_entity=auth_entity.key,
                   url=actor.get('url'),
                   name=actor.get('displayName'),
                   picture=actor.get('image', {}).get('url'),
                   **kwargs)





[docs]
  def silo_url(self):
    """Returns the Twitter account URL, e.g. https://twitter.com/foo."""
    return self.gr_source.user_url(self.username)





[docs]
  def label_name(self):
    """Returns the username."""
    return self.username








[docs]
  def get_like(self, activity_user_id, activity_id, like_user_id, **kwargs):
    """Returns an ActivityStreams 'like' activity object for a favorite.

    We get Twitter favorites by scraping HTML, and we only get the first page,
    which only has 25. So, use a :class:`models.Response` in the datastore
    first, if we have one, and only re-scrape HTML as a fallback.

    Args:
      activity_user_id (str): id of the user who posted the original activity
      activity_id (str): activity id
      like_user_id (str): id of the user who liked the activity
      kwargs: passed to :meth:`granary.source.Source.get_comment`
    """
    id = self.gr_source.tag_uri(f'{activity_id}_favorited_by_{like_user_id}')
    resp = models.Response.get_by_id(id)
    if resp:
      return json_loads(resp.response_json)





[docs]
  def is_private(self):
    """Returns True if this Twitter account is protected.

    * https://dev.twitter.com/rest/reference/get/users/show#highlighter_25173
    * https://support.twitter.com/articles/14016
    * https://support.twitter.com/articles/20169886
    """
    return json_loads(self.auth_entity.get().user_json).get('protected')





[docs]
  def canonicalize_url(self, url, activity=None, **kwargs):
    """Normalize ``/statuses/`` to ``/status/``.

    https://github.com/snarfed/bridgy/issues/618
    """
    url = url.replace('/statuses/', '/status/')
    return super().canonicalize_url(url, **kwargs)








[docs]
class Auth():
  """Base OAuth handler class."""



[docs]
  def start_oauth_flow(self, feature):
    """Redirects to Twitter's OAuth endpoint to start the OAuth flow.

    Args:
      feature: ``listen`` or ``publish``
    """
    features = feature.split(',') if feature else []
    for feature in features:
      if feature not in util.FEATURES:
        error(f'Unknown feature: {feature}')

    # pass explicit 'write' instead of None for publish so that oauth-dropins
    # (and tweepy) don't use signin_with_twitter ie /authorize. this works
    # around a twitter API bug: https://dev.twitter.com/discussions/21281
    access_type = 'write' if 'publish' in features else 'read'
    view = util.oauth_starter(oauth_twitter.Start, feature=feature)(
      '/twitter/add', access_type=access_type)
    return view.dispatch_request()






class Add(oauth_twitter.Callback, Auth):
  def finish(self, auth_entity, state=None):
    util.maybe_add_or_delete_source(Twitter, auth_entity, state)




[docs]
class Start(oauth_twitter.Start, Auth):
  """Custom OAuth start handler that uses ``access_type=read`` for ``state=listen``.

  Tweepy converts access_type to ``x_auth_access_type`` for Twitter's
  oauth/request_token endpoint. Details:
  https://dev.twitter.com/docs/api/1/post/oauth/request_token
  """
  def dispatch_request(self):
    return self.start_oauth_flow(request.form['feature'])




app.add_url_rule('/twitter/start', view_func=Start.as_view('twitter_start', '/twitter/add'), methods=['POST'])
app.add_url_rule('/twitter/add', view_func=Add.as_view('twitter_add', 'unused'))
app.add_url_rule('/twitter/delete/finish', view_func=oauth_twitter.Callback.as_view('twitter_delete_finish', '/delete/finish'))
app.add_url_rule('/twitter/publish/start', view_func=oauth_twitter.Start.as_view('twitter_publish_finish', '/publish/twitter/finish'), methods=['POST'])