adding followers/following and other improvements

duolingo.py

@@ -129,17 +129,17 @@ def _check_login(self):
         resp = self._make_req(self.get_user_url())
         return resp.status_code == 200
 
-    def get_user_url_by_id(self, fields=None):
+    def get_user_url_by_id(self, fields=None, user_id=None):
         if fields is None:
             fields = []
-        url = 'https://www.duolingo.com/2017-06-30/users/{}'.format(self.user_data.id)
+        url = f'https://www.duolingo.com/2017-06-30/users/{user_id or self.user_data.id}'
         fields_params = requests.utils.requote_uri(','.join(fields))
         if fields_params:
             url += '?fields={}'.format(fields_params)
         return url
 
-    def get_user_url(self):
-        return "https://duolingo.com/users/%s" % self.username
+    def get_user_url(self, username=None):
+        return "https://duolingo.com/users/%s" % (username or self.username)
 
     def set_username(self, username):
         self.username = username
@@ -221,7 +221,7 @@ def buy_streak_freeze(self):
             return True
         except AlreadyHaveStoreItemException:
             return False
-        
+
     def buy_weekend_amulet(self):
         """
         figure out the users current learning language
@@ -235,7 +235,6 @@ def buy_weekend_amulet(self):
             return True
         except AlreadyHaveStoreItemException:
             return False
-
 
     def _switch_language(self, lang):
         """
@@ -256,35 +255,60 @@ def _switch_language(self, lang):
         except ValueError:
             raise DuolingoException('Failed to switch language')
 
-    def get_data_by_user_id(self, fields=None):
+    def get_data_by_user_id(self, fields=None, user_id=None):
         """
         Get user's data from ``https://www.duolingo.com/2017-06-30/users/<user_id>``.
         """
         if fields is None:
             fields = []
-        get = self._make_req(self.get_user_url_by_id(fields))
+        get = self._make_req(self.get_user_url_by_id(fields, user_id))
         if get.status_code == 404:
             raise DuolingoException('User not found')
         else:
             return get.json()
 
-    def _get_data(self):
+    def _get_data(self, username=None):
         """
         Get user's data from ``https://www.duolingo.com/users/<username>``.
         """
-        get = self._make_req(self.get_user_url())
+        get = self._make_req(self.get_user_url(username))
         if get.status_code == 404:
             raise Exception('User not found')
         else:
             return get.json()
 
+    def load_other_user(self, username=None, user_id=None):
+        """
+        Get another user's data from ``https://www.duolingo.com/users/<username>`` and set it as the loaded username (can differ from authenticated user).
+        :return: Object with new user's data
+        :rtype: obj
+        """
+        if not user_id and not username:
+            raise Exception('Either username or user_id must be provided')
+
+        if username:
+            self.user_data = Struct(**self._get_data(username))
+            self.username = username
+        else:
+            self.user_data = Struct(**self.get_data_by_user_id(user_id=user_id))
+        return self.user_data
+
+    def get_user_id_from_username(self, username):
+        """
+        Get the userId from the given username
+        """
+        if username == self.username:
+            return self.user_data.id
+        else:
+            return Struct(**self._get_data(username)).id
+
     @staticmethod
     def _make_dict(keys, array):
         data = {}
 
         for key in keys:
             if type(array) == dict:
-                data[key] = array[key]
+                data[key] = array.get(key, None)
             else:
                 data[key] = getattr(array, key, None)
 
@@ -419,6 +443,7 @@ def get_friends(self):
         """Get user's friends."""
         for k, v in self.user_data.language_data.items():
             data = []
