Presentazione.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "7d69daf7-c471-47ec-a3c2-dad926449692",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Twitch AI Powered Chat Analizer\n",
    "<img src=\"NotebookPictures/TwitchLogo.png\" alt=\"Alternative text\" width=\"500\" height=\"340\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42ac00f1-ab40-4460-8d7e-081a677c94ed",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "## Cos'è Twitch?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c0a785d5-181f-4d63-b34a-798af5d59627",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Twitch1.png\" alt=\"Twitch Homepage\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bea68217-4ddf-4fc6-9cff-d38c21d84fe3",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "- Livestreams di varie categorie\n",
    "- Gaming\n",
    "- Talk Show\n",
    "- Utenti: Abbonati (Sub) e Non Abbonati (Not Sub)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9349ca7b-39be-4b78-98be-36ac1b78970d",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/TwitchChat.png\" alt=\"Twitch Chat\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "913d7434-45f8-43ad-bce0-8cb3753f7dfa",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "## Scopo del progetto"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9b4800b8-661f-4435-a222-ec4ee427d041",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "fragment"
    },
    "tags": []
   },
   "source": [
    "- Fornire uno strumento ai creator o utenti \"curiosi\"\n",
    "- Arricchire i messaggi della chat analizzandone **sentiment** ed **emotion**\n",
    "- Interfaccia semplice che dia informazioni sulla chat a colpo d'occhio"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1d6bfae-423b-4fb6-8c04-0bf7135b88c2",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/AIChat.png\" alt=\"AI Powered Chat\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed9f70db-020a-462b-a1a1-5fdb2d4644fe",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Fornire **metriche** sull'andamento della chat:\n",
    "- Variazione del sentiment del tempo, in generale, per i sub, e per i not sub\n",
    "- Attività della chat\n",
    "- Proporzione delle emotion\n",
    "- Proporzione di messaggi ricevuti da Sub e non Sub"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e79db99-903e-4d2c-9fd0-60e6e3c71ba8",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/GeneralSentiment.png\" alt=\"General Sentiment\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3b433d46-f1fe-4ca1-b739-364649c0e889",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/SubSentiment.png\" alt=\"Sub Sentiment\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fb53d616-15a8-47ac-ba4c-ad87cc3d705d",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/NotSubSentiment.png\" alt=\"Not Sub Sentiment\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "caa71b45-9bf5-47fd-9a2f-f9008685d572",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/ChatActivity.png\" alt=\"Chat Activity\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dbcf82f4-461e-4e27-9674-da44d71133fe",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/SubPortion.png\" alt=\"Sub Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "23ab65e8-15d4-4325-b1ed-c3aa7045af0b",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/EmotionPortion.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42737672-10bd-4768-89fd-191b31796f3c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Live Demo!"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dfa06ac2-1932-4580-9a6c-0e162f5cf0d0",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Struttura del progetto"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "03862364-94fc-4e41-bf16-9ccd966d285d",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Tecnologie usate:\n",
    "- Logstash\n",
    "- Kafka\n",
    "- Kafka Streams\n",
    "- Spark\n",
    "- Elastic Search\n",
    "- Kibana"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1022128d-6199-4787-8775-7de53e61e97c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Altri moduli/pacchetti utilizzati:\n",
    "- Flask\n",
    "- TwitchIO APIs\n",
    "- Twitch REST APIs\n",
    "- Java Watchservice\n",
    "- OpenAI APIs (Chat GPT 3.5 Turbo)\n",
    "- SocketIO"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "001f2e6d-7f83-4c81-9683-a5a28e3d0f31",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura1.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15ccd5d5-c7c8-4c0e-99cb-bdd7e41b17d5",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura2.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ccc798a9-58cf-4c1e-b7b1-a64ef9d8073d",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura3.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f91fdcf-1a6c-489c-80b5-aa6984825cc9",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura4.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b338bcab-161f-4f89-be13-70947f5ab091",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura5.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f7c99d8-c88b-481e-9e66-a0c53bde093c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura6.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4d8fb76b-0c29-46ef-bd1b-885016e7adb5",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura7.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cef34bd2-45c9-4921-963d-0e62b5fe5edd",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura8.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93891e2f-491c-477d-ac24-0a099730c3b0",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "<img src=\"NotebookPictures/Struttura9.png\" alt=\"Emotion Portion\" height=\"720\" width=\"1280\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4feec1f5-0485-42cd-bf5b-181575015337",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Approfondimenti sul codice"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d666eb1-df36-4b85-90dd-af8b30e8dcef",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Tutto il sistema gira su container docker, i quali vengono creati con un docker compose:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "18c42bbd-7bea-4201-8289-09cee0a45408",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "version: '3.8'\n",
    "services:\n",
    "\n",
    "  zookeeper:\n",
    "    hostname: zookeeper\n",
    "    image: confluentinc/cp-zookeeper\n",
    "    container_name: zookeeper\n",
    "    ports:\n",
    "      - \"2181:2181\"\n",
    "    environment:\n",
    "      ZOOKEEPER_CLIENT_PORT: 2181\n",
    "\n",
    "  kafka:\n",
    "    image: confluentinc/cp-kafka\n",
    "    hostname: kafka\n",
    "    container_name: kafka\n",
    "    ports:\n",
    "      - \"9092:9092\"\n",
    "    environment:\n",
    "      KAFKA_ADVERTISED_HOST_NAME: host.docker.internal\n",
    "      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\n",
    "      ALLOW_PLAINTEXT_LISTENER: yes\n",
    "      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://host.docker.internal:9092\n",
    "      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1\n",
    "    depends_on:\n",
    "      - zookeeper"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb6c7364-201b-424a-be4b-8ddbb1fc4e49",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "twitch-chat-ingestor:\n",
    "    hostname: twitch-chat-ingestor\n",
    "    image: docker.elastic.co/logstash/logstash:8.13.0\n",
    "    container_name: twitch-chat-ingestor\n",
    "    environment:\n",
    "      XPACK_MONITORING_ENABLED: \"false\"\n",
    "    ports:\n",
    "      - 9090:9090\n",
    "    volumes:\n",
    "      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf\n",
    "    depends_on:\n",
    "      - kafka\n",
    "\n",
    "  init-kafka:\n",
    "    image: confluentinc/cp-kafka\n",
    "    entrypoint: ['/bin/sh', '-c']\n",
    "    command: |\n",
    "      \"\n",
    "      echo -e 'Creating kafka topics...'\n",
    "      kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic general --replication-factor 1 --partitions 1\n",
    "      kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic no_emotes_general --replication-factor 1 --partitions 1\n",
    "      kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic enriched_chat --replication-factor 1 --partitions 1\n",
    "\n",
    "      echo -e 'Successfully created the following topics:'\n",
    "      kafka-topics --bootstrap-server kafka:9092 --list\n",
    "      \"\n",
    "    depends_on:\n",
    "      - kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5bd3a193-9965-4002-a969-6eb41f859d7c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "kafka-ui:\n",
    "    image: provectuslabs/kafka-ui:latest\n",
    "    container_name: kafka-ui\n",
    "    ports:\n",
    "      - \"8080:8080\"\n",
    "    environment:\n",
    "      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092\n",
    "      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181\n",
    "    depends_on:\n",
    "      - kafka\n",
    "\n",
    "  emote_downloader:\n",
    "    build:\n",
    "      context: ./python_emote_downloader\n",
    "      dockerfile: Dockerfile\n",
    "    container_name: emote_downloader\n",
    "    hostname: emote_downloader\n",
    "    ports:\n",
    "      - \"12345:12345\"\n",
    "    volumes:\n",
    "      - E:\\Uni\\TAP\\twitchsentimentanalisys/python_emote_downloader:/app\n",
    "    command: >\n",
    "      python getEmotes.py\n",
    "    depends_on:\n",
    "      - kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bc30fec4-4590-41fe-844a-5eaf9b0a9414",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "kafka-stream:\n",
    "    hostname: kafka-stream\n",
    "    container_name: kafka-stream\n",
    "    build:\n",
    "      context: ./filter_emotes\n",
    "      dockerfile: Dockerfile\n",
    "    volumes:\n",
    "      - E:\\Uni\\TAP\\TwitchSentimentAnalisys\\filter_emotes\\app\\build\\libs\\app-1.0.jar:/app/app-1.0.jar\n",
    "      - E:\\Uni\\TAP\\twitchsentimentanalisys/python_emote_downloader:/app\n",
    "    command: bash -c \"ls && java -jar app-1.0.jar\"\n",
    "    depends_on:\n",
    "      - emote_downloader\n",
    "      - init-kafka\n",
    "\n",
    "  elasticsearch:\n",
    "    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.4\n",
    "    container_name: elasticsearch\n",
    "    ports: \n",
    "      - \"9200:9200\"\n",
    "      - \"9300:9300\"\n",
    "    environment:\n",
    "      - \"discovery.type=single-node\"\n",
    "      - \"xpack.security.enabled=false\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "45f11b89-f4d6-4eb5-b03c-063a632d5e47",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "kibana:\n",
    "    hostname: kibana\n",
    "    image: docker.elastic.co/kibana/kibana:8.13.4\n",
    "    ports:\n",
    "      - 5601:5601\n",
    "    environment:\n",
    "      - \"xpack.security.enabled=false\"\n",
    "    volumes:\n",
    "      - E:\\Uni\\TAP\\twitchsentimentanalisys/spark_sentiment/kibana.yml:/usr/share/kibana/config/kibana.yml\n",
    "    depends_on:\n",
    "      - elasticsearch\n",
    "\n",
    "  spark:\n",
    "    build:\n",
    "      context: ./spark_sentiment\n",
    "      dockerfile: Dockerfile\n",
    "    container_name: spark\n",
    "    hostname: spark\n",
    "    ports:\n",
    "      - \"4040:4040\"\n",
    "    volumes:\n",
    "      - E:\\Uni\\TAP\\twitchsentimentanalisys/spark_sentiment:/opt/twitchsentimentanalisys\n",
    "    command: >\n",
    "      /opt/spark/bin/spark-submit --conf spark.driver.extraJavaOptions=\"-Divy.cache.dir=/tmp -Divy.home=/tmp\" --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.2,org.elasticsearch:elasticsearch-spark-30_2.12:8.13.4 /opt/twitchsentimentanalisys/spark_sentiment.py\n",
    "    depends_on:\n",
    "      init-kafka:\n",
    "        condition: service_completed_successfully\n",
    "      elasticsearch:\n",
    "        condition: service_started"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a662c92f-cfe8-48ae-8d8a-b9345f476c80",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "flask:\n",
    "    build:\n",
    "      context: ./python_chat_listener\n",
    "      dockerfile: Dockerfile\n",
    "    ports:\n",
    "      - \"5000:5000\"\n",
    "    command: python web_interface.py\n",
    "    volumes:\n",
    "      - ./python_chat_listener:/app\n",
    "      - ./python_emote_downloader/.env:/app/.env\n",
    "    depends_on:\n",
    "      spark:\n",
    "        condition: service_started\n",
    "      kafka:\n",
    "        condition: service_started\n",
    "      init-kafka:\n",
    "        condition: service_completed_successfully\n",
    "      elasticsearch:\n",
    "        condition: service_started"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9f5a42a-a038-4f8f-89ba-8acf612ae49c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Web Server Flask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc3e54a0-a366-479d-b073-c715cd1624c3",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "@app.route('/')\n",
    "def home():\n",
    "    return render_template('index.html')\n",
    "\n",
    "@app.route('/script-status') # Endpoint per verificare lo stato del processo chat_listener\n",
    "def get_script_status():\n",
    "    if process is None:\n",
    "        return jsonify(status=False)\n",
    "    else:\n",
    "        return jsonify(status=True)\n",
    "\n",
    "@app.route('/update-channel', methods=['POST']) # Endpoint per aggiornare il canale nel file .env\n",
    "def update_channel():\n",
    "    new_channel = request.json['channel']\n",
    "    with open('.env', 'r') as file:\n",
    "        lines = file.readlines()\n",
    "    with open('.env', 'w') as file:\n",
    "        for line in lines:\n",
    "            if line.startswith('CHANNEL='):\n",
    "                file.write(f'CHANNEL={new_channel}\\n')\n",
    "            else:\n",
    "                file.write(line)\n",
    "    return jsonify(message='Canale aggiornato con successo!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7fc809bc-2fbc-4741-8749-6e4c2831d761",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "@app.route('/run-script') # Endpoint per avviare lo script chat_listener\n",
    "def run_script():\n",
    "    global process\n",
    "    script_path = './chat_listener.py'\n",
    "    \n",
    "    try:\n",
    "        # Esegui lo script in background\n",
    "        if process is None:\n",
    "            process = subprocess.Popen(['python', script_path])\n",
    "            print(f'[WEB_INTERFACE] - Script path: {script_path}', flush=True)\n",
    "            print(f'[WEB_INTERFACE] - Script avviato con PID: {process.pid}', flush=True)\n",
    "\n",
    "            response = requests.get('http://host.docker.internal:12345/get_emotes') # Richiede al servizio emote_downloader di scaricare le emote relative al canale\n",
    "            if response.status_code == 200:\n",
    "                print('Emote ottenute con successo:', response.json())\n",
    "            else:\n",
    "                print('Errore nella richiesta delle emote:', response.status_code)\n",
    "\n",
    "            return jsonify(message='Script avviato in background.'), 200\n",
    "        else: \n",
    "            print('Script già in esecuzione', flush=True)\n",
    "            return jsonify(message='Script già in esecuzione.'), 400\n",
    "        \n",
    "    except Exception as e:\n",
    "        print('Errore durante l\\'esecuzione del script:',str(e), flush=True)\n",
    "        return jsonify(message=str(e)), 500\n",
    "    \n",
    "@app.route('/stop-script') # Endpoint per fermare lo script chat_listener\n",
    "def stop_script():\n",
    "    global process\n",
    "    if process is not None:\n",
    "        process.terminate()  \n",
    "        process.wait()  \n",
    "        process = None\n",
    "        return jsonify(message='Script fermato'), 200\n",
    "    else:\n",
    "        return jsonify(message='Nessun script in esecuzione'), 400"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec1e2d9e-ce2d-4168-9595-a004e9d35c59",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Thread kafka consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6993925-1234-406a-93d2-943f67934381",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "@socketio.on('connect')\n",
    "def handle_connect():\n",
    "    emit('status', {'message': 'Connected to WebSocket'})\n",
    "\n",
    "def consume_messages(): # Funzione per consumare i messaggi dal servizio Kafka\n",
    "    consumer = kafka.KafkaConsumer('enriched_chat', bootstrap_servers=['host.docker.internal:9092'])\n",
    "    for message in consumer:\n",
    "        socketio.emit('new_message', {'message': message.value.decode()})\n",
    "        print(f'[CONSUMER] - {message.value.decode()}', flush=True)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "\n",
    "    thread = threading.Thread(target=consume_messages) # Avvia il thread per consumare messaggi da Kafka\n",
    "    thread.daemon = True  # Imposta il thread come daemon così si chiude con il programma\n",
    "    thread.start()\n",
    "\n",
    "    # Avvia il server Flask\n",
    "    socketio.run(app, debug=True, allow_unsafe_werkzeug=True, host='0.0.0.0')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "005968fc-0472-4e4f-b5be-adbf812c57f8",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Chat Listener Bot"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "686c8c9b-5082-47cf-87d6-2e0166b877f5",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "    async def event_message(self, message):\n",
    "        #if message.echo:\n",
    "        #    return\n",
    "\n",
    "        # Definisco il messaggio\n",
    "        msg = { \"channel\" : message.channel.name,\n",
    "                \"timestamp\" : message.timestamp.strftime('%d-%m-%Y %H:%M:%S'),\n",
    "                \"chatter_name\" : message.author.display_name,\n",
    "                \"is_broadcaster\" : message.author.is_broadcaster,\n",
    "                \"is_mod\" : message.author.is_mod,\n",
    "                \"is_subscriber\" : message.author.is_subscriber,\n",
    "                \"is_vip\" : message.author.is_vip,\n",
    "                \"content\" : message.content\n",
    "                }\n",
    "\n",
    "        print(msg, flush=True)\n",
    "        \n",
    "        try:\n",
    "            # Invia il messaggio al servizio di ingestion (logstash)\n",
    "            async with aiohttp.ClientSession() as session:\n",
    "                async with session.post(os.getenv('INGESTOR_URL'), json=msg) as response:\n",
    "                    response_text = await response.text()\n",
    "                    print(f\"Sent msg to '{os.getenv('INGESTOR_URL')}', RESPONSE = {response_text}\", flush=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3866aa3-6d0d-4959-a3d5-8e30dfe12709",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Configurazione Logstash"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b979630-09e2-4cdb-b8ca-a54e4c0abc42",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "input {\n",
    "  http {\n",
    "    id => \"kart_http_in\"\n",
    "    port => 9090\n",
    "  }\n",
    "}\n",
    "\n",
    "filter {\n",
    "    if [http][method] != \"POST\"{\n",
    "        drop {}\n",
    "    }\n",
    "\n",
    "    mutate {\n",
    "      remove_field => [\"user_agent\", \"event\", \"url\", \"http\", \"@timestamp\", \"host\", \"@version\"]\n",
    "    }\n",
    "}\n",
    "\n",
    "output {\n",
    "  kafka {\n",
    "    codec => json\n",
    "    topic_id => \"general\"\n",
    "    bootstrap_servers => \"kafka:9092\"\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1a08a99f-80db-477a-8aa8-b2d405e53bb1",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Emote Downloader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "874e172c-fe24-4ec7-8810-1c033cbe0b4f",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "@app.route('/get_emotes')\n",
    "def get_emotes():\n",
    "\n",
    "    emotes_list.clear()\n",
    "\n",
    "    dotenv_file = dotenv.find_dotenv('.env')\n",
    "    dotenv.load_dotenv(dotenv_file, override=True)\n",
    "    client_id = os.getenv('CLIENT_ID')\n",
    "    client_secret = os.getenv('SECRET')\n",
    "    nome_canale = os.getenv('CHANNEL')\n",
    "    print(nome_canale, client_id, client_secret)\n",
    "\n",
    "    channel_id = get_channel_id(nome_canale, client_id, client_secret)\n",
    "\n",
    "    get_twitch_emotes(channel_id, client_id, client_secret)\n",
    "    get_7tv_emotes(channel_id)\n",
    "    print('Done!', emotes_list, flush=True)\n",
    "\n",
    "    with open('emotes.json', 'w') as file:\n",
    "        json.dump(emotes_list, file, indent=4) # Dump on file\n",
    "\n",
    "    print('Emotes wrote on file \"emotes.json\"', flush=True)\n",
    "\n",
    "    return jsonify(emotes_list), 200"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6381b5e6-8b89-4462-8206-fb230574e808",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Kafka Streams"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01928074-c50b-4156-a1ed-99f3d6e19d06",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "private static void watchFileChanges(Path path) throws IOException, InterruptedException {\n",
    "        Path dirPath = path.toAbsolutePath().getParent();\n",
    "        if (dirPath == null) {\n",
    "            System.err.println(\"Il percorso assoluto non ha un genitore.\");\n",
    "            return;\n",
    "        }\n",
    "\n",
    "        WatchService watchService = FileSystems.getDefault().newWatchService();\n",
    "        dirPath.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY);\n",
    "\n",
    "        long lastModifiedTime = 0;\n",
    "        final long debounceTime = 3000; // Tempo di debounce in millisecondi\n",
    "\n",
    "        while (true) {\n",
    "            WatchKey key = watchService.take();\n",
    "            for (WatchEvent<?> event : key.pollEvents()) {\n",
    "                WatchEvent.Kind<?> kind = event.kind();\n",
    "                if (kind.equals(StandardWatchEventKinds.ENTRY_MODIFY)) {\n",
    "                    Path changed = (Path) event.context();\n",
    "                    if (changed.endsWith(path.getFileName())) {\n",
    "                        long currentTime = System.currentTimeMillis();\n",
    "                        if (currentTime - lastModifiedTime > debounceTime) {\n",
    "                            System.out.println(\"File emotes.json modificato. Riavvio del servizio...\");\n",
    "                            try {\n",
    "                                restartKafkaService(); // Se il file viene modificato riavvia il servizio\n",
    "                            } catch (Exception e) {\n",
    "                                System.err.println(\"Errore nel riavvio del servizio Kafka: \" + e.getMessage());\n",
    "                                e.printStackTrace();\n",
    "                            }\n",
    "                            lastModifiedTime = currentTime;\n",
    "                        }\n",
    "                    }\n",
    "                }\n",
    "            }\n",
    "            if (!key.reset()) {\n",
    "                break;\n",
    "            }\n",
    "        }\n",
    "    }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "59a6a3dc-14d4-4874-915f-66c57c822a00",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "private static void startKafkaService() throws Exception {\n",
    "        StreamsBuilder builder = new StreamsBuilder();\n",
    "        ObjectMapper objectMapper = new ObjectMapper();\n",
    "        ObjectMapper mapper = new ObjectMapper();\n",
    "\n",
    "        Properties properties = new Properties();\n",
    "        properties.put(\"bootstrap.servers\", \"host.docker.internal:9092\");\n",
    "        properties.put(\"application.id\", \"emoji-filter-app\");\n",
    "\n",
    "        try {\n",
    "\n",
    "            // Carica le emote dal file emotes.json\n",
    "            final Set<String>[] emojis = new Set[1];\n",
    "            emojis[0] = mapper.readValue(new File(\"emotes.json\"), Set.class);\n",
    "\n",
    "            System.out.println(emojis[0]);\n",
    "\n",
    "            // Filtra i messaggi che contengono emote\n",
    "            builder.<String, String>stream(\"general\", Consumed.with(Serdes.String(), Serdes.String()))\n",
    "                    .filter((key, value) -> {\n",
    "                        String content = \"\";\n",
    "                        try {\n",
    "                            JsonNode jsonNode = objectMapper.readTree(value);\n",
    "                            content = jsonNode.get(\"content\").asText();\n",
    "                        } catch (Exception e) {\n",
    "                            e.printStackTrace();\n",
    "                        }\n",
    "                        System.out.println('\\n' + content + '\\n');\n",
    "                        return Arrays.stream(content.split(\"\\\\s+\"))\n",
    "                                .noneMatch(word -> emojis[0].contains(word));\n",
    "                    })\n",
    "                    .to(\"no_emotes_general\");\n",
    "\n",
    "        } catch (IOException e) {\n",
    "            e.printStackTrace();\n",
    "        }\n",
    "\n",
    "        try {\n",
    "            streams = new KafkaStreams(builder.build(), properties);\n",
    "            streams.start();\n",
    "        } catch (Exception e) {\n",
    "            e.printStackTrace();\n",
    "        }\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "446e0e07-ac96-4669-b11f-3d82c7282216",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "## Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3605a1c5-2895-404e-93b9-def3fa016350",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "source": [
    "Configurazione di elastic search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4ae1cea-e120-400b-83e5-b4b284d9cb44",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "es = Elasticsearch(\"http://elasticsearch:9200\")\n",
    "sparkConf = SparkConf().set(\"es.nodes\", \"elasticsearch\") \\\n",
    "                        .set(\"es.port\", \"9200\") \\\n",
    "                        .set(\"es.index.auto.create\", \"true\")\n",
    "\n",
    "mapping = {\n",
    "    \"mappings\": {\n",
    "        \"properties\": {\n",
    "            \"timestamp\": {\n",
    "                \"type\": \"date\",  # Definisce il campo come tipo date\n",
    "                \"format\": \"epoch_millis\"  # Formati di data accettati\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "}\n",
    "\n",
    "elastic_index=\"twitchmessages\"\n",
    "\n",
    "if not es.indices.exists(index=elastic_index):\n",
    "    es.indices.create(index=elastic_index, body=mapping)\n",
    "else:\n",
    "    # Se l'indice esiste già e vuoi aggiornare il mapping\n",
    "    es.indices.put_mapping(index=elastic_index, body=mapping['mappings'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73aa4235-3ae8-4e61-9e36-88250ef8bd58",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Configurazione per leggere lo stream kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c20ef8e-57f8-4a9c-b6f6-742723341c7c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "kafkaServer = \"host.docker.internal:9092\"\n",
    "inputTopic = \"no_emotes_general\"\n",
    "kafkaOutputServer = \"host.docker.internal:9092\"\n",
    "outputTopic = \"enriched_chat\"\n",
    "\n",
    "df = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", kafkaServer) \\\n",
    "    .option(\"subscribe\", inputTopic) \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f4ccc943-e8f9-4964-b479-dac69aa27bae",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Operazioni sullo stream di messaggi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e70655b-a7af-4ee3-9cb1-844e374c5693",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Select the value and timestamp fields and cast them to string\n",
    "df = df.selectExpr(\"CAST(value AS STRING)\", \"CAST(timestamp AS TIMESTAMP)\")\n",
    "\n",
    "# Parse the JSON content and extract fields\n",
    "json_df = df.withColumn(\"json_data\", from_json(col(\"value\"), json_schema)) \\\n",
    "            .select(\"json_data.*\", col(\"timestamp\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "092e5deb-4e1c-4ac0-b1b0-4d7e453750ca",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "Arricchimento con ChatGPT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8fcd493c-3dd0-43dd-ab16-f4001ae94d33",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def chatgpt_sentiment(text):\n",
    "    if text is not None and text.strip() != \"\":\n",
    "        client = OpenAI(api_key=OPENAI_API_KEY)   \n",
    "        response = client.chat.completions.create(\n",
    "            model='gpt-3.5-turbo',\n",
    "            messages=[{'role': 'system', 'content': 'Classify the sentiment of the following message. Sentiment is a float from -1 to 1, where -1 is negative, 0 is neutral and 1 is positive. Emotion is a word representing the emotion of the message, can be one of the following: happy, sad, angry, neutral, surprised, disgusted, fearful. The response must be a JSON with two fields: sentiment and emotion'}, {\"role\": \"user\", \"content\": text}],\n",
    "            max_tokens=100,\n",
    "            temperature=0\n",
    "        )\n",
    "\n",
    "        data = json.loads(response.model_dump_json())\n",
    "        print(json.dumps(data, indent=4))\n",
    "        \n",
    "        # Estrai il sentiment o l'emozione dalla risposta\n",
    "        parsed_data = json.loads(data['choices'][0]['message']['content'])\n",
    "        return f\"{parsed_data['sentiment']}, {parsed_data['emotion']}\"\n",
    "    else:\n",
    "        return \"0, neutral\"\n",
    "    \n",
    "chatgpt_sentiment_udf = udf(chatgpt_sentiment, StringType())\n",
    "    \n",
    "json_df = json_df.withColumn(\"gpt_sentiment\", chatgpt_sentiment_udf(col(\"content\")))\n",
    "\n",
    "# Split the gpt_sentiment column into sentiment and emotion columns\n",
    "split_col = split(col(\"gpt_sentiment\"), \", \")\n",
    "json_df = json_df.withColumn(\"sentiment\", split_col.getItem(0).cast(\"float\")) \\\n",
    "                 .withColumn(\"emotion\", split_col.getItem(1))\n",
    "\n",
    "# Rimuovi la colonna gpt_sentiment\n",
    "json_df = json_df.drop(\"gpt_sentiment\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "860f25b0-6dea-4290-85d1-00b5cd69e29e",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "kafka_compatible_df = json_df.select(to_json(struct(*[col for col in json_df.columns])).alias(\"value\")) # Creazione di un df compatibile con Kafka\n",
    "\n",
    "# Scrivi il DataFrame a Elasticsearch\n",
    "elasticQuery = json_df.writeStream \\\n",
    "   .option(\"checkpointLocation\", \"/tmp/\") \\\n",
    "   .format(\"es\") \\\n",
    "   .start(elastic_index)\n",
    "\n",
    "kafkaQuery = kafka_compatible_df.writeStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", kafkaOutputServer) \\\n",
    "    .option(\"topic\", outputTopic) \\\n",
    "    .option(\"checkpointLocation\", \"/tmp/kafka_checkpoint\") \\\n",
    "    .start()\n",
    "\n",
    "elasticQuery.awaitTermination()\n",
    "kafkaQuery.awaitTermination()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7a1a09bd-7b9d-4412-9bbd-95121c58d2c5",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "slide"
    },
    "tags": []
   },
   "source": [
    "# Conclusioni"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae1746a2-64d4-4ec3-89a4-454ff36c6617",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "subslide"
    },
    "tags": []
   },
   "source": [
    "- Affidabilità grazie a tecnologie quali Logstash e Kafka"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e154dae-56b4-4467-8fcf-80f88d322f9b",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "fragment"
    },
    "tags": []
   },
   "source": [
    "- Scalabilità"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "521326b8-3962-48e0-91e9-8890bd199345",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "fragment"
    },
    "tags": []
   },
   "source": [
    "- Possibilità di ampliare le capacità del modello passando a GPT-4"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "395431dc-7b83-4535-8faf-dd3f6a6e424c",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": "fragment"
    },
    "tags": []
   },
   "source": [
    "- Costi ridotti: ~ $0.40/h analizzando una livestream di media affluenza (GPT 3.5)."
   ]
  }
 ],
 "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}