添加tags, 修改 存储结构 更加接近制直播的数据存储架构.

extractor/openrec/openrec_test.go

@@ -103,14 +103,6 @@ func TestExtractor(t *testing.T) {
 	ai.SetAnchorId(anchorId)
 	ai.SetPlatform(string(intimate.Popenrec))
 
-	collect.InsertAnchorInfo(ai)
-
-	// if source != nil {
-	// 	defer store.Restore(source)
-	// }
-	if err != nil {
-		t.Error(err)
-	}
 	sdata := source.GetExt().([]byte)
 
 	if gjson.ValidBytes(sdata) {
@@ -143,9 +135,12 @@ func TestExtractor(t *testing.T) {
 		} else {
 			t.Error(err)
 		}
+
 		t.Error(source.GetSource())
 		t.Error(anchorName)
 
+		ai.SetAnchorName(anchorName)
+
 		// c-contents
 		xp, err = extractor.XPathResult("//ul[@class='c-contents']//p[@class='c-thumbnailVideo__footer__liveCount']/text()")
 		if err != nil {
@@ -233,6 +228,16 @@ func TestExtractor(t *testing.T) {
 			}
 		}
 
+		LiveUrl := "https://www.openrec.tv/live/" + anchorId
+		ai.SetLiveUrl(sql.NullString{String: LiveUrl, Valid: true})
+
+		Uid, err := collect.InsertAnchorInfo(ai)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+
+		clog.SetUid(Uid)
 		clog.SetGratuity(sql.NullInt64{Int64: gratuity, Valid: true})
 		clog.SetPlatform(string(intimate.Popenrec))
 		clog.SetFollowers(sql.NullInt64{Int64: int64(followersInt), Valid: true})

sql/intimate_extractor.sql

@@ -8,7 +8,7 @@ CREATE TABLE IF NOT EXISTS `anchor_info` (
   `anchor_name`   varchar(255) NOT NULL,
   `live_url`      text,
   `channel`       varchar(128) DEFAULT NULL,
-  `show_type`     varchar(255) DEFAULT NULL,
+  `tags`          json DEFAULT NULL,
   `ext`           json DEFAULT NULL,
   `update_time`   timestamp   NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
   PRIMARY KEY (`uid`),

store.go

@@ -181,12 +181,50 @@ func NewExtractorStore() *ExtractorStore {
 */
 
 // InsertAnchorInfo AnchorInfo表, 插入数据
-func (store *ExtractorStore) InsertAnchorInfo(isource IGetAnchorInfo) error {
+func (store *ExtractorStore) InsertAnchorInfo(isource IGetAnchorInfo) (Uid int64, err error) {
-	_, err := store.db.Exec("insert  into "+AnchorTable+"(platform, anchor_id, anchor_name,  live_url, channel, show_type, ext) values(?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE", isource.GetPlatform(), isource.GetAnchorId(), isource.GetAnchorName(), isource.GetLiveUrl(), isource.GetChannel(), isource.GetShowType(), isource.GetExt())
+	// select uid from table where platform = ? and anchor_id = ?
-	store.errorAlarm(err)
+	selectSQL := "select uid from " + AnchorTable + " where platform = ? and anchor_id = ?"
-	return err
+	tx, err := store.db.Begin()
+	if err != nil {
+		log.Println(err)
+		return 0, err
+	}
+
+	row := tx.QueryRow(selectSQL+` limit 1 for update`, isource.GetPlatform(), isource.GetAnchorId())
+
+	if row != nil {
+		var uid int64
+		row.Scan(&uid)
+		return uid, nil
+	}
+
+	result, err := tx.Exec("insert  into "+AnchorTable+"(platform, anchor_id, anchor_name,  live_url, channel, show_type, ext) values(?,?,?,?,?,?,?);", isource.GetPlatform(), isource.GetAnchorId(), isource.GetAnchorName(), isource.GetLiveUrl(), isource.GetChannel(), isource.GetShowType(), isource.GetExt())
+	log.Println(result.LastInsertId())
+
+	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 result.LastInsertId()
 }
 
+// InsertAnchorInfo AnchorInfo表, 插入数据
+// func (store *ExtractorStore) InsertAnchorInfo(isource IGetAnchorInfo) error {
+// 	_, err := store.db.Exec("insert  into "+AnchorTable+"(platform, anchor_id, anchor_name,  live_url, channel, show_type, ext) values(?,?,?,?,?,?,?)", isource.GetPlatform(), isource.GetAnchorId(), isource.GetAnchorName(), isource.GetLiveUrl(), isource.GetChannel(), isource.GetShowType(), isource.GetExt())
+// 	store.errorAlarm(err)
+// 	return err
+// }
+
 /*
 	  `uid`      bigint,
   `platform` varchar(255) NOT NULL,