init

Author: xwang233

.gitignore

@@ -0,0 +1,5 @@
+*.json 
+__pycache__ 
+*.ipynb 
+.ipynb_checkpoints
+*.db

DataBase.py

@@ -0,0 +1,60 @@
+import json
+from pandas.io.json import json_normalize
+import dataset
+import threading
+from numpy import int64
+
+
+def serialize_list(j):
+    jcp = {}
+    for key in j:
+        if type(j[key]) == list:
+            if len(j[key]) == 0:
+                pass
+            elif len(j[key]) == 1:
+                jcp[key] = j[key][0]
+            else:
+                for i in range(len(j[key])):
+                    jcp['{}_{}'.format(key, i)] = j[key][i]
+        elif type(j[key]) == dict:
+            jcp[key] = serialize_list(j[key])
+        else:
+            jcp[key] = j[key]
+    return jcp
+
+
+def flatten(j: 'json'):
+    j1 = serialize_list(j)
+    j2 = json_normalize(j1, sep='$').iloc[0].to_dict()
+
+    for key in j2:
+        if type(j2[key]) == int64:
+            j2[key] = j2[key].item()
+
+        if j2[key] == False:
+            j2[key] = 'False'
+        elif j2[key] == True:
+            j2[key] = 'True'
+    return j2
+
+
+class database:
+    def __init__(self, db_str, table_str, *args, **kwargs):
+        self.db = dataset.connect(db_str)
+
+        if table_str not in self.db.tables:
+            self.db.create_table(table_str, primary_id='pid')
+        self.table = self.db[table_str]
+
+    # pass the full origin response json, the database will do the formatting
+    def insert(self, j):
+        assert type(j) == dict
+
+        def insert_thread():
+            flatten_j = flatten(j)
+            self.table.insert(flatten_j)
+
+        #t = threading.Thread(target = insert_thread)
+        # t.start()
+
+        insert_thread()

FeedlyClient.py

@@ -0,0 +1,115 @@
+import requests
+import json
+import sys
+import urllib
+import time
+import random
+from datetime import datetime
+
+
+class FeedlyClient:
+    def __init__(self, file_str, db, *args, **kwargs):
+        self.file_str = file_str
+        self.db = db
+        self.prefix = 'https://cloud.feedly.com/v3'
+
+        with open(file_str, 'r') as f:
+            options = json.load(f)
+        # print(options)
+
+        self.client_id = options.get('client_id')
+        self.client_secret = options.get('client_secret')
+        self.access_token = options.get('access_token')
+        self.refresh_token = options.get('refresh_token')
+        self.id = options.get('id')
+        self.last_fetch = options.get('last_fetch')
+        self.my_stream_id1 = options.get('my_stream_id1')
+
+    def tag_fetch(self):
+        headers = self.auth_header()
+        continuation = None
+
+        total_fetched = 0
+
+        while True:
+            params = {'streamId': self.my_stream_id1,
+                      'continuation': continuation,
+                      'ranked': 'oldest',
+                      'newerThan': self.last_fetch}
+
+            res = self._get('/streams/contents',
+                            params=params, headers=headers)
+
+            total_fetched += len(res['items'])
+            current_latest = -1
+            for item in res['items']:
+                self.db.insert(item)
+
+                self.last_fetch = max(self.last_fetch, item['actionTimestamp'])
+                current_latest = max(current_latest, item['actionTimestamp'])
+
+            print('{} entries fetched, current latest {}!'.format(
+                len(res['items']), datetime.fromtimestamp(current_latest/1000)))
+
+            if 'continuation' not in res:
+                break
+            continuation = res['continuation']
+
+            time.sleep(random.randint(2, 4))
+
+        self.last_fetch += 1
+        self._config_update('last_fetch', self.last_fetch)
+        print('total {} entries fetched! latest {}'.format(
+            total_fetched, datetime.fromtimestamp(self.last_fetch/1000)))
+
+        return total_fetched
+
+    def _get(self, endpoint, params=None, headers=None):
+        path = self.prefix + endpoint
+        r = requests.get(path, params=params, headers=headers)
+
+        if r.status_code == 401:
+            self._renew_access_token()
+            headers['Authorization'] = self.auth_header()['Authorization']
+            return self._get(endpoint, params, headers)
+
+        try:
+            r.raise_for_status()
+        except:
+            print(r.json(), flush=True, file=sys.stderr)
+            raise
+
+        return r.json()
+
+    def auth_header(self):
+        return {'Authorization': 'Bearer {}'.format(self.access_token)}
+
+    def _renew_access_token(self):
+        data = {
+            'refresh_token': self.refresh_token,
+            'client_id': self.client_id,
+            'client_secret': self.client_secret,
+            'grant_type': 'refresh_token'
+        }
+
+        r = requests.post(self.prefix+'/auth/token', data=data)
+
+        try:
+            r.raise_for_status()
+        except:
+            print(r.json(), flush=True, file=sys.stderr)
+            raise
+
+        jr = r.json()
+        self.access_token = jr['access_token']
+        self._config_update('access_token', self.access_token)
+
+        print('access_token is successfully updated',
+              flush=True, file=sys.stderr)
+
+    def _config_update(self, entry, updated_value):
+        with open(self.file_str, 'r') as f:
+            options = json.load(f)
+        options[entry] = updated_value
+        with open(self.file_str, 'w') as f:
+            json.dump(options, f)

main.py

@@ -0,0 +1,20 @@
+import traceback
+import json
+
+import FeedlyClient
+import DataBase
+
+
+def main():
+    db = DataBase.database('sqlite:///feedly.db', 'feedly')
+
+    fdc = FeedlyClient.FeedlyClient('config.json', db)
+
+    fdc.tag_fetch()
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except:
+        traceback.print_exc()