Twitter Sentimental Analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b57a52b5",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "import tweepy\n",
    "from tweepy import OAuthHandler\n",
    "from textblob import TextBlob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "4f772305",
   "metadata": {},
   "outputs": [],
   "source": [
    "class TwitterClient(object):\n",
    "    '''\n",
    "    Generic Twitter Class for sentiment analysis.\n",
    "    '''\n",
    "    def __init__(self):\n",
    "        '''\n",
    "        Class constructor or initialization method.\n",
    "        '''\n",
    "        # keys and tokens from the Twitter Dev Console\n",
    "        consumer_key = '1x2JwsBFLyvo0XMx9ypE9YPS9'\n",
    "        consumer_secret = 'D9gddVi4jCNNATuRgq7ubmI9tLwxkzP59RrGSyA5x9CUo50y36'\n",
    "        access_token = '1305735738131206144-Z9tG1RVMMtUrFOy1VRb5JyvF5Rinem'\n",
    "        access_token_secret = 'tkdeVKseVDjDAlB6lSgvkp5Pf1T0am3xcU63WQGK6uYqc'\n",
    " \n",
    "        # attempt authentication\n",
    "        try:\n",
    "            # create OAuthHandler object\n",
    "            self.auth = OAuthHandler(consumer_key, consumer_secret)\n",
    "            # set access token and secret\n",
    "            self.auth.set_access_token(access_token, access_token_secret)\n",
    "            # create tweepy API object to fetch tweets\n",
    "            self.api = tweepy.API(self.auth)\n",
    "        except:\n",
    "            print(\"Error: Authentication Failed\")\n",
    " \n",
    "    def clean_tweet(self, tweet):\n",
    "        '''\n",
    "        Utility function to clean tweet text by removing links, special characters\n",
    "        using simple regex statements.\n",
    "        '''\n",
    "        return ' '.join(re.sub(\"(@[A-Za-z0-9]+)|([^0-9A-Za-z \\t]) |(\\w+:\\/\\/\\S+)\", \" \", tweet).split())\n",
    " \n",
    "    def get_tweet_sentiment(self, tweet):\n",
    "        '''\n",
    "        Utility function to classify sentiment of passed tweet\n",
    "        using textblob's sentiment method\n",
    "        '''\n",
    "        # create TextBlob object of passed tweet text\n",
    "        analysis = TextBlob(self.clean_tweet(tweet))\n",
    "        # set sentiment\n",
    "        if analysis.sentiment.polarity > 0:\n",
    "            return 'positive'\n",
    "        elif analysis.sentiment.polarity == 0:\n",
    "            return 'neutral'\n",
    "        else:\n",
    "            return 'negative'\n",
    " \n",
    "    def get_tweets(self, query, count = 10):\n",
    "        '''\n",
    "        Main function to fetch tweets and parse them.\n",
    "        '''\n",
    "        # empty list to store parsed tweets\n",
    "        tweets = []\n",
    " \n",
    "        try:\n",
    "            # call twitter api to fetch tweets\n",
    "            fetched_tweets = self.api.search_tweets(q = query, count = count)\n",
    " \n",
    "            # parsing tweets one by one\n",
    "            for tweet in fetched_tweets:\n",
    "                # empty dictionary to store required params of a tweet\n",
    "                parsed_tweet = {}\n",
    " \n",
    "                # saving text of tweet\n",
    "                parsed_tweet['text'] = tweet.text\n",
    "                # saving sentiment of tweet\n",
    "                parsed_tweet['sentiment'] = self.get_tweet_sentiment(tweet.text)\n",
    " \n",
    "                # appending parsed tweet to tweets list\n",
    "                if tweet.retweet_count > 0:\n",
    "                    # if tweet has retweets, ensure that it is appended only once\n",
    "                    if parsed_tweet not in tweets:\n",
    "                        tweets.append(parsed_tweet)\n",
    "                else:\n",
    "                    tweets.append(parsed_tweet)\n",
    " \n",
    "            # return parsed tweets\n",
    "            return tweets\n",
    " \n",
    "        except tweepy.errors.TweepyException as e:\n",
    "            # print error (if any)\n",
    "            print(\"Error : \" + str(e))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "44aa4bcc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Error : 403 Forbidden\n",
      "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/twitter-api/getting-started/about-twitter-api#v2-access-leve\n"
     ]
    },
    {
     "ename": "TypeError",
     "evalue": "'NoneType' object is not iterable",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[19], line 31\u001b[0m\n\u001b[0;32m     27\u001b[0m         \u001b[38;5;28mprint\u001b[39m(tweet[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mtext\u001b[39m\u001b[38;5;124m'\u001b[39m])\n\u001b[0;32m     29\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;18m__name__\u001b[39m \u001b[38;5;241m==\u001b[39m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m__main__\u001b[39m\u001b[38;5;124m\"\u001b[39m:\n\u001b[0;32m     30\u001b[0m     \u001b[38;5;66;03m# calling main function\u001b[39;00m\n\u001b[1;32m---> 31\u001b[0m     \u001b[43mmain\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "Cell \u001b[1;32mIn[19], line 8\u001b[0m, in \u001b[0;36mmain\u001b[1;34m()\u001b[0m\n\u001b[0;32m      5\u001b[0m tweets \u001b[38;5;241m=\u001b[39m api\u001b[38;5;241m.\u001b[39mget_tweets(query \u001b[38;5;241m=\u001b[39m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mElon Musk\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m      7\u001b[0m \u001b[38;5;66;03m# picking positive tweets from tweets\u001b[39;00m\n\u001b[1;32m----> 8\u001b[0m ptweets \u001b[38;5;241m=\u001b[39m [tweet \u001b[38;5;28;01mfor\u001b[39;00m tweet \u001b[38;5;129;01min\u001b[39;00m tweets \u001b[38;5;28;01mif\u001b[39;00m tweet[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124msentiment\u001b[39m\u001b[38;5;124m'\u001b[39m] \u001b[38;5;241m==\u001b[39m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mpositive\u001b[39m\u001b[38;5;124m'\u001b[39m]\n\u001b[0;32m      9\u001b[0m \u001b[38;5;66;03m# percentage of positive tweets\u001b[39;00m\n\u001b[0;32m     10\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mPositive tweets percentage: \u001b[39m\u001b[38;5;132;01m{}\u001b[39;00m\u001b[38;5;124m \u001b[39m\u001b[38;5;124m%\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39mformat(\u001b[38;5;241m100\u001b[39m\u001b[38;5;241m*\u001b[39m\u001b[38;5;28mlen\u001b[39m(ptweets)\u001b[38;5;241m/\u001b[39m\u001b[38;5;28mlen\u001b[39m(tweets)))\n",
      "\u001b[1;31mTypeError\u001b[0m: 'NoneType' object is not iterable"
     ]
    }
   ],
   "source": [
    "def main():\n",
    "    # creating object of TwitterClient Class\n",
    "    api = TwitterClient()\n",
    "    # calling function to get tweets\n",
    "    tweets = api.get_tweets(query = 'Elon Musk')\n",
    "\n",
    "    # picking positive tweets from tweets\n",
    "    ptweets = [tweet for tweet in tweets if tweet['sentiment'] == 'positive']\n",
    "    # percentage of positive tweets\n",
    "    print(\"Positive tweets percentage: {} %\".format(100*len(ptweets)/len(tweets)))\n",
    "    # picking negative tweets from tweets\n",
    "    ntweets = [tweet for tweet in tweets if tweet['sentiment'] == 'negative']\n",
    "    # percentage of negative tweets\n",
    "    print(\"Negative tweets percentage: {} %\".format(100*len(ntweets)/len(tweets)))\n",
    "    # percentage of neutral tweets\n",
    "    print(\"Neutral tweets percentage: {} % \\\n",
    "        \".format(100*(len(tweets) -(len( ntweets )+len( ptweets)))/len(tweets)))\n",
    " \n",
    "    # printing first 5 positive tweets\n",
    "    print(\"\\n\\nPositive tweets:\")\n",
    "    for tweet in ptweets[:10]:\n",
    "        print(tweet['text'])\n",
    " \n",
    "    # printing first 5 negative tweets\n",
    "    print(\"\\n\\nNegative tweets:\")\n",
    "    for tweet in ntweets[:10]:\n",
    "        print(tweet['text'])\n",
    " \n",
    "if __name__ == \"__main__\":\n",
    "    # calling main function\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "60c4b7c8",
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}