updateInfoPleroma.py
def process_tweets(self, tweets_to_post):
        """Transforms tweets for posting them to Pleroma
        Expands shortened URLs
        Downloads tweet related media and prepares them for upload

        :param tweets_to_post: Dict of tweet objects to be processed
        :type tweets_to_post: dict
        :returns: Tweets ready to be published
        :rtype: list
        """
        for tweet in tweets_to_post['data']:
            media = []
            # Replace shortened links
            try:
                for url_entity in tweet['entities']['urls']:
                    matching_pattern = url_entity['url']
                    matches = re.findall(matching_pattern, tweet['text'])
                    for match in matches:
                        tweet['text'] = re.sub(match, url_entity['expanded_url'], tweet['text'])
            except KeyError:
                # URI regex
                matching_pattern = r'(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)(?:[^\s()<>]+|\(([' \
                                   r'^\s()<>]+|(\([^\s()<>]+\)))*\))+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[' \
                                   r'\]{};:\'".,<>?«»“”‘’]))'
                matches = re.finditer(matching_pattern, tweet['text'])
                for matchNum, match in enumerate(matches, start=1):
                    # don't be brave trying to unwound an URL when it gets cut off
                    if not match.group().__contains__("…"):
                        session = requests.Session()  # so connections are recycled
                        response = session.head(match.group(), allow_redirects=True)
                        if not response.ok:
                            response.raise_for_status()
                        expanded_url = response.url
                        tweet['text'] = re.sub(match.group(), expanded_url, tweet['text'])
            if hasattr(self, "rich_text"):
                if self.rich_text:
                    matches = re.findall(r'\B\@\w+', tweet['text'])
                    for match in matches:
                        mention_link = "[" + match + "](https://twitter.com/" + match[1:] + ")"
                        tweet['text'] = re.sub(match, mention_link, tweet['text'])
            try:
                if self.nitter:
                    matching_pattern = "https://twitter.com"
                    matches = re.findall(matching_pattern, tweet['text'])
                    for match in matches:
                        tweet['text'] = re.sub(match, "https://nitter.net", tweet['text'])
            except AttributeError:
                pass
            try:
                for item in tweet['attachments']['media_keys']:
                    for media_include in tweets_to_post['includes']['media']:
                        if item == media_include['media_key']:
                            # Video download not implemented in v2 yet
                            # fallback to v1.1
                            if media_include['type'] == 'video' or media_include['type'] == 'animated_gif':
                                tweet_video = self._get_tweets('v1.1', tweet['id'])
                                for extended_media in tweet_video['extended_entities']['media']:
                                    media.append(extended_media)
                            else:
                                media.append(media_include)
            except KeyError:
                pass
            # Create folder to store attachments related to the tweet ID
            tweet_path = os.path.join(self.tweets_temp_path, tweet['id'])
            if not os.path.isdir(tweet_path):
                os.mkdir(tweet_path)
            # Download media only if we plan to upload it later
            if self.media_upload:
                for idx, item in enumerate(media):
                    if item['type'] != 'video' and item['type'] != 'animated_gif':
                        media_url = item['url']
                    else:
                        bitrate = 0
                        for variant in item['video_info']['variants']:
                            try:
                                if variant['bitrate'] >= bitrate:
                                    media_url = variant['url']
                            except KeyError:
                                pass
                    response = requests.get(media_url, stream=True)
                    if not response.ok:
                        response.raise_for_status()
                    response.raw.decode_content = True
                    filename = str(idx) + mimetypes.guess_extension(response.headers['Content-Type'])
                    with open(os.path.join(self.tweets_temp_path, tweet['id'], filename), 'wb') as outfile:
                        shutil.copyfileobj(response.raw, outfile)

            # Process poll if exists and no media is used
            try:
                if tweet['attachments']['poll_ids'] and not media:

                    # tweet_poll = tweet['includes']['polls']
                    poll_url = self.twitter_base_url_v2 + '/tweets'

                    params = {
                        'ids': tweet['id'],
                        'expansions': 'attachments.poll_ids',
                        'poll.fields': 'duration_minutes,'
                                       'options'
                    }

                    response = requests.get(
                        poll_url, headers=self.header_twitter, params=params)
                    if not response.ok:
                        response.raise_for_status()
                    tweet_poll = json.loads(response.content)[
                        'includes']['polls'][0]

                    pleroma_poll = {'options': [option['label']
                                                for option in tweet_poll['options']],
                                    'expires_in': tweet_poll['duration_minutes'] * 60}

                    # Add poll to tweet
                    tweet['polls'] = pleroma_poll

                else:
                    tweet['polls'] = None
            except KeyError:
                tweet['polls'] = None
                pass
        return tweets_to_post
