Merge branch 'release/v0.5.0'

v0.4.0

.gitignore

@@ -1,9 +1,12 @@
 *.log
+log
 screenlog.*
 *.tar
 *.7z
 intimate
 *.gz
 debug.test
-
+myblock
+run.sh
+stop.sh
 

build.sh

@@ -0,0 +1,19 @@
+
+source_tasks=./tasks/*/*
+extractor_tasks=./extractor/*
+
+src=`pwd`
+
+for path in `ls -d $source_tasks`
+do
+    echo $path
+    cd $path && go build
+    cd $src
+done
+
+for path in `ls -d $extractor_tasks`
+do
+    echo $path
+    cd $path && go build
+    cd $src
+done

config.go

@@ -17,7 +17,7 @@ func init() {
 	InitConfig.Load()
 	// storeOpenrec = NewStore()
 
-	log.SetFlags(log.Llongfile | log.Ldate)
+	log.SetFlags(log.Llongfile | log.Ltime)
 }
 
 // Config 配置

config.yaml

@@ -1,3 +1,3 @@
 database:
-  source_uri: "root:@tcp(127.0.0.1:4000)/intimate_source?parseTime=true"
+  source_uri: "root:@tcp(127.0.0.1:4000)/intimate_source?parseTime=true&loc=Local"
-  extractor_uri: "root:@tcp(127.0.0.1:4000)/intimate_extractor?parseTime=true"
+  extractor_uri: "root:@tcp(127.0.0.1:4000)/intimate_extractor?parseTime=true&loc=Local"

crx/myblock.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDSG09DSvB03TOe
+eOmQwfiCIf0wa2WRB31ewxa6i/PRgEKeJSUvIsIuaECUer2ss+J3rwSS2lDpGuiw
+FnsVyZqKI/+Rcuc83YJGYg6OAzVMz6UL8YCWhXu3huTJ+V+a5iNereIC69ZERRJt
+nXlWqsq6HKya+6BP9sX9CI4GTHQrnWBysAxsswhdnnnRvu+GxglWafSIzuS6OizT
+1M1CmkZxNvDJhTSOR7SJlIYm2kM5/fIL53BdndF2IGAjfV1WV7AjwhTfun5cViEO
+i8niQUIMY4L0AiO9grFD1g1xIYkeuVBoLxOUBzPxJwQmb64gseb9Dvt0BKLRGoou
+SIOyE+KVAgMBAAECggEAI4b6J2kR0VUBEDwmVHO0K38HUstqNHSVgrNO0dLt8sAz
+I44o5DhGqPW4a9L4ZS5SrkWyKonPcic6buISRIwfPVoacjQBfVWAXJnil6lbtyYK
+ZMNcqLcgBRfCcpOgEq91DiKta6yIwekDFXVyCdFd78v+9ML1J+hUsLVkXJTLdP88
+PGamRWVd6vGy3QMRjyM29GLPgS+/6Vrp1cptSuYNqYhlszohmu8lBvzjH9jbPh9d
+GFrrd8Bs7IRCdtKZig/3fbln4JEyyOYE+gcT2jplPksB6mR/5DBIdkVbeuFwGB0+
+h1/PKlprNQt7+Ei0HhHnTib7lZP8WGo4HkSi7PsAGQKBgQD1Ptho0wJiI2+6gL1O
+iNsEJVKIQ2Sxdx3wI/qudphM99t6xKCpPyVI2Nd9PBf2jbZjGAaz+P/KQYxEqb6i
+PRcQ+i99wCQoRfnRvUbKA4goEpKwRXmvn+499dm6D5pEuumOXGQYCmaFXuLTRN/I
+BL6GNgLtoZAlLjUXaWtk8TszGQKBgQDbUf3p3HLpCjRvRDW/vA5xj+08t7xtF9uO
+NilGK79uOA4VnxE2w3ioYqQ7t3I8J/0rAzGKq3tylg4QX6UpQ4b2koRr2B3cqoAk
+dsRdNWAHwCNepz8hTLsZyuihzbNv2nHmoqhzjK/FcrBHx5NAM+T6OBpLzQBnbUzk
+3wIcqm223QKBgQDo/IRxyY0pGMtLXoT6ODACF0b6JzRhGG37tuKvngGAlbQQRP7w
+6wmL1F2cH1wQon7UU34CupqfVnhgvvZZgToJqfU2PTTcgeYc6Pl4b7SJhWOQTOCX
+BZQ7jvYCulHv27aIxaNd53uQVx2cYoFKr58lN+i+QtADUoujq0YYxshb+QKBgQDW
+ZOti7kZCeuBRGIu2V56C8uBFp5MBzf2polZsqx1iIFfcWPfZ4fGUIYFMgwKfvbOl
+lWSbmxB9LiSnaugoU0OezBG43rYqXV4Qxy0jtKagTPoGcFWtNrX7+7e3XD8Zi6Am
+hkFHW3MEAB5EvNq8Oz6OP8Os78SCVn2BimMlJJFF3QKBgQCF+aEAiBv+ivcmHUeP
+2eBq9nLltPFAfXJ/p31MMQ6Jgo36DBqUeoLeyq/WfIXvwqbVbP9fANZrKoTPbI97
+dilCHUoO33rafXJy6jtaggtpz14tt9soecTop0vM/rU7tGtfBe6NXg9LRl+oDJCU
+37I3a9Is+2CLyAUXWCk9mLfFsQ==
+-----END PRIVATE KEY-----

extractor/openrec_extractor/main.go

@@ -1,5 +1,9 @@
 package main
 
+import (
+	_ "net/http/pprof"
+)
+
 /*
   `uid`           varchar(36)  NOT NULL,
   `platform`      varchar(255) NOT NULL,

extractor/openrec_extractor/openrec_extractor.go

@@ -17,6 +17,9 @@ import (
 	"github.com/tidwall/gjson"
 )
 
+var estore = intimate.NewStoreExtractor()
+var sstore = intimate.NewStoreSource(string(intimate.STOpenrec))
+
 // OpenrecExtractor 提取方法
 type OpenrecExtractor struct {
 	user       *intimate.ExtractorSource
@@ -35,83 +38,71 @@ func (oe *OpenrecExtractor) Execute() {
 		atomic.StoreInt32(&loop, 0)
 	}()
 
-	collect := intimate.NewExtractorStore()
-	store := intimate.NewSourceStore("source_openrec")
 	var lasterr error = nil
-
 	for atomic.LoadInt32(&loop) > 0 {
+		var err error
 
-		source, err := store.Pop(string(intimate.TTOpenrecRanking), 100)
+		source, err := sstore.Pop(intimate.TOpenrecUser, 0)
 		if err != nil {
 			if err != lasterr {
 				log.Println(err, lasterr)
 				lasterr = err
 			}
-			time.Sleep(time.Second * 2)
+			time.Sleep(time.Second * 5)
 			continue
 		}
 
-		source.SetOperator(int32(intimate.OperatorError))
+		sdata := source.Ext.([]byte)
-		anchorId := source.GetSource().String
+		datamap := gjson.ParseBytes(sdata).Map()
 
-		ai := &intimate.AnchorInfo{}
+		source.Operator = int32(intimate.OperatorError)
-		ai.SetAnchorId(anchorId)
+		userId := datamap["var_user_id"].String()
-		ai.SetPlatform(string(intimate.Popenrec))
 
-		sdata := source.GetExt().([]byte)
+		streamer := &intimate.Streamer{}
-		if gjson.ValidBytes(sdata) {
+		streamer.UserId = userId
-			result := gjson.ParseBytes(sdata)
+		streamer.Platform = intimate.Popenrec
-			datamap := result.Map()
 
-			oe.user = intimate.NewExtractorSource(datamap["user"])
+		htmlUser := datamap["html_user"]
+		oe.user = intimate.NewExtractorSource(&htmlUser)
 		oe.user.CreateExtractor()
 
-			oe.userLive = intimate.NewExtractorSource(datamap["user_live"])
+		htmlLive := datamap["html_live"]
+		oe.userLive = intimate.NewExtractorSource(&htmlLive)
 		oe.userLive.CreateExtractor()
 
-			oe.supporters = intimate.NewExtractorSource(datamap["supporters"])
+		jsonSupporters := datamap["json_supporters"]
-
+		oe.supporters = intimate.NewExtractorSource(&jsonSupporters)
 		clog := &intimate.CollectLog{}
 
-			log.Println(anchorId)
+		// log.Println(anchorId)
 
 		oe.extractFollowers(clog)
-			oe.extractAnchorName(ai)
+		oe.extractUserName(streamer)
 		oe.extractViewsAndLiveStreaming(clog)
 		oe.extractGiversAndGratuity(clog)
 		oe.extractLive(clog)
 		oe.extractTags(clog)
 
-			ai.Set("UpdateTime", source.GetUpdateTime())
+		streamer.Uid = source.StreamerId.Int64
+		streamer.UpdateTime = source.UpdateTime
+		streamer.Tags = clog.Tags
 
-			LiveUrl := "https://www.openrec.tv/live/" + anchorId
+		clog.Platform = string(intimate.Popenrec)
-			ai.Set("LiveUrl", sql.NullString{String: LiveUrl, Valid: true})
+		clog.UserId = userId
+		clog.UpdateTime = source.UpdateTime
 
-			Uid, err := collect.InsertAnchorInfo(ai)
+		logUid := estore.InsertClog(clog)
-			if err != nil {
+
-				log.Println(err)
+		LiveUrl := "https://www.openrec.tv/live/" + userId
-				source.SetErrorMsg(sql.NullString{String: err.Error(), Valid: true})
+		streamer.LiveUrl = sql.NullString{String: LiveUrl, Valid: true}
-				store.UpdateOperator(source)
+		streamer.LatestLogUid = logUid
-				return
+		streamer.Operator = 0
+		estore.UpdateStreamer(streamer)
+
+		source.Operator = int32(intimate.OperatorExtractorOK)
+		sstore.UpdateOperator(source)
 	}
 
-			clog.Set("Uid", Uid)
-			clog.Set("Platform", string(intimate.Popenrec))
-			clog.Set("AnchorId", anchorId)
-			clog.Set("UpdateTime", source.GetUpdateTime())
-
-			if err = collect.InsertCollectLog(clog); err != nil {
-				source.SetErrorMsg(sql.NullString{String: err.Error(), Valid: true})
-				store.UpdateOperator(source)
-				return
-			}
-
-			source.SetOperator(int32(intimate.OperatorExtractorOK))
-			store.UpdateOperator(source)
-		} else {
-			log.Println("data is not json:\n", string(sdata))
-		}
-	}
 }
 
 func (oe *OpenrecExtractor) extractFollowers(clog intimate.ISet) {
@@ -133,14 +124,16 @@ func (oe *OpenrecExtractor) extractFollowers(clog intimate.ISet) {
 	clog.Set("Followers", sql.NullInt64{Int64: followersInt, Valid: true})
 }
 
-func (oe *OpenrecExtractor) extractAnchorName(ai intimate.ISet) {
+func (oe *OpenrecExtractor) extractUserName(streamer intimate.ISet) {
 	extractor := oe.user.GetExtractor()
-	xp, err := extractor.XPathResult("//p[@class='c-global__user__profile__list__name__text official-icon--after']/text()")
+	xp, err := extractor.XPathResult("//p[ contains(@class, 'c-global__user__profile__list__name__text')]/text()")
-	if xp.NodeIter().Next() {
+	if err != nil {
-		anchorName := xp.String()
-		ai.Set("AnchorName", anchorName)
-	} else {
 		log.Println(err)
+	} else {
+		if xp.NodeIter().Next() {
+			userName := xp.String()
+			streamer.Set("UserName", sql.NullString{String: userName, Valid: true})
+		}
 	}
 }
 
@@ -151,6 +144,7 @@ func (oe *OpenrecExtractor) extractViewsAndLiveStreaming(clog intimate.ISet) {
 	if err != nil {
 		log.Println(err)
 	}
+
 	if xp.NodeIter().Next() {
 		views := regexp.MustCompile(`[0-9,]+`).FindString(xp.String())
 		views = strings.ReplaceAll(views, ",", "")
@@ -160,7 +154,7 @@ func (oe *OpenrecExtractor) extractViewsAndLiveStreaming(clog intimate.ISet) {
 		}
 
 		clog.Set("Views", sql.NullInt64{Int64: int64(viewsint), Valid: true})
-		clog.Set("IsLiveStreaming", int32(1))
+		clog.Set("IsLiveStreaming", true)
 	}
 }
 
@@ -207,6 +201,7 @@ func (oe *OpenrecExtractor) extractLive(clog intimate.ISet) {
 			if err != nil {
 				log.Println(err)
 			}
+			// log.Println(iter.Node().NodeValue(), tm.Local())
 			clog.Set("LiveStartTime", sql.NullTime{Time: tm.Local(), Valid: true})
 
 			duration, err := extractor.XPathResult("//meta[@itemprop='duration']/@content")
@@ -234,7 +229,6 @@ func (oe *OpenrecExtractor) extractTags(clog intimate.ISet) {
 	for _, m := range matheslist {
 		tags = append(tags, m[1])
 	}
-	log.Println(tags)
 	tagsBytes, err := json.Marshal(tags)
 	if err != nil {
 		log.Println(err)

extractor/openrec_extractor/openrec_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/474420502/hunter"
 	"github.com/lestrrat-go/libxml2"
 )
 
@@ -90,6 +91,28 @@ func TestCase(t *testing.T) {
 	t.Error(xr)
 }
 
+func TestUserName(t *testing.T) {
+	f, err := os.Open("test.html")
+	if err != nil {
+		panic(err)
+	}
+	data, err := ioutil.ReadAll(f)
+	if err != nil {
+		panic(err)
+	}
+
+	extractor := hunter.NewExtractor(data)
+	xp, err := extractor.XPathResult("//p[ contains(@class, 'c-global__user__profile__list__name__text')]/text()")
+	if err != nil {
+		t.Error(err)
+	} else {
+		if xp.NodeIter().Next() {
+			userName := xp.String()
+			t.Error(userName)
+		}
+	}
+}
+
 func TestExtractor(t *testing.T) {
 	oe := &OpenrecExtractor{}
 	oe.Execute()

extractor/twitch_extractor/.gitignore

@@ -0,0 +1,4 @@
+*.html
+log
+screenlog.*
+twitch_extractor

extractor/twitch_extractor/tiwtch_extractor.go

@@ -0,0 +1,247 @@
+package main
+
+import (
+	"database/sql"
+	"encoding/json"
+	"intimate"
+	"log"
+	"regexp"
+	"time"
+
+	"github.com/tebeka/selenium"
+)
+
+// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitch))
+
+// estore 解析存储连接实例
+var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+
+func main() {
+	wd := intimate.GetChromeDriver(3030)
+	ps := intimate.NewPerfectShutdown()
+
+	counter := intimate.NewCounter()
+	counter.SetMaxLimit(200)
+	counter.SetMaxToDo(func(olist ...interface{}) error {
+		owd := olist[0].(*selenium.WebDriver)
+		(*owd).Close()
+		(*owd).Quit()
+		*owd = intimate.GetChromeDriver(3030)
+		return nil
+	}, &wd)
+
+	var lasterr error = nil
+	// var err error
+
+	for !ps.IsClose() {
+		streamer, err := estore.Pop(intimate.Ptwitch, 0)
+		if streamer == nil || err != nil {
+			if err != lasterr {
+				log.Println(err, lasterr)
+				lasterr = err
+			}
+			time.Sleep(time.Second * 2)
+			continue
+		}
+
+		var updateUrl map[string]string
+		json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl)
+		liveUrl := updateUrl["live"]
+		log.Println(liveUrl)
+
+		// err = wd.Get("https://www.twitch.tv/zoe_0601" + "/about")
+		err = wd.Get(liveUrl + "/about")
+		if err != nil {
+			log.Println(err)
+			estore.UpdateError(streamer, err)
+			time.Sleep(time.Second * 5)
+			continue
+		}
+
+		streamer.LiveUrl = sql.NullString{String: liveUrl, Valid: true}
+		clog := &intimate.CollectLog{}
+		clog.UserId = streamer.UserId
+		clog.Gratuity = sql.NullInt64{Int64: 0, Valid: false}
+
+		time.Sleep(time.Millisecond * 500)
+		err = extractUserName(wd, streamer)
+		if err != nil {
+			continue
+		}
+		err = extractFollowers(wd, clog)
+		if err != nil {
+			continue
+		}
+
+		err = extractViews(wd, clog) // views + tags + gratuity
+		if err != nil {
+			// 不直播时提取礼物 gratuity
+			wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+				channelchat, err := wd.FindElement(selenium.ByXPATH, `//a[@data-a-target="channel-home-tab-Chat"]`)
+				btn, _ := web.FindElement(selenium.ByXPATH, `//button[@data-test-selector="expand-grabber"]`)
+				if (err == nil && channelchat != nil) || btn != nil {
+					if channelchat != nil {
+						channelchat.Click()
+					}
+					time.Sleep(time.Second)
+					extractGratuity(wd, clog)
+					return true, nil
+				}
+				return false, nil
+
+			}, time.Second*4)
+		}
+
+		streamer.Platform = intimate.Ptwitch
+		clog.Platform = string(streamer.Platform)
+		clog.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
+		lastClogId := estore.InsertClog(clog)
+
+		streamer.Operator = 10
+		streamer.LatestLogUid = lastClogId
+		if clog.Tags != nil {
+			streamer.Tags = clog.Tags
+		}
+
+		switch fl := clog.Followers.Int64; {
+		case fl > 100000:
+			streamer.UpdateInterval = 120
+		case fl > 10000:
+			streamer.UpdateInterval = 240
+		case fl > 1000:
+			streamer.UpdateInterval = 360
+		case fl > 100:
+			streamer.UpdateInterval = 720
+		case fl > 0:
+			streamer.UpdateInterval = 1440
+		}
+
+		streamer.UpdateTime = clog.UpdateTime
+		estore.UpdateStreamer(streamer)
+		counter.AddWithReset(1)
+	}
+
+	wd.Close()
+	wd.Quit()
+}
+
+func extractUserName(wd selenium.WebDriver, streamer *intimate.Streamer) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		label, err := web.FindElement(selenium.ByXPATH, "//a[@class='tw-interactive']//h1")
+		if err == nil {
+			if ltxt, err := label.Text(); err == nil && ltxt != "" {
+				// log.Println("label:", ltxt)
+				streamer.UserName = sql.NullString{String: ltxt, Valid: true}
+				return true, nil
+			}
+		}
+		return false, err
+	}, 15*time.Second)
+}
+
+func extractFollowers(wd selenium.WebDriver, clog *intimate.CollectLog) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		efollowers, err := web.FindElement(selenium.ByXPATH, "//div[@data-a-target='about-panel']//div[@class='tw-align-center']")
+		if err != nil {
+			return false, err
+		}
+		followers, err := efollowers.Text()
+		if err != nil || followers == "" {
+			return false, err
+		}
+		followers = regexp.MustCompile(`[\d,]+`).FindString(followers)
+		fint, _ := intimate.ParseNumber(followers)
+		clog.Followers = sql.NullInt64{Int64: int64(fint), Valid: true}
+		// log.Println("followers: ", followers, fint)
+		return true, nil
+	}, 4*time.Second)
+}
+
+func extractViews(wd selenium.WebDriver, clog *intimate.CollectLog) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		views, err := web.FindElement(selenium.ByXPATH, "//a[@data-a-target='home-live-overlay-button']/span")
+		if views != nil {
+			if txt, err := views.Text(); err == nil {
+
+				vint, _ := intimate.ParseNumber(txt)
+				clog.Views = sql.NullInt64{Int64: vint, Valid: true}
+				// log.Println("views:", txt)
+				views.Click()
+
+				extractTags(wd, clog)
+				extractTitle(wd, clog)
+				extractGratuity(wd, clog)
+
+				return true, nil
+			}
+		}
+		return false, err
+	}, time.Second*4)
+}
+
+func extractTitle(wd selenium.WebDriver, clog *intimate.CollectLog) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		title, err := web.FindElement(selenium.ByXPATH, `//h2[@data-a-target='stream-title']`)
+		if err == nil {
+			if txt, err := title.Text(); err == nil {
+				clog.LiveTitle = sql.NullString{String: txt, Valid: true}
+				return true, nil
+			}
+		}
+		return false, err
+	}, time.Second*4)
+}
+
+func extractTags(wd selenium.WebDriver, clog *intimate.CollectLog) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		tags, err := web.FindElements(selenium.ByXPATH, "//a[@aria-label and @data-a-target and @href]/div[@class and text()]")
+		if len(tags) == 0 {
+			return false, err
+		}
+
+		var stags []string
+		for _, tag := range tags {
+			if txt, err := tag.Text(); err == nil {
+				stags = append(stags, txt)
+			} else {
+				log.Println(err)
+			}
+		}
+		if len(stags) > 0 {
+			if tagbuf, err := json.Marshal(stags); err == nil {
+				clog.Tags = tagbuf
+			} else {
+				log.Println(err)
+			}
+		}
+
+		return true, nil
+	}, time.Second*4)
+}
+
+func extractGratuity(wd selenium.WebDriver, clog *intimate.CollectLog) error {
+	return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+		btn, err := web.FindElement(selenium.ByXPATH, `//button[@data-test-selector="expand-grabber"]`)
+		if err == nil {
+			btn.Click()
+			time.Sleep(time.Second)
+			gifcount, err := web.FindElements(selenium.ByXPATH, `//div[@class="sub-gift-count tw-flex"]/p`)
+			if err == nil {
+				var gratuity int64 = 0
+				for _, gc := range gifcount {
+					if gtxt, err := gc.Text(); err == nil {
+						gint, _ := intimate.ParseNumber(gtxt)
+						gratuity += gint
+					} else {
+						log.Println(err)
+					}
+				}
+				clog.Gratuity = sql.NullInt64{Int64: gratuity, Valid: true}
+			}
+			return true, nil
+		}
+
+		return false, err
+	}, time.Second*4)
+}

extractor/twitch_extractor/twitch_test.go

@@ -0,0 +1,9 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestCase0(t *testing.T) {
+	main()
+}

extractor_field.go

@@ -8,181 +8,51 @@ import (
 	"github.com/tidwall/gjson"
 )
 
-type ISetAnchorInfo interface {
+type GetSet struct {
-	SetUid(int64)               //
-	SetPlatform(string)         //
-	SetAnchorId(string)         //
-	SetAnchorName(string)       //
-	SetLiveUrl(sql.NullString)  //
-	SetChannel(sql.NullString)  //
-	SetTags(interface{})        //
-	SetExt(interface{})         //
-	SetUpdateTime(sql.NullTime) //
 }
 
-type IGetAnchorInfo interface {
+type Streamer struct {
-	GetUid() int64              //
-	GetPlatform() string        //
-	GetAnchorId() string        //
-	GetAnchorName() string      //
-	GetLiveUrl() sql.NullString //
-	GetChannel() sql.NullString //
-	GetTags() interface{}
-	GetExt() interface{}         //
-	GetUpdateTime() sql.NullTime //
-}
-
-type AnchorInfo struct {
 	Uid      int64    //
-	Platform   string         //
+	Platform Platform //
-	AnchorId   string         //
+	UserId   string   //
-	AnchorName string         //
+
+	UserName sql.NullString //
 	LiveUrl  sql.NullString //
 	Channel  sql.NullString //
 	Tags     interface{}
 	Ext      interface{} //
+
+	IsUpdateStreamer bool // 更新上面的内容
+	IsUpdateUrl      bool
+	UpdateInterval   int32
+	UpdateUrl        interface{}
+	LatestLogUid     int64
 	UpdateTime       sql.NullTime //
+
+	ErrorMsg sql.NullString
+	Operator int32
+
+	LastOperator int32
+}
+
+// Get Simple Value
+func (ai *Streamer) Get(field string) interface{} {
+	return reflect.ValueOf(ai).Elem().FieldByName(field).Interface()
 }
 
 // Set Simple Value
-func (ai *AnchorInfo) Set(field string, value interface{}) {
+func (ai *Streamer) Set(field string, value interface{}) {
 	reflect.ValueOf(ai).Elem().FieldByName(field).Set(reflect.ValueOf(value))
 }
 
-// GetTags Get return Tags interface{}
-func (ai *AnchorInfo) GetTags() interface{} {
-	return ai.Tags
-}
-
-// SetTags Set Tags interface{}
-func (ai *AnchorInfo) SetTags(Tags interface{}) {
-	ai.Tags = Tags
-}
-
-// GetUpdateTime Get return UpdateTime time.Time
-func (ai *AnchorInfo) GetUpdateTime() sql.NullTime {
-	return ai.UpdateTime
-}
-
-// SetUpdateTime Set UpdateTime time.Time
-func (ai *AnchorInfo) SetUpdateTime(UpdateTime sql.NullTime) {
-	ai.UpdateTime = UpdateTime
-}
-
-// GetExt Get return Ext interface{}
-func (ai *AnchorInfo) GetExt() interface{} {
-	return ai.Ext
-}
-
-// SetExt Set Ext interface{}
-func (ai *AnchorInfo) SetExt(Ext interface{}) {
-	ai.Ext = Ext
-}
-
-// GetChannel Get return Channel sql.NullString
-func (ai *AnchorInfo) GetChannel() sql.NullString {
-	return ai.Channel
-}
-
-// SetChannel Set Channel sql.NullString
-func (ai *AnchorInfo) SetChannel(Channel sql.NullString) {
-	ai.Channel = Channel
-}
-
-// GetLiveUrl Get return LiveUrl sql.NullString
-func (ai *AnchorInfo) GetLiveUrl() sql.NullString {
-	return ai.LiveUrl
-}
-
-// SetLiveUrl Set LiveUrl sql.NullString
-func (ai *AnchorInfo) SetLiveUrl(LiveUrl sql.NullString) {
-	ai.LiveUrl = LiveUrl
-}
-
-// GetAnchorName Get return AnchorName string
-func (ai *AnchorInfo) GetAnchorName() string {
-	return ai.AnchorName
-}
-
-// SetAnchorName Set AnchorName string
-func (ai *AnchorInfo) SetAnchorName(AnchorName string) {
-	ai.AnchorName = AnchorName
-}
-
-// GetAnchorId Get return AnchorId string
-func (ai *AnchorInfo) GetAnchorId() string {
-	return ai.AnchorId
-}
-
-// SetAnchorId Set AnchorId string
-func (ai *AnchorInfo) SetAnchorId(AnchorId string) {
-	ai.AnchorId = AnchorId
-}
-
-// GetPlatform Get return Platform string
-func (ai *AnchorInfo) GetPlatform() string {
-	return ai.Platform
-}
-
-// SetPlatform Set Platform string
-func (ai *AnchorInfo) SetPlatform(Platform string) {
-	ai.Platform = Platform
-}
-
-// GetUid Get return Uid int64
-func (ai *AnchorInfo) GetUid() int64 {
-	return ai.Uid
-}
-
-// SetUid Set Uid int64
-func (ai *AnchorInfo) SetUid(Uid int64) {
-	ai.Uid = Uid
-}
-
-type IGetCollectLog interface {
-	GetUid() int64                  //
-	GetPlatform() string            //
-	GetAnchorId() string            //
-	GetIsLiveStreaming() int32      //
-	GetIsError() int32              //
-	GetFollowers() sql.NullInt64    //
-	GetViews() sql.NullInt64        //
-	GetGiver() interface{}          //
-	GetGratuity() sql.NullInt64     //
-	GetLiveTitle() sql.NullString   //
-	GetLiveStartTime() sql.NullTime //
-	GetLiveEndTime() sql.NullTime   //
-	GetUpdateTime() sql.NullTime    //
-	GetTags() interface{}           //
-	GetExt() interface{}            //
-	GetErrorMsg() sql.NullString    //
-}
-
-type ISetCollectLog interface {
-	SetUid(int64)                  //
-	SetPlatform(string)            //
-	SetAnchorId(string)            //
-	SetIsLiveStreaming(int32)      //
-	SetIsError(int32)              //
-	SetFollowers(sql.NullInt64)    //
-	SetViews(sql.NullInt64)        //
-	SetGiver(interface{})          //
-	SetGratuity(sql.NullInt64)     //
-	SetLiveTitle(sql.NullString)   //
-	SetLiveStartTime(sql.NullTime) //
-	SetLiveEndTime(sql.NullTime)   //
-	SetUpdateTime(sql.NullTime)    //
-	SetTags(interface{})           //
-	SetExt(interface{})            //
-	SetErrorMsg(sql.NullString)    //
-}
-
 type CollectLog struct {
-	Uid             int64          //
+	LogUid      int64 // 日志id
+	StreamerUid int64 // StreamerId 表id与
+
 	Platform        string         //
-	AnchorId        string         //
+	UserId          string         // 平台的UserId
-	IsLiveStreaming int32          //
+	IsLiveStreaming bool           //
-	IsError         int32          //
+	IsError         bool           //
 	Followers       sql.NullInt64  //
 	Views           sql.NullInt64  //
 	Giver           interface{}    //
@@ -196,187 +66,42 @@ type CollectLog struct {
 	ErrorMsg        sql.NullString //
 }
 
+// Get Simple Value
+func (cl *CollectLog) Get(field string) interface{} {
+	return reflect.ValueOf(cl).Elem().FieldByName(field).Interface()
+}
+
 // Set Simple Value
 func (cl *CollectLog) Set(field string, value interface{}) {
 	reflect.ValueOf(cl).Elem().FieldByName(field).Set(reflect.ValueOf(value))
 }
 
-// GetTags Get return Tags interface{}
-func (cl *CollectLog) GetTags() interface{} {
-	return cl.Tags
-}
-
-// SetTags Set Tags interface{}
-func (cl *CollectLog) SetTags(Tags interface{}) {
-	cl.Tags = Tags
-}
-
-// GetErrorMsg Get return Error sql.NullString
-func (cl *CollectLog) GetErrorMsg() sql.NullString {
-	return cl.ErrorMsg
-}
-
-// SetErrorMsg Set Error sql.NullString
-func (cl *CollectLog) SetErrorMsg(ErrorMsg sql.NullString) {
-	cl.ErrorMsg = ErrorMsg
-}
-
-// GetExt Get return Ext interface{}
-func (cl *CollectLog) GetExt() interface{} {
-	return cl.Ext
-}
-
-// SetExt Set Ext interface{}
-func (cl *CollectLog) SetExt(Ext interface{}) {
-	cl.Ext = Ext
-}
-
-// GetUpdateTime Get return UpdateTime time.Time
-func (cl *CollectLog) GetUpdateTime() sql.NullTime {
-	return cl.UpdateTime
-}
-
-// SetUpdateTime Set UpdateTime time.Time
-func (cl *CollectLog) SetUpdateTime(UpdateTime sql.NullTime) {
-	cl.UpdateTime = UpdateTime
-}
-
-// GetLiveEndTime Get return ShowEndTime sql.NullTime
-func (cl *CollectLog) GetLiveEndTime() sql.NullTime {
-	return cl.LiveEndTime
-}
-
-// SetLiveEndTime Set ShowEndTime sql.NullTime
-func (cl *CollectLog) SetLiveEndTime(ShowEndTime sql.NullTime) {
-	cl.LiveEndTime = ShowEndTime
-}
-
-// GetLiveStartTime Get return ShowStartTime sql.NullTime
-func (cl *CollectLog) GetLiveStartTime() sql.NullTime {
-	return cl.LiveStartTime
-}
-
-// SetLiveStartTime Set ShowStartTime sql.NullTime
-func (cl *CollectLog) SetLiveStartTime(ShowStartTime sql.NullTime) {
-	cl.LiveStartTime = ShowStartTime
-}
-
-// GetLiveTitle Get return ShowTitle sql.NullString
-func (cl *CollectLog) GetLiveTitle() sql.NullString {
-	return cl.LiveTitle
-}
-
-// SetLiveTitle Set ShowTitle sql.NullString
-func (cl *CollectLog) SetLiveTitle(ShowTitle sql.NullString) {
-	cl.LiveTitle = ShowTitle
-}
-
-// GetGratuity Get return Gratuity sql.NullInt32
-func (cl *CollectLog) GetGratuity() sql.NullInt64 {
-	return cl.Gratuity
-}
-
-// SetGratuity Set Gratuity sql.NullInt32
-func (cl *CollectLog) SetGratuity(Gratuity sql.NullInt64) {
-	cl.Gratuity = Gratuity
-}
-
-// GetGiver Get return Giver interface{}
-func (cl *CollectLog) GetGiver() interface{} {
-	return cl.Giver
-}
-
-// SetGiver Set Giver interface{}
-func (cl *CollectLog) SetGiver(Giver interface{}) {
-	cl.Giver = Giver
-}
-
-// GetViews Get return Views sql.NullInt64
-func (cl *CollectLog) GetViews() sql.NullInt64 {
-	return cl.Views
-}
-
-// SetViews Set Views sql.NullInt64
-func (cl *CollectLog) SetViews(Views sql.NullInt64) {
-	cl.Views = Views
-}
-
-// GetFollowers Get return Followers sql.NullInt64
-func (cl *CollectLog) GetFollowers() sql.NullInt64 {
-	return cl.Followers
-}
-
-// SetFollowers Set Followers sql.NullInt32
-func (cl *CollectLog) SetFollowers(Followers sql.NullInt64) {
-	cl.Followers = Followers
-}
-
-// GetIsError Get return IsError int32
-func (cl *CollectLog) GetIsError() int32 {
-	return cl.IsError
-}
-
-// SetIsError Set IsError int32
-func (cl *CollectLog) SetIsError(IsError int32) {
-	cl.IsError = IsError
-}
-
-// GetIsLiveStreaming Get return IsShowing int32
-func (cl *CollectLog) GetIsLiveStreaming() int32 {
-	return cl.IsLiveStreaming
-}
-
-// SetIsLiveStreaming Set IsShowing int32
-func (cl *CollectLog) SetIsLiveStreaming(IsLive int32) {
-	cl.IsLiveStreaming = IsLive
-}
-
-// GetAnchorId Get return AnchorId string
-func (cl *CollectLog) GetAnchorId() string {
-	return cl.AnchorId
-}
-
-// SetAnchorId Set AnchorId string
-func (cl *CollectLog) SetAnchorId(AnchorId string) {
-	cl.AnchorId = AnchorId
-}
-
-// GetPlatform Get return Platform string
-func (cl *CollectLog) GetPlatform() string {
-	return cl.Platform
-}
-
-// SetPlatform Set Platform string
-func (cl *CollectLog) SetPlatform(Platform string) {
-	cl.Platform = Platform
-}
-
-// GetUid Get return Uid int64
-func (cl *CollectLog) GetUid() int64 {
-	return cl.Uid
-}
-
-// SetUid Set Uid int64
-func (cl *CollectLog) SetUid(Uid int64) {
-	cl.Uid = Uid
-}
-
 type ExtractorSource struct {
-	source    gjson.Result
+	source    *gjson.Result
 	extractor *hunter.Extractor
 }
 
-func NewExtractorSource(gr gjson.Result) *ExtractorSource {
+func NewExtractorSource(gr *gjson.Result) *ExtractorSource {
 	es := &ExtractorSource{}
-	es.source = gr
+	es.SetSource(gr)
 	return es
 }
 
-func (es *ExtractorSource) CreateExtractor() {
+func (es *ExtractorSource) SetSource(gr *gjson.Result) {
-	es.extractor = hunter.NewExtractor([]byte(es.source.Str))
+	es.source = gr
+	es.extractor = nil
 }
 
-func (es *ExtractorSource) GetSource() gjson.Result {
+func (es *ExtractorSource) Clear() {
+	es.source = nil
+	es.extractor = nil
+}
+
+func (es *ExtractorSource) CreateExtractor() {
+	es.extractor = hunter.NewExtractor([]byte(es.source.String()))
+}
+
+func (es *ExtractorSource) GetSource() *gjson.Result {
 	return es.source
 }
 

go.mod

@@ -3,10 +3,13 @@ module intimate
 go 1.14
 
 require (
+	github.com/474420502/focus v0.12.0
 	github.com/474420502/gcurl v0.1.2
-	github.com/474420502/hunter v0.3.0
+	github.com/474420502/hunter v0.3.4
+	github.com/474420502/requests v1.6.0
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/lestrrat-go/libxml2 v0.0.0-20200215080510-6483566f52cb
+	github.com/tebeka/selenium v0.9.9
 	github.com/tidwall/gjson v1.6.0
 	github.com/tidwall/pretty v1.0.1 // indirect
 	golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect

go.sum

@@ -6,8 +6,8 @@ github.com/474420502/focus v0.12.0 h1:+icbmj7IEOefvTegHt5EpcHt6WFbe2miIrceUJx2Ev
 github.com/474420502/focus v0.12.0/go.mod h1:d0PMjtMxFz1a9HIhwyFPkWa+JF+0LgOrEUfd8iZka6s=
 github.com/474420502/gcurl v0.1.2 h1:ON9Yz3IgAdtDlFlHfkAJ3aIEBDxH0RiViPE5ST5ohKg=
 github.com/474420502/gcurl v0.1.2/go.mod h1:hws5q/Ao64bXLLDnldz9VyTQUndTWc/i5DzdEazFfoM=
-github.com/474420502/hunter v0.3.0 h1:0VPi1MInxjHOta3da4v0ALWK0y3/X4/6nUSLFvdbiFU=
+github.com/474420502/hunter v0.3.4 h1:fyLAgI84jWe3IcqsISC53j1w3CXI1FERxX//Potns0M=
-github.com/474420502/hunter v0.3.0/go.mod h1:pe4Xr/I+2agvq339vS/OZV+EiHAWtpXQs75rioSW9oA=
+github.com/474420502/hunter v0.3.4/go.mod h1:pe4Xr/I+2agvq339vS/OZV+EiHAWtpXQs75rioSW9oA=
 github.com/474420502/requests v1.6.0 h1:f4h4j40eT0P5whhg9LdkotD8CaKjtuDu/vz9iSUkCgY=
 github.com/474420502/requests v1.6.0/go.mod h1:SLXrQ5dL9c7dkIeKNUCBAjOIt3J9KFCS2RQjWJecNwo=
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=

platform_list.go

@@ -4,6 +4,9 @@ package intimate
 type Platform string
 
 const (
-	// Popenrec openrec源table名称
+	// Popenrec openrec 平台
 	Popenrec Platform = "openrec"
+
+	// Ptwitch twitch 平台
+	Ptwitch Platform = "twitch"
 )

source_field.go

@@ -2,144 +2,35 @@ package intimate
 
 import (
 	"database/sql"
+	"reflect"
 )
 
-// IGetSource 源接口结构
-type IGetSource interface {
-	GetUid() int64               //
-	GetUrl() string              //
-	GetTargetType() string       //
-	GetSource() sql.NullString   //
-	GetPassGob() sql.NullString  //
-	GetExt() interface{}         //
-	GetUpdateTime() sql.NullTime //
-	GetOperator() int32          //
-	GetErrorMsg() sql.NullString //
-}
-
-type IUpdateSource interface {
-	IGetSource
-
-	GetLastOperator() int32
-
-	SetPassGob(sql.NullString)
-	SetExt(ext interface{})          //
-	SetUpdateTime(ut sql.NullTime)   //
-	SetOperator(operator int32)      //
-	SetErrorMsg(emsg sql.NullString) //
-}
-
 // Source 的结构体
 type Source struct {
 	Uid int64 //
+
 	Url string //
-	TargetType string         //
+
+	StreamerId sql.NullInt64 //
+
 	Source     sql.NullString //
 	PassGob    sql.NullString //
 	Ext        interface{}    //
 	UpdateTime sql.NullTime   //
-	Operator   int32          //
 	ErrorMsg   sql.NullString //
 
-	lastOperator int32
+	Target   Target //
+	Operator int32  //
+
+	LastOperator int32
 }
 
-// GetPassGob Get return PassGob sql.NullString
+// Get Simple Value
-func (so *Source) GetPassGob() sql.NullString {
+func (so *Source) Get(field string) interface{} {
-	return so.PassGob
+	return reflect.ValueOf(so).Elem().FieldByName(field).Interface()
 }
 
-// SetPassGob Set PassGob sql.NullString
+// Set Simple Value
-func (so *Source) SetPassGob(PassGob sql.NullString) {
+func (so *Source) Set(field string, value interface{}) {
-	so.PassGob = PassGob
+	reflect.ValueOf(so).Elem().FieldByName(field).Set(reflect.ValueOf(value))
-}
-
-// GetLastOperator Get return lastOperator int32
-func (so *Source) GetLastOperator() int32 {
-	return so.lastOperator
-}
-
-// SetLastOperator Set lastOperator int32
-func (so *Source) SetLastOperator(lastOperator int32) {
-	so.lastOperator = lastOperator
-}
-
-// GetErrorMsg Get return ErrorMsg sql.NullString
-func (so *Source) GetErrorMsg() sql.NullString {
-	return so.ErrorMsg
-}
-
-// SetErrorMsg Set ErrorMsg sql.NullString
-func (so *Source) SetErrorMsg(ErrorMsg sql.NullString) {
-	so.ErrorMsg = ErrorMsg
-}
-
-// GetOperator Get return Operator sql.NullInt32
-func (so *Source) GetOperator() int32 {
-	return so.Operator
-}
-
-// SetOperator Set Operator sql.NullInt32
-func (so *Source) SetOperator(Operator int32) {
-	so.Operator = Operator
-}
-
-// GetUpdateTime Get return UpdateTime time.Time
-func (so *Source) GetUpdateTime() sql.NullTime {
-	return so.UpdateTime
-}
-
-// SetUpdateTime Set UpdateTime time.Time
-func (so *Source) SetUpdateTime(UpdateTime sql.NullTime) {
-	so.UpdateTime = UpdateTime
-}
-
-// GetExt Get return Ext interface{}
-func (so *Source) GetExt() interface{} {
-	return so.Ext
-}
-
-// SetExt Set Ext interface{}
-func (so *Source) SetExt(Ext interface{}) {
-	so.Ext = Ext
-}
-
-// GetSource Get return Source sql.NullString
-func (so *Source) GetSource() sql.NullString {
-	return so.Source
-}
-
-// SetSource Set Source sql.NullString
-func (so *Source) SetSource(Source sql.NullString) {
-	so.Source = Source
-}
-
-// GetTargetType Get return TargetType string
-func (so *Source) GetTargetType() string {
-	return so.TargetType
-}
-
-// SetTargetType Set TargetType string
-func (so *Source) SetTargetType(TargetType string) {
-	so.TargetType = TargetType
-}
-
-// GetUrl Get return Url string
-func (so *Source) GetUrl() string {
-	return so.Url
-}
-
-// SetUrl Set Url string
-func (so *Source) SetUrl(Url string) {
-	so.Url = Url
-}
-
-// GetUid Get return Uid int64
-func (so *Source) GetUid() int64 {
-	return so.Uid
-}
-
-// SetUid Set Uid int64
-func (so *Source) SetUid(Uid int64) {
-	so.Uid = Uid
 }

sql/intimate_extractor.sql

@@ -1,50 +1,63 @@
 create database if not exists `intimate_extractor`;
 use intimate_extractor;
 
-CREATE TABLE IF NOT EXISTS `anchor_info` (
+CREATE TABLE IF NOT EXISTS `streamer` (
-  `uid`           bigint AUTO_INCREMENT,
+  `uid`           bigint AUTO_INCREMENT COMMENT '自增UID, 便于查询定位',
-  `platform`      varchar(255) NOT NULL,
+  `platform`      varchar(255) NOT NULL COMMENT '平台',
-  `anchor_id`     varchar(255) NOT NULL,
+  `user_id`     varchar(255) NOT NULL COMMENT '用户唯一UID',
-  `anchor_name`   varchar(255) NOT NULL,
+  `user_name`   varchar(255) DEFAULT NULL COMMENT '用户名字 区别于ID',
-  `live_url`      text,
+  `live_url`      text COMMENT '直播的url', 
-  `channel`       varchar(128) DEFAULT NULL,
+  `channel`       varchar(128) DEFAULT NULL COMMENT'所属 频道,分类 未必所有平台都有明确的标签',
-  `tags`          json DEFAULT NULL,
+  `tag`           json DEFAULT NULL COMMENT 'streamer 最新的tag',
-  `ext`           json DEFAULT NULL,
+  `ext`           json DEFAULT NULL COMMENT '扩展类型, 把一些可能需要但是没字段的数据放在json扩展',
-  `update_time`   timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+  `is_update_streamer` tinyint(1) DEFAULT 0 COMMENT '是否需要持续更新streamer的信息. 1为需要,0则否',
+  `is_update_url` tinyint(1) DEFAULT 0 COMMENT '是否需要持续更新update_url. 1为需要,0则否',
+  `update_url`    json DEFAULT NULL COMMENT '更新数据的url, 如直播url, profile url等',
+  `update_interval` int  DEFAULT 30 COMMENT '分钟单位, 默认30分钟, 下次更新的时间间隔',
+  `update_time`   Timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+  `latest_log_uid` bigint COMMENT '最新更新的日志表的uid, 方便关联',  
+
+  `error_msg` text DEFAULT NULL COMMENT '错误信息',
+  `operator`      int DEFAULT 0 COMMENT '操作标志位, 根据不同解析方法有不同标志',
+
   PRIMARY KEY (`uid`),
-  UNIQUE KEY `platform_anchor_id_idx` (`platform`, `anchor_id`),
+  UNIQUE KEY `platform_user_id_idx` (`platform`, `user_id`),
   KEY `platform_idx` (`platform`),
-  KEY `anchor_id_idx` (`anchor_id`),
+  KEY `user_id_idx` (`user_id`),
-  KEY `anchor_name_idx` (`anchor_name`),
+  KEY `user_name_idx` (`user_name`),
   KEY `channel_idx` (`channel`),
-  KEY `update_time_idx` (`update_time`)
+  KEY `update_time_idx` (`update_time`),
+  KEY `operator_idx` (`operator`)
 );
 
 CREATE TABLE IF NOT EXISTS `collect_log` (
-  `uid`     bigint,
+  `log_uid`        bigint AUTO_INCREMENT COMMENT '日志自增UID',
-  `platform` varchar(255) NOT NULL,
+  `streamer_uid`     bigint COMMENT '对应streamer表的UID',
-  `anchor_id` varchar(255) NOT NULL,
+  `platform` varchar(255) NOT NULL COMMENT '平台名称, 方便于搜索日志分类',
+  `user_id` varchar(255) NOT NULL COMMENT '用户UID',
 
-  `is_live_streaming` tinyint(1) DEFAULT 0,
+  `is_live_streaming` tinyint(1) DEFAULT 0 COMMENT '是否正在直播',
-  `is_error` tinyint(1) DEFAULT 0,
+  `is_error` tinyint(1) DEFAULT 0 COMMENT '是否采集数据的时候出错, 便于定位错误',
 
-  `followers` bigint(11) DEFAULT NULL,
+  `followers` bigint(11) DEFAULT NULL COMMENT '关注数',
-  `views` bigint(11) DEFAULT NULL,
+  `views` bigint(11) DEFAULT NULL COMMENT '当前直播的观众',
-  `giver` json DEFAULT NULL,
+  `giver` json DEFAULT NULL COMMENT '打赏礼物者,和一些金额数据等, 数据类型异于平台',
-  `gratuity` bigint(11) DEFAULT NULL,
+  `gratuity` bigint(11) DEFAULT NULL COMMENT '打赏值的总值, 数据类型异于平台',
 
-  `live_title` text DEFAULT NULL,
+  `live_title` text DEFAULT NULL COMMENT '直播标题',
-  `live_start_time` timestamp   NULL DEFAULT NULL,
+  `live_start_time` Timestamp   NULL DEFAULT NULL COMMENT '直播开始时间',
-  `live_end_time` timestamp  NULL DEFAULT NULL,
+  `live_end_time` Timestamp  NULL DEFAULT NULL COMMENT '直播结束时间',
-  `update_time` timestamp  NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  `update_time` Timestamp  NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '数据日志更新时间',
-  `tags` json DEFAULT NULL,
+  `tags` json DEFAULT NULL COMMENT '主播直播的类型标签, 非永久固定',
-  `ext` json DEFAULT NULL,
+  `ext` json DEFAULT NULL COMMENT '扩展字段, 用于一些数据不存在的字段, 便于记录扩展',
   
-  `error_msg` text DEFAULT NULL,
+  `error_msg` text DEFAULT NULL COMMENT '错误信息',
 
-  KEY `uid_idx` (`uid`),
+  PRIMARY KEY (`log_uid`),
+  KEY `streamer_uid_idx` (`streamer_uid`),
   KEY `platform_idx` (`platform`),
-  KEY `anchor_id_idx` (`anchor_id`),
+  KEY `user_id_idx` (`user_id`),
   KEY `is_live_streaming_idx` (`is_live_streaming`),
   KEY `is_error_idx` (`is_error`),
   KEY `followers_idx` (`followers`),

sql/intimate_source.sql

@@ -2,15 +2,38 @@ create database if not exists `intimate_source`;
 use intimate_source;
 
 CREATE TABLE IF NOT EXISTS `source_openrec` (
-	uid bigint AUTO_INCREMENT,
+	uid bigint AUTO_INCREMENT COMMENT '自增UID',
-	`url` text NOT NULL,
+
-	`target_type` varchar(64) NOT NULL,
+	`streamer_id` bigint DEFAULT NULL COMMENT 'streamer uid, 关联主播',
-	`source` longtext DEFAULT NULL,
+	`url` text NOT NULL COMMENT '获取源数据地址',
-	`ext` json DEFAULT NULL,
+	`source` longtext DEFAULT NULL COMMENT '源数据',
-	`pass_gob` blob DEFAULT NULL,
+	`ext` json DEFAULT NULL COMMENT '扩展字段',
-	`update_time` timestamp  NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+	`serialize` blob DEFAULT NULL COMMENT '需要给下个任务传递 序列花数据, 非必要不用',
-	`operator` int DEFAULT 0,
+	`update_time` Timestamp  NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新数据',
-	`error_msg` text DEFAULT NULL,
+	`error_msg` text DEFAULT NULL COMMENT '错误信息',
+
+	`target_type` varchar(64) NOT NULL COMMENT '目标类型',
+	`operator` int DEFAULT 0 COMMENT '操作标志位, 根据不同解析方法有不同标志',
+	PRIMARY KEY(`uid`),
+	KEY `operator_idx` (`operator`),
+	KEY `update_time_idx` (`update_time`),
+	KEY `target_type_idx` (`target_type`)
+ );
+
+
+ CREATE TABLE IF NOT EXISTS `source_twitch` (
+	uid bigint AUTO_INCREMENT COMMENT '自增UID',
+
+	`streamer_id` bigint DEFAULT NULL COMMENT 'streamer uid, 关联主播',
+	`url` text NOT NULL COMMENT '获取源数据地址',
+	`source` longtext DEFAULT NULL COMMENT '源数据',
+	`ext` json DEFAULT NULL COMMENT '扩展字段',
+	`serialize` blob DEFAULT NULL COMMENT '需要给下个任务传递 序列花数据, 非必要不用',
+	`update_time` Timestamp  NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新数据',
+	`error_msg` text DEFAULT NULL COMMENT '错误信息',
+
+	`target_type` varchar(64) NOT NULL COMMENT '目标类型',
+	`operator` int DEFAULT 0 COMMENT '操作标志位, 根据不同解析方法有不同标志',
 	PRIMARY KEY(`uid`),
 	KEY `operator_idx` (`operator`),
 	KEY `update_time_idx` (`update_time`),

sql/remake_database.sh

@@ -0,0 +1,10 @@
+# /bin/bash
+USER=root
+HOST=127.0.0.1
+PORT=4000
+
+# mysql -h $HOST -u $USER -P $PORT -c "drop database intimate_source";
+# mysql -h $HOST -u $USER -P $PORT -c "drop database intimate_extractor";
+
+mysql -h $HOST -u $USER -P $PORT < ./intimate_extractor.sql;
+mysql -h $HOST -u $USER -P $PORT < ./intimate_source.sql;

store.go

@@ -3,6 +3,7 @@ package intimate
 import (
 	"database/sql"
 	"log"
+	"time"
 
 	_ "github.com/go-sql-driver/mysql"
 )
@@ -25,24 +26,43 @@ type ISet interface {
 	Set(string, interface{})
 }
 
+type IGet interface {
+	Get(string) interface{}
+}
+
+type IGetSet interface {
+	ISet
+	IGet
+}
+
 // SourceStore 储存
-type SourceStore struct {
+type StoreSource struct {
 	table string
 	db    *sql.DB
+
+	popCount   int
 	errorCount int
 	errorLimit int
 }
 
+func (store *StoreSource) PopCount() int {
+	return store.popCount
+}
+
+func (store *StoreSource) Close() error {
+	return store.db.Close()
+}
+
 // NewSourceStore 创建一个存储实例
-func NewSourceStore(table string) *SourceStore {
+func NewStoreSource(table string) *StoreSource {
 	db, err := sql.Open("mysql", InitConfig.Database.SourceURI)
 	if err != nil {
 		panic(err)
 	}
-	return &SourceStore{table: table, db: db}
+	return &StoreSource{table: table, db: db}
 }
 
-func (store *SourceStore) errorAlarm(err error) {
+func (store *StoreSource) errorAlarm(err error) {
 	if err != nil {
 		log.Println("store error: ", err)
 		// 报警. 如果数据插入有问题
@@ -58,46 +78,67 @@ func (store *SourceStore) errorAlarm(err error) {
 }
 
 // Insert 插入数据
-func (store *SourceStore) Insert(isource IGetSource) {
+func (store *StoreSource) Insert(isource IGet) {
-	_, err := store.db.Exec("insert into "+store.table+"(url, target_type, source,  ext, operator, error_msg) values(?,?,?,?,?,?)", isource.GetUrl(), isource.GetTargetType(), isource.GetSource(), isource.GetExt(), isource.GetOperator(), isource.GetErrorMsg())
+	_, err := store.db.Exec("insert into "+store.table+"(url, target_type, source,  ext, operator, error_msg, streamer_id) values(?,?,?,?,?,?,?)", isource.Get("Url"), isource.Get("Target"), isource.Get("Source"), isource.Get("Ext"), isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("StreamerId"))
-	store.errorAlarm(err)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// Deduplicate 去重
+func (store *StoreSource) Deduplicate(target Target, field string) {
+	sql := `DELETE FROM ` + store.table + ` WHERE uid NOT IN (SELECT MAX(s.uid) FROM (SELECT uid, ` + field + ` FROM ` + store.table + ` force index(target_type_idx) WHERE target_type = "` + string(target) + `" ) s GROUP BY s.` + string(field) + `) ;`
+	_, err := store.db.Exec(sql)
+	if err != nil {
+		panic(err)
+	}
 }
 
 // Update 更新数据
-func (store *SourceStore) Update(isource IUpdateSource) {
+func (store *StoreSource) Update(isource IGet) {
-	_, err := store.db.Exec("update "+store.table+" set  ext = ?, pass_gob = ?, operator = ?, error_msg = ? where uid = ?", isource.GetExt(), isource.GetPassGob(), isource.GetOperator(), isource.GetErrorMsg(), isource.GetUid())
+	_, err := store.db.Exec("update "+store.table+" set  ext = ?, pass_gob = ?, operator = ?, error_msg = ? where uid = ?", isource.Get("Ext"), isource.Get("PassGob"), isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	store.errorAlarm(err)
+	if err != nil {
+		panic(err)
+	}
 }
 
 // UpdateOperator 更新数据操作标志位
-func (store *SourceStore) UpdateOperator(isource IUpdateSource) {
+func (store *StoreSource) UpdateOperator(isource IGet) {
-	_, err := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.GetOperator(), isource.GetErrorMsg(), isource.GetUid())
+	_, err := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	store.errorAlarm(err)
+	if err != nil {
+		panic(err)
+	}
 }
 
 // UpdateError 更新错误数据
-func (store *SourceStore) UpdateError(isource IUpdateSource, err error) {
+func (store *StoreSource) UpdateError(isource IGetSet, err error) {
-	isource.SetOperator(int32(OperatorError))
+	isource.Set("Operator", int32(OperatorError)+isource.Get("Operator").(int32))
-	isource.SetErrorMsg(sql.NullString{String: err.Error(), Valid: true})
+	isource.Set("ErrorMsg", sql.NullString{String: err.Error(), Valid: true})
-	_, dberr := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.GetOperator(), isource.GetErrorMsg(), isource.GetUid())
+	_, dberr := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	store.errorAlarm(dberr)
+	if dberr != nil {
+		// email tell owner to deal with
+		panic(err)
+	}
 }
 
 // Restore 恢复Operator数据状态
-func (store *SourceStore) Restore(isource IUpdateSource) {
+func (store *StoreSource) Restore(isource IGet) {
-	_, err := store.db.Exec("update "+store.table+" set  operator = ?  where uid = ?", isource.GetLastOperator(), isource.GetUid())
+	_, dberr := store.db.Exec("update "+store.table+" set  operator = ?  where uid = ?", isource.Get("LastOperator"), isource.Get("Uid"))
-	store.errorAlarm(err)
+	if dberr != nil {
+		// email tell owner to deal with
+		panic(dberr)
+	}
 }
 
 // Pop 弹出一条未处理的数据
-func (store *SourceStore) Pop(targetType string, operators ...int32) (IUpdateSource, error) {
+func (store *StoreSource) Pop(targetType Target, operators ...int32) (*Source, error) {
 
 	tx, err := store.db.Begin()
 	if err != nil {
 		return nil, err
 	}
-	var args = []interface{}{targetType}
+	var args = []interface{}{string(targetType)}
-	selectSQL := `select uid, url, target_type, source, ext, operator, update_time from ` + store.table + ` where target_type = ?`
+	selectSQL := `select uid, url, target_type, source, ext, operator, update_time, streamer_id from ` + store.table + ` where target_type = ?`
 	if len(operators) == 0 {
 		selectSQL += " and operator = ?"
 		args = append(args, 0)
@@ -120,35 +161,46 @@ func (store *SourceStore) Pop(targetType string, operators ...int32) (IUpdateSou
 				log.Println(err)
 			}
 		}
+		store.popCount++
 	}()
 
 	s := &Source{}
 	// uid, url, target_type, source, ext, operator
-	err = row.Scan(&s.Uid, &s.Url, &s.TargetType, &s.Source, &s.Ext, &s.Operator, &s.UpdateTime)
+	err = row.Scan(&s.Uid, &s.Url, &s.Target, &s.Source, &s.Ext, &s.Operator, &s.UpdateTime, &s.StreamerId)
 	if err != nil {
 		return nil, err
 	}
-	s.SetLastOperator(s.Operator)
+
+	s.Set("LastOperator", s.Operator)
 	_, err = tx.Exec("update "+store.table+" set operator = ? where uid = ?", OperatorWait, s.Uid)
 	return s, nil
 }
 
-// AnchorTable 主播表名称
+// StreamerTable 主播表名称
-const AnchorTable string = "anchor_info"
+const StreamerTable string = "streamer"
 
 // CollectLogTable 采集日志表
 const CollectLogTable string = "collect_log"
 
-type ExtractorStore struct {
+type StoreExtractor struct {
 	db *sql.DB
 
+	popCount   int
 	errorCount int
 	errorLimit int
 }
 
-func (store *ExtractorStore) errorAlarm(err error) {
+func (store *StoreExtractor) PopCount() int {
+	return store.popCount
+}
+
+func (store *StoreExtractor) Close() error {
+	return store.db.Close()
+}
+
+func (store *StoreExtractor) errorAlarm(err error) {
 	if err != nil {
-		log.Panic("store error: ", err)
+		log.Println("store error: ", err)
 		// 报警. 如果数据插入有问题
 		store.errorCount++
 		if store.errorCount >= store.errorLimit {
@@ -161,66 +213,181 @@ func (store *ExtractorStore) errorAlarm(err error) {
 	}
 }
 
-func NewExtractorStore() *ExtractorStore {
+// NewStoreExtractor 生成一个extractor库的相关链接
+func NewStoreExtractor() *StoreExtractor {
 	db, err := sql.Open("mysql", InitConfig.Database.ExtractorURI)
 	if err != nil {
 		panic(err)
 	}
-	return &ExtractorStore{db: db}
+	return &StoreExtractor{db: db}
 }
 
-/*
+// Pop 弹出一条未处理的数据
-  `uid`           bigint,
+func (store *StoreExtractor) Pop(platform Platform, operators ...int32) (*Streamer, error) {
-  `platform`      varchar(255) NOT NULL,
-  `anchor_id`     varchar(255) NOT NULL,
-  `anchor_name`   varchar(255) NOT NULL,
-  `live_url`      text,
-  `channel`       varchar(128) DEFAULT NULL,
-  `show_type`     varchar(255) DEFAULT NULL,
-*/
 
