blog:2022:0512_nervhome_twitter_handler

Quick project: twitter api [python]

, ,

In this new post, I want to build myself a new component to be able to access/use my twitter accounts from python. I like twitter, it's handy to just quickly share a few words and then forget about it. But, I feel it's a bit of a pain for me to have to open my browser to send something there! 😝 So I would like to be able to use twitter easily directly from a command line! And given my recent journey in my NervProj framework, there is no better place to try to implement that from my perspective.

Setting up Twitter developper account and keys

  • Actually I'm not quite sure what I can do exactly with the twitter API, but let's just do something simple for a start: Just trying to send a message from your account. Should be pretty standard, right ? 😊
  • I found this page with indications on how to use tweepy: https://www.jcchouinard.com/twitter-api/#How_to_Post_Tweets_using_Twitter_API
  • ⇒ So let me try to follow those notes…
  • Next question I'm wondering is: how would I set the permissions to read or post anything ?
  • ⇒ Okay, so I can create the access token for my twitter user account on the “keys & token” tab of my app:

  • Now for the permissions (we are in read-only mode by default): we have this article: https://developer.twitter.com/en/docs/apps/app-permissions
  • ⇒ This is not clearly explaining how to setup the permissions 😅, but anyway, I found that first I have to setup the authentication methods for the app: that's were we decide if the app needs read only/read write/read write with direct message permission,
  • And only after that we generate the access token for our twitter user ⇒ should reflect the correct permissions then.
  • Note: I also activated OAuth 2.0 support in my app, so I got a client_id and a client_secret, but not sure yet what I will do with these, we'll see…
  • Another thing I find a bit puzzling for the moment are those callback URLS (you must provide one when setting up the app authentication): I simply used my wiki website address for now: https://wiki.nervtech.org/ but I have no idea yet why this is needed.

Initial component skeleton

  • Okay, so, in theory I should now be able to use tweepy to send a post ? [😆]
  • ⇒ Let's try to prepare a minimal component for that:
    • Defining a new social_env: 
          "social_env": {
      "packages": ["requests", "jstyleson", "xxhash", "tweepy"]
    }
    • And installing that env: 
      $ nvp pyenv setup social_env
    • Also defining a new script for the new twitter component: 
          "tweet": {
      "custom_python_env": "social_env",
      "cmd": "${PYTHON} ${PROJECT_ROOT_DIR}/nvh/social/twitter_handler.py",
      "python_path": ["${PROJECT_ROOT_DIR}", "${NVP_ROOT_DIR}"]
    }
    • And here is the initial skeleton component: 
      """Module for TwitterHandler class definition"""

import logging

from nvp.nvp_context import NVPContext
from nvp.nvp_component import NVPComponent

logger = logging.getLogger(__name__)


class TwitterHandler(NVPComponent):
    """TwitterHandler component class"""

    def __init__(self, ctx: NVPContext):
        """class constructor"""
        NVPComponent.__init__(self, ctx)

    def process_command(self, cmd):
        """Check if this component can process the given command"""

        if cmd == 'send':
            logger.info("Should send a message here.")
            return True

        return False


if __name__ == "__main__":
    # Create the context:
    context = NVPContext()

    # Add our component:
    comp = context.register_component("TwitterHandler", TwitterHandler(context))

    context.define_subparsers("main", {
        'send': None,
    })

    psr = context.get_parser('main.send')
    psr.add_argument("message", type=str,
                     help="Simple message that we should send")

    comp.run()
  • Running this component works as desired (ie. doing nothing for now): 
    $ nvp run tweet send "Hello manu!"
2022/05/12 14:35:05 [__main__] INFO: Should send a message here.

Adding support for authentication

  • Now I have just added a function called setup_api() where we check if we can properly establish a connection to the twitter API: 
        def process_command(self, cmd):
        """Check if this component can process the given command"""

        if cmd == 'send':
            logger.info("Should send a message here.")
            self.setup_api()
            return True

        return False

    def setup_api(self):
        """Setup the Twitter API object"""

        # We read the key/secret from our config:
        api_key = self.config["api_key"]
        api_secrets = self.config["api_key_secret"]
        access_token = self.config["access_token"]
        access_secret = self.config["access_token_secret"]

        # Authenticate to Twitter
        auth = tweepy.OAuthHandler(api_key, api_secrets)
        auth.set_access_token(access_token, access_secret)

        self.api = tweepy.API(auth)

        # try:
        self.api.verify_credentials()
        logger.info('Successful Authentication')
        # except:
        #     logger.info('Failed authentication')
  • Let's see what we get: 
    $ nvp run tweet send "hello world"