Transforms tweets for posting them to Pleroma
        Expands shortened URLs
        Downloads tweet related media and prepares them for upload

        :param tweets_to_post: Dict of tweet objects to be processed
        :type tweets_to_post: dict
        :returns: Tweets ready to be published
        :rtype: list
        
Transforms tweets for posting them to Pleroma
Expands shortened URLs
Downloads tweet related media and prepares them for upload
Transforms tweets for posting them to Pleroma Expands shortened URLs Downloads tweet related media and prepares them for upload
def process_tweets(self, tweets_to_post):
        for tweet in tweets_to_post['data']:
            media = []
            try:
                for url_entity in tweet['entities']['urls']:
                    matching_pattern = url_entity['url']
                    matches = re.findall(matching_pattern, tweet['text'])
                    for match in matches:
                        tweet['text'] = re.sub(match, url_entity['expanded_url'], tweet['text'])
            except KeyError:
                matching_pattern = r'(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)(?:[^\s()<>]+|\(([' \
                                   r'^\s()<>]+|(\([^\s()<>]+\)))*\))+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[' \
                                   r'\]{};:\'".,<>?«»“”‘’]))'
                matches = re.finditer(matching_pattern, tweet['text'])
                for matchNum, match in enumerate(matches, start=1):
                    if not match.group().__contains__("…"):
                        session = requests.Session()  
                        response = session.head(match.group(), allow_redirects=True)
                        if not response.ok:
                            response.raise_for_status()
                        expanded_url = response.url
                        tweet['text'] = re.sub(match.group(), expanded_url, tweet['text'])
            if hasattr(self, "rich_text"):
                if self.rich_text:
                    matches = re.findall(r'\B\@\w+', tweet['text'])
                    for match in matches:
                        mention_link = "[" + match + "](https://twitter.com/" + match[1:] + ")"
                        tweet['text'] = re.sub(match, mention_link, tweet['text'])
            try:
                if self.nitter:
                    matching_pattern = "https://twitter.com"
                    matches = re.findall(matching_pattern, tweet['text'])
                    for match in matches:
                        tweet['text'] = re.sub(match, "https://nitter.net", tweet['text'])
            except AttributeError:
                pass
            try:
                for item in tweet['attachments']['media_keys']:
                    for media_include in tweets_to_post['includes']['media']:
                        if item == media_include['media_key']:
                            if media_include['type'] == 'video' or media_include['type'] == 'animated_gif':
                                tweet_video = self._get_tweets('v1.1', tweet['id'])
                                for extended_media in tweet_video['extended_entities']['media']:
                                    media.append(extended_media)
                            else:
                                media.append(media_include)
            except KeyError:
                pass
            tweet_path = os.path.join(self.tweets_temp_path, tweet['id'])
            if not os.path.isdir(tweet_path):
                os.mkdir(tweet_path)
            if self.media_upload:
                for idx, item in enumerate(media):
                    if item['type'] != 'video' and item['type'] != 'animated_gif':
                        media_url = item['url']
                    else:
                        bitrate = 0
                        for variant in item['video_info']['variants']:
                            try:
                                if variant['bitrate'] >= bitrate:
                                    media_url = variant['url']
                            except KeyError:
                                pass
                    response = requests.get(media_url, stream=True)
                    if not response.ok:
                        response.raise_for_status()
                    response.raw.decode_content = True
                    filename = str(idx) + mimetypes.guess_extension(response.headers['Content-Type'])
                    with open(os.path.join(self.tweets_temp_path, tweet['id'], filename), 'wb') as outfile:
                        shutil.copyfileobj(response.raw, outfile)
            try:
                if tweet['attachments']['poll_ids'] and not media:
                    poll_url = self.twitter_base_url_v2 + '/tweets'
                    params = {
                        'ids': tweet['id'],
                        'expansions': 'attachments.poll_ids',
                        'poll.fields': 'duration_minutes,'
                                       'options'
                    }
                    response = requests.get(
                        poll_url, headers=self.header_twitter, params=params)
                    if not response.ok:
                        response.raise_for_status()
                    tweet_poll = json.loads(response.content)[
                        'includes']['polls'][0]
                    pleroma_poll = {'options': [option['label']
                                                for option in tweet_poll['options']],
                                    'expires_in': tweet_poll['duration_minutes'] * 60}
                    tweet['polls'] = pleroma_poll
                else:
                    tweet['polls'] = None
            except KeyError:
                tweet['polls'] = None
                pass
        return tweets_to_post
