无需身份验证即可抓取Twitter前端API

twitter/setup.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# Note: To use the 'upload' functionality of this file, you must:
+#   $ pip install twine
+
+import io
+import os
+import sys
+from shutil import rmtree
+
+from setuptools import find_packages, setup, Command
+
+# Package meta-data.
+NAME = 'twitter-scraper'
+DESCRIPTION = 'Scrape the Twitter Frontend API without authentication.'
+URL = 'https://github.com/kennethreitz/twitter-scraper'
+EMAIL = 'me@kennethreitz.org'
+AUTHOR = 'Kenneth Reitz'
+VERSION = '0.2.1'
+
+# What packages are required for this module to be executed?
+REQUIRED = [
+    'requests-html'
+]
+
+# The rest you shouldn't have to touch too much :)
+# ------------------------------------------------
+# Except, perhaps the License and Trove Classifiers!
+# If you do change the License, remember to change the Trove Classifier for that!
+
+here = os.path.abspath(os.path.dirname(__file__))
+
+# Import the README and use it as the long-description.
+# Note: this will only work if 'README.rst' is present in your MANIFEST.in file!
+with io.open(os.path.join(here, 'README.rst'), encoding='utf-8') as f:
+    long_description = '\n' + f.read()
+
+
+class UploadCommand(Command):
+    """Support setup.py upload."""
+
+    description = 'Build and publish the package.'
+    user_options = []
+
+    @staticmethod
+    def status(s):
+        """Prints things in bold."""
+        print('\033[1m{0}\033[0m'.format(s))
+
+    def initialize_options(self):
+        pass
+
+    def finalize_options(self):
+        pass
+
+    def run(self):
+        try:
+            self.status('Removing previous builds…')
+            rmtree(os.path.join(here, 'dist'))
+        except OSError:
+            pass
+
+        self.status('Building Source and Wheel (universal) distribution…')
+        os.system('{0} setup.py sdist bdist_wheel --universal'.format(sys.executable))
+
+        self.status('Uploading the package to PyPi via Twine…')
+        os.system('twine upload dist/*')
+
+        sys.exit()
+
+
+# Where the magic happens:
+setup(
+    name=NAME,
+    version=VERSION,
+    description=DESCRIPTION,
+    long_description=long_description,
+    author=AUTHOR,
+    author_email=EMAIL,
+    url=URL,
+    # If your package is a single module, use this instead of 'packages':
+    py_modules=['twitter_scraper'],
+
+    # entry_points={
+    #     'console_scripts': ['mycli=mymodule:cli'],
+    # },
+    install_requires=REQUIRED,
+    include_package_data=True,
+    license='MIT',
+    classifiers=[
+        # Trove classifiers
+        # Full list: https://pypi.python.org/pypi?%3Aaction=list_classifiers
+        'License :: OSI Approved :: MIT License',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 3.6',
+        'Programming Language :: Python :: Implementation :: CPython',
+        'Programming Language :: Python :: Implementation :: PyPy'
+    ],
+    # $ setup.py publish support.
+    cmdclass={
+        'upload': UploadCommand,
+    },
+)

twitter/twitter.py

@@ -1,59 +0,0 @@
-from requests_oauthlib import OAuth1Session
-
-consumer_key = 'xxxxxxxxxxxxxxxxxxxxx'
-consumer_secret = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
-resource_owner_key = ''
-resource_owner_secret = ''
-
-# 获取请求令牌
-def twitter_get_oauth_request_token():
-    global resource_owner_key
-    global resource_owner_secret
-    request_token = OAuth1Session(client_key=consumer_key, client_secret=consumer_secret)
-    url = 'https://api.twitter.com/oauth/request_token'
-    data = request_token.get(url)
-    print(data.text)
-    data_token = str.split(data.text, '&')
-    ro_key = str.split(data_token[0], '=')
-    ro_secret = str.split(data_token[1], '=')
-    resource_owner_key = ro_key[1]
-    resource_owner_secret = ro_secret[1]
-    resource = [resource_owner_key, resource_owner_secret]
-    return resource
-
-
-# 获取访问令牌
-def twitter_get_oauth_token(verifier, ro_key, ro_secret):
-    oauth_token = OAuth1Session(client_key=consumer_key,
-                                client_secret=consumer_secret,
-                                resource_owner_key=ro_key,
-                                resource_owner_secret=ro_secret)
-    url = 'https://api.twitter.com/oauth/access_token'
-    data = {"oauth_verifier": verifier}
-    print(ro_key)
-    print(ro_secret)
-    access_token_data = oauth_token.post(url, data=data)
-    print(access_token_data.text)
-    access_token_list = str.split(access_token_data.text, '&')
-    return access_token_list
-
-
-# 获取用户数据
-def twitter_get_access_token(access_token_list):
-    access_token_key = str.split(access_token_list[0], '=')
-    access_token_secret = str.split(access_token_list[1], '=')
-    access_token_name = str.split(access_token_list[3], '=')
-    access_token_id = str.split(access_token_list[2], '=')
-    key = access_token_key[1]
-    secret = access_token_secret[1]
-    name = access_token_name[1]
-    id = access_token_id[1]
-    oauth_user = OAuth1Session(client_key=consumer_key,
-                               client_secret=consumer_secret,
-                               resource_owner_key=key,
-                               resource_owner_secret=secret)
-    url_user = 'https://api.twitter.com/1.1/account/verify_credentials.json'
-    params = {"include_email": 'true'}
-    user_data = oauth_user.get(url_user, params=params)
-    print(user_data.json())
-    return user_data.json()