2022/05/12 14:38:19 [__main__] INFO: Should send a message here.
Traceback (most recent call last):
  File "D:\Projects\NervHome\nvh\social\twitter_handler.py", line 68, in <module>
    comp.run()
  File "D:\Projects\NervProj\nvp\nvp_component.py", line 69, in run
    res = self.process_command(cmd)
  File "D:\Projects\NervHome\nvh\social\twitter_handler.py", line 26, in process_command
    self.setup_api()
  File "D:\Projects\NervHome\nvh\social\twitter_handler.py", line 47, in setup_api
    self.api.verify_credentials()
  File "D:\Projects\NervProj\.pyenvs\social_env\lib\site-packages\tweepy\api.py", line 46, in wrapper
    return method(*args, **kwargs)
  File "D:\Projects\NervProj\.pyenvs\social_env\lib\site-packages\tweepy\api.py", line 2649, in verify_credentials
    return self.request(
  File "D:\Projects\NervProj\.pyenvs\social_env\lib\site-packages\tweepy\api.py", line 259, in request
    raise Forbidden(resp)
tweepy.errors.Forbidden: 403 Forbidden
453 - You currently have Essential access which includes access to Twitter API v2 endpoints only. If you need access to this endpoint,
 you’ll need to apply for Elevated access via the Developer Portal. You can learn more here: https://developer.twitter.com/en/docs/twi
tter-api/getting-started/about-twitter-api#v2-access-leve
  • ⇒ Of course… 😂
  • So, let's just say we bypass verifying credentials for now, could we at least send a simple post ?
  • Okay, so same error if I just try to call update_status as follow: 
                status = "This is my first post to Twitter using the API. I am still learning, please be kind :)"
            self.api.update_status(status=status)
I tried to look for another package that will support the API v2 in python but could not find anything so far.
  • Waaaooohh… amazing! It works 🥳!
  • ⇒ I removed the tweepy and then added the requests-oauthlib package in my env instead: 
        "social_env": {
      "packages": ["requests", "jstyleson", "xxhash", "requests-oauthlib"]
    }
  • And updated the code with a minimal “send_message” function: 
        def process_command(self, cmd):
        """Check if this component can process the given command"""

        if cmd == 'send':
            logger.info("Should send a message here.")
            # self.setup_api()

            # status = "This is my first post to Twitter using the API. I am still learning, please be kind :)"
            # self.api.update_status(status=status)
            msg = self.get_param("message")
            self.send_message(msg)
            return True

        return False

    def send_message(self, msg):
        """Send a given message"""
        # logger.info("Should send message '%s'", msg)
        # We read the key/secret from our config:
        api_key = self.config["api_key"]
        api_secrets = self.config["api_key_secret"]
        access_token = self.config["access_token"]
        access_secret = self.config["access_token_secret"]

        # Make the request
        oauth = OAuth1Session(
            api_key,
            client_secret=api_secrets,
            resource_owner_key=access_token,
            resource_owner_secret=access_secret,
        )

        # Be sure to add replace the text of the with the text you wish to Tweet. You can also add parameters
        # to post polls, quote Tweets, Tweet with reply settings, and Tweet to Super Followers
        # in addition to other features.
        payload = {"text": msg}

        # Making the request
        response = oauth.post(
            "https://api.twitter.com/2/tweets",
            json=payload,
        )

        if response.status_code != 201:
            raise Exception(f"Request returned an error: {response.status_code} {response.text}")

        logger.info("Response code: %d", response.status_code)

        # Saving the response as JSON
        json_response = response.json()
        logger.info(json.dumps(json_response, indent=4, sort_keys=True))
  • And I could send a message from the command line: 
    $ nvp run tweet send "hello world from python."
2022/05/12 16:01:18 [__main__] INFO: Should send a message here.
2022/05/12 16:01:19 [__main__] INFO: Response code: 201
2022/05/12 16:01:19 [__main__] INFO: {
    "data": {
        "id": "1524766675698077698",
        "text": "hello world from python."
    }
}
  • And this will appear on my twitter account:

  • Yeepeeee! 🥳✌!

Uploading a media file

  • ⇒ let's see how I can integrate that in my component…
  • Hmmm.. first things first, it seems this example code is using the API v1.1 so not good for me (?) Let's see if we have something similar in API v2.
  • Well, no… 😰 Apparently there is currently no support to import media files in the API version 2.
  • ⇒ So, I decided I should not make my life any harder, and just made a request to get an elevated access to the twitter API. We will see how this goes.

Build an initial GUI application to handle the tweets

  • Until my elevated access request above is accepted I think I should not focus that much on the API usage itself.
  • But what I could do instead would be to start building a minimal GUI to manage my twitter access.
  • Thus, I started with creating a kind of “base skeleton” that I may then use for multiple application, which I called AppBase naturally: 
    """Module for AppBase class definition"""

import logging
import sys

from PyQt5.QtWidgets import QApplication

from nvp.nvp_context import NVPContext
from nvp.nvp_component import NVPComponent

from nvh.gui.main_window import MainWindow

logger = logging.getLogger(__name__)


class AppBase(NVPComponent):
    """AppBase component class"""

    def __init__(self, ctx: NVPContext, app_name: str):
        """class constructor, should provide an app name that will be used
        to select the config element to load."""

        NVPComponent.__init__(self, ctx)
        self.app_name = app_name
        self.config = self.config[app_name]
        self.app = None
        self.main_window = None

    def get_app(self):
        """Retrieve the Qapplication"""
        return self.app

    def get_config(self):
        """Retrieve the config for this application"""
        return self.config

    def get_main_window(self):
        """Retrieve the main window of the application"""
        return self.main_window

    def build_app(self):
        """Build the application."""

        # Simple base build of an empty application:
        self.app = QApplication(sys.argv)

        self.main_window = MainWindow(self)

    def run_app(self):
        """Run the application."""

        # First we build the app:
        self.build_app()

        # Display the main window:
        self.main_window.show()

        # Run the event loop:
        self.app.exec()
  • In this AppBase I provide the minimal support to setup the application level config, create a MainWindow, and also run the application with the run_app method.
  • Then of course I also had to build that generic MainWindow class: 
    ""Module for the main window of the FCITuner application"""

import logging
from PyQt5.QtCore import Qt
from PyQt5 import QtCore
import PyQt5.QtWidgets as qwd

from nvp.nvp_context import NVPContext

logger = logging.getLogger(__name__)


class MainWindow(qwd.QMainWindow):
    """Main Window class for FCITuner application"""

    def __init__(self, app):
        """Constructor of the MainWindow class"""
        super().__init__()

        self.app = app

        # Retrieve the app config:
        self.config = app.get_config()

        # keep ref on the context:
        self.ctx = NVPContext.get()

        title = self.config["application_title"]
        self.setWindowTitle(title)
  • As you can see, this main window will already read its application title from the application config provided by the AppBase.
  • Then I started to build a specific application, which I called NervGate on top of the AppBase: 
    """Module for NervGate class definition"""

import logging

from nvp.nvp_context import NVPContext

from nvh.gui.app_base import AppBase

logger = logging.getLogger(__name__)


class NervGate(AppBase):
    """NervGate component class"""

    def __init__(self, ctx: NVPContext):
        """class constructor"""
        AppBase.__init__(self, ctx, "nervgate")

    def process_command(self, cmd):
        """Check if this component can process the given command"""

        if cmd == 'run':
            self.run_app()
            return True

        return False

    def build_app(self):
        """Re-implementation of the build function"""
        super().build_app()

        win = self.get_main_window()
        # win.setWindowTitle("NervGate")

        win.setFixedSize(800, 600)


if __name__ == "__main__":
    # Create the context:
    context = NVPContext()

    # Add our component:
    comp = context.register_component("NervGate", NervGate(context))

    context.define_subparsers("main", {
        'run': None,
    })

    psr = context.get_parser('main.run')
    # psr.add_argument("message", type=str,
    #                  help="Simple message that we should send")

    comp.run()
  • Also added a dedicated python env and a new script to start that application: 
        "nvgate": {
      "custom_python_env": "nvgate_env",
      "cmd": "${PYTHON} ${PROJECT_ROOT_DIR}/nvh/app/nerv_gate.py run",
      "python_path": ["${PROJECT_ROOT_DIR}", "${NVP_ROOT_DIR}"]
    }
  • And now, as expected I can display an empty window with the desired title by runnign the command: 
    $ nvp run nvgate

Add a Twitter panel to the NervGate app

  • Naturally, next I should add a “twitter” panel/widget to send some tweets 😝
  • I started with the definition of a DockWidget class to support flexible docking in my app: 
    """Base DockWidget module"""

# cf. https://stackoverflow.com/questions/63638969/pyqt5-move-qdockwidget-by-dragging-tab

import logging
import PyQt5.QtWidgets as qwd
# from PyQt5 import QtCore

logger = logging.getLogger(__name__)


class DockWidget(qwd.QDockWidget):
    """Re-implementation of Dock widget."""

    def __init__(self, title: str):
        """Constructor for dock widget."""
        super().__init__(title)
        self.title = title
        self.setMinimumSize(200, 100)
        self.setTitleBarWidget(qwd.QWidget(self))
        self.dockLocationChanged.connect(self.on_dock_location_changed)

    def on_dock_location_changed(self):
        """Handle the display/hiding of the tab titles"""

        main: qwd.QMainWindow = self.parent()
        all_dock_widgets = main.findChildren(qwd.QDockWidget)

        for dock_widget in all_dock_widgets:
            sibling_tabs = main.tabifiedDockWidgets(dock_widget)
            # If you pull a tab out of a group the other tabs still see it as a sibling while dragging...
            sibling_tabs = [s for s in sibling_tabs if not s.isFloating()]

            if len(sibling_tabs) != 0:
                # Hide title bar
                # logger.info("Showing title for %s", dock_widget.title)
                dock_widget.setTitleBarWidget(qwd.QWidget(self))
                # dock_widget.titleBarWidget().setFixedHeight(30)

            else:
                # Re-enable title bar
                # logger.info("Hidding title for %s", dock_widget.title)
                # dock_widget.titleBarWidget().setFixedHeight(20)
                dock_widget.setTitleBarWidget(None)

    # def minimumSizeHint(self) -> QtCore.QSize:
    #     """Return min size hint"""
    #     return QtCore.QSize(100, 100)
  • And then started to build a minimal TwitterPanel widget: 
    """TwitterPanel widget module"""

import logging
from PyQt5 import QtCore
from PyQt5.QtCore import Qt
import PyQt5.QtWidgets as qwd

logger = logging.getLogger(__name__)


class TwitterPanel(qwd.QWidget):
    """TwitterPanel class"""

    def __init__(self):
        """Constructor of the TwitterPanel class"""
        super().__init__()

        lyt = qwd.QVBoxLayout()
        lyt.setContentsMargins(0, 0, 0, 0)

        # Add a simple label:
        label = qwd.QLabel(self)
        label.setText("This is the twitter panel")
        label.setAlignment(Qt.AlignCenter)
        lyt.addWidget(label)

        self.setLayout(lyt)

        setattr(self, "sizeHint", lambda: QtCore.QSize(360, 300))
  • Then I simply added that new widget in a dock inside the main window on construction: 
        def build_app(self):
        """Re-implementation of the build function"""
        super().build_app()

        win = self.get_main_window()
        # win.setWindowTitle("NervGate")

        # win.setFixedSize(800, 600)

        dock1 = DockWidget("Twitter")
        dock1.setWidget(TwitterPanel())

        dock1.setAllowedAreas(Qt.AllDockWidgetAreas)

        win.addDockWidget(Qt.RightDockWidgetArea, dock1)
        win.setDockOptions(win.GroupedDragging | win.AllowTabbedDocks | win.AllowNestedDocks)
        win.setTabPosition(Qt.AllDockWidgetAreas, qwd.QTabWidget.North)

        # Set a size hint on the window:
        setattr(win, "sizeHint", lambda: QtCore.QSize(600, 300))
  • And this is the result I get so far:

  • ⇒ The resulting panel is detachable, draggable, re-attachable, etc… so, all good so far ✌!
  • Next, let just add a text area to type in a message and a send button to send the tweet message: 
            lyt = qwd.QVBoxLayout()
        lyt.setContentsMargins(5, 5, 5, 5)
        lyt.setSpacing(5)

        lyt.addStretch(1)

        # Add a simple label:
        label = qwd.QLabel(self)
        label.setText("Message text:")
        label.setAlignment(Qt.AlignLeft)
        lyt.addWidget(label)

        wid = qwd.QTextEdit()
        wid.setMaximumHeight(100)
        lyt.addWidget(wid, stretch=0)

        row = qwd.QHBoxLayout()
        row.setContentsMargins(0, 0, 0, 0)
        row.setSpacing(5)

        row.addStretch(1)

        btn = gui.create_bitmap_button("mail-send", text=" Send")
        row.addWidget(btn)
        lyt.addLayout(row, stretch=0)

        self.setLayout(lyt)
While I was at it, I also added support to set icons on the “Send” button and for the application itself 😜
  • So here is the view we get now:

Handling the Send button action

  • Now we should handle the user click on the send button
  • So connecting the button to an action: 
            btn = gui.create_bitmap_button("mail-send", text=" Send")
        btn.pressed.connect(self.send_message)
  • And defining the action: 
        def send_message(self):
        """Send the message currently typed in the message edit text"""
        msg = self.message_edit.toPlainText()

        logger.info("Should send the message: '%s'", msg)

        self.message_edit.setText("")
  • ⇒ This workds correctly.
  • But to actually send the message, I need access to the twitter_handler I created previously. I could construct/load that component when building the application, but I think I have a better idea on this already: I could instead just prepare a list of “components” that the application may need, and then only lazily load those components when requested with the get_component() method, how about that 🤪?! Let's check if this will work as expected…
  • So in my NervProj config file I'm now adding a list of “dynamic components” like that: 
      "+components": {
    "twitter": "nvh.social.twitter_handler"
  },
The “+” in front of the “components” key here is just the mechanism I use to notify that the content of this dict, which is read from an additional “user config” file in my case, not just the NervProj “base config” file, should be appended/added to the existing dict instead of simply replacing it.
  • Next I need the NVPContext to discover those dynamic components when needed: 
        def get_component(self, cname, do_init=True):
        """Retrieve a component by name or create it if missing"""

        proj = self.get_current_project()
        if proj is not None and proj.has_component(cname):
            return proj.get_component(cname, do_init)

        if cname in self.components:
            comp = self.components[cname]
            if do_init:
                comp.initialize()
            return comp

        # If the requested component is not found, then it might be a dynamic component:
        # So we search for it in our config:
        dyn_comps = self.config.get("components", {})

        if cname not in dyn_comps:
            logger.warning("Cannot find component %s", cname)
            return None

        # We have a dyn component module name, so we try to load it:
        mname = dyn_comps[cname]
        logger.info("Loading dynamic component %s from module %s", cname, mname)

        comp_module = import_module(mname)
        comp = comp_module.create_component(self)

        # And we register that component now:
        self.register_component(cname, comp)

        if do_init:
            comp.initialize()
        return comp
  • Which also means that I need a create_component() function in each component file that I should be able to load dynamically: 
    def create_component(ctx: NVPContext):
    """Create an instance of the component"""
    return TwitterHandler(ctx)
  • And with that in place, I updated the action for the send button on my gui: 
        def send_message(self):
        """Send the message currently typed in the message edit text"""
        msg = self.message_edit.toPlainText()

        # logger.info("Should send the message: '%s'", msg)
        twitter: TwitterHandler = NVPContext.get().get_component('twitter')

        twitter.send_message(msg)

        self.message_edit.setText("")
  • And finally, time to give it a try (crossing fingers 🤞): 
    $ nvp run nvgate
2022/05/13 10:39:44 [nvp.nvp_context] INFO: Loading dynamic component twitter from module nvh.social.twitter_handler
2022/05/13 10:39:44 [nvh.social.twitter_handler] INFO: Response code: 201
2022/05/13 10:39:44 [nvh.social.twitter_handler] INFO: {
    "data": {
        "id": "1525048133083865088",
        "text": "Hello twitter world! What a wonderfull day it is today to try some twitter automation, isn't it ?"
    }
}
  • ⇒ Ohh my god… this works just fine already 😭 [i'm so happy that I'm crying!]. Here is what I see on my twitter account:

Conclusion

⇒ Okay, this is a pretty large article/dev session already, so i think I should stop it here for the moment. We will get back to this when I get an elevated twitte access to implement support to send images, etc. Meanwhile, Have a nice day everyone! 🖐

  • blog/2022/0512_nervhome_twitter_handler.txt
  • Last modified: 2022/05/13 10:02
  • by 127.0.0.1