def process_tweets ( self , tweets_to_post ) : for tweet in tweets_to_post [ 'data' ] : media = [ ] try : for url_entity in tweet [ 'entities' ] [ 'urls' ] : matching_pattern = url_entity [ 'url' ] matches = re . findall ( matching_pattern , tweet [ 'text' ] ) for match in matches : tweet [ 'text' ] = re . sub ( match , url_entity [ 'expanded_url' ] , tweet [ 'text' ] ) except KeyError : matching_pattern = r'(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)(?:[^\s()<>]+|\(([' r'^\s()<>]+|(\([^\s()<>]+\)))*\))+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[' r'\]{};:\'".,<>?«»“”‘’]))' matches = re . finditer ( matching_pattern , tweet [ 'text' ] ) for matchNum , match in enumerate ( matches , start = 1 ) : if not match . group ( ) . __contains__ ( "…"): session = requests . Session ( ) response = session . head ( match . group ( ) , allow_redirects = True ) if not response . ok : response . raise_for_status ( ) expanded_url = response . url tweet [ 'text' ] = re . sub ( match . group ( ) , expanded_url , tweet [ 'text' ] ) if hasattr ( self , "rich_text" ) : if self . rich_text : matches = re . findall ( r'\B\@\w+' , tweet [ 'text' ] ) for match in matches : mention_link = "[" + match + "](https://twitter.com/" + match [ 1 : ] + ")" tweet [ 'text' ] = re . sub ( match , mention_link , tweet [ 'text' ] ) try : if self . nitter : matching_pattern = "https://twitter.com" matches = re . findall ( matching_pattern , tweet [ 'text' ] ) for match in matches : tweet [ 'text' ] = re . sub ( match , "https://nitter.net" , tweet [ 'text' ] ) except AttributeError : pass try : for item in tweet [ 'attachments' ] [ 'media_keys' ] : for media_include in tweets_to_post [ 'includes' ] [ 'media' ] : if item == media_include [ 'media_key' ] : if media_include [ 'type' ] == 'video' or media_include [ 'type' ] == 'animated_gif' : tweet_video = self . _get_tweets ( 'v1.1' , tweet [ 'id' ] ) for extended_media in tweet_video [ 'extended_entities' ] [ 'media' ] : media . append ( extended_media ) else : media . append ( media_include ) except KeyError : pass tweet_path = os . path . join ( self . tweets_temp_path , tweet [ 'id' ] ) if not os . path . isdir ( tweet_path ) : os . mkdir ( tweet_path ) if self . media_upload : for idx , item in enumerate ( media ) : if item [ 'type' ] != 'video' and item [ 'type' ] != 'animated_gif' : media_url = item [ 'url' ] else : bitrate = 0 for variant in item [ 'video_info' ] [ 'variants' ] : try : if variant [ 'bitrate' ] >= bitrate : media_url = variant [ 'url' ] except KeyError : pass response = requests . get ( media_url , stream = True ) if not response . ok : response . raise_for_status ( ) response . raw . decode_content = True filename = str ( idx ) + mimetypes . guess_extension ( response . headers [ 'Content-Type' ] ) with open ( os . path . join ( self . tweets_temp_path , tweet [ 'id' ] , filename ) , 'wb' ) as outfile : shutil . copyfileobj ( response . raw , outfile ) try : if tweet [ 'attachments' ] [ 'poll_ids' ] and not media : poll_url = self . twitter_base_url_v2 + '/tweets' params = { 'ids' : tweet [ 'id' ] , 'expansions' : 'attachments.poll_ids' , 'poll.fields' : 'duration_minutes,' 'options' } response = requests . get ( poll_url , headers = self . header_twitter , params = params ) if not response . ok : response . raise_for_status ( ) tweet_poll = json . loads ( response . content ) [ 'includes' ] [ 'polls' ] [ 0 ] pleroma_poll = { 'options' : [ option [ 'label' ] for option in tweet_poll [ 'options' ] ] , 'expires_in' : tweet_poll [ 'duration_minutes' ] * 60 } tweet [ 'polls' ] = pleroma_poll else : tweet [ 'polls' ] = None except KeyError : tweet [ 'polls' ] = None pass return tweets_to_post
Transforms tweets for posting them to Pleroma
Expands shortened URLs
Downloads tweet related media and prepares them for upload
Transforms tweets for posting them to Pleroma Expands shortened URLs Downloads tweet related media and prepares them for upload
"""Transforms tweets for posting them to Pleroma
        Expands shortened URLs
        Downloads tweet related media and prepares them for upload

        :param tweets_to_post: Dict of tweet objects to be processed
        :type tweets_to_post: dict
        :returns: Tweets ready to be published
        :rtype: list
        """ # Replace shortened links # URI regex # don't be brave trying to unwound an URL when it gets cut off # so connections are recycled # Video download not implemented in v2 yet # fallback to v1.1 # Create folder to store attachments related to the tweet ID # Download media only if we plan to upload it later # Process poll if exists and no media is used # tweet_poll = tweet['includes']['polls'] # Add poll to tweet
Tweets ready to be published
Tweets ready to be published
list
self
tweets_to_post
Dict of tweet objects to be processed
Dict of tweet objects to be processed