fix: stable

add mirrativ

autostore.go

@@ -34,9 +34,10 @@ type Table struct {
 	name    string
 	setting interface{}
 
-	updatesql string
+	updatesql    string
-	selectsql string
+	selectsql    string
-	insertsql string
+	insertsql    string
+	duplicatesql string
 }
 
 func NewStore(uri string) *Store {
@@ -54,6 +55,7 @@ func (store *Store) Table(name string) *Table {
 	table.name = name
 
 	table.insertsql = `INSERT INTO ` + table.name + `(%s) values(%s)`
+	table.duplicatesql = `INSERT INTO ` + table.name + `(%s) values(%s) ON DUPLICATE KEY UPDATE %s`
 	table.updatesql = `UPDATE ` + table.name + ` SET %s WHERE %s = ?`
 	table.selectsql = `SELECT %s FROM ` + table.name + ` WHERE %s `
 	return table
@@ -143,7 +145,7 @@ func (queue *Queue) Pop() (result interface{}, err error) {
 	rows, err := tx.Query(selectsql, queue.cond.CondArgs...)
 
 	if err != nil {
-		return nil, fmt.Errorf("table: %s queue is empty", queue.table.name)
+		return nil, fmt.Errorf("table: %s queue is empty. %s", queue.table.name, err.Error())
 	}
 
 	var fields = make([]interface{}, len(queue.fieldIndex))
@@ -226,6 +228,83 @@ func (t *Table) Insert(obj interface{}) error {
 	return err
 }
 
+// DUpdate ON DUPLICATE KEY UPDATE struct. Field对应的tag field字段
+type DUpdate struct {
+	Field string // selected 字段
+	Value interface{}
+}
+
+// InsertOrUpdate   nil 不插入. 不支持嵌套. 必须是Ptr类型
+func (t *Table) InsertOrUpdate(obj interface{}, updates ...DUpdate) error {
+	ov := reflect.ValueOf(obj).Elem()
+	ot := reflect.TypeOf(obj)
+
+	fieldsql := ""
+	argssql := ""
+
+	var SourceUpdate []*DUpdate
+	var OtherUpdate []*DUpdate
+	for _, u := range updates {
+		if u.Value == nil {
+			SourceUpdate = append(SourceUpdate, &u)
+		} else {
+			OtherUpdate = append(OtherUpdate, &u)
+		}
+	}
+
+	var args []interface{}
+	for i := 0; i < ov.NumField(); i++ {
+		field := ov.Field(i)
+		ftype := ot.Elem().Field(i)
+
+		if fname, ok := ftype.Tag.Lookup("field"); ok {
+			// if flag, ok := ftype.Tag.Lookup("uid"); ok {
+			// 	if flag == "auto" {
+			// 		continue
+			// 	}
+			// }
+
+			k := ftype.Type.Kind()
+			if k == reflect.Ptr || k == reflect.Interface {
+				if !field.IsNil() {
+					felem := field.Elem()
+					args = append(args, felem.Interface())
+					fieldsql += fname + ","
+					argssql += "?,"
+				}
+			} else {
+				args = append(args, field.Interface())
+				fieldsql += fname + ","
+				argssql += "?,"
+			}
+
+			for _, u := range SourceUpdate {
+				if u.Field == fname {
+					u.Value = args[len(args)-1]
+					break
+				}
+			}
+
+		}
+
+	}
+
+	var duplicateSet string = ""
+	for _, u := range SourceUpdate {
+		duplicateSet += u.Field + " = ?,"
+		args = append(args, u.Value)
+	}
+
+	for _, u := range OtherUpdate {
+		duplicateSet += u.Field + " = ?,"
+		args = append(args, u.Value)
+	}
+
+	ssql := fmt.Sprintf(t.duplicatesql, fieldsql[:len(fieldsql)-1], argssql[:len(argssql)-1], duplicateSet[:len(duplicateSet)-1])
+	_, err := t.store.db.Exec(ssql, args...)
+	return err
+}
+
 // InsertRetAutoID   nil 不插入. 不支持嵌套. 并返回auto uid
 func (t *Table) InsertRetAutoID(obj interface{}) (int64, error) {
 	ov := reflect.ValueOf(obj).Elem()
@@ -348,10 +427,10 @@ func (t *Table) UpdateError(obj interface{}, err error) {
 		}
 	}
 
-	_, dberr := t.store.db.Exec("update "+t.name+" set  operator = ?, error_msg = ? where ? = ?", 10000, sql.NullString{String: err.Error(), Valid: true}, uidname, uidvalue)
+	_, dberr := t.store.db.Exec("update "+t.name+" set  operator = ?, error_msg = ? where "+uidname+" = ?", 10000, sql.NullString{String: err.Error(), Valid: true}, uidvalue)
 	if dberr != nil {
 		// email tell owner to deal with
-		panic(err)
+		panic(dberr)
 	}
 }
 

autostore_test.go

@@ -5,26 +5,24 @@ import (
 	"encoding/json"
 	"testing"
 	"time"
-
-	"github.com/davecgh/go-spew/spew"
 )
 
-func estAutoStore(t *testing.T) {
+func TestAutoStore(t *testing.T) {
 	uri := "root:@tcp(127.0.0.1:4000)/test?parseTime=true&loc=Local&charset=utf8mb4&collation=utf8mb4_unicode_ci"
 	store := NewStore(uri)
 
-	queue := store.Table("streamer").Queue(TSreamer{}, CondWhere{Condition: "operator = 0"})
+	// queue := store.Table("streamer").Queue(TSreamer{}, CondWhere{Condition: "operator = 0"})
-	re, _ := queue.Pop()
+	// re, _ := queue.Pop()
 
-	pstreamer := re.(*TSreamer)
+	// pstreamer := re.(*TSreamer)
-	m := make(map[string]interface{})
+	// m := make(map[string]interface{})
-	json.Unmarshal(pstreamer.Iface.([]byte), &m)
+	// json.Unmarshal(pstreamer.Iface.([]byte), &m)
-	spew.Println(re.(*TSreamer), m)
+	// spew.Println(re.(*TSreamer), m)
 
 	streamer := &TSreamer{}
 
-	streamer.Uid = 2
+	streamer.Uid = 1
-	streamer.UserID = &sql.NullString{String: "hehe", Valid: true}
+	streamer.UserID = &sql.NullString{String: "xixi", Valid: true}
 	streamer.Name = "streamer"
 	streamer.Operator = 0
 	streamer.Bit = 0b11
@@ -41,7 +39,7 @@ func estAutoStore(t *testing.T) {
 	now := time.Now()
 	streamer.UpdateTime = &now
 
-	err = store.Table("streamer").Insert(streamer)
+	err = store.Table("streamer").InsertOrUpdate(streamer, DUpdate{Field: "userid"})
 	if err != nil {
 		t.Error(err)
 	}

extractor/mirrativ_extractor/mirrativ_extractor.go

@@ -0,0 +1,163 @@
+package main
+
+import (
+	"database/sql"
+	"encoding/json"
+	"intimate"
+	"log"
+	"time"
+
+	"github.com/474420502/gcurl"
+	"github.com/tidwall/gjson"
+)
+
+func main() {
+
+	sessionstr := `
+	-H 'authority: www.mirrativ.com' 
+	-H 'accept: application/json' 
+	-H 'x-timezone: Asia/Shanghai' 
+	-H 'x-csrf-token: F3Ojd6RBtApP6YAZzVn-9jWN1of159VxAqOQL1Zn' 
+	-H 'user-agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.89 Safari/537.36' 
+	-H 'content-type: application/json' 
+	-H 'sec-fetch-site: same-origin' 
+	-H 'sec-fetch-mode: cors' 
+	-H 'sec-fetch-dest: empty' 
+	-H 'referer: https://www.mirrativ.com/live/O5Ia4iX9c5CeZj7DFtg52Q' 
+	-H 'accept-language: zh-CN,zh;q=0.9,ja;q=0.8' 
+	-H 'cookie: f=A2D75F0E-D218-11EA-A042-452BF6D21CE8; _ga=GA1.2.689947597.1596081392; mr_id=kxb65LddGMZf5C28jkR_tGCZD_ZFOAepD5gfXO7eNjfPMB8EKYvU1Vg_Y29V1lsa; _gid=GA1.2.2116692650.1600139685; lang=ja'`
+
+	ps := intimate.NewPerfectShutdown()
+	gprofile := gcurl.Parse(`curl 'https://www.mirrativ.com/api/user/profile?user_id=103383701'` + sessionstr)
+	tpProfile := gprofile.CreateTemporary(nil)
+	tpProfileUserID := tpProfile.QueryParam("user_id")
+
+	g := gcurl.Parse(`culr 'https://www.mirrativ.com/api/live/live?live_id=O5Ia4iX9c5CeZj7DFtg52Q'` + sessionstr)
+	tpLive := g.CreateTemporary(nil)
+	tpLiveID := tpLive.QueryParam("live_id")
+
+	var lasterr error
+	queue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.PMirrativ))
+	for !ps.IsClose() {
+		istreamer, err := queue.Pop()
+		if err != nil {
+			if lasterr != err {
+				lasterr = err
+				log.Println(err)
+			}
+			ps.Wait(time.Second * 5)
+			continue
+		}
+
+		now := &sql.NullTime{Time: time.Now(), Valid: true}
+		streamer := istreamer.(*intimate.Streamer)
+		streamer.UpdateTime = now
+		userid := *streamer.UserId
+		log.Println(userid)
+
+		tpProfileUserID.StringSet(userid)
+		resp, err := tpProfile.Execute()
+		if err != nil {
+			log.Println(err)
+			time.Sleep(time.Second)
+			continue
+		}
+
+		clog := &intimate.CollectLog{}
+		clog.Platform = intimate.PMirrativ
+		clog.UpdateTime = now
+		clog.UserId = userid
+		clog.StreamerUid = streamer.Uid
+
+		profilejson := gjson.ParseBytes(resp.Content())
+		if result := profilejson.Get("follower_num"); result.Exists() {
+			clog.Followers = &sql.NullInt64{Int64: result.Int(), Valid: true}
+		}
+
+		if result := profilejson.Get("onlive.live_id"); result.Exists() {
+			liveID := result.String()
+			tpLiveID.StringSet(liveID)
+			resp, err = tpLive.Execute()
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+
+			livejson := gjson.ParseBytes(resp.Content())
+			if result := livejson.Get("total_viewer_num"); result.Exists() {
+				clog.Views = &sql.NullInt64{Int64: result.Int(), Valid: true}
+			}
+
+			if result := livejson.Get("max_online_viewer_num"); result.Exists() {
+				clog.PCU = &sql.NullInt64{Int64: result.Int(), Valid: true}
+			}
+
+			if result := livejson.Get("title"); result.Exists() {
+				clog.LiveTitle = &sql.NullString{String: result.String(), Valid: true}
+			}
+
+			if result := livejson.Get("started_at"); result.Exists() {
+				clog.LiveStartTime = &sql.NullTime{Time: time.Unix(result.Int(), 0), Valid: true}
+			}
+
+			if result := livejson.Get("heartbeated_at"); result.Exists() {
+				clog.LiveEndTime = &sql.NullTime{Time: time.Unix(result.Int(), 0), Valid: true}
+			}
+
+			if result := livejson.Get("app_id"); result.Exists() {
+				streamer.Channel = &sql.NullString{String: result.String(), Valid: true}
+			}
+
+			if result := livejson.Get("timeline.#.app.short_title"); result.Exists() {
+				for _, tl := range result.Array() {
+					var tags []string = []string{tl.String()}
+					jtags, _ := json.Marshal(tags)
+					streamer.Tags = jtags
+					clog.Tags = jtags
+					break
+				}
+			} else {
+				log.Println(string(resp.Content()))
+				return
+			}
+
+			if result := livejson.Get("gift_ranking_url"); result.Exists() {
+				// streamer.Channel = &sql.NullString{String: result.String(), Valid: true}
+				gifturl := "curl '" + result.String() + "&type=monthly&cursor='" + sessionstr
+				ggift := gcurl.Parse(gifturl)
+				tp := ggift.CreateTemporary(nil)
+				tp.SetURLRawPath("/api/gift/ranking")
+				pcursor := tp.QueryParam("cursor")
+				var gratuity int64 = 0
+
+				for {
+					giftdata, err := tp.Execute()
+					giftjson := gjson.ParseBytes(giftdata.Content())
+					if err != nil {
+						log.Println(err)
+					} else {
+						for _, rpoint := range giftjson.Get("ranking.#.point").Array() {
+							gratuity += rpoint.Int()
+						}
+					}
+					ncursor := giftjson.Get("next_cursor").String()
+					if ncursor == "" {
+						break
+					}
+					pcursor.StringSet(ncursor)
+				}
+				// https://www.mirrativ.com/gift/ranking?live_id=O5Ia4iX9c5CeZj7DFtg52Q&obfuscated_user_id=PgIBEgc6jVc
+				clog.Gratuity = &sql.NullInt64{Int64: gratuity, Valid: true}
+			}
+
+			cid, err := intimate.TClog.InsertRetAutoID(clog)
+			if err != nil {
+				log.Println(err)
+			}
+			streamer.LatestLogUid = cid
+		}
+
+		intimate.TStreamer.Update(streamer)
+		time.Sleep(time.Second * 2)
+	}
+}

extractor/mirrativ_extractor/mirrativ_extractor_test.go

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

extractor/nimo_extractor/nimo_extractor.go

@@ -11,10 +11,10 @@ import (
 )
 
 // sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STNimo))
+// var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STNimo))
 
-// estore 解析存储连接实例
+// // estore 解析存储连接实例
-var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+// var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
 
 func main() {
 	Execute()
@@ -28,21 +28,25 @@ type LiveInfo struct {
 }
 
 func Execute() {
-	wd := intimate.GetChromeDriver(3030)
+	adriver := intimate.GetChromeDriver()
 	count := 0
 	countlimit := 200
-
+	wd := adriver.Webdriver
 	waitfor := intimate.NewWaitFor(wd)
 	ps := intimate.NewPerfectShutdown()
 
+	queue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.PNimo))
+
 	for !ps.IsClose() {
-		streamer, err := estore.Pop(intimate.PNimo)
+		istreamer, err := queue.Pop()
 		if err != nil {
 			log.Println(err)
-			estore.UpdateError(streamer, err)
+			intimate.TStreamer.UpdateError(istreamer, err)
 			continue
 		}
 
+		streamer := istreamer.(*intimate.Streamer)
+
 		wd.Get(streamer.LiveUrl.String)
 		// wd.Get("https://www.nimo.tv/live/1253835677")
 
@@ -71,22 +75,25 @@ func Execute() {
 
 		clog := &intimate.CollectLog{}
 		clog.Platform = intimate.PNimo
-		clog.Followers = sql.NullInt64{Int64: li.Followers, Valid: true}
+		clog.Followers = &sql.NullInt64{Int64: li.Followers, Valid: true}
-		clog.Views = sql.NullInt64{Int64: li.Views, Valid: true}
+		clog.Views = &sql.NullInt64{Int64: li.Views, Valid: true}
-		clog.UpdateTime = utime
+		clog.UpdateTime = &utime
 		clog.StreamerUid = streamer.Uid
 
 		var sum int64 = 0
 		for _, v := range li.Gratuity {
 			sum += v
 		}
-		clog.Gratuity = sql.NullInt64{Int64: sum, Valid: true}
+		clog.Gratuity = &sql.NullInt64{Int64: sum, Valid: true}
 
-		cuid := estore.InsertClog(clog)
+		cuid, err := intimate.TClog.InsertRetAutoID(clog)
+		if err != nil {
+			panic(err)
+		}
 
-		streamer.Channel = sql.NullString{String: li.Channel, Valid: true}
+		streamer.Channel = &sql.NullString{String: li.Channel, Valid: true}
 		streamer.LatestLogUid = cuid
-		streamer.UpdateTime = utime
+		streamer.UpdateTime = &utime
 		streamer.Operator = 0
 
 		switch {
@@ -102,14 +109,17 @@ func Execute() {
 			streamer.UpdateInterval = 60
 		}
 
-		estore.Update(streamer, "update_interval", streamer.UpdateInterval, "operator", streamer.Operator, "channel", streamer.Channel, "latest_log_uid", streamer.LatestLogUid, "update_time", streamer.UpdateTime)
+		// estore.Update(streamer, "update_interval", streamer.UpdateInterval, "operator", streamer.Operator, "channel", streamer.Channel, "latest_log_uid", streamer.LatestLogUid, "update_time", streamer.UpdateTime)
+		err = intimate.TStreamer.Update(streamer)
+		if err != nil {
+			panic(err)
+		}
 
 		count++
 		if count >= countlimit {
 			count = 0
-			wd.Close()
+			adriver.Close()
-			wd.Quit()
+			adriver = intimate.GetChromeDriver()
-			wd = intimate.GetChromeDriver(3030)
 		}
 	}
 }

extractor/openrec_extractor/openrec_extractor.go

@@ -18,7 +18,7 @@ import (
 type UserInfo struct {
 	UserName  string `exp:"//p[ contains(@class, 'c-global__user__profile__list__name__text')]"`
 	Followers int64  `exp:"//p[@class='c-global__user__count__row__right js-userCountFollowers']" mth:"r:ParseNumber"`
-	Views     int64  `exp:"//ul[@class='c-contents']//p[@class='c-thumbnailVideo__footer__liveCount']" mth:"r:ExtractNumber"`
+	PCU       int64  `exp:"//ul[@class='c-contents']//p[@class='c-thumbnailVideo__footer__liveCount']" mth:"r:ExtractNumber"`
 }
 
 //UserLive 提取信息的结构体
@@ -27,6 +27,7 @@ type UserLive struct {
 	LiveStartTime string   `exp:"//meta[@itemprop='uploadDate']/@content"`
 	LiveEndTime   string   `exp:"//meta[@itemprop='duration']/@content"`
 	Tags          []string `exp:"//div[contains(@class,'MovieMetaContent__TagContainer')]//a[@role ='button']"`
+	Views         int64    `exp:"//meta[@itemprop='interactionCount']/@content"`
 }
 
 // Execute 执行
@@ -173,10 +174,7 @@ func Extractor(streamer *intimate.Streamer, userId string, htmlUser, htmlLive st
 
 	if ok1 {
 		clog.Followers = &sql.NullInt64{Int64: ui.Followers, Valid: true}
-		clog.Views = &sql.NullInt64{Int64: ui.Views, Valid: true}
+		clog.PCU = &sql.NullInt64{Int64: ui.PCU, Valid: true}
-		if ui.Views != 0 {
-			clog.IsLiveStreaming = true
-		}
 		streamer.UserName = &sql.NullString{String: ui.UserName, Valid: true}
 
 		// giverjson := jsonSupporters
@@ -208,6 +206,7 @@ func Extractor(streamer *intimate.Streamer, userId string, htmlUser, htmlLive st
 
 	//log.Println(ul)
 	if ok2 {
+		clog.Views = &sql.NullInt64{Int64: ul.Views, Valid: true}
 		clog.LiveTitle = &sql.NullString{String: ul.Title, Valid: true}
 
 		startTime, err := time.ParseInLocation("2006-01-02T15:04:05Z07:00", ul.LiveStartTime, time.Local)

extractor/twitcasting_extractor/twitcasting_extractor.go

@@ -13,12 +13,6 @@ import (
 	"github.com/474420502/requests"
 )
 
-// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec))
-
-// estore 解析存储连接实例
-var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
-
 type LiveData struct {
 	UserName     string   `exp:"//span[@class='tw-live-author__info-username']" method:"Text"`
 	Follower     int64    `exp:"(//span[@class='tw-user-nav-list-count'])[2]" method:"r:ExtractNumber"`
@@ -34,13 +28,17 @@ func main() {
 	ps := intimate.NewPerfectShutdown()
 	ses := requests.NewSession()
 	streamerQueue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.Ptwitcasting))
-
+	var lasterr error
 	for !ps.IsClose() {
 
 		// streamer, err := estore.Pop(intimate.Ptwitcasting)
 		isteamer, err := streamerQueue.Pop()
 		if err != nil {
-			log.Println(err, isteamer)
+			if lasterr != err {
+				lasterr = err
+				log.Println(err)
+			}
+			time.Sleep(time.Minute)
 			continue
 		}
 

extractor/twitch_extractor/tiwtch_extractor.go

@@ -3,134 +3,326 @@ package main
 import (
 	"database/sql"
 	"encoding/json"
+	"fmt"
 	"intimate"
 	"log"
 	"regexp"
-	"strings"
 	"time"
 
 	"github.com/tebeka/selenium"
 )
 
-// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
+// // sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitch))
+// var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitch))
 
-// estore 解析存储连接实例
+// // estore 解析存储连接实例
-var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+// var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
 
 func main() {
-	wd := intimate.GetChromeDriver(3030)
+
+	adriver := intimate.GetChromeDriver()
+
 	ps := intimate.NewPerfectShutdown()
 
+	slqueue := intimate.TStreamerList.Queue(intimate.StreamerList{}, intimate.ConditionDefault(intimate.Ptwitch))
+	squeue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.Ptwitch))
+
 	var count = 0
 	var countlimt = 200
+	var recreate = time.Now()
 
 	var lasterr error = nil
 	// var err error
-
 	for !ps.IsClose() {
-		streamer, err := estore.Pop(intimate.Ptwitch)
+
-		if streamer == nil || err != nil {
+		wd := adriver.Webdriver
-			if err != lasterr {
+		// sourceChannel, err := sstore.Pop(intimate.TTwitchChannel)
-				log.Println(err, lasterr)
+		isl, err := slqueue.Pop()
+		if err != nil {
+			if lasterr != err {
 				lasterr = err
+				log.Println(err)
+			}
+
+			istreamer, err := squeue.Pop()
+			if err != nil {
+				if lasterr != err {
+					lasterr = err
+					log.Println(err)
+					ps.Wait(time.Minute)
+					continue
+				}
+			}
+
+			streamer := istreamer.(*intimate.Streamer)
+			Extractor(wd, streamer)
+			if err = intimate.TStreamer.Update(streamer); err != nil {
+				log.Println(err)
+			}
+
+			count++
+			if count >= countlimt || time.Now().Sub(recreate) >= time.Minute*120 {
+				count = 0
+				adriver.Close()
+				adriver = intimate.GetChromeDriver()
+				recreate = time.Now()
 			}
-			time.Sleep(time.Second * 2)
 			continue
 		}
 
-		var updateUrl map[string]string
+		streamerlist := isl.(*intimate.StreamerList)
-		json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl)
-		liveUrl := updateUrl["live"]
-		liveUrl = strings.Replace(liveUrl, "/watchparty", "", -1)
-		log.Println(liveUrl)
 
-		// err = wd.Get("https://www.twitch.tv/zoe_0601" + "/about")
+		weburl := streamerlist.Url + "?sort=VIEWER_COUNT"
-		err = wd.Get(liveUrl + "/about")
+		err = wd.Get(weburl)
 		if err != nil {
 			log.Println(err)
-			estore.UpdateError(streamer, err)
+			// sstore.UpdateError(sourceChannel, err)
-			time.Sleep(time.Second * 5)
+			intimate.TStreamerList.UpdateError(streamerlist, err)
+			time.Sleep(time.Second * 10)
 			continue
 		}
 
-		streamer.LiveUrl = sql.NullString{String: liveUrl, Valid: true}
+		wd.WaitWithTimeout(func(wd selenium.WebDriver) (bool, error) {
-		clog := &intimate.CollectLog{}
+			_, err := wd.FindElement(selenium.ByXPATH, "(//div/p[@class=''])[last()]")
-		clog.UserId = streamer.UserId
+			if err != nil {
-		clog.Gratuity = sql.NullInt64{Int64: 0, Valid: false}
+				return false, err
-
-		time.Sleep(time.Millisecond * 500)
-		err = extractUserName(wd, streamer)
-		if err != nil {
-			_, err = wd.FindElement(selenium.ByXPATH, "//a[@data-a-target='browse-channels-button']")
-			if err == nil {
-				log.Println(streamer.UserId, "may be cancell")
-				streamer.Operator = 5
-				streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
-				estore.UpdateStreamer(streamer)
 			}
-			continue
+			return true, nil
-		}
+		}, time.Second*10)
-		err = extractFollowers(wd, clog)
-		if err != nil {
-			continue
-		}
 
-		err = extractViews(wd, clog) // views + tags + gratuity
+		btn, err := wd.FindElement(selenium.ByXPATH, "//button[@data-a-target='browse-sort-menu']")
 		if err != nil {
-			// 不直播时提取礼物 gratuity
+			log.Println(err)
-			wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
+			continue
-				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"]`)
+		btn.Click()
-				if (err == nil && channelchat != nil) || btn != nil {
+
-					if channelchat != nil {
+		var elements []selenium.WebElement
-						channelchat.Click()
+		var liveurls = 0
-					}
+		var delayerror = 2
-					time.Sleep(time.Second)
+		for i := 0; i < 200 && !ps.IsClose(); i++ {
-					extractGratuity(wd, clog)
+			elements, err = wd.FindElements(selenium.ByXPATH, "(//div/p[@class=''])[last()]")
-					return true, nil
+			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
 				}
-				return false, nil
+			} else {
-
+				delayerror = 2
-			}, time.Second*4)
+			}
+			liveurls = len(elements)
+		}
+		articles, err := wd.FindElements(selenium.ByXPATH, "//article")
+		if err != nil {
+			log.Println(err)
+			continue
 		}
 
-		streamer.Platform = intimate.Ptwitch
+		var streamers []*intimate.Streamer
-		clog.Platform = streamer.Platform
+		for _, article := range articles {
-		clog.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
-		lastClogId := estore.InsertClog(clog)
 
-		streamer.Operator = 10
+			e, err := article.FindElement(selenium.ByXPATH, ".//a[@data-a-target='preview-card-title-link' and @href]")
-		streamer.LatestLogUid = lastClogId
+			if err != nil {
-		if clog.Tags != nil {
+				log.Println(err)
-			streamer.Tags = clog.Tags
+				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 {
+				mc := matches[1]
+				streamer.UserId = &mc
+			} else {
+				log.Println(href)
+				continue
+			}
+
+			jtags, err := json.Marshal(tags)
+			if err != nil {
+				log.Println(err)
+			} else {
+				streamer.Tags = jtags
+			}
+
+			streamer.Platform = intimate.Ptwitch
+			streamer.LiveUrl = &sql.NullString{String: href, Valid: true}
+			streamer.Operator = 0
+
+			streamers = append(streamers, streamer)
+
+			// if estore.InsertStreamer(streamer) {
+			// 	// log.Println("streamer update tags", streamer.Uid, tags)
+			// 	if streamer.Tags != nil {
+			// 		estore.Update(streamer, "Tags", streamer.Tags)
+			// 	}
+			// }
 		}
 
-		switch fl := clog.Followers.Int64; {
+		for _, streamer := range streamers {
-		case fl > 100000:
+			Extractor(wd, streamer)
-			streamer.UpdateInterval = 120
+			streamer.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
-		case fl > 10000:
+			if err = intimate.TStreamer.InsertOrUpdate(streamer,
-			streamer.UpdateInterval = 240 * 2
+				intimate.DUpdate{Field: "tags"},
-		case fl > 1000:
+				intimate.DUpdate{Field: "update_time"},
-			streamer.UpdateInterval = 360 * 2
+				intimate.DUpdate{Field: "update_interval"},
-		case fl > 100:
+			); err != nil {
-			streamer.UpdateInterval = 720 * 2
+				log.Println(err)
-		case fl > 0:
+			}
-			streamer.UpdateInterval = 1440 * 4
+		}
+
+		log.Println("streamer find", len(articles))
+		if len(articles) == 0 {
+			intimate.TStreamerList.UpdateError(streamerlist, fmt.Errorf(""))
 		}
 
-		streamer.UpdateTime = clog.UpdateTime
-		estore.UpdateStreamer(streamer)
 		count++
-		if count >= countlimt {
+		if count >= countlimt || time.Now().Sub(recreate) >= time.Minute*120 {
 			count = 0
-			// wd.Quit()
+			adriver.Close()
-			wd = intimate.GetChromeDriver(3030)
+			adriver = intimate.GetChromeDriver()
+			recreate = time.Now()
 		}
+
 	}
 
-	wd.Close()
+	adriver.Close()
-	wd.Quit()
+}
+
+func Extractor(wd selenium.WebDriver, streamer *intimate.Streamer) {
+	// streamer, err := estore.Pop(intimate.Ptwitch)
+	// 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 := "https://www.twitch.tv/" + (*streamer.UserId)
+	// liveUrl = strings.Replace(liveUrl, "/watchparty", "", -1)
+	log.Println(liveUrl)
+
+	// err = wd.Get("https://www.twitch.tv/zoe_0601" + "/about")
+	err := wd.Get(liveUrl + "/about")
+	if err != nil {
+		errstr := fmt.Errorf("%s: %s", err.Error(), liveUrl+"/about")
+		log.Println(errstr)
+		intimate.TStreamer.UpdateError(streamer, errstr)
+		time.Sleep(time.Second * 5)
+		return
+	}
+
+	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 {
+		_, err = wd.FindElement(selenium.ByXPATH, "//a[@data-a-target='browse-channels-button']")
+		if err == nil {
+			log.Println(*streamer.UserId, "may be cancell")
+			streamer.Operator = 5
+			intimate.TStreamer.UpdateError(streamer, fmt.Errorf(*streamer.UserId, "may be cancell"))
+		}
+		return
+	}
+	err = extractFollowers(wd, clog)
+	if err != nil {
+		// log.Println(err)
+		streamer.UpdateInterval += 30
+
+		return
+	}
+
+	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 = streamer.Platform
+	clog.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
+	// clog.StreamerUid = streamer.Uid
+	lastClogId, err := intimate.TClog.InsertRetAutoID(clog)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	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 * 2
+	case fl > 1000:
+		streamer.UpdateInterval = 360 * 2
+	case fl > 100:
+		streamer.UpdateInterval = 720 * 2
+	case fl > 0:
+		streamer.UpdateInterval = 1440 * 4
+	}
+
+	streamer.UpdateTime = clog.UpdateTime
+	// intimate.TStreamer.InsertOrUpdate(streamer)
+	// count++
+	// if count >= countlimt {
+	// 	count = 0
+	// 	// wd.Quit()
+	// 	wd = intimate.GetChromeDriver(3030)
+	// }
 }
 
 func extractUserName(wd selenium.WebDriver, streamer *intimate.Streamer) error {
@@ -139,7 +331,7 @@ func extractUserName(wd selenium.WebDriver, streamer *intimate.Streamer) error {
 		if err == nil {
 			if ltxt, err := label.Text(); err == nil && ltxt != "" {
 				// log.Println("label:", ltxt)
-				streamer.UserName = sql.NullString{String: ltxt, Valid: true}
+				streamer.UserName = &sql.NullString{String: ltxt, Valid: true}
 				return true, nil
 			}
 		}
@@ -159,7 +351,7 @@ func extractFollowers(wd selenium.WebDriver, clog *intimate.CollectLog) error {
 		}
 		followers = regexp.MustCompile(`[\d,]+`).FindString(followers)
 		fint, _ := intimate.ParseNumber(followers)
-		clog.Followers = sql.NullInt64{Int64: int64(fint), Valid: true}
+		clog.Followers = &sql.NullInt64{Int64: int64(fint), Valid: true}
 		// log.Println("followers: ", followers, fint)
 		return true, nil
 	}, 4*time.Second)
@@ -172,7 +364,8 @@ func extractViews(wd selenium.WebDriver, clog *intimate.CollectLog) error {
 			if txt, err := views.Text(); err == nil {
 
 				vint, _ := intimate.ParseNumber(txt)
-				clog.Views = sql.NullInt64{Int64: vint, Valid: true}
+				clog.Views = &sql.NullInt64{Int64: vint, Valid: true}
+				clog.PCU = clog.Views
 				// log.Println("views:", txt)
 				views.Click()
 
@@ -192,7 +385,7 @@ func extractTitle(wd selenium.WebDriver, clog *intimate.CollectLog) 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}
+				clog.LiveTitle = &sql.NullString{String: txt, Valid: true}
 				return true, nil
 			}
 		}
@@ -244,7 +437,7 @@ func extractGratuity(wd selenium.WebDriver, clog *intimate.CollectLog) error {
 						log.Println(err)
 					}
 				}
-				clog.Gratuity = sql.NullInt64{Int64: gratuity, Valid: true}
+				clog.Gratuity = &sql.NullInt64{Int64: gratuity, Valid: true}
 			}
 			return true, nil
 		}

extractor_field.go

@@ -9,9 +9,9 @@ type GetSet struct {
 }
 
 type StreamerList struct {
-	UrlHash  string `field:"urlhash" `  //
+	UrlHash  string `field:"urlhash" uid:"true"` //
-	Platform string `field:"platform" ` //
+	Platform string `field:"platform" `          //
-	Url      string `field:"url" `      //
+	Url      string `field:"url" `               //
 
 	Label *sql.NullString `field:"label" ` //
 
@@ -46,6 +46,7 @@ type Streamer struct {
 	Channel  *sql.NullString `field:"channel"`   //
 	Tags     interface{}     `field:"tags"`
 	Ext      interface{}     `field:"ext"` //
+	// Comments interface{}     `field:"comments"`
 
 	IsUpdateStreamer bool // 更新上面的内容
 	IsUpdateUrl      bool
@@ -74,21 +75,22 @@ type CollectLog struct {
 	LogUid      int64 `field:"log_uid"`      // 日志id
 	StreamerUid int64 `field:"streamer_uid"` // StreamerId 表id与
 
-	Platform        Platform        `field:"platform"`          //
+	Platform      Platform        `field:"platform"`        //
-	UserId          string          `field:"user_id"`           // 平台的UserId
+	UserId        string          `field:"user_id"`         // 平台的UserId
-	IsLiveStreaming bool            `field:"is_live_streaming"` //
+	PCU           *sql.NullInt64  `field:"pcu"`             //
-	IsError         bool            `field:"is_error"`          //
+	IsError       bool            `field:"is_error"`        //
-	Followers       *sql.NullInt64  `field:"followers"`         //
+	Followers     *sql.NullInt64  `field:"followers"`       //
-	Views           *sql.NullInt64  `field:"views"`             //
+	Views         *sql.NullInt64  `field:"views"`           //
-	Giver           interface{}     `field:"giver"`             //
+	Giver         interface{}     `field:"giver"`           //
-	Gratuity        *sql.NullInt64  `field:"gratuity"`          //
+	Gratuity      *sql.NullInt64  `field:"gratuity"`        //
-	LiveTitle       *sql.NullString `field:"live_title"`        //
+	LiveTitle     *sql.NullString `field:"live_title"`      //
-	LiveStartTime   *sql.NullTime   `field:"live_start_time"`   //
+	LiveStartTime *sql.NullTime   `field:"live_start_time"` //
-	LiveEndTime     *sql.NullTime   `field:"live_end_time"`     //
+	LiveEndTime   *sql.NullTime   `field:"live_end_time"`   //
-	UpdateTime      *sql.NullTime   `field:"update_time"`       //
+	UpdateTime    *sql.NullTime   `field:"update_time"`     //
-	Tags            interface{}     `field:"tags"`
+	Tags          interface{}     `field:"tags"`
-	Ext             interface{}     `field:"ext"`       //
+	Ext           interface{}     `field:"ext"`       //
-	ErrorMsg        *sql.NullString `field:"error_msg"` //
+	ErrorMsg      *sql.NullString `field:"error_msg"` //
+	Comments      interface{}     `field:"comments"`  //
 }
 
 // Get Simple Value

platform_list.go

@@ -15,4 +15,7 @@ const (
 
 	// PNimo PNimo 平台
 	PNimo Platform = "nimo"
+
+	// PMirrativ PNimo 平台
+	PMirrativ Platform = "mirrativ"
 )

store.go

@@ -1,12 +1,7 @@
 package intimate
 
 import (
-	"crypto/md5"
 	"database/sql"
-	"fmt"
-	"log"
-	"strings"
-	"time"
 
 	_ "github.com/go-sql-driver/mysql"
 )
@@ -47,441 +42,3 @@ type StoreSource struct {
 	errorCount int
 	errorLimit int
 }
-
-func (store *StoreSource) PopCount() int {
-	return store.popCount
-}
-
-func (store *StoreSource) Close() error {
-	return store.db.Close()
-}
-
-// NewSourceStore 创建一个存储实例
-func NewStoreSource(table string) *StoreSource {
-	db, err := sql.Open("mysql", InitConfig.Database.SourceURI)
-	if err != nil {
-		panic(err)
-	}
-	return &StoreSource{table: table, db: db}
-}
-
-func (store *StoreSource) errorAlarm(err error) {
-	if err != nil {
-		log.Println("store error: ", err)
-		// 报警. 如果数据插入有问题
-		store.errorCount++
-		if store.errorCount >= store.errorLimit {
-			// 数据库频繁操作初问题 报警, 减少没意义的请求
-		}
-	} else {
-		if store.errorCount > 0 {
-			store.errorCount--
-		}
-	}
-}
-
-// Insert 插入数据
-func (store *StoreSource) Insert(isource IGet) {
-	_, 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"))
-	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 *StoreSource) Update(isource IGet) {
-	_, 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"))
-	if err != nil {
-		panic(err)
-	}
-}
-
-// UpdateOperator 更新数据操作标志位
-func (store *StoreSource) UpdateOperator(isource IGet) {
-	_, err := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	if err != nil {
-		panic(err)
-	}
-}
-
-// UpdateError 更新错误数据
-func (store *StoreSource) UpdateError(isource IGetSet, err error) {
-	isource.Set("Operator", int32(OperatorError)+isource.Get("Operator").(int32))
-	isource.Set("ErrorMsg", sql.NullString{String: err.Error(), Valid: true})
-	_, dberr := store.db.Exec("update "+store.table+" set  operator = ?, error_msg = ? where uid = ?", isource.Get("Operator"), isource.Get("ErrorMsg"), isource.Get("Uid"))
-	if dberr != nil {
-		// email tell owner to deal with
-		panic(err)
-	}
-}
-
-// Restore 恢复Operator数据状态
-func (store *StoreSource) Restore(isource IGet) {
-	_, dberr := store.db.Exec("update "+store.table+" set  operator = ?  where uid = ?", isource.Get("LastOperator"), isource.Get("Uid"))
-	if dberr != nil {
-		// email tell owner to deal with
-		panic(dberr)
-	}
-}
-
-// Pop 弹出一条未处理的数据
-func (store *StoreSource) Pop(targetType Target, operators ...int32) (*Source, error) {
-
-	tx, err := store.db.Begin()
-	if err != nil {
-		return nil, err
-	}
-	var args = []interface{}{string(targetType)}
-	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)
-	} else {
-		for _, operator := range operators {
-			selectSQL += " and operator = ?"
-			args = append(args, operator)
-		}
-	}
-
-	// log.Println(selectSQL + ` limit 1 for update`)
-	row := tx.QueryRow(selectSQL+` limit 1 for update`, args...)
-
-	defer func() {
-		err := tx.Commit()
-		if err != nil {
-			log.Println(err)
-			err = tx.Rollback()
-			if err != nil {
-				log.Println(err)
-			}
-		}
-		store.popCount++
-	}()
-
-	s := &Source{}
-	// uid, url, target_type, source, ext, operator
-	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.Set("LastOperator", s.Operator)
-	_, err = tx.Exec("update "+store.table+" set operator = ? where uid = ?", OperatorWait, s.Uid)
-	return s, nil
-}
-
-// StreamerTable 主播表名称
-const StreamerTable string = "streamer"
-
-// CollectLogTable 采集日志表
-const CollectLogTable string = "collect_log"
-
-// StreamerListTable 主播表名称
-const StreamerListTable string = "streamer_list"
-
-type StoreExtractor struct {
-	db *sql.DB
-
-	popCount   int
-	errorCount int
-	errorLimit int
-}
-
-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.Println("store error: ", err)
-		// 报警. 如果数据插入有问题
-		store.errorCount++
-		if store.errorCount >= store.errorLimit {
-			// 数据库频繁操作初问题 报警, 减少没意义的请求
-		}
-	} else {
-		if store.errorCount > 0 {
-			store.errorCount--
-		}
-	}
-}
-
-// NewStoreExtractor 生成一个extractor库的相关链接
-func NewStoreExtractor() *StoreExtractor {
-	db, err := sql.Open("mysql", InitConfig.Database.ExtractorURI)
-	if err != nil {
-		panic(err)
-	}
-	return &StoreExtractor{db: db}
-}
-
-// PopNoWait 弹出一个不用按时间间隔更新的主播信息, 主要用来测试.
-func (store *StoreExtractor) PopNoWait(platform Platform, condition string, operators ...int32) (*Streamer, error) {
-	tx, err := store.db.Begin()
-	if err != nil {
-		return nil, err
-	}
-	var args = []interface{}{string(platform)}
-	selectSQL := `select uid, update_time, user_id, tags, live_url, update_url, is_update_streamer, update_interval from ` + StreamerTable + ` where platform = ? and ` + condition
-	if len(operators) == 0 {
-		selectSQL += " and operator = ?"
-		args = append(args, 0)
-	} else {
-		for _, operator := range operators {
-			selectSQL += " and operator = ?"
-			args = append(args, operator)
-		}
-	}
-
-	defer func() {
-		err := tx.Commit()
-		if err != nil {
-			log.Println(err)
-			err = tx.Rollback()
-			if err != nil {
-				log.Println(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.Tags, &s.LiveUrl, &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
-}
-
-// Pop 弹出一条未处理的数据
-func (store *StoreExtractor) Pop(platform Platform, operators ...int32) (*Streamer, error) {
-
-	tx, err := store.db.Begin()
-	if err != nil {
-		return nil, err
-	}
-	var args = []interface{}{string(platform)}
-	selectSQL := `select uid, update_time, user_id, tags, live_url, 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)
-		}
-	}
-
-	defer func() {
-		err := tx.Commit()
-		if err != nil {
-			log.Println(err)
-			err = tx.Rollback()
-			if err != nil {
-				log.Println(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.Tags, &s.LiveUrl, &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
-}
-
-// UpdateStreamerList streamerlist表, 更新数据
-func (store *StoreExtractor) UpdateStreamerList(streamer IGet, fieldvalues ...interface{}) {
-	updateSQL := "UPDATE " + StreamerListTable + " 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 urlhash = ?"
-	values = append(values, streamer.Get("UrlHash"))
-	_, err := store.db.Exec(updateSQL, values...)
-	if err != nil {
-		panic(err)
-	}
-}
-
-// InsertStreamer streamerlist表, 插入数据
-func (store *StoreExtractor) InsertStreamerList(streamerlist IGet) (isExists bool) {
-	urlstr := streamerlist.Get("Url").(string)
-
-	_, err := store.db.Exec("insert into streamer_list(urlhash, url, platform, label, serialize, update_interval, error_msg, operator) values(?,?,?,?,?,?,?,?)",
-		fmt.Sprintf("%x", md5.Sum([]byte(urlstr))),
-		urlstr,
-		streamerlist.Get("Platform"),
-		streamerlist.Get("Label"),
-		streamerlist.Get("Serialize"),
-		streamerlist.Get("UpdateInterval"),
-		streamerlist.Get("ErrorMsg"),
-		streamerlist.Get("Operator"),
-	)
-
-	if err != nil {
-		if !strings.HasPrefix(err.Error(), "Error 1062") {
-			log.Println(err)
-		}
-		return true
-	}
-
-	return false
-}
-
-// InsertStreamer Streamer表, 插入数据
-func (store *StoreExtractor) InsertStreamer(streamer *Streamer) (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)
-	}
-
-	defer func() {
-		err = tx.Commit()
-		if err != nil {
-			rerr := tx.Rollback()
-			if rerr != nil {
-				log.Println(rerr)
-			}
-			panic(err)
-		}
-	}()
-
-	streamer.UpdateTime = &sql.NullTime{Time: time.Now().Add(-time.Hour * 100000), Valid: true}
-	_, err = tx.Exec("INSERT IGNORE INTO "+StreamerTable+"(platform, user_id, user_name, live_url, update_url, tags, update_time) VALUES(?,?,?,?,?,?,?);",
-		streamer.Platform,
-		streamer.UserId,
-		streamer.UserName,
-		streamer.LiveUrl,
-		streamer.UpdateUrl,
-		streamer.Tags,
-		streamer.UpdateTime,
-	)
-
-	if err != nil {
-		panic(err)
-	}
-	return false
-}
-
-// UpdateError 更新错误数据
-func (store *StoreExtractor) UpdateError(isource IGetSet, err error) {
-	isource.Set("Operator", int32(OperatorError)+isource.Get("Operator").(int32))
-	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"))
-	if dberr != nil {
-		// 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) {
-	// log.Printf("UPDATE "+StreamerTable+" SET user_name = %v,  live_url = %v, channel = %v, latest_log_uid = %v, tags = %v, ext = %v, operator = %v, update_time = %v, update_interval = %v WHERE uid = %v", 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"))
-	_, 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 = updateSQL[0 : len(updateSQL)-1]
-	updateSQL += "WHERE uid = ?"
-	values = append(values, streamer.Get("Uid"))
-	_, err := store.db.Exec(updateSQL, values...)
-	if err != nil {
-		log.Println(updateSQL)
-		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
-}

supervisor_conf/clear_log.sh

@@ -1,6 +1,6 @@
 CURPATH=`pwd`
 BINPATH=$(dirname "$CURPATH")/bin
-find $BINPATH -type f -name 'log' -exec rm {} +
+find $BINPATH -type f -name 'log' -exec truncate -s 0 {} +
 
 
 

supervisor_conf/mirrativ_extractor.conf

@@ -0,0 +1,10 @@
+[supervisord]
+nodaemon=true
+
+[program:mirrativ_extractor]
+directory = MYPATH/bin/mirrativ_extractor/
+command= MYPATH/bin/mirrativ_extractor/mirrativ_extractor
+autorestart=true
+stderr_logfile=MYPATH/bin/mirrativ_extractor/log
+stderr_logfile_maxbytes=0
+stopsignal=QUIT

supervisor_conf/mirrativ_task1.conf

@@ -0,0 +1,13 @@
+[supervisord]
+nodaemon=false
+
+[program:mirrativ_task1]
+environment=DISPLAY=":99"
+directory = MYPATH/bin/mirrativ_task1/
+command= MYPATH/bin/mirrativ_task1/mirrativ_task1 
+# process_name=%(program_name)s_%(process_num)02d              ;多进程名称
+# numprocs=1                                                   ;启动多个进程
+autorestart=true
+stderr_logfile=MYPATH/bin/mirrativ_task1/log
+stderr_logfile_maxbytes=0
+stopsignal=QUIT

supervisor_conf/openrec_task2.conf

@@ -1,12 +0,0 @@
-[supervisord]
-nodaemon=true
-
-[program:openrec_source]
-directory = MYPATH/bin/openrec_task2/
-command= MYPATH/bin/openrec_task2/openrec_task2
-process_name=%(program_name)s_%(process_num)02d              ;多进程名称
-numprocs=4                                                   ;启动多个进程
-autorestart=true
-stderr_logfile=MYPATH/bin/openrec_task2/log
-stderr_logfile_maxbytes=0
-stopsignal=QUIT

supervisor_conf/twitch_extractor.conf

@@ -1,5 +1,5 @@
 [supervisord]
-nodaemon=false
+nodaemon=true
 
 [program:twitch_extractor]
 environment=DISPLAY=":99"

supervisor_conf/twitch_extractor_p2.conf

@@ -1,13 +0,0 @@
-[supervisord]
-nodaemon=false
-
-[program:twitch_extractor_p2]
-environment=DISPLAY=":99",pac_proxy=http://localhost:1090/pac1
-directory = MYPATH/bin/twitch_extractor
-command= MYPATH/bin/twitch_extractor/twitch_extractor 
-process_name=%(program_name)s_%(process_num)02d              ;多进程名称
-numprocs=2                                                  ;启动多个进程
-autorestart=true
-stderr_logfile=MYPATH/bin/twitch_extractor/log
-stderr_logfile_maxbytes=0
-stopsignal=QUIT

supervisor_conf/twitch_task1.conf

@@ -0,0 +1,13 @@
+[supervisord]
+nodaemon=false
+
+[program:twitch_task1]
+environment=DISPLAY=":99"
+directory = MYPATH/bin/twitch_task1
+command= MYPATH/bin/twitch_task1/twitch_task1 
+# process_name=%(program_name)s_%(process_num)02d              ;多进程名称
+# numprocs=1                                                   ;启动多个进程
+autorestart=true
+stderr_logfile=MYPATH/bin/twitch_task1/log
+stderr_logfile_maxbytes=0
+stopsignal=QUIT

supervisor_conf/twitch_task2.conf

@@ -1,13 +0,0 @@
-[supervisord]
-nodaemon=false
-
-[program:twitch_task2]
-environment=DISPLAY=":99"
-directory = MYPATH/bin/twitch_task2
-command= MYPATH/bin/twitch_task2/twitch_task2 
-process_name=%(program_name)s_%(process_num)02d              ;多进程名称
-numprocs=6                                                   ;启动多个进程
-autorestart=true
-stderr_logfile=MYPATH/bin/twitch_task2/log
-stderr_logfile_maxbytes=0
-stopsignal=QUIT

tasks/mirrativ/mirrativ_task1/mirrativ_task1.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"database/sql"
+	"intimate"
+	"log"
+	"time"
+
+	"github.com/474420502/gcurl"
+	"github.com/tidwall/gjson"
+)
+
+func main() {
+	bcurl := `curl 'https://www.mirrativ.com/api/live/catalog?id=2&cursor=%s' \
+	-H 'authority: www.mirrativ.com' \
+	-H 'accept: application/json' \
+	-H 'x-timezone: Asia/Shanghai' \
+	-H 'x-csrf-token: F3Ojd6RBtApP6YAZzVn-9jWN1of159VxAqOQL1Zn' \
+	-H 'user-agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.89 Safari/537.36' \
+	-H 'content-type: application/json' \
+	-H 'sec-fetch-site: same-origin' \
+	-H 'sec-fetch-mode: cors' \
+	-H 'sec-fetch-dest: empty' \
+	-H 'referer: https://www.mirrativ.com/' \
+	-H 'accept-language: zh-CN,zh;q=0.9,ja;q=0.8' \
+	-H 'cookie: f=A2D75F0E-D218-11EA-A042-452BF6D21CE8; _ga=GA1.2.689947597.1596081392; mr_id=kxb65LddGMZf5C28jkR_tGCZD_ZFOAepD5gfXO7eNjfPMB8EKYvU1Vg_Y29V1lsa; _gid=GA1.2.2116692650.1600139685; lang=ja' \
+	--compressed`
+
+	curl := gcurl.Parse(bcurl)
+	tp := curl.CreateTemporary(nil)
+	cursor := tp.QueryParam(`cursor`)
+	cursor.StringSet("")
+
+	ps := intimate.NewPerfectShutdown()
+	for !ps.IsClose() {
+
+		log.Println(tp.ParsedURL.String())
+		resp, err := tp.Execute()
+		if err != nil {
+			log.Println(err)
+			time.Sleep(time.Second * 2)
+			continue
+		}
+
+		apijson := gjson.ParseBytes(resp.Content())
+		next := apijson.Get("next_cursor").String()
+
+		for _, liveinfo := range apijson.Get("list").Array() {
+			var prekey string
+			if liveinfo.Get("live_preview").Exists() {
+				prekey = "live_preview"
+			} else {
+				prekey = "live"
+			}
+			owner := liveinfo.Get(prekey + ".owner")
+			if guserid := owner.Get("user_id"); guserid.String() != "" {
+				streamer := &intimate.Streamer{}
+				streamer.Platform = intimate.PMirrativ
+				streamer.Operator = 0
+				streamer.UserId = &guserid.Str
+				streamer.UserName = &sql.NullString{String: owner.Get("name").String(), Valid: true}
+				streamer.UpdateInterval = 600
+				streamer.UpdateTime = intimate.GetUpdateTimeNow()
+
+				err = intimate.TStreamer.InsertOrUpdate(
+					streamer,
+					intimate.DUpdate{Field: "update_time"},
+				)
+
+				if err != nil {
+					log.Println(err)
+					panic(err)
+				}
+			}
+		}
+		if next == "" {
+			ps.Wait(time.Minute * 10)
+		} else {
+			ps.Wait(time.Second * 2)
+		}
+		cursor.StringSet(next)
+	}
+}

tasks/mirrativ/mirrativ_task1/mirrativ_task1_test.go

@@ -1,6 +1,8 @@
 package main
 
-import "testing"
+import (
+	"testing"
+)
 
 func TestMain(t *testing.T) {
 	main()

tasks/nimo/nimo_task1/nimo_task1.go

@@ -12,7 +12,7 @@ import (
 )
 
 // estore 解析存储连接实例
-var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
+// var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
 
 // Execute 执行
 func Execute() {
@@ -95,7 +95,10 @@ func Execute() {
 				}
 
 				streamer.UpdateInterval = 120
-				estore.InsertStreamer(streamer)
+				err = intimate.TStreamer.Insert(streamer)
+				if err != nil {
+					panic(err)
+				}
 
 			} else {
 				log.Println("userid is null.", room.String())

tasks/twitcasting/twitcasting_task1/twitcasting.go

@@ -14,12 +14,6 @@ import (
 	"github.com/474420502/requests"
 )
 
-// sstore  源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
-var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STTwitcasting))
-
-// estore 解析存储连接实例
-var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
-
 type SearchProfile struct {
 	UserName string   `exp:".//span[@class='username']" method:"Text"`
 	UserId   string   // `exp:".//span[@class='fullname']" method:"Text"`
@@ -70,11 +64,11 @@ func Execute() {
 			if ok := queuedict[wurl]; !ok {
 				log.Println(wurl)
 				sl := &intimate.StreamerList{}
-				sl.Platform = intimate.Ptwitcasting
+				sl.Platform = string(intimate.Ptwitcasting)
 				sl.Url = wurl
 				sl.Operator = 0
 				sl.UpdateInterval = 120
-				sl.UpdateTime = time.Now()
+				sl.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
 				sl.UrlHash = intimate.GetUrlHash(sl.Url)
 
 				intimate.TStreamerList.Insert(sl)
@@ -104,11 +98,11 @@ func Execute() {
 				sp.TagUrl[i] = wurl
 				if ok := queuedict[wurl]; !ok {
 					sl := &intimate.StreamerList{}
-					sl.Platform = intimate.Ptwitcasting
+					sl.Platform = string(intimate.Ptwitcasting)
 					sl.Url = wurl
 					sl.Operator = 0
 					sl.UpdateInterval = 120
-					sl.UpdateTime = time.Now()
+					sl.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
 					sl.UrlHash = intimate.GetUrlHash(sl.Url)
 					intimate.TStreamerList.Insert(sl)
 

tasks/twitch/twitch_task1/task_twitch.go

@@ -20,10 +20,13 @@ import (
 func Execute() {
 
 	ps := intimate.NewPerfectShutdown()
+	var adriver *intimate.AutoCloseDriver
 
 	for !ps.IsClose() {
+
 		var err error
-		wd := intimate.GetChromeDriver(3030)
+		adriver = intimate.GetChromeDriver()
+		wd := adriver.Webdriver
 
 		weburl := "https://www.twitch.tv/directory?sort=VIEWER_COUNT"
 		err = wd.Get(weburl)
@@ -107,21 +110,11 @@ func Execute() {
 			if err != nil {
 				log.Println(err)
 			}
-
-			// 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")
+		adriver.Close()
-
+		ps.Wait(time.Minute * 5)
-		wd.Close()
-		wd.Quit()
-		time.Sleep(time.Minute * 30)
 	}
+
 }

tasks/twitch/twitch_task2/.gitignore

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

tasks/twitch/twitch_task2/main.go

@@ -1,6 +0,0 @@
-package main
-
-func main() {
-
-	Execute()
-}

tasks/twitch/twitch_task2/task_twitch.go

@@ -1,175 +0,0 @@
-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()
-
-// 获取类型的所有频道链接
-
-// Execute 执行任务
-func 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)
-	defer wd.Quit()
-	ps := intimate.NewPerfectShutdown()
-	counter := intimate.NewCounter()
-	counter.SetMaxLimit(100)
-	counter.SetMaxToDo(func(olist ...interface{}) error {
-		owd := olist[0].(*selenium.WebDriver)
-		if err := (*owd).Quit(); err != nil {
-			log.Println(err)
-		}
-		*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 {
-				mc := matches[1]
-				streamer.UserId = &mc
-			} 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)
-				if streamer.Tags != nil {
-					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)
-	}
-}

utils.go

@@ -5,9 +5,10 @@ import (
 	"database/sql"
 	"fmt"
 	"log"
+	"net"
 	"os"
+	"os/exec"
 	"os/signal"
-	"runtime"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -93,13 +94,43 @@ func ParseDuration(dt string) (time.Duration, error) {
 	return tdt.Sub(zeroTime), nil
 }
 
-func GetChromeDriver(port int) selenium.WebDriver {
+type AutoCloseDriver struct {
+	Webdriver selenium.WebDriver
+	Port      int
+}
+
+func (adriver *AutoCloseDriver) Close() {
+
+	data, err := exec.Command("/bin/bash", "-c", fmt.Sprintf(`pgrep -f "port=%d"`, adriver.Port)).Output()
+	if err != nil {
+		log.Println(err)
+		log.Println(string(data))
+		return
+	}
+	// log.Println(string(data))
+
+	killshell := fmt.Sprintf("pkill -9 -P %s", data)
+	err = exec.Command("/bin/bash", "-c", killshell).Run()
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	err = exec.Command("/bin/bash", "-c", fmt.Sprintf("kill -9 %s", data)).Run()
+	if err != nil {
+		log.Println(err)
+		return
+	}
+}
+
+func GetChromeDriver() *AutoCloseDriver {
+
+	port := GetFreePort()
+
 	var err error
 	caps := selenium.Capabilities{"browserName": "chrome"}
 
 	chromecaps := chrome.Capabilities{}
-
-	// chromecaps.AddExtension("/home/eson/test/myblock.crx")
 	for _, epath := range []string{"../../../crx/myblock.crx", "../../crx/myblock.crx"} {
 		_, err := os.Stat(epath)
 		if err == nil {
@@ -131,28 +162,27 @@ func GetChromeDriver(port int) selenium.WebDriver {
 
 	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 {
+	adriver := &AutoCloseDriver{}
-			log.Println(err)
+	adriver.Port = port
-		}
+	adriver.Webdriver = wd
-		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
+	return adriver
 }
 
 // PerfectShutdown 完美关闭程序
@@ -180,6 +210,18 @@ func (ps *PerfectShutdown) IsClose() bool {
 	return atomic.LoadInt32(&ps.loop) == 0
 }
 
+// Wait 判断是否要关闭
+func (ps *PerfectShutdown) Wait(tm time.Duration) bool {
+	now := time.Now()
+	for time.Now().Sub(now) <= tm {
+		if ps.IsClose() {
+			return false
+		}
+		time.Sleep(time.Second)
+	}
+	return true
+}
+
 type Counter struct {
 	dcount   int
 	count    int
@@ -305,3 +347,17 @@ func (wf *WaitFor) WaitWithTimeout(xpath string, timeout time.Duration, do func(
 
 	}, timeout)
 }
+
+func GetFreePort() int {
+	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
+	if err != nil {
+		panic(err)
+	}
+
+	l, err := net.ListenTCP("tcp", addr)
+	if err != nil {
+		panic(err)
+	}
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port
+}