path: root/libs/apprise/plugins/NotifyTwitter.py

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

# See https://developer.twitter.com/en/docs/direct-messages/\
#           sending-and-receiving/api-reference/new-event.html
import re
import requests
from copy import deepcopy
from datetime import datetime
from datetime import timezone
from requests_oauthlib import OAuth1
from json import dumps
from json import loads
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
from ..common import NotifyType
from ..utils import parse_list
from ..utils import parse_bool
from ..utils import validate_regex
from ..AppriseLocale import gettext_lazy as _
from ..attachment.AttachBase import AttachBase

IS_USER = re.compile(r'^\s*@?(?P<user>[A-Z0-9_]+)$', re.I)


class TwitterMessageMode:
    """
    Twitter Message Mode
    """
    # DM (a Direct Message)
    DM = 'dm'

    # A Public Tweet
    TWEET = 'tweet'


# Define the types in a list for validation purposes
TWITTER_MESSAGE_MODES = (
    TwitterMessageMode.DM,
    TwitterMessageMode.TWEET,
)


class NotifyTwitter(NotifyBase):
    """
    A wrapper to Twitter Notifications

    """

    # The default descriptive name associated with the Notification
    service_name = 'Twitter'

    # The services URL
    service_url = 'https://twitter.com/'

    # The default secure protocol is twitter.
    secure_protocol = ('x', 'twitter', 'tweet')

    # A URL that takes you to the setup/help of the specific protocol
    setup_url = 'https://github.com/caronc/apprise/wiki/Notify_twitter'

    # Support attachments
    attachment_support = True

    # Do not set body_maxlen as it is set in a property value below
    # since the length varies depending if we are doing a direct message
    # or a tweet
    # body_maxlen = see below @propery defined

    # Twitter does have titles when creating a message
    title_maxlen = 0

    # Twitter API Reference To Acquire Someone's Twitter ID
    twitter_lookup = 'https://api.twitter.com/1.1/users/lookup.json'

    # Twitter API Reference To Acquire Current Users Information
    twitter_whoami = \
        'https://api.twitter.com/1.1/account/verify_credentials.json'

    # Twitter API Reference To Send A Private DM
    twitter_dm = 'https://api.twitter.com/1.1/direct_messages/events/new.json'

    # Twitter API Reference To Send A Public Tweet
    twitter_tweet = 'https://api.twitter.com/1.1/statuses/update.json'

    # it is documented on the site that the maximum images per tweet
    # is 4 (unless it's a GIF, then it's only 1)
    __tweet_non_gif_images_batch = 4

    # Twitter Media (Attachment) Upload Location
    twitter_media = 'https://upload.twitter.com/1.1/media/upload.json'

    # Twitter is kind enough to return how many more requests we're allowed to
    # continue to make within it's header response as:
    # X-Rate-Limit-Reset: The epoc time (in seconds) we can expect our
    #                    rate-limit to be reset.
    # X-Rate-Limit-Remaining: an integer identifying how many requests we're
    #                        still allow to make.
    request_rate_per_sec = 0

    # For Tracking Purposes
    ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)

    # Default to 1000; users can send up to 1000 DM's and 2400 tweets a day
    # This value only get's adjusted if the server sets it that way
    ratelimit_remaining = 1

    templates = (
        '{schema}://{ckey}/{csecret}/{akey}/{asecret}',
        '{schema}://{ckey}/{csecret}/{akey}/{asecret}/{targets}',
    )

    # Define our template tokens
    template_tokens = dict(NotifyBase.template_tokens, **{
        'ckey': {
            'name': _('Consumer Key'),
            'type': 'string',
            'private': True,
            'required': True,
        },
        'csecret': {
            'name': _('Consumer Secret'),
            'type': 'string',
            'private': True,
            'required': True,
        },
        'akey': {
            'name': _('Access Key'),
            'type': 'string',
            'private': True,
            'required': True,
        },
        'asecret': {
            'name': _('Access Secret'),
            'type': 'string',
            'private': True,
            'required': True,
        },
        'target_user': {
            'name': _('Target User'),
            'type': 'string',
            'prefix': '@',
            'map_to': 'targets',
        },
        'targets': {
            'name': _('Targets'),
            'type': 'list:string',
        },
    })

    # Define our template arguments
    template_args = dict(NotifyBase.template_args, **{
        'mode': {
            'name': _('Message Mode'),
            'type': 'choice:string',
            'values': TWITTER_MESSAGE_MODES,
            'default': TwitterMessageMode.DM,
        },
        'cache': {
            'name': _('Cache Results'),
            'type': 'bool',
            'default': True,
        },
        'to': {
            'alias_of': 'targets',
        },
        'batch': {
            'name': _('Batch Mode'),
            'type': 'bool',
            'default': True,
        },
    })

    def __init__(self, ckey, csecret, akey, asecret, targets=None,
                 mode=TwitterMessageMode.DM, cache=True, batch=True, **kwargs):
        """
        Initialize Twitter Object

        """
        super().__init__(**kwargs)

        self.ckey = validate_regex(ckey)
        if not self.ckey:
            msg = 'An invalid Twitter Consumer Key was specified.'
            self.logger.warning(msg)
            raise TypeError(msg)

        self.csecret = validate_regex(csecret)
        if not self.csecret:
            msg = 'An invalid Twitter Consumer Secret was specified.'
            self.logger.warning(msg)
            raise TypeError(msg)

        self.akey = validate_regex(akey)
        if not self.akey:
            msg = 'An invalid Twitter Access Key was specified.'
            self.logger.warning(msg)
            raise TypeError(msg)

        self.asecret = validate_regex(asecret)
        if not self.asecret:
            msg = 'An invalid Access Secret was specified.'
            self.logger.warning(msg)
            raise TypeError(msg)

        # Store our webhook mode
        self.mode = self.template_args['mode']['default'] \
            if not isinstance(mode, str) else mode.lower()

        if self.mode not in TWITTER_MESSAGE_MODES:
            msg = 'The Twitter message mode specified ({}) is invalid.' \
                .format(mode)
            self.logger.warning(msg)
            raise TypeError(msg)

        # Set Cache Flag
        self.cache = cache

        # Prepare Image Batch Mode Flag
        self.batch = batch

        # Track any errors
        has_error = False

        # Identify our targets
        self.targets = []
        for target in parse_list(targets):
            match = IS_USER.match(target)
            if match and match.group('user'):
                self.targets.append(match.group('user'))
                continue

            has_error = True
            self.logger.warning(
                'Dropped invalid Twitter user ({}) specified.'.format(target),
            )

        if has_error and not self.targets:
            # We have specified that we want to notify one or more individual
            # and we failed to load any of them.  Since it's also valid to
            # notify no one at all (which means we notify ourselves), it's
            # important we don't switch from the users original intentions
            msg = 'No Twitter targets to notify.'
            self.logger.warning(msg)
            raise TypeError(msg)

        # Initialize our cache values
        self._whoami_cache = None
        self._user_cache = {}

        return

    def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
             **kwargs):
        """
        Perform Twitter Notification
        """

        # Build a list of our attachments
        attachments = []

        if attach and self.attachment_support:
            # We need to upload our payload first so that we can source it
            # in remaining messages
            for attachment in attach:

                # Perform some simple error checking
                if not attachment:
                    # We could not access the attachment
                    self.logger.error(
                        'Could not access attachment {}.'.format(
                            attachment.url(privacy=True)))
                    return False

                if not re.match(r'^image/.*', attachment.mimetype, re.I):
                    # Only support images at this time
                    self.logger.warning(
                        'Ignoring unsupported Twitter attachment {}.'.format(
                            attachment.url(privacy=True)))
                    continue

                self.logger.debug(
                    'Preparing Twitter attachment {}'.format(
                        attachment.url(privacy=True)))

                # Upload our image and get our id associated with it
                # see: https://developer.twitter.com/en/docs/twitter-api/v1/\
                #         media/upload-media/api-reference/post-media-upload
                postokay, response = self._fetch(
                    self.twitter_media,
                    payload=attachment,
                )

                if not postokay:
                    # We can't post our attachment
                    return False

                if not (isinstance(response, dict)
                        and response.get('media_id')):
                    self.logger.debug(
                        'Could not attach the file to Twitter: %s (mime=%s)',
                        attachment.name, attachment.mimetype)
                    continue

                # If we get here, our output will look something like this:
                # {
                #   "media_id": 710511363345354753,
                #   "media_id_string": "710511363345354753",
                #   "media_key": "3_710511363345354753",
                #   "size": 11065,
                #   "expires_after_secs": 86400,
                #   "image": {
                #     "image_type": "image/jpeg",
                #     "w": 800,
                #     "h": 320
                #   }
                # }

                response.update({
                    # Update our response to additionally include the
                    # attachment details
                    'file_name': attachment.name,
                    'file_mime': attachment.mimetype,
                    'file_path': attachment.path,
                })

                # Save our pre-prepared payload for attachment posting
                attachments.append(response)

        # - calls _send_tweet if the mode is set so
        # - calls _send_dm (direct message) otherwise
        return getattr(self, '_send_{}'.format(self.mode))(
            body=body, title=title, notify_type=notify_type,
            attachments=attachments, **kwargs)

    def _send_tweet(self, body, title='', notify_type=NotifyType.INFO,
                    attachments=None, **kwargs):
        """
        Twitter Public Tweet
        """

        # Error Tracking
        has_error = False

        payload = {
            'status': body,
        }

        payloads = []
        if not attachments:
            payloads.append(payload)

        else:
            # Group our images if batch is set to do so
            batch_size = 1 if not self.batch \
                else self.__tweet_non_gif_images_batch

            # Track our batch control in our message generation
            batches = []
            batch = []
            for attachment in attachments:
                batch.append(str(attachment['media_id']))

                # Twitter supports batching images together.  This allows
                # the batching of multiple images together.  Twitter also
                # makes it clear that you can't batch `gif` files; they need
                # to be separate.  So the below preserves the ordering that
                # a user passed their attachments in.  if 4-non-gif images
                # are passed, they are all part of a single message.
                #
                # however, if they pass in image, gif, image, gif.  The
                # gif's inbetween break apart the batches so this would
                # produce 4 separate tweets.
                #
                # If you passed in, image, image, gif, image. <- This would
                # produce 3 images (as the first 2 images could be lumped
                # together as a batch)
                if not re.match(
                        r'^image/(png|jpe?g)', attachment['file_mime'], re.I) \
                        or len(batch) >= batch_size:
                    batches.append(','.join(batch))
                    batch = []

            if batch:
                batches.append(','.join(batch))

            for no, media_ids in enumerate(batches):
                _payload = deepcopy(payload)
                _payload['media_ids'] = media_ids

                if no or not body:
                    # strip text and replace it with the image representation
                    _payload['status'] = \
                        '{:02d}/{:02d}'.format(no + 1, len(batches))
                payloads.append(_payload)

        for no, payload in enumerate(payloads, start=1):
            # Send Tweet
            postokay, response = self._fetch(
                self.twitter_tweet,
                payload=payload,
                json=False,
            )

            if not postokay:
                # Track our error
                has_error = True

                errors = []
                try:
                    errors = ['Error Code {}: {}'.format(
                        e.get('code', 'unk'), e.get('message'))
                        for e in response['errors']]

                except (KeyError, TypeError):
                    pass

                for error in errors:
                    self.logger.debug(
                        'Tweet [%.2d/%.2d] Details: %s',
                        no, len(payloads), error)
                continue

            try:
                url = 'https://twitter.com/{}/status/{}'.format(
                    response['user']['screen_name'],
                    response['id_str'])

            except (KeyError, TypeError):
                url = 'unknown'

            self.logger.debug(
                'Tweet [%.2d/%.2d] Details: %s', no, len(payloads), url)

            self.logger.info(
                'Sent [%.2d/%.2d] Twitter notification as public tweet.',
                no, len(payloads))

        return not has_error

    def _send_dm(self, body, title='', notify_type=NotifyType.INFO,
                 attachments=None, **kwargs):
        """
        Twitter Direct Message
        """

        # Error Tracking
        has_error = False

        payload = {
            'event': {
                'type': 'message_create',
                'message_create': {
                    'target': {
                        # This gets assigned
                        'recipient_id': None,
                    },
                    'message_data': {
                        'text': body,
                    }
                }
            }
        }

        # Lookup our users (otherwise we look up ourselves)
        targets = self._whoami(lazy=self.cache) if not len(self.targets) \
            else self._user_lookup(self.targets, lazy=self.cache)

        if not targets:
            # We failed to lookup any users
            self.logger.warning(
                'Failed to acquire user(s) to Direct Message via Twitter')
            return False

        payloads = []
        if not attachments:
            payloads.append(payload)

        else:
            for no, attachment in enumerate(attachments):
                _payload = deepcopy(payload)
                _data = _payload['event']['message_create']['message_data']
                _data['attachment'] = {
                    'type': 'media',
                    'media': {
                        'id': attachment['media_id']
                    },
                    'additional_owners':
                    ','.join([str(x) for x in targets.values()])
                }
                if no or not body:
                    # strip text and replace it with the image representation
                    _data['text'] = \
                        '{:02d}/{:02d}'.format(no + 1, len(attachments))
                payloads.append(_payload)

        for no, payload in enumerate(payloads, start=1):
            for screen_name, user_id in targets.items():
                # Assign our user
                target = payload['event']['message_create']['target']
                target['recipient_id'] = user_id

                # Send Twitter DM
                postokay, response = self._fetch(
                    self.twitter_dm,
                    payload=payload,
                )

                if not postokay:
                    # Track our error
                    has_error = True
                    continue

                self.logger.info(
                    'Sent [{:02d}/{:02d}] Twitter DM notification to @{}.'
                    .format(no, len(payloads), screen_name))

        return not has_error

    def _whoami(self, lazy=True):
        """
        Looks details of current authenticated user

        """

        if lazy and self._whoami_cache is not None:
            # Use cached response
            return self._whoami_cache

        # Contains a mapping of screen_name to id
        results = {}

        # Send Twitter DM
        postokay, response = self._fetch(
            self.twitter_whoami,
            method='GET',
            json=False,
        )

        if postokay:
            try:
                results[response['screen_name']] = response['id']
                self._whoami_cache = {
                    response['screen_name']: response['id'],
                }

                self._user_cache.update(results)

            except (TypeError, KeyError):
                pass

        return results

    def _user_lookup(self, screen_name, lazy=True):
        """
        Looks up a screen name and returns the user id

        the screen_name can be a list/set/tuple as well
        """

        # Contains a mapping of screen_name to id
        results = {}

        # Build a unique set of names
        names = parse_list(screen_name)

        if lazy and self._user_cache:
            # Use cached response
            results = {
                k: v for k, v in self._user_cache.items() if k in names}

            # limit our names if they already exist in our cache
            names = [name for name in names if name not in results]

        if not len(names):
            # They're is nothing further to do
            return results

        # Twitters API documents that it can lookup to 100
        # results at a time.
        # https://developer.twitter.com/en/docs/accounts-and-users/\
        #     follow-search-get-users/api-reference/get-users-lookup
        for i in range(0, len(names), 100):
            # Look up our names by their screen_name
            postokay, response = self._fetch(
                self.twitter_lookup,
                payload={
                    'screen_name': names[i:i + 100],
                },
                json=False,
            )

            if not postokay or not isinstance(response, list):
                # Track our error
                continue

            # Update our user index
            for entry in response:
                try:
                    results[entry['screen_name']] = entry['id']

                except (TypeError, KeyError):
                    pass

        # Cache our response for future use; this saves on un-nessisary extra
        # hits against the Twitter API when we already know the answer
        self._user_cache.update(results)

        return results

    def _fetch(self, url, payload=None, method='POST', json=True):
        """
        Wrapper to Twitter API requests object
        """

        headers = {
            'User-Agent': self.app_id,
        }

        data = None
        files = None

        # Open our attachment path if required:
        if isinstance(payload, AttachBase):
            # prepare payload
            files = {'media': (payload.name, open(payload.path, 'rb'))}

        elif json:
            headers['Content-Type'] = 'application/json'
            data = dumps(payload)

        else:
            data = payload

        auth = OAuth1(
            self.ckey,
            client_secret=self.csecret,
            resource_owner_key=self.akey,
            resource_owner_secret=self.asecret,
        )

        # Some Debug Logging
        self.logger.debug('Twitter {} URL: {} (cert_verify={})'.format(
            method, url, self.verify_certificate))
        self.logger.debug('Twitter Payload: %s' % str(payload))

        # By default set wait to None
        wait = None

        if self.ratelimit_remaining == 0:
            # Determine how long we should wait for or if we should wait at
            # all. This isn't fool-proof because we can't be sure the client
            # time (calling this script) is completely synced up with the
            # Twitter server.  One would hope we're on NTP and our clocks are
            # the same allowing this to role smoothly:

            now = datetime.now(timezone.utc).replace(tzinfo=None)
            if now < self.ratelimit_reset:
                # We need to throttle for the difference in seconds
                # We add 0.5 seconds to the end just to allow a grace
                # period.
                wait = (self.ratelimit_reset - now).total_seconds() + 0.5

        # Default content response object
        content = {}

        # Always call throttle before any remote server i/o is made;
        self.throttle(wait=wait)

        # acquire our request mode
        fn = requests.post if method == 'POST' else requests.get
        try:
            r = fn(
                url,
                data=data,
                files=files,
                headers=headers,
                auth=auth,
                verify=self.verify_certificate,
                timeout=self.request_timeout,
            )

            try:
                content = loads(r.content)

            except (AttributeError, TypeError, ValueError):
                # ValueError = r.content is Unparsable
                # TypeError = r.content is None
                # AttributeError = r is None
                content = {}

            if r.status_code != requests.codes.ok:
                # We had a problem
                status_str = \
                    NotifyTwitter.http_response_code_lookup(r.status_code)

                self.logger.warning(
                    'Failed to send Twitter {} to {}: '
                    '{}error={}.'.format(
                        method,
                        url,
                        ', ' if status_str else '',
                        r.status_code))

                self.logger.debug(
                    'Response Details:\r\n{}'.format(r.content))

                # Mark our failure
                return (False, content)

            try:
                # Capture rate limiting if possible
                self.ratelimit_remaining = \
                    int(r.headers.get('x-rate-limit-remaining'))
                self.ratelimit_reset = datetime.fromtimestamp(
                    int(r.headers.get('x-rate-limit-reset')), timezone.utc
                ).replace(tzinfo=None)

            except (TypeError, ValueError):
                # This is returned if we could not retrieve this information
                # gracefully accept this state and move on
                pass

        except requests.RequestException as e:
            self.logger.warning(
                'Exception received when sending Twitter {} to {}: '.
                format(method, url))
            self.logger.debug('Socket Exception: %s' % str(e))

            # Mark our failure
            return (False, content)

        except (OSError, IOError) as e:
            self.logger.warning(
                'An I/O error occurred while handling {}.'.format(
                    payload.name if isinstance(payload, AttachBase)
                    else payload))
            self.logger.debug('I/O Exception: %s' % str(e))
            return (False, content)

        finally:
            # Close our file (if it's open) stored in the second element
            # of our files tuple (index 1)
            if files:
                files['media'][1].close()

        return (True, content)

    @property
    def body_maxlen(self):
        """
        The maximum allowable characters allowed in the body per message
        This is used during a Private DM Message Size (not Public Tweets
        which are limited to 280 characters)
        """
        return 10000 if self.mode == TwitterMessageMode.DM else 280

    def url(self, privacy=False, *args, **kwargs):
        """
        Returns the URL built dynamically based on specified arguments.
        """

        # Define any URL parameters
        params = {
            'mode': self.mode,
            'batch': 'yes' if self.batch else 'no',
            'cache': 'yes' if self.cache else 'no',
        }

        # Extend our parameters
        params.update(self.url_parameters(privacy=privacy, *args, **kwargs))

        return '{schema}://{ckey}/{csecret}/{akey}/{asecret}' \
            '/{targets}/?{params}'.format(
                schema=self.secure_protocol[0],
                ckey=self.pprint(self.ckey, privacy, safe=''),
                csecret=self.pprint(
                    self.csecret, privacy, mode=PrivacyMode.Secret, safe=''),
                akey=self.pprint(self.akey, privacy, safe=''),
                asecret=self.pprint(
                    self.asecret, privacy, mode=PrivacyMode.Secret, safe=''),
                targets='/'.join(
                    [NotifyTwitter.quote('@{}'.format(target), safe='@')
                     for target in self.targets]),
                params=NotifyTwitter.urlencode(params))

    def __len__(self):
        """
        Returns the number of targets associated with this notification
        """
        targets = len(self.targets)
        return targets if targets > 0 else 1

    @staticmethod
    def parse_url(url):
        """
        Parses the URL and returns enough arguments that can allow
        us to re-instantiate this object.

        """
        results = NotifyBase.parse_url(url, verify_host=False)
        if not results:
            # We're done early as we couldn't load the results
            return results

        # Acquire remaining tokens
        tokens = NotifyTwitter.split_path(results['fullpath'])

        # The consumer token is stored in the hostname
        results['ckey'] = NotifyTwitter.unquote(results['host'])

        #
        # Now fetch the remaining tokens
        #

        # Consumer Secret
        results['csecret'] = tokens.pop(0) if tokens else None
        # Access Token Key
        results['akey'] = tokens.pop(0) if tokens else None
        # Access Token Secret
        results['asecret'] = tokens.pop(0) if tokens else None

        # The defined twitter mode
        if 'mode' in results['qsd'] and len(results['qsd']['mode']):
            results['mode'] = \
                NotifyTwitter.unquote(results['qsd']['mode'])

        elif results['schema'].startswith('tweet'):
            results['mode'] = TwitterMessageMode.TWEET

        results['targets'] = []

        # if a user has been defined, add it to the list of targets
        if results.get('user'):
            results['targets'].append(results.get('user'))

        # Store any remaining items as potential targets
        results['targets'].extend(tokens)

        # Get Cache Flag (reduces lookup hits)
        if 'cache' in results['qsd'] and len(results['qsd']['cache']):
            results['cache'] = \
                parse_bool(results['qsd']['cache'], True)

        # Get Batch Mode Flag
        results['batch'] = \
            parse_bool(results['qsd'].get(
                'batch', NotifyTwitter.template_args['batch']['default']))

        # The 'to' makes it easier to use yaml configuration
        if 'to' in results['qsd'] and len(results['qsd']['to']):
            results['targets'] += \
                NotifyTwitter.parse_list(results['qsd']['to'])

        return results