test/intimate
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

for save.

Browse Source
This commit is contained in:
eson 2020-07-10 18:31:17 +08:00
parent d5151f92bf
commit d258cc51d3
5 changed files with 29 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
extractor/openrec/main.go
View File

@ -6,7 +6,7 @@ package main
`anchor_id` varchar(255) NOT NULL, `anchor_id` varchar(255) NOT NULL,
`anchor_name` varchar(255) NOT NULL, `anchor_name` varchar(255) NOT NULL,
`live_url` text, `live_url` text,
`channel` varchar(128) DEFAULT NULL, `channel` varchar(128) DEFAULT NULL, // 没有分类
`show_type` varchar(255) DEFAULT NULL, `show_type` varchar(255) DEFAULT NULL,
*/ */

10
extractor/openrec/openrec_test.go
View File

@ -2,13 +2,13 @@ package main
import ( import (
"intimate" "intimate"
"os"
"testing" "testing"
"github.com/tidwall/gjson" "github.com/tidwall/gjson"
) )
func TestExtractor(t *testing.T) { func TestExtractor(t *testing.T) {
collect := intimate.NewExtractorStore()
store := intimate.NewSourceStore("source_openrec") store := intimate.NewSourceStore("source_openrec")
source, err := store.Pop("openrec_user", 100) source, err := store.Pop("openrec_user", 100)
if source != nil { if source != nil {
@ -24,11 +24,9 @@ func TestExtractor(t *testing.T) {
m := result.Map() m := result.Map()
for key := range m { for key := range m {
t.Error(key) t.Error(key)
f, err := os.OpenFile("./openrec_"+key+".html", os.O_CREATE|os.O_RDWR|os.O_TRUNC, os.ModePerm) ai := &intimate.CollectLog{}
if err != nil { ai.SetAnchorId("123")
panic(err) collect.InsertCollectLog(ai)
}
f.WriteString(m[key].String())
} }
} else { } else {
t.Error("data is not json:\n", string(sdata)) t.Error("data is not json:\n", string(sdata))

18
extractor_field.go
View File

@ -168,7 +168,7 @@ type IGetCollectLog interface {
GetShowEndTime() sql.NullTime // GetShowEndTime() sql.NullTime //
GetUpdateTime() time.Time // GetUpdateTime() time.Time //
GetExt() interface{} // GetExt() interface{} //
GetError() sql.NullString // GetErrorMsg() sql.NullString //
} }
type ISetCollectLog interface { type ISetCollectLog interface {
@ -186,7 +186,7 @@ type ISetCollectLog interface {
SetShowEndTime(sql.NullTime) // SetShowEndTime(sql.NullTime) //
SetUpdateTime(time.Time) // SetUpdateTime(time.Time) //
SetExt(interface{}) // SetExt(interface{}) //
SetError(sql.NullString) // SetErrorMsg(sql.NullString) //
} }
/* /*
@ -238,17 +238,17 @@ type CollectLog struct {
ShowEndTime sql.NullTime // ShowEndTime sql.NullTime //
UpdateTime time.Time // UpdateTime time.Time //
Ext interface{} // Ext interface{} //
Error sql.NullString // ErrorMsg sql.NullString //
} }
// GetError Get return Error sql.NullString // GetErrorMsg Get return Error sql.NullString
func (cl *CollectLog) GetError() sql.NullString { func (cl *CollectLog) GetErrorMsg() sql.NullString {
return cl.Error return cl.ErrorMsg
} }
// SetError Set Error sql.NullString // SetErrorMsg Set Error sql.NullString
func (cl *CollectLog) SetError(Error sql.NullString) { func (cl *CollectLog) SetErrorMsg(ErrorMsg sql.NullString) {
cl.Error = Error cl.ErrorMsg = ErrorMsg
} }
// GetExt Get return Ext interface{} // GetExt Get return Ext interface{}

2
sql/intimate_extractor.sql
View File

@ -40,7 +40,7 @@ CREATE TABLE IF NOT EXISTS `collect_log` (
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`ext` json DEFAULT NULL, `ext` json DEFAULT NULL,
`error` text DEFAULT NULL, `error_msg` text DEFAULT NULL,
KEY `uid_idx` (`uid`), KEY `uid_idx` (`uid`),
KEY `platform_idx` (`platform`), KEY `platform_idx` (`platform`),

18
store.go
View File

@ -141,7 +141,7 @@ type ExtractorStore struct {
func (store *ExtractorStore) errorAlarm(err error) { func (store *ExtractorStore) errorAlarm(err error) {
if err != nil { if err != nil {
log.Println("store error: ", err) log.Panic("store error: ", err)
// 报警. 如果数据插入有问题 // 报警. 如果数据插入有问题
store.errorCount++ store.errorCount++
if store.errorCount >= store.errorLimit { if store.errorCount >= store.errorLimit {
@ -154,6 +154,14 @@ func (store *ExtractorStore) errorAlarm(err error) {
} }
} }
func NewExtractorStore() *ExtractorStore {
db, err := sql.Open("mysql", InitConfig.Database.ExtractorURI)
if err != nil {
panic(err)
}
return &ExtractorStore{db: db}
}
/* /*
`uid` bigint, `uid` bigint,
`platform` varchar(255) NOT NULL, `platform` varchar(255) NOT NULL,
@ -189,11 +197,13 @@ func (store *ExtractorStore) InsertAnchorInfo(isource IGetAnchorInfo) {
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, --时间戳从源数据里获取 `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, --时间戳从源数据里获取
`ext` json DEFAULT NULL, `ext` json DEFAULT NULL,
`error` text DEFAULT NULL, `error_msg` text DEFAULT NULL,
*/ */
// InsertCollectLog CollectLog表插入数据 // InsertCollectLog CollectLog表插入数据
func (store *ExtractorStore) InsertCollectLog(isource IGetAnchorInfo) { func (store *ExtractorStore) InsertCollectLog(isource IGetCollectLog) {
_, err := store.db.Exec("insert into "+CollectLogTable+"(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()) _, err := store.db.Exec("insert into "+CollectLogTable+"(uid, platform, anchor_id, is_showing, is_error, followers, views, giver, gratuity, show_title, show_start_time, show_end_time, update_time, ext, error_msg) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
isource.GetUid(), isource.GetPlatform(), isource.GetAnchorId(), isource.GetIsShowing(), isource.GetIsError(), isource.GetFollowers(), isource.GetViews(), isource.GetGiver(), isource.GetGratuity(), isource.GetShowTitle(), isource.GetShowStartTime(), isource.GetShowEndTime(), isource.GetUpdateTime(), isource.GetExt(), isource.GetErrorMsg(),
)
store.errorAlarm(err) store.errorAlarm(err)
} }