-// InsertAnchorInfo AnchorInfo表, 插入数据
-func (store *ExtractorStore) InsertAnchorInfo(isource IGetAnchorInfo) (Uid int64, err error) {
-	// select uid from table where platform = ? and anchor_id = ?
-	selectSQL := "select uid from " + AnchorTable + " where platform = ? and anchor_id = ?"
 	tx, err := store.db.Begin()
 	if err != nil {
-		log.Println(err)
+		return nil, err
-		return 0, err
+	}
+	var args = []interface{}{string(platform)}
+	selectSQL := `select uid, update_time, user_id, update_url, is_update_streamer, update_interval from ` + StreamerTable + ` where platform = ? and  TIMESTAMPDIFF(MINUTE , update_time, CURRENT_TIMESTAMP()) >= update_interval`
+	if len(operators) == 0 {
+		selectSQL += " and operator = ?"
+		args = append(args, 0)
+	} else {
+		for _, operator := range operators {
+			selectSQL += " and operator = ?"
+			args = append(args, operator)
+		}
 	}
 
-	row := tx.QueryRow(selectSQL+` limit 1 for update`, isource.GetPlatform(), isource.GetAnchorId())
+	defer func() {
-
+		err := tx.Commit()
-	var uid int64
-	if err = row.Scan(&uid); err == nil {
-		return uid, nil
-	}
-
-	result, err := tx.Exec("insert  into "+AnchorTable+"(platform, anchor_id, anchor_name,  live_url, channel, tags, ext) values(?,?,?,?,?,?,?);", isource.GetPlatform(), isource.GetAnchorId(), isource.GetAnchorName(), isource.GetLiveUrl(), isource.GetChannel(), isource.GetTags(), isource.GetExt())
-
-	if err != nil {
-		log.Println(err)
-		return 0, nil
-	}
-
-	err = tx.Commit()
 		if err != nil {
 			log.Println(err)
 			err = tx.Rollback()
 			if err != nil {
 				log.Println(err)
 			}
-		return 0, err
+		}
+		store.popCount++
+	}()
+
+	// log.Println(selectSQL + ` limit 1 for update`)
+	row := tx.QueryRow(selectSQL+` limit 1 for update`, args...)
+
+	s := &Streamer{}
+	// uid, url, target_type, source, ext, operator
+	err = row.Scan(&s.Uid, &s.UpdateTime, &s.UserId, &s.UpdateUrl, &s.IsUpdateStreamer, &s.UpdateInterval)
+	if err != nil {
+		return nil, err
+	}
+	s.Set("LastOperator", s.Operator)
+	_, err = tx.Exec("update "+StreamerTable+" set operator = ? where uid = ?", OperatorWait, s.Uid)
+	return s, nil
+}
+
+// InsertStreamer Streamer表, 插入数据
+func (store *StoreExtractor) InsertStreamer(streamer IGet) (isExists bool) {
+	// select uid from table where platform = ? and user_id = ?
+	selectSQL := "SELECT is_update_url, uid FROM " + StreamerTable + " WHERE platform = ? AND user_id = ?"
+	tx, err := store.db.Begin()
+	if err != nil {
+		panic(err)
 	}
 
-	return result.LastInsertId()
+	defer func() {
+		err = tx.Commit()
+		if err != nil {
+			rerr := tx.Rollback()
+			if rerr != nil {
+				log.Println(rerr)
+			}
+			panic(err)
+		}
+	}()
+
+	row := tx.QueryRow(selectSQL+` LIMIT 1 FOR UPDATE`, streamer.Get("Platform"), streamer.Get("UserId"))
+	var isUpdateUrl bool
+	var Uid int64
+	if err = row.Scan(&isUpdateUrl, &Uid); err == nil {
+		if isUpdateUrl {
+			tx.Exec("UPDATE "+StreamerTable+" SET update_url = ?", streamer.Get("UpdateUrl"))
+		}
+		streamer.(ISet).Set("Uid", Uid)
+		return true
+	}
+
+	_, err = tx.Exec("INSERT INTO "+StreamerTable+"(platform, user_id, update_url, update_time) VALUES(?,?,?,?);", streamer.Get("Platform"), streamer.Get("UserId"), streamer.Get("UpdateUrl"), time.Now().Add(-time.Minute*60))
+	if err != nil {
+		panic(err)
+	}
+	return false
 }
 
