v0.0.5

Author: HoodyH

my_feed/modules/post.py

@@ -20,6 +20,9 @@ def __init__(self, post_id, title, created_at, url):
         self.media: List[MediaModel] = []
         self.description: str = ''
 
+    def __repr__(self):
+        return f'{self.url}'
+
     def add_media(self, media_id, media_url) -> None:
         """
         Append a new media to the Post

my_feed/my_feed.py

@@ -58,7 +58,7 @@ def __init__(self, platform: Platforms, channel_id: str):
         self.is_active = True
 
         # The platform update type (reddit, instagram, etc)
-        self.platform: str = platform.identifier.value
+        self.platform: PlatformsId = platform.identifier
         self.updater = platform.callable  # the updater Class
 
         # initialize last update with an old time
@@ -67,39 +67,53 @@ def __init__(self, platform: Platforms, channel_id: str):
         self.__temp_last_update = None
 
         # update the data every minutes interval
-        self.update_interval: int = 1
+        self.update_interval: int = 5
 
         # how to identify the last update, to not send again the same data
         # this value can be a string, slug, int or tuple based on the platform that you are using
-        self.last_update_id = None
+        self.__last_update_id = None
         # temporally save the last update id
         self.__temp_last_update_id = None
 
         # the channel specification, this must match che update requirements in the platform api
         self.id: str = channel_id
 
+    def __repr__(self):
+        return f'{self.platform.value}/{self.id}'
+
     @property
     def is_time_to_update(self) -> bool:
         if datetime.now() - self.last_update > timedelta(minutes=self.update_interval) and self.is_active:
             return True
         return False
 
+    def reset_last_update_id(self):
+        self.__last_update_id = None
+
     def update(self) -> List[PostModel]:
         updater = self.updater()  # create the class
-        out = updater.update(self.id, self.last_update_id)
+        out = updater.update(self.id, self.__last_update_id)
         self.__temp_last_update = datetime.now()
 
         # temporally save the last update id
         # after all the operations on the data are done call set_last_update_now()
         self.__temp_last_update_id = updater.last_post_id
 
         # if is the first time the last_update_id is none, so dont show any post
-        if self.last_update_id:
+        if self.__last_update_id:
             return out
         else:
             return []
 
     def set_last_update_now(self) -> None:
         # update the last id, after all the operation on the data is done
-        self.last_update_id = self.__temp_last_update_id
+        self.__last_update_id = self.__temp_last_update_id
         self.last_update = self.__temp_last_update
+
+    def auto_update(self):
+        """Auto update the feed, this can be done also manually"""
+        out = []
+        if self.is_time_to_update:
+            out = self.update()
+            self.set_last_update_now()
+        return out

my_feed/platforms/instagram.py

@@ -0,0 +1,147 @@
+from enum import Enum
+from my_feed.modules.post import PostModel
+from my_feed.modules.types import PostType
+from my_feed.platforms import PlatformInterface, PlatformsId
+from my_feed.platforms.instagram.credentials import credentials_manager
+from my_feed.platforms.instagram.login import InstagramLogin
+
+
+class MediaType(Enum):
+    video = 'video'
+    image = 'image'
+
+
+class Instagram(PlatformInterface):
+
+    def __init__(self):
+        super().__init__()
+
+        self.api: InstagramLogin
+
+    def __repr__(self):
+        return PlatformsId.INSTAGRAM.value
+
+    def _login(self):
+        user = credentials_manager.credentials[0]
+        il = InstagramLogin()
+        self.api = il.login(user.username, user.password, 'configs/{}_creds.json'.format(user.username))
+
+    @staticmethod
+    def get_media_url(media_data):
+        media = media_data.get('standard_resolution', {})
+        return media.get('url')
+
+    def story(self, user_id, last_update_id) -> [PostModel]:
+        """
+        story can be get only if you are logged in
+        Load all the stories from the user loaded
+        :return: InstagramStory list
+        """
+        out = []  # list of all the items in the story
+
+        stories = self.api.user_story_feed(user_id)
+
+        reels = stories.get('reel')
+        if not reels:
+            return []
+
+        reels = reels.get('items')
+        for reel in reels:
+
+            post_id = reel.get('pk')
+            if post_id == last_update_id:
+                break
+
+            post = PostModel(
+                post_id=post_id,
+                title='',
+                created_at=reel.get('taken_at'),
+                url=reel.get('link')
+            )
+
+            media_type = MediaType(reel.get('type'))
+            post.type = PostType.VIDEO if media_type == MediaType.video else PostType.IMAGE
+            media = reel.get('videos' if media_type == MediaType.video else 'images')
+            post.add_media(media_id=None, media_url=self.get_media_url(media))
+
+            out.append(post)
+
+        return out
+
+    def post(self, user_id, last_update_id) -> [PostModel]:
+        """
+        Get all the post from the user loaded
+        Store them as a list of InstagramPost obj
+        :return: PostModel list
+        """
+        out = []
+
+        media = self.api.user_feed(user_id)
+
+        items = media.get('items')
+        for item in items:
+
+            post_id = item.get('code')
+            if post_id == last_update_id:
+                break
+
+            # get description of the post
+            caption = item.get('caption', {})
+            text = ''
+            if caption:
+                text = caption.get('text', '')
+
+            post = PostModel(
+                post_id=post_id,
+                title=text,
+                created_at=item.get('created_time'),
+                url=item.get('link')
+            )
+
+            """
+            Check if is a post with multiple elements
+                - multi element post has media list inside carousel_media
+                - if not the object media is no encapsulated
+            """
+            carousel = item.get('carousel_media')
+            media_type = MediaType(item.get('type'))
+            post.type = PostType.VIDEO if media_type == MediaType.video else PostType.IMAGE
+            if carousel:
+                for c in carousel:
+                    media = c.get('videos' if media_type == MediaType.video else 'images')
+                    post.add_media(media_id=None, media_url=self.get_media_url(media))
+
+            else:
+                media = item.get('videos' if media_type == MediaType.video else 'images')
+                post.add_media(media_id=None, media_url=self.get_media_url(media))
+
+            out.append(post)
+
+        return out
+
+    def update(self, target, last_update_id):
+
+        # decouple the last update id
+        # this cause instagram have 2 feed sources, stories and posts
+        last_update_id_story = None
+        last_update_id_post = None
+        if last_update_id:
+            last_update_id_story, last_update_id_post = last_update_id
+
+        # do login every time
+        self._login()
+
+        # get the data from the api
+        story_feed = self.story(target, last_update_id_story)
+        post_feed = self.post(target, last_update_id_post)
+
+        # get the post id from the feed
+        last_update_id_story = self._get_last_post_id(story_feed, last_update_id_story)
+        last_update_id_post = self._get_last_post_id(post_feed, last_update_id_post)
+
+        # the set feed will also revert it
+        self._set_feed(post_feed + story_feed)
+
+        # update the post id
+        self._last_post_id = (last_update_id_story, last_update_id_post)
+        return self._feed

my_feed/platforms/instagram/__init__.py

@@ -0,0 +1,32 @@
+import json
+
+file_name = 'configs/instagram_creds.json'
+with open(file_name, 'r', encoding='utf8') as f:
+    file_data = json.loads(f.read())
+    f.close()
+
+
+class CredentialsManager:
+    """
+    Load all the Users Credentials in an array
+    """
+    class _UserCredentials:
+        def __init__(self, username, password):
+            self.username = username
+            self.password = password
+            self.pass_code = None
+
+    def __init__(self):
+        """
+        Load the credentials from the json file
+        Store them ar list of _UserCredentials obj
+        """
+        self.credentials = []
+
+        _credentials = file_data.get('creds')
+        for el in _credentials:
+            key = list(el.keys())[0]
+            self.credentials.append(self._UserCredentials(key, el[key]))
+
+
+credentials_manager = CredentialsManager()