fix: 时区问题. sqluri导致解析错误.

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"

extractor/openrec_extractor/openrec_extractor.go

@@ -79,7 +79,7 @@ func (oe *OpenrecExtractor) Execute() {
 			// log.Println(anchorId)
 
 			oe.extractFollowers(clog)
-			oe.extractAnchorName(streamer)
+			oe.extractUserName(streamer)
 			oe.extractViewsAndLiveStreaming(clog)
 			oe.extractGiversAndGratuity(clog)
 			oe.extractLive(clog)
@@ -138,7 +138,7 @@ 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(ai intimate.ISet) {
 	extractor := oe.user.GetExtractor()
 	xp, err := extractor.XPathResult("//p[@class='c-global__user__profile__list__name__text official-icon--after']/text()")
 	if err != nil {
@@ -214,6 +214,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")

store.go

@@ -210,19 +210,33 @@ func (store *ExtractorStore) UpdateStreamerLogUid(logUid, streamerUid int64) err
 
 // InsertStreamer Streamer表, 插入数据
 func (store *ExtractorStore) InsertStreamer(isource IGet) (Uid int64, err error) {
-	// select uid from table where platform = ? and anchor_id = ?
+	// select uid from table where platform = ? and user_id = ?
-	selectSQL := "select uid from " + StreamerTable + " where platform = ? and anchor_id = ?"
+	selectSQL := "select uid from " + StreamerTable + " where platform = ? and user_id = ?"
 	tx, err := store.db.Begin()
 	if err != nil {
 		log.Println(err)
 		return 0, err
 	}
 
+	defer func() {
+		err = tx.Commit()
+		if err != nil {
+			log.Println(err)
+			err = tx.Rollback()
+			if err != nil {
+				log.Println(err)
+			}
+			Uid = 0
+		}
+	}()
+
 	row := tx.QueryRow(selectSQL+` limit 1 for update`, isource.Get("Platform"), isource.Get("UserId"))
 
 	var uid int64
 	if err = row.Scan(&uid); err == nil {
 		return uid, nil
+	} else {
+		log.Println(err)
 	}
 
 	result, err := tx.Exec("insert  into "+StreamerTable+"(platform, user_id, user_name,  live_url, channel, latest_log_uid, ext) values(?,?,?,?,?,?,?);", isource.Get("Platform"), isource.Get("UserId"), isource.Get("UserName"), isource.Get("LiveUrl"), isource.Get("Channel"), isource.Get("LatestLogUid"), isource.Get("Ext"))
@@ -232,16 +246,6 @@ func (store *ExtractorStore) InsertStreamer(isource IGet) (Uid int64, err error)
 		return 0, nil
 	}
 
-	err = tx.Commit()
-	if err != nil {
-		log.Println(err)
-		err = tx.Rollback()
-		if err != nil {
-			log.Println(err)
-		}
-		return 0, err
-	}
-
 	return result.LastInsertId()
 }
 

tasks/openrec/openrec_task1/task_openrec.go

@@ -60,7 +60,8 @@ func (or *OpenrecRanking) Execute(cxt *hunter.TaskContext) {
 		resp, err := cxt.Hunt()
 		if err != nil {
 			log.Println(err)
-			break
+			time.Sleep(time.Second * 2)
+			continue
 		}
 
 		tp := cxt.Temporary()
@@ -69,7 +70,7 @@ func (or *OpenrecRanking) Execute(cxt *hunter.TaskContext) {
 		if len(content) <= 200 { // 末页退出
 			finishpoint := time.Now()
 			log.Println("任务结束休眠, 下次启动时间:", finishpoint.Add(time.Minute*30))
-			for time.Now().Sub(finishpoint) < time.Minute*30 {
+			for time.Now().Sub(finishpoint) < time.Minute*60 {
 				time.Sleep(time.Second)
 				if atomic.LoadInt32(&loop) > 0 {
 					return