1  package main
  2  
  3  import (
  4  	"fmt"
  5  	"log"
  6  	"strings"
  7  	"time"
  8  )
  9  
 10  type SampleData struct {
 11  	ChannelName   string
 12  	ChannelId     uint64
 13  	Bits          uint64
 14  	AvgViewers    uint64
 15  	Topic         string
 16  	Title         string
 17  	Length        uint64
 18  	StartTime     string
 19  	MessageCount  uint64
 20  	QuestionCount uint64
 21  }
 22  
 23  func filterMessage(sender, message string) bool {
 24  	isBot := sender == "nightbot" || sender == "streamelements"
 25  	isQuestion := strings.Contains(message, "?")
 26  	hasLink := strings.Contains(message, "http")
 27  	isSignificant := len(message) > 15
 28  
 29  	return !isBot && !hasLink && isQuestion && isSignificant
 30  }
 31  
 32  func CollectSample(app App, stream StreamInfo, sampling chan<- string) {
 33  	sample := SampleData{}
 34  	sample.ChannelName = stream.UserName
 35  	sample.ChannelId = stream.ChannelId
 36  	sample.AvgViewers = stream.Viewers
 37  	sample.Topic = stream.Topic
 38  	sample.Title = stream.Title
 39  	sample.StartTime = stream.StartTS
 40  
 41  	start := time.Now()
 42  	interrupt := make(chan struct{}, 1)
 43  	terminate := make(chan struct{}, 1)
 44  	period := time.NewTicker(time.Minute * 10)
 45  	offline := make(chan struct{}, 1)
 46  	chats := make(chan ChatMessage)
 47  
 48  	// log.Printf("Connecting to %s's chat...", stream.UserName)
 49  	go ConnectTwitchChat(app, stream.UserName, chats, interrupt, terminate)
 50  
 51  	for {
 52  		select {
 53  		case c := <-chats:
 54  			if filterMessage(c.User, c.Message) || c.Bits > 0 {
 55  				if c.Bits > 0 {
 56  					sample.Bits = sample.Bits + c.Bits
 57  				} else {
 58  					sample.QuestionCount++
 59  				}
 60  				// log.Printf("%18s | (%3d) %s: %s", stream.UserName, c.Bits, c.User, c.Message)
 61  				_, err := app.InsertQuestionStmt.Exec(stream.ChannelId, c.Id, c.User, c.Message)
 62  				if err != nil {
 63  					log.Printf("%24s | Failed to Insert Question: %s\n", stream.UserName, err)
 64  				}
 65  			}
 66  			sample.MessageCount++
 67  
 68  		case <-period.C:
 69  			streams, _ := HelixGetStreams(app, "", stream.UserName)
 70  			if len(streams) == 0 {
 71  				offline <- struct{}{}
 72  			} else {
 73  				s := streams[0]
 74  				log.Printf("%24s | new average viewers: %d (prev: %d, curr: %d)\n", s.UserName, (sample.AvgViewers+s.Viewers)/2, sample.AvgViewers, s.Viewers)
 75  				sample.AvgViewers = (sample.AvgViewers + s.Viewers) / 2
 76  				sample.Topic = s.Topic
 77  				sample.Title = s.Title
 78  			}
 79  
 80  		case <-offline:
 81  			log.Printf("%24s | channel is offline...\n", stream.UserName)
 82  			interrupt <- struct{}{}
 83  
 84  		case <-terminate:
 85  			finish := time.Now()
 86  			sample.Length = uint64(finish.Sub(start).Truncate(time.Second).Minutes())
 87  			if sample.Length > 5 {
 88  				_, err := app.InsertSampleStmt.Exec(
 89  					sample.ChannelId,
 90  					sample.ChannelName,
 91  					sample.Bits,
 92  					sample.AvgViewers,
 93  					sample.Topic,
 94  					sample.Title,
 95  					sample.Length,
 96  					sample.StartTime,
 97  					sample.MessageCount,
 98  					sample.QuestionCount,
 99  				)
100  				if err != nil {
101  					fmt.Println(err)
102  				}
103  			}
104  
105  			sampling <- stream.UserName
106  			return
107  		}
108  	}
109  }