-// InsertCollectLog CollectLog表插入数据
+// UpdateError 更新错误数据
-func (store *ExtractorStore) InsertCollectLog(isource IGetCollectLog) error {
+func (store *StoreExtractor) UpdateError(isource IGetSet, err error) {
-	_, err := store.db.Exec("insert  into "+CollectLogTable+"(uid, platform, anchor_id,  is_live_streaming, is_error, followers, views, giver, gratuity, live_title, live_start_time, live_end_time, update_time, tags, ext, error_msg) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+	isource.Set("Operator", int32(OperatorError)+isource.Get("Operator").(int32))
-		isource.GetUid(), isource.GetPlatform(), isource.GetAnchorId(), isource.GetIsLiveStreaming(), isource.GetIsError(), isource.GetFollowers(), isource.GetViews(), isource.GetGiver(), isource.GetGratuity(), isource.GetLiveTitle(), isource.GetLiveStartTime(), isource.GetLiveEndTime(), isource.GetUpdateTime(), isource.GetTags(), isource.GetExt(), isource.GetErrorMsg(),
+	isource.Set("ErrorMsg", sql.NullString{String: err.Error(), Valid: true})
-	)
+	_, dberr := store.db.Exec("update "+StreamerTable+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	store.errorAlarm(err)
+	if dberr != nil {
-	return err
+		// email tell owner to deal with
+		panic(err)
+	}
+}
+
+// UpdateStreamerLog 只更新Streamer的关联日志和时间戳
+func (store *StoreExtractor) UpdateStreamerLog(latestUid int64, streamerUid int64) {
+	_, err := store.db.Exec("UPDATE "+StreamerTable+" SET  latest_log_uid = ?,  update_time = CURRENT_TIMESTAMP() WHERE uid = ?", latestUid, streamerUid)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// UpdateOperator Streamer表, 插入数据
+func (store *StoreExtractor) UpdateOperator(isource IGet) {
+	_, err := store.db.Exec("update "+StreamerTable+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
+	if err != nil {
+		panic(err)
+	}
+}
+
+// UpdateStreamer Streamer表, 插入数据
+func (store *StoreExtractor) UpdateStreamer(streamer IGet) {
+	_, err := store.db.Exec("UPDATE "+StreamerTable+" SET user_name = ?,  live_url = ?, channel = ?, latest_log_uid = ?, tags = ?, ext = ?, operator = ?, update_time = ?, update_interval = ? WHERE uid = ?;",
+		streamer.Get("UserName"), streamer.Get("LiveUrl"), streamer.Get("Channel"), streamer.Get("LatestLogUid"), streamer.Get("Tags"), streamer.Get("Ext"), streamer.Get("Operator"), streamer.Get("UpdateTime"), streamer.Get("UpdateInterval"), streamer.Get("Uid"))
+	if err != nil {
+		panic(err)
+	}
+}
+
+// Update Streamer表, 更新指定的字段
+func (store *StoreExtractor) Update(streamer IGet, fieldvalues ...interface{}) {
+	updateSQL := "UPDATE " + StreamerTable + " SET "
+	var values []interface{}
+	for i := 0; i < len(fieldvalues); i += 2 {
+		field := fieldvalues[i]
+		values = append(values, fieldvalues[i+1])
+		updateSQL += field.(string) + " = ? "
+	}
+	updateSQL += "WHERE uid = ?"
+	values = append(values, streamer.Get("Uid"))
+	_, err := store.db.Exec(updateSQL, values...)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// InsertClog CollectLog表插入数据
+func (store *StoreExtractor) InsertClog(clog IGet) int64 {
+	tx, err := store.db.Begin()
+
+	defer func() {
+		if err := recover(); err != nil {
+			tx.Rollback()
+			log.Panic(err)
+		}
+	}()
+
+	if err != nil {
+		panic(err)
+	}
+
+	result, err := tx.Exec("insert  into "+CollectLogTable+"(streamer_uid, platform, user_id,  is_live_streaming, is_error, followers, views, giver, gratuity, live_title, live_start_time, live_end_time, update_time, tags, ext, error_msg) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+		clog.Get("StreamerUid"), clog.Get("Platform"), clog.Get("UserId"), clog.Get("IsLiveStreaming"), clog.Get("IsError"), clog.Get("Followers"), clog.Get("Views"), clog.Get("Giver"), clog.Get("Gratuity"), clog.Get("LiveTitle"), clog.Get("LiveStartTime"), clog.Get("LiveEndTime"), clog.Get("UpdateTime"), clog.Get("Tags"), clog.Get("Ext"), clog.Get("ErrorMsg"),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	logUid, err := result.LastInsertId()
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = tx.Exec("update "+StreamerTable+" set latest_log_uid = ? where uid = ?", logUid, clog.Get("StreamerUid"))
+	if err = tx.Commit(); err != nil {
+		panic(err)
+	}
+	return logUid
 }

store_test.go

@@ -2,8 +2,6 @@ package intimate
 
 import (
 	"testing"
-
-	"github.com/tidwall/gjson"
 )
 
 func TestStoreInsert(t *testing.T) {
@@ -26,20 +24,20 @@ func TestStoreInsertCase1(t *testing.T) {
 }
 
 func TestStorePopCase1(t *testing.T) {
-	store := NewSourceStore("source_openrec")
+	// store := NewSourceStore("source_openrec")
-	source, err := store.Pop(string(TTOpenrecRanking))
+	// source, err := store.Pop(string(TTOpenrecRanking))
-	if err != nil {
+	// if err != nil {
-		t.Error(err)
+	// 	t.Error(err)
-	}
+	// }
-	t.Error(source.GetOperator())
+	// t.Error(source.GetOperator())
-	t.Error(gjson.Valid(source.GetSource().String))
+	// t.Error(gjson.Valid(source.GetSource().String))
-	result := gjson.Parse(source.GetSource().String)
+	// result := gjson.Parse(source.GetSource().String)
-	if result.IsArray() {
+	// if result.IsArray() {
-		for _, User := range result.Array() {
+	// 	for _, User := range result.Array() {
-			t.Error(User.Get("channel.id").String())
+	// 		t.Error(User.Get("channel.id").String())
-		}
+	// 	}
-	} else {
+	// } else {
-		t.Error("array error")
+	// 	t.Error("array error")
-	}
+	// }
 
 }

table_list.go

@@ -6,5 +6,7 @@ type SourceTable string
 const (
 	// STOpenrec openrec源table名称
 	STOpenrec SourceTable = "source_openrec"
-)
 
+	// STTwitch twitch源table名称
+	STTwitch SourceTable = "source_twitch"
+)

target_type_list.go

@@ -1,12 +1,18 @@
 package intimate
 
-// TargetType 源的 目标类型 列表
+// Target 源的 目标类型 列表
-type TargetType string
+type Target string
 
 const (
-	// TTOpenrecRanking openrec源TargetType名称
+	// TOpenrecRanking 获取排名 Target名称
-	TTOpenrecRanking TargetType = "openrec_ranking"
+	TOpenrecRanking Target = "openrec_ranking"
 
-	// TTOpenrecUser openrec源TargetType名称
+	// TOpenrecUser  获取用户列表 源Target名称
-	TTOpenrecUser TargetType = "openrec_ranking"
+	TOpenrecUser Target = "openrec_user"
+
+	// TTwitchChannel twitch 获取类别操作目标
+	TTwitchChannel Target = "twitch_channel"
+
+	// TTwitchUser twitch 获取类别操作目标
+	TTwitchUser Target = "twitch_user"
 )

tasks/openrec/openrec_task1/source_openrec.go

@@ -1,19 +0,0 @@
-package main
-
-/*
-CREATE TABLE  `source_openrec` (
-	uid bigint AUTO_INCREMENT,
-	`url` text NOT NULL,
-	`target_type` varchar(64) NOT NULL,
-	`source` longtext DEFAULT NULL,
-	`ext` json DEFAULT NULL,
-
-	`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
-	`operator` int DEFAULT 0,
-	`error_msg` text DEFAULT NULL,
-	PRIMARY KEY(`uid`),
-	KEY `operator_idx` (`operator`),
-	KEY `update_time_idx` (`update_time`),
-	KEY `target_type_idx` (`target_type`)
- );
-*/

tasks/openrec/openrec_task1/task_openrec.go

@@ -1,7 +1,7 @@
 package main
 
 import (
-	"database/sql"
+	"encoding/json"
 	"intimate"
 	"log"
 	"os"
@@ -17,8 +17,11 @@ import (
 
 var openrecRanking *OpenrecRanking
 
-// store  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var store *intimate.SourceStore = intimate.NewSourceStore(string(intimate.STOpenrec))
+var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec))
+
+// estore 解析存储连接实例
+var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
 
 func init() {
 
@@ -60,55 +63,77 @@ func (or *OpenrecRanking) Execute(cxt *hunter.TaskContext) {
 		resp, err := cxt.Hunt()
 		if err != nil {
 			log.Println(err)
-			break
+			time.Sleep(time.Second * 2)
+			continue
 		}
 
-		wf := cxt.Temporary()
+		tp := cxt.Temporary()
 
 		content := resp.Content()
-		if len(content) <= 200 {
+		if len(content) <= 200 { //末页时没有内容返回, 末页退出
+			finishpoint := time.Now()
+			log.Println("任务Ranking UserId结束休眠, 下次启动时间:", finishpoint.Add(time.Minute*120))
+			for time.Now().Sub(finishpoint) < time.Minute*120 {
+				time.Sleep(time.Second)
+				if atomic.LoadInt32(&loop) <= 0 {
 					return
 				}
+			}
+
+			log.Println("获取Ranking UserId启动:", time.Now())
+			querys := tp.GetQuery()
+			querys.Set("page", strconv.Itoa(1))
+			tp.SetQuery(querys)
+
+			continue
+		}
 
 		result := gjson.ParseBytes(content)
 		if result.IsArray() {
 			for _, User := range result.Array() {
-				data := &intimate.Source{}
+
 				userid := User.Get("channel.id").String()
 
-				data.SetSource(sql.NullString{String: userid, Valid: len(userid) > 0})
+				// data := &intimate.Source{}
-				data.SetUrl(wf.GetRawURL())
+				// data.Source = sql.NullString{String: userid, Valid: len(userid) > 0}
-				data.SetTargetType(string(intimate.TTOpenrecUser))
+				// data.Url = tp.GetRawURL()
-				store.Insert(data)
+				// data.TargetType = string(intimate.TTOpenrecUser)
+				// sstore.Insert(data)
+
+				streamer := &intimate.Streamer{}
+				streamer.UserId = userid
+				streamer.Platform = intimate.Popenrec
+
+				updateUrl := make(map[string]interface{})
+
+				supportersUrl := "curl 'https://www.openrec.tv/viewapp/api/v6/supporters?identify_id=sumomo_xqx&month=&Uuid=B96EE988-E3A2-4A44-A543-611A8B4BC683&Token=46598c320408bd69ae3c63298f6f4a3a97354175&Random=AZVXNAAXQVMOSVWNDPIQ&page_number=1' -H 'accept: application/json, text/javascript, */*; q=0.01'   -H 'user-agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36'  -H 'cookie: uuid=B96EE988-E3A2-4A44-A543-611A8B4BC683;'   --compressed"
+				updateUrl["supporters"] = supportersUrl
+				updateUrl["user"] = "https://www.openrec.tv/user/" + userid
+				updateUrl["live"] = "https://www.openrec.tv/live/" + userid
+
+				updateUrlBytes, err := json.Marshal(updateUrl)
+				if err != nil {
+					estore.UpdateError(streamer, err)
+					continue
+				}
+
+				streamer.UpdateUrl = updateUrlBytes
+				estore.InsertStreamer(streamer)
 			}
 		}
 
-		querys := wf.GetQuery()
+		// 修改url query 参数的page递增. 遍历所有页面
+		querys := tp.GetQuery()
 		page, err := strconv.Atoi(querys.Get("page"))
 		if err != nil {
 			log.Println(err)
 			return
 		}
-		return
+
 		page++
 		querys.Set("page", strconv.Itoa(page))
-		wf.SetQuery(querys)
+		tp.SetQuery(querys)
-		time.Sleep(time.Second * 2)
+
+		time.Sleep(time.Second * 1)
 	}
 }
-
-// OpenrecUser 获取用户信息
-type OpenrecUser struct {
-	hunter.PreCurlUrl
-}
-
-// Execute 执行方法
-func (or *OpenrecUser) Execute(cxt *hunter.TaskContext) {
-	resp, err := cxt.Hunt()
-	if err != nil {
-		log.Println(err)
-		return
-	}
-
-	resp.Content()
-}

tasks/openrec/openrec_task1/task_openrec_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"testing"
+	"time"
 
 	"github.com/tidwall/gjson"
 
@@ -56,6 +57,12 @@ func TestRanking(t *testing.T) {
 	ht.Execute()
 }
 
+func TestTimeAdd(t *testing.T) {
+	finishpoint := time.Now()
+	time.Sleep(time.Second * 2)
+	t.Error(time.Now().Sub(finishpoint) > time.Second*1)
+}
+
 func TestRankingInsert(t *testing.T) {
 	ht := hunter.NewHunter(openrecRanking)
 	ht.Execute()

tasks/openrec/openrec_task2/.gitignore

@@ -1 +1,2 @@
 openrec_task2
+log

tasks/openrec/openrec_task2/task_openrec.go

@@ -20,8 +20,11 @@ import (
 
 var oer *OpenrecExtratorRanking
 
-// store  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var store *intimate.SourceStore = intimate.NewSourceStore(string(intimate.STOpenrec))
+var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec))
+
+// estore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_extractor.sql
+var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
 
 func init() {
 	oer = &OpenrecExtratorRanking{}
@@ -44,41 +47,53 @@ func (oer *OpenrecExtratorRanking) Execute(cxt *hunter.TaskContext) {
 		atomic.StoreInt32(&loop, 0)
 	}()
 
+	var lasterr error = nil
+
 	for atomic.LoadInt32(&loop) > 0 {
 
-		source, err := store.Pop(string(intimate.TTOpenrecUser))
+		streamer, err := estore.Pop(intimate.Popenrec) //队列里弹出一个streamer行. 进行解析
 
-		if source == nil || err != nil {
+		if streamer == nil || err != nil {
-			log.Println(err)
+			if err != lasterr {
+				log.Println(err, lasterr)
+				lasterr = err
+			}
 			time.Sleep(time.Second * 2)
 			continue
 		}
 
-		userSource := &intimate.Source{}
+		userId := streamer.UserId
-		userid := source.GetSource().String
-		userUrl := "https://www.openrec.tv/user/" + userid
-		userSource.SetUrl(userUrl)
 
-		wf := cxt.Session().Get(userUrl)
+		var updateUrl map[string]string
-		resp, err := wf.Execute()
+
-		source.SetUpdateTime(sql.NullTime{Time: time.Now(), Valid: true})
+		err = json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl) // 反序列化update_url, 里面存了需要采集的url
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		// Check Userid
+
+		userUrl := updateUrl["user"]
+		tp := cxt.Session().Get(userUrl) // 获取user url页面数据
+		resp, err := tp.Execute()
+		streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
 
 		if err != nil {
 			log.Println(err)
-			store.UpdateError(source, err)
+			estore.UpdateError(streamer, err)
 			continue
 		}
 
-		cookies := cxt.Session().GetCookies(wf.GetParsedURL())
+		cookies := cxt.Session().GetCookies(tp.GetParsedURL())
 
-		scurl := "https://www.openrec.tv/viewapp/api/v6/supporters?identify_id=sumomo_xqx&month=&Uuid=B96EE988-E3A2-4A44-A543-611A8B4BC683&Token=46598c320408bd69ae3c63298f6f4a3a97354175&Random=AZVXNAAXQVMOSVWNDPIQ&page_number=1 -H 'accept: application/json, text/javascript, */*; q=0.01'   -H 'user-agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36'  -H 'cookie: uuid=B96EE988-E3A2-4A44-A543-611A8B4BC683;'   --compressed"
+		scurl := updateUrl["supporters"] //获取打赏者的数据
 		curl := gcurl.ParseRawCURL(scurl)
 		supportersSession := curl.CreateSession()
 
 		temporary := curl.CreateTemporary(supportersSession)
 		supportersSession.SetCookies(temporary.GetParsedURL(), cookies)
 		var supporters []string
-		for {
+		for { // supporters 数据需要登录信息. 下面为赋值 supporters链接获取的uid token random码
 
 			supportersQuery := temporary.GetQuery()
 
@@ -99,7 +114,7 @@ func (oer *OpenrecExtratorRanking) Execute(cxt *hunter.TaskContext) {
 				}
 			}
 
-			supportersQuery.Set("identify_id", source.GetSource().String)
+			supportersQuery.Set("identify_id", userId)
 			temporary.SetQuery(supportersQuery)
 
 			resp, err := temporary.Execute()
@@ -107,13 +122,13 @@ func (oer *OpenrecExtratorRanking) Execute(cxt *hunter.TaskContext) {
 				log.Println(err)
 			}
 			supporterjson := gjson.ParseBytes(resp.Content())
-			supporterdata := supporterjson.Get("data")
+			supporterdata := supporterjson.Get("data") //解析supporters获取的json数据
 			if supporterdata.Type == gjson.Null {
 				break
 			}
 			supporters = append(supporters, string(resp.Content()))
 
-			page := supportersQuery.Get("page_number")
+			page := supportersQuery.Get("page_number") // page_number 加1
 			pageint, err := strconv.Atoi(page)
 			if err != nil {
 				log.Println(err)
@@ -128,28 +143,36 @@ func (oer *OpenrecExtratorRanking) Execute(cxt *hunter.TaskContext) {
 		// cookies := cxt.Session().GetCookies(wf.GetParsedURL())
 		ext := make(map[string]interface{})
 
-		ext["supporters"] = supporters
+		ext["json_supporters"] = supporters
-		ext["user"] = string(resp.Content())
+		ext["html_user"] = string(resp.Content())
 
-		wf = cxt.Session().Get("https://www.openrec.tv/live/" + userid)
+		liveUrl := updateUrl["live"]
-		resp, err = wf.Execute()
+		tp = cxt.Session().Get(liveUrl)
+		resp, err = tp.Execute()
 		if err != nil {
 			log.Println(err)
-			store.UpdateError(source, err)
+			estore.UpdateError(streamer, err)
 			continue
 		}
-		ext["user_live"] = string(resp.Content())
+		ext["html_live"] = string(resp.Content())
+		ext["var_user_id"] = userId
 
 		extJsonBytes, err := json.Marshal(ext)
 		if err != nil {
 			log.Println(err)
-			store.UpdateError(source, err)
+			estore.UpdateError(streamer, err)
 			continue
 		}
 
-		source.SetOperator(int32(intimate.OperatorOK))
+		streamer.Operator = int32(intimate.OperatorOK)
-		source.SetExt(string(extJsonBytes))
+
-		store.Update(source)
+		source := &intimate.Source{}
+		source.Target = intimate.TOpenrecUser
+		source.Ext = string(extJsonBytes)
+		source.StreamerId = sql.NullInt64{Int64: streamer.Uid, Valid: true}
+		sstore.Insert(source)
+
+		estore.UpdateOperator(streamer)
 	}
 
 }

tasks/twitch/twitch_task1/.gitignore

@@ -0,0 +1,2 @@
+twitch_task1
+log

tasks/twitch/twitch_task1/main.go

@@ -0,0 +1,6 @@
+package main
+
+func main() {
+	e := ChannelLink{}
+	e.Execute()
+}

tasks/twitch/twitch_task1/task_twitch.go

@@ -0,0 +1,113 @@
+package main
+
+import (
+	"database/sql"
+	"intimate"
+	"log"
+	"time"
+
+	"github.com/tebeka/selenium"
+)
+
+// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitch))
+
+// estore 解析存储连接实例
+var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+
+// 获取类型的所有频道链接
+
+// ChannelLink 频道链接
+type ChannelLink struct {
+}
+
+// Execute 执行任务
+func (cl *ChannelLink) Execute() {
+	var err error
+	wd := intimate.GetChromeDriver(3030)
+	ps := intimate.NewPerfectShutdown()
+
+	weburl := "https://www.twitch.tv/directory?sort=VIEWER_COUNT"
+	err = wd.Get(weburl)
+	if err != nil {
+		panic(err)
+	}
+
+	cardCondition := func(wd selenium.WebDriver) (bool, error) {
+		elements, err := wd.FindElements(selenium.ByXPATH, "//span/a[contains(@data-a-target,'card-') and @href]")
+		if err != nil {
+			return false, err
+		}
+		return len(elements) > 0, nil
+	}
+	wd.WaitWithTimeout(cardCondition, time.Second*15)
+	time.Sleep(time.Second)
+
+	e, err := wd.FindElement(selenium.ByXPATH, "//button[@data-a-target='browse-sort-menu']")
+	if err != nil {
+		panic(err)
+	}
+	e.Click()
+
+	var hrefs map[string]bool = make(map[string]bool)
+	var delayerror = 5
+	var samecount = 0
+	for i := 0; i <= 200; i++ {
+		cards, err := wd.FindElements(selenium.ByXPATH, "//span/a[contains(@data-a-target,'card-') and @href]")
+		if err != nil {
+			log.Println(err)
+			break
+		}
+
+		if len(cards) == samecount {
+			delayerror--
+			if delayerror <= 0 {
+				break
+			}
+		} else {
+			delayerror = 5
+		}
+
+		for ii := 0; ii < 10; ii++ {
+			for _, card := range cards {
+				href, err := card.GetAttribute("href")
+				if err != nil {
+					log.Println(href, err)
+					continue
+				} else {
+					hrefs[href] = true
+				}
+			}
+			break
+		}
+		samecount = len(cards)
+		if ps.IsClose() {
+			break
+		}
+
+		if len(cards) > 10 {
+			log.Println(len(cards))
+			wd.ExecuteScript(`items = document.evaluate("//div[@data-target='directory-page__card-container']/../self::div[@data-target and @style]", document, null, XPathResult.ORDERED_NODE_SNAPSHOT_TYPE, null); 
+		for (var i = 0; i < items.snapshotLength - 10; i++) { item = items.snapshotItem(i); item.remove() ;};`, nil)
+		}
+		time.Sleep(time.Millisecond * 200)
+		wd.KeyDown(selenium.EndKey)
+		time.Sleep(time.Millisecond * 200)
+		wd.KeyUp(selenium.EndKey)
+		time.Sleep(time.Millisecond * 2500)
+	}
+
+	for href := range hrefs {
+
+		// TODO: Save href
+		source := &intimate.Source{}
+		source.Source = sql.NullString{String: href, Valid: true}
+		source.Operator = 0
+		source.Target = intimate.TTwitchChannel
+		source.Url = weburl
+		sstore.Insert(source)
+	}
+
+	log.Println("hrefs len:", len(hrefs))
+	sstore.Deduplicate(intimate.TTwitchChannel, "source")
+}

tasks/twitch/twitch_task1/task_twitch_test.go

@@ -0,0 +1,14 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestCase1(t *testing.T) {
+	e := ChannelLink{}
+	e.Execute()
+}
+
+func TestLiveUrl(t *testing.T) {
+
+}

tasks/twitch/twitch_task2/.gitignore

@@ -0,0 +1,2 @@
+twitch_task2
+log

tasks/twitch/twitch_task2/main.go

@@ -0,0 +1,6 @@
+package main
+
+func main() {
+	ul := UserList{}
+	ul.Execute()
+}

tasks/twitch/twitch_task2/task_twitch.go

@@ -0,0 +1,177 @@
+package main
+
+import (
+	"database/sql"
+	"encoding/json"
+	"intimate"
+	"log"
+	"regexp"
+	"time"
+
+	"github.com/tebeka/selenium"
+)
+
+// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitch))
+
+// estore 解析存储连接实例
+var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+
+// 获取类型的所有频道链接
+
+// UserList 频道链接
+type UserList struct {
+}
+
+// Execute 执行任务
+func (cl *UserList) Execute() {
+	// DELETE FROM source_twitch WHERE uid NOT IN (SELECT MAX(s.uid) FROM (SELECT uid, source FROM source_twitch ) s GROUP BY s.source) ;
+	//article//a[@data-a-target='preview-card-title-link']
+
+	wd := intimate.GetChromeDriver(3030)
+	ps := intimate.NewPerfectShutdown()
+	counter := intimate.NewCounter()
+	counter.SetMaxLimit(100)
+	counter.SetMaxToDo(func(olist ...interface{}) error {
+		owd := olist[0].(*selenium.WebDriver)
+		(*owd).Close()
+		(*owd).Quit()
+		*owd = intimate.GetChromeDriver(3030)
+		return nil
+	}, &wd)
+
+	for !ps.IsClose() {
+
+		var err error
+		sourceChannel, err := sstore.Pop(intimate.TTwitchChannel)
+		if err != nil {
+			panic(err)
+		}
+
+		weburl := sourceChannel.Source.String + "?sort=VIEWER_COUNT"
+		err = wd.Get(weburl)
+		if err != nil {
+			log.Println(err)
+			sstore.UpdateError(sourceChannel, err)
+			time.Sleep(time.Second * 10)
+			continue
+		}
+
+		wd.WaitWithTimeout(func(wd selenium.WebDriver) (bool, error) {
+			_, err := wd.FindElement(selenium.ByXPATH, "(//div/p[@class=''])[last()]")
+			if err != nil {
+				return false, err
+			}
+			return true, nil
+		}, time.Second*10)
+
+		btn, err := wd.FindElement(selenium.ByXPATH, "//button[@data-a-target='browse-sort-menu']")
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		btn.Click()
+
+		var elements []selenium.WebElement
+		var liveurls = 0
+		var delayerror = 2
+		for i := 0; i < 200 && !ps.IsClose(); i++ {
+			elements, err = wd.FindElements(selenium.ByXPATH, "(//div/p[@class=''])[last()]")
+			if err != nil {
+				log.Println(err)
+				break
+			}
+			time.Sleep(time.Millisecond * 200)
+			wd.KeyDown(selenium.EndKey)
+			time.Sleep(time.Millisecond * 200)
+			wd.KeyUp(selenium.EndKey)
+			time.Sleep(time.Millisecond * 2000)
+			if len(elements) == liveurls {
+				delayerror--
+				if delayerror <= 0 {
+					break
+				}
+			} else {
+				delayerror = 2
+			}
+			liveurls = len(elements)
+		}
+		articles, err := wd.FindElements(selenium.ByXPATH, "//article")
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+
+		for _, article := range articles {
+
+			e, err := article.FindElement(selenium.ByXPATH, ".//a[@data-a-target='preview-card-title-link' and @href]")
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+
+			href, err := e.GetAttribute("href")
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+
+			btns, err := article.FindElements(selenium.ByXPATH, ".//div[@class='tw-full-width tw-inline-block']//button")
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+
+			var tags []string
+			for _, btn := range btns {
+				tag, err := btn.GetAttribute("data-a-target")
+				if err == nil {
+					tags = append(tags, tag)
+				}
+			}
+
+			streamer := &intimate.Streamer{}
+
+			matches := regexp.MustCompile(`https://www.twitch.tv/(\w+)`).FindStringSubmatch(href)
+			if len(matches) == 2 {
+				streamer.UserId = matches[1]
+			} else {
+				log.Println(href)
+				continue
+			}
+
+			jtags, err := json.Marshal(tags)
+			if err != nil {
+				log.Println(err)
+			} else {
+				streamer.Tags = jtags
+			}
+
+			streamer.Platform = intimate.Ptwitch
+
+			updateUrl := make(map[string]string)
+			updateUrl["live"] = href
+			streamer.LiveUrl = sql.NullString{String: href, Valid: true}
+			data, err := json.Marshal(updateUrl)
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+			streamer.UpdateUrl = data
+			streamer.Operator = 0
+			if estore.InsertStreamer(streamer) {
+				// log.Println("streamer update tags", streamer.Uid, tags)
+				estore.Update(streamer, "Tags", streamer.Tags)
+			}
+		}
+		log.Println("streamer find", len(articles))
+		if len(articles) == 0 {
+			sourceChannel.Operator = 5
+			sstore.UpdateOperator(sourceChannel)
+		}
+		counter.AddWithReset(1)
+	}
+
+	wd.Close()
+	wd.Quit()
+}

tasks/twitch/twitch_task2/task_twitch_test.go

@@ -0,0 +1,7 @@
+package main
+
+import "testing"
+
+func TestMain(t *testing.T) {
+	main()
+}

utils.go

@@ -1,8 +1,19 @@
 package intimate
 
 import (
+	"fmt"
 	"log"
+	"os"
+	"os/signal"
+	"runtime"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"syscall"
 	"time"
+
+	"github.com/tebeka/selenium"
+	"github.com/tebeka/selenium/chrome"
 )
 
 var zeroTime time.Time
@@ -17,6 +28,12 @@ func init() {
 
 }
 
+// ParseNumber 去逗号解析数字
+func ParseNumber(number string) (int64, error) {
+	number = strings.ReplaceAll(number, ",", "")
+	return strconv.ParseInt(number, 10, 64)
+}
+
 // ParseDuration time to duration eg: 1:40:00 -> time.Duration
 func ParseDuration(dt string) (time.Duration, error) {
 
@@ -37,8 +54,186 @@ func ParseDuration(dt string) (time.Duration, error) {
 
 	tdt, err := time.Parse("15:04:05", string(parse))
 	if err != nil {
-
 		return time.Duration(0), err
 	}
 	return tdt.Sub(zeroTime), nil
 }
+
+func GetChromeDriver(port int) selenium.WebDriver {
+	var err error
+	caps := selenium.Capabilities{"browserName": "chrome"}
+
+	chromecaps := chrome.Capabilities{}
+	for _, epath := range []string{"../../../crx/myblock.crx", "../../crx/myblock.crx"} {
+		_, err := os.Stat(epath)
+		if err == nil {
+			err := chromecaps.AddExtension(epath)
+			if err != nil {
+				panic(err)
+			}
+			break
+		}
+	}
+
+	if proxy := os.Getenv("chrome_proxy"); proxy != "" {
+		log.Println("proxy-server", proxy)
+		chromecaps.Args = append(chromecaps.Args, "--proxy-server="+proxy)
+	}
+
+	if proxy := os.Getenv("pac_proxy"); proxy != "" {
+		log.Println("--proxy-pac-url=" + proxy)
+		chromecaps.Args = append(chromecaps.Args, "--proxy-pac-url="+proxy)
+	}
+
+	// chromecaps.Args = append(chromecaps.Args, "--proxy-pac-url=http://127.0.0.1:1081/pac")
+	chromecaps.Args = append(chromecaps.Args, "--disk-cache-dir=/tmp/chromedriver-cache")
+	chromecaps.Args = append(chromecaps.Args, "--disable-gpu", "--disable-images", "--start-maximized", "--disable-infobars")
+	// chromecaps.Args = append(chromecaps.Args, "--headless")
+	chromecaps.Args = append(chromecaps.Args, "--no-sandbox")
+	chromecaps.Args = append(chromecaps.Args, "--disable-dev-shm-usage", "--mute-audio", "--safebrowsing-disable-auto-update")
+
+	chromecaps.ExcludeSwitches = append(chromecaps.ExcludeSwitches, "enable-automation")
+	caps.AddChrome(chromecaps)
+	_, err = selenium.NewChromeDriverService("/usr/bin/chromedriver", port)
+	if err != nil {
+		panic(err)
+	}
+	wd, err := selenium.NewRemote(caps, fmt.Sprintf("http://localhost:%d/wd/hub", port))
+	if err != nil {
+		panic(err)
+	}
+	runtime.SetFinalizer(wd, func(obj interface{}) {
+
+		if err := obj.(selenium.WebDriver).Close(); err != nil {
+			log.Println(err)
+		}
+		if err := obj.(selenium.WebDriver).Quit(); err != nil {
+			log.Println(err)
+		}
+
+	})
+	wd.ExecuteScript("windows.navigator.webdriver = undefined", nil)
+	if err != nil {
+		panic(err)
+	}
+
+	return wd
+}
+
+// PerfectShutdown 完美关闭程序
+type PerfectShutdown struct {
+	loop int32
+}
+
+// NewPerfectShutdown 创建完美关闭程序
+func NewPerfectShutdown() *PerfectShutdown {
+	ps := &PerfectShutdown{}
+	ps.loop = 1
+
+	go func() {
+		signalchan := make(chan os.Signal)
+		signal.Notify(signalchan, syscall.SIGINT, syscall.SIGKILL, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGSTOP)
+		log.Println("accept stop command:", <-signalchan)
+		atomic.StoreInt32(&ps.loop, 0)
+	}()
+
+	return ps
+}
+
+// IsClose 判断是否要关闭
+func (ps *PerfectShutdown) IsClose() bool {
+	return atomic.LoadInt32(&ps.loop) == 0
+}
+
+type Counter struct {
+	dcount   int
+	count    int
+	maxLimit int
+	minLimit int
+
+	minobj       []interface{}
+	maxobj       []interface{}
+	maxLimitToDo func(obj ...interface{}) error
+	minLimitToDo func(obj ...interface{}) error
+}
+
+func NewCounter() *Counter {
+	c := &Counter{}
+	return c
+}
+
+// SetDefault 设置默认值
+func (c *Counter) SetDefault(n int) {
+	c.dcount = n
+}
+
+// Reset 最置count为defaultCount值
+func (c *Counter) Reset() {
+	c.count = c.dcount
+}
+
+// SetCount 设置count到最大值的时候执行do函数
+func (c *Counter) SetCount(count int) {
+	c.count = count
+}
+
+// GetCount 设置count到最大值的时候执行do函数
+func (c *Counter) GetCount() int {
+	return c.count
+}
+
+// SetMinLimit 设置最小限制
+func (c *Counter) SetMinLimit(n int) {
+	c.minLimit = n
+}
+
+// SetMaxLimit 设置最大限制
+func (c *Counter) SetMaxLimit(n int) {
+	c.maxLimit = n
+}
+
+// SetMaxToDo 设置count到最大值的时候执行do函数
+func (c *Counter) SetMaxToDo(do func(obj ...interface{}) error, obj ...interface{}) {
+	c.maxLimitToDo = do
+	c.maxobj = obj
+}
+
+// SetMinToDo 设置count到最小值的时候执行do函数
+func (c *Counter) SetMinToDo(do func(obj ...interface{}) error, obj ...interface{}) {
+	c.minLimitToDo = do
+	c.minobj = obj
+}
+
+// AddWithReset 操作 count 默认值为0, 当触发限制时, 重置为默认值
+func (c *Counter) AddWithReset(n int) error {
+	c.count += n
+	if c.maxLimitToDo != nil {
+		if c.count >= c.maxLimit {
+			defer c.Reset()
+			return c.maxLimitToDo(c.maxobj...)
+		}
+	}
+	if c.minLimitToDo != nil {
+		if c.count <= c.minLimit {
+			defer c.Reset()
+			return c.minLimitToDo(c.minobj...)
+		}
+	}
+	return nil
+}
+
+// Add 操作 count 默认值为0
+func (c *Counter) Add(n int) error {
+	c.count += n
+	if c.maxLimitToDo != nil {
+		if c.count >= c.maxLimit {
+			return c.maxLimitToDo(c.maxobj...)
+		}
+	}
+	if c.minLimitToDo != nil {
+		if c.count <= c.minLimit {
+			return c.minLimitToDo(c.minobj...)
+		}
+	}
+	return nil
+}

xvfb.sh

@@ -0,0 +1 @@
+screen -dmS xvfb-99  Xvfb :99 -screen 0 1280x720x24 -ac -nolisten tcp -dpi 96 +extension RANDR -nolisten tcp