twitch_chat_logstash.yml

gistfile1.txt

input {
  file {
    path => "/tmp/irssi_raw.log"
    start_position => "end"
    type => "irc"
    tags => ["irc"]
  }
  http {
    port => 31311 # default: 8080
    tags => ["http"]
  }
}

filter {
  if "irc" in [tags] {
    grok {
        add_tag => [ "chat_message" ]
        match => [
          "message", "^>>.*PRIVMSG"
        ]
    }
    if "chat_message" not in [tags] {
      drop {}
    }

    mutate {
      remove_tag => [ "chat_message" ]
    }

    grok {
      match => [
        "message", ">> @(?<badges_raw>[^ ]+) :(?<username>[^!]+)!(?<identity>[^ ]+) PRIVMSG #(?<channel>[^ ]+) :(?<msg>.*)"
      ]
    }

    grok {
        add_tag => [ "clipit" ]
        match => [
          "message", "TTours"
        ]
    }


    kv { value_split => "="  field_split => ";" source => "badges_raw" target => "badges" }

    mutate {
      split => { "[badges][emotes]" => "," }
    }

    ruby {
      code => "event.set('number_of_smiles', !event.get('[badges][emotes]').nil? ? event.get('[badges][emotes]').length : 0)"
    }

    ruby {
      code => "event.set('msg_time', Time.now().strftime('%Y-%m-%dT%H:%M:%S+04:00'))"
    }
    ruby {
      code => "
        require 'json'
        require 'date'
        file = File.open '/tmp/twitch_data.json'
        data = JSON.load file
        user_name = event.get('channel')
        event.set('viewers', 0)
        event.set('stream_time', 0)
        for user in data['data']
          if user_name == user['user_login']
            event.set('viewers', user['viewer_count'])
            started_at_time = Time.strptime(user['started_at'], '%Y-%m-%dT%H:%M:%S%z')
            stream_time = Time.now.getlocal().strftime('%s').to_i - (started_at_time.strftime('%s').to_i)
            event.set('stream_time', stream_time)

            break
          end
        end
        file.close
      "
    }
    prune {
      whitelist_names => [
        "^channel$",
        "^viewers$",
        "^stream_time$",
        "^username$",
        "^number_of_smiles$",
        "^msg$",
        "^msg_time$",
        "^tags$"
      ]
    }
  }

}

output {
#    file {
#      path => "/tmp/output_logstash.txt"
#      codec => line { format => "%{stream_time} #%{channel}(%{viewers}) %{username} -> %{msg} ..... %{number_of_smiles}"}
#    }

    if "irc" in [tags] {
      elasticsearch {
        hosts => ['localhost:9200']
        user => 'elastic'
        password => 'changeme'
        # index => "twitch_clipz_%{+yyyy.MM.dd}"
        index => "logstash-%{+yyyy.MM.dd}"
      }
      kafka {
        codec => json
        bootstrap_servers => "localhost:9092"
        topic_id => "twitch-clipz"
      }
    }
    if "http" in [tags] or "clipit" in [tags] {
      kafka {
        codec => json
        bootstrap_servers => "localhost:9092"
        topic_id => "auto-clipz"
      }
#      exec {
#        command => "/home/bebebeko/spaces/twitch-dl/bin/auto_clip.sh %{channel} %{time}"
        #command => "su - bebebeko /bin/sh -c \"/home/bebebeko/spaces/twitch-dl/bin/auto_clip.sh %{channel} %{time}\""
#      }
    }


}

Nice that was it! It was the /rawlog I reckon that was the issue. Huge thanks! 🙏