twitter/twitter_scraper.py

@@ -0,0 +1,78 @@
+import re
+from requests_html import HTMLSession, HTML
+from datetime import datetime
+
+session = HTMLSession()
+
+
+def get_tweets(user, pages=25):
+    """Gets tweets for a given user, via the Twitter frontend API."""
+
+    url = f'https://twitter.com/i/profiles/show/{user}/timeline/tweets?include_available_features=1&include_entities=1&include_new_items_bar=true'
+    headers = {
+        'Accept': 'application/json, text/javascript, */*; q=0.01',
+        'Referer': f'https://twitter.com/{user}',
+        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/603.3.8 (KHTML, like Gecko) Version/10.1.2 Safari/603.3.8',
+        'X-Twitter-Active-User': 'yes',
+        'X-Requested-With': 'XMLHttpRequest'
+    }
+
+    def gen_tweets(pages):
+        r = session.get(url, headers=headers)
+
+        while pages > 0:
+            try:
+                html = HTML(html=r.json()['items_html'],
+                            url='bunk', default_encoding='utf-8')
+            except KeyError:
+                raise ValueError(
+                    f'Oops! Either "{user}" does not exist or is private.')
+
+            comma = ","
+            dot = "."
+            tweets = []
+            for tweet in html.find('.stream-item'):
+                text = tweet.find('.tweet-text')[0].full_text
+                tweetId = tweet.find(
+                    '.js-permalink')[0].attrs['data-conversation-id']
+                time = datetime.fromtimestamp(
+                    int(tweet.find('._timestamp')[0].attrs['data-time-ms'])/1000.0)
+                interactions = [x.text for x in tweet.find(
+                    '.ProfileTweet-actionCount')]
+                replies = int(interactions[0].split(" ")[0].replace(comma, "").replace(dot,""))
+                retweets = int(interactions[1].split(" ")[
+                               0].replace(comma, "").replace(dot,""))
+                likes = int(interactions[2].split(" ")[0].replace(comma, "").replace(dot,""))
+                hashtags = [hashtag_node.full_text for hashtag_node in tweet.find('.twitter-hashtag')]
+                urls = [url_node.attrs['data-expanded-url'] for url_node in tweet.find('a.twitter-timeline-link:not(.u-hidden)')]
+                photos = [photo_node.attrs['data-image-url'] for photo_node in tweet.find('.AdaptiveMedia-photoContainer')]
+                
+                videos = []
+                video_nodes = tweet.find(".PlayableMedia-player")
+                for node in video_nodes:
+                    styles = node.attrs['style'].split()
+                    for style in styles:
+                        if style.startswith('background'):
+                            tmp = style.split('/')[-1]
+                            video_id = tmp[:tmp.index('.jpg')]
+                            videos.append({'id': video_id})
+                tweets.append({'tweetId': tweetId, 'time': time, 'text': text,
+                               'replies': replies, 'retweets': retweets, 'likes': likes, 
+                               'entries': {
+                                    'hashtags': hashtags, 'urls': urls,
+                                    'photos': photos, 'videos': videos
+                                }
+                               })
+
+            last_tweet = html.find('.stream-item')[-1].attrs['data-item-id']
+
+            for tweet in tweets:
+                if tweet:
+                    tweet['text'] = re.sub('http', ' http', tweet['text'], 1)
+                    yield tweet
+
+            r = session.get(
+                url, params = {'max_position': last_tweet}, headers = headers)
+            pages += -1
+
+    yield from gen_tweets(pages)