+            if not 'points_ranking_data' in v: continue
             for friend in v['points_ranking_data']:
                 temp = {'username': friend['username'],
                         'id': friend['id'],
@@ -428,6 +453,29 @@ def get_friends(self):
                 data.append(temp)
 
             return data
+        return []
+
+    def _get_friends(self, username=None, user_id=None, _type=None):
+        if _type not in ['followers', 'following']:
+            raise Exception(f'Type of friends must be followers or following, not [{_type}]')
+        if not user_id and not username:
+            raise Exception('Either username or user_id must be provided')
+        if not user_id:
+            user_id = self.get_user_id_from_username(username)
+
+        get = self._make_req(f'https://friends-prod.duolingo.com/users/{user_id}/{_type}?pageSize=5000')
+        if get.status_code == 404:
+            raise Exception(f'{_type} not found')
+        else:
+            return get.json()[_type]["users"]
+
+    def get_followers(self, username=None, user_id=None):
+        """Get a user's list of followers."""
+        return self._get_friends(username, user_id, "followers")
+
+    def get_following(self, username=None, user_id=None):
+        """Get the list of users followed by a user."""
+        return self._get_friends(username, user_id, "following")
 
     def get_known_words(self, lang):
         """Get a list of all words learned by user in a language."""
@@ -609,7 +657,7 @@ def get_audio_url(self, word, language_abbr=None, rand=True, voice=None):
             self._populate_voice_url_dictionary(language_abbr)
         # If no audio exists for a word, return None
         if word not in self.voice_url_dict[language_abbr]:
-            return None
+            return ""
         # Get word audio links
         word_links = list(self.voice_url_dict[language_abbr][word])
         # If a voice is specified, get that one or None
@@ -676,7 +724,7 @@ def get_related_words(self, word, language_abbr=None):
                 related_lexemes = word_data['related_lexemes']
                 return [w for w in overview['vocab_overview']
                         if w['lexeme_id'] in related_lexemes]
-
+        return []
 
     def get_word_definition_by_id(self, lexeme_id):
         """
@@ -716,7 +764,7 @@ def get_daily_xp_progress(self):
         update_cutoff = round((reported_midnight + time_discrepancy).timestamp())
 
         lessons = [lesson for lesson in daily_progress['xpGains'] if
-                lesson['time'] > update_cutoff]
+                   lesson['time'] > update_cutoff]
 
         return {
             "xp_goal": daily_progress['xpGoal'],

tests.py

@@ -5,6 +5,7 @@
 
 import duolingo
 
+
 USERNAME = os.environ.get('DUOLINGO_USER', 'ferguslongley')
 PASSWORD = os.environ.get('DUOLINGO_PASSWORD')
 USERNAME2 = os.environ.get("DUOLINGO_USER_2", "Spaniard")
@@ -18,7 +19,8 @@ def _example_word(lang):
     """
     return {
         "de": "mann",
-        "es": "hombre"
+        "es": "hombre",
+        "fr": "garçon"
     }.get(lang)
 
 
@@ -278,7 +280,7 @@ def test_get_audio_url(self):
         response = self.lingo.get_audio_url(word, self.lang)
         assert isinstance(response, str)
         response = self.lingo.get_audio_url("zz")
-        assert response is None
+        assert response == ""
 
     def test_get_word_definition_by_id(self):
         response = self.lingo.get_word_definition_by_id("52383869a8feb3e5cf83dbf7fab9a018")
@@ -301,6 +303,7 @@ def setUpClass(cls):
         cls.lingo = duolingo.Duolingo(USERNAME, PASSWORD)
         cls.lingo.set_username(USERNAME2)
         cls.lang = cls.lingo.user_data.learning_language
+        cls.USERNAME_TEST = "duo"
 
     def test_get_daily_xp_progress(self):
         try:
@@ -325,6 +328,46 @@ def test_get_related_words(self):
         except duolingo.OtherUserException as e:
             assert "Vocab cannot be listed when the user has been switched" in str(e)
 
+    def test_get_user_id_from_username(self):
+        user_id = self.lingo.get_user_id_from_username(USERNAME2)
+        print(f'userid={user_id}')
+        assert isinstance(user_id, int), "user_id should be a number"
+
+    def test_load_other_user(self):
+        other = self.lingo.load_other_user(USERNAME2)
+        assert other == self.lingo.user_data, "should return obj with user_data"
+        assert self.lingo.username == USERNAME2, "username from other user should have been updated"
+        response = self.lingo.get_user_info()
+        assert isinstance(response, dict)
+        assert "avatar" in response
+        assert "id" in response
+        assert "location" in response
+        assert "learning_language_string" in response
+
+    def test_get_followers(self):
+        try: self.lingo.get_followers()
+        except Exception as e:
+            assert "Either username or user_id must be provided" in str(e)
+        f = self.lingo.get_followers(self.USERNAME_TEST)
+        assert isinstance(f, list)
+        assert len(f) > 0
+        assert "userId" in f[0]
+        user_id = self.lingo.get_user_id_from_username(self.USERNAME_TEST)
+        f_id = self.lingo.get_followers(user_id=user_id)
+        assert f == f_id
+
+    def test_get_following(self):
+        try: self.lingo.get_following()
+        except Exception as e:
+            assert "Either username or user_id must be provided" in str(e)
+        f = self.lingo.get_following(self.USERNAME_TEST)
+        assert isinstance(f, list)
+        assert len(f) > 0
+        assert "userId" in f[0]
+        user_id = self.lingo.get_user_id_from_username(self.USERNAME_TEST)
+        f_id = self.lingo.get_following(user_id=user_id)
+        assert f == f_id
+
 
 if __name__ == '__main__':
     unittest.main()