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

Merge branch 'release/v0.6.0'

Browse Source
This commit is contained in:
eson 2020-09-09 17:26:51 +08:00
commit 4693296671
53 changed files with 14468 additions and 897 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

578
autostore.go Normal file
View File

@ -0,0 +1,578 @@
package intimate
import (
"database/sql"
"encoding/binary"
"fmt"
"log"
"reflect"
"strconv"
"time"
)
// StoreExtractorDB 全局的Extractor DB 库链接
var StoreExtractorDB *Store
// TStreamer 全局的Streamer. 在config init 完成初始化
var TStreamer *Table
// TClog 全局的Clog
var TClog *Table
// TStreamerList 全局的streamer list 这个表存的url. 进去可以找到主播的列表. 便于动态更新
var TStreamerList *Table
/*Store 结构体. 必须使用tag. field 数据库字段标签 uid 唯一id字段标签必须存在
*/
type Store struct {
db *sql.DB
}
// Table 表
type Table struct {
store *Store
name string
setting interface{}
updatesql string
selectsql string
insertsql string
duplicatesql string
}
func NewStore(uri string) *Store {
db, err := sql.Open("mysql", uri)
if err != nil {
panic(err)
}
s := &Store{db: db}
return s
}
// Table 选择表.
func (store *Store) Table(name string) *Table {
table := &Table{store: store}
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
}
// Queue mysql 队列结构
type Queue struct {
table *Table
obj reflect.Type
fieldIndex []int
selected string
cond CondWhere
uidname string
uididx int
}
type CondWhere struct {
Condition string
CondArgs []interface{}
}
// OperatorType 字典Operator 标志位的类型
type OperatorType string
const (
// OpOK 正常
OpOK OperatorType = "0"
// OpWAIT 等待处理
OpWAIT OperatorType = "1000"
// OpERROR 错误处理
OpERROR OperatorType = "10000"
)
// ConditionDefault 默认的条件
func ConditionDefault(platform Platform) CondWhere {
return CondWhere{
Condition: "platform = ? and operator = 0 and TIMESTAMPDIFF(MINUTE , update_time, CURRENT_TIMESTAMP()) >= update_interval",
CondArgs: []interface{}{string(platform)},
}
}
// Queue 根据Table生成一个队列. 处理结构. 每次弹出一个 obj 是要处理的结构体 自定义的whereCondition条件
func (t *Table) Queue(obj interface{}, whereCondition CondWhere) *Queue {
q := &Queue{}
q.cond = whereCondition
q.obj = reflect.TypeOf(obj)
q.table = t
q.fieldIndex = []int{} // select 需要配对字段变量的对应index位置
for i := 0; i < q.obj.NumField(); i++ {
field := q.obj.Field(i)
if fname, ok := field.Tag.Lookup("field"); ok {
q.selected += fname + ","
if _, ok := field.Tag.Lookup("uid"); ok {
q.uididx = i
q.uidname = fname
}
q.fieldIndex = append(q.fieldIndex, i)
}
}
q.selected = q.selected[:len(q.selected)-1]
return q
}
// Pop 队列弹出一个数据(任务). 参考队列处理 不支持嵌套.
func (queue *Queue) Pop() (result interface{}, err error) {
db := queue.table.store.db
tx, err := db.Begin()
if err != nil {
return nil, err
}
defer func() {
cerr := tx.Commit()
if cerr != nil {
log.Println(cerr)
log.Println(tx.Rollback())
}
}()
selectsql := `SELECT ` + queue.selected + ` FROM ` + queue.table.name + ` WHERE ` + queue.cond.Condition + " limit 1 for update"
rows, err := tx.Query(selectsql, queue.cond.CondArgs...)
if err != nil {
return nil, fmt.Errorf("table: %s queue is empty", queue.table.name)
}
var fields = make([]interface{}, len(queue.fieldIndex))
for i := range fields {
var iv interface{}
fields[i] = &iv
}
if rows.Next() {
err = rows.Scan(fields...)
if err != nil {
return nil, err
}
}
columntypes, err := rows.ColumnTypes()
if err != nil {
return nil, err
}
if err = rows.Close(); err != nil {
return nil, err
}
_, err = tx.Exec("UPDATE "+queue.table.name+" SET operator = "+string(OpWAIT)+" WHERE "+queue.uidname+" = ?", fields[queue.uididx])
if err != nil {
log.Println(err)
return nil, err
}
obj := reflect.New(queue.obj).Elem()
for i, idx := range queue.fieldIndex {
field := obj.Field(idx)
convert(*fields[i].(*interface{}), field, columntypes[i])
}
return obj.Addr().Interface(), err
}
// Insert nil 不插入. 不支持嵌套. 必须是Ptr类型
func (t *Table) Insert(obj interface{}) error {
ov := reflect.ValueOf(obj).Elem()
ot := reflect.TypeOf(obj)
fieldsql := ""
argssql := ""
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 += "?,"
}
}
}
ssql := fmt.Sprintf(t.insertsql, fieldsql[:len(fieldsql)-1], argssql[:len(argssql)-1])
_, err := t.store.db.Exec(ssql, args...)
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()
ot := reflect.TypeOf(obj)
fieldsql := ""
argssql := ""
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 += "?,"
}
}
}
ssql := fmt.Sprintf(t.insertsql, fieldsql[:len(fieldsql)-1], argssql[:len(argssql)-1])
result, err := t.store.db.Exec(ssql, args...)
if err != nil {
return 0, err
}
return result.LastInsertId()
}
// Update 结构体更新
func (t *Table) Update(obj interface{}) error {
ov := reflect.ValueOf(obj).Elem()
ot := reflect.TypeOf(obj)
fieldsql := ""
var uidname string
var uidvalue interface{}
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 _, ok := ftype.Tag.Lookup("uid"); ok {
if uidvalue != nil {
panic(fmt.Errorf("uid must unique, %s and %s", uidname, fname))
}
uidname = fname
uidvalue = field.Interface()
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 + " = ?,"
}
} else {
args = append(args, field.Interface())
fieldsql += fname + " = ?,"
}
}
}
if uidvalue == nil {
panic(fmt.Errorf("update must contain `uid` tag"))
}
usql := fmt.Sprintf(t.updatesql, fieldsql[:len(fieldsql)-1], uidname)
args = append(args, uidvalue)
_, err := t.store.db.Exec(usql, args...)
return err
}
// UpdateError 更新错误数据
func (t *Table) UpdateError(obj interface{}, err error) {
ov := reflect.ValueOf(obj).Elem()
ot := reflect.TypeOf(obj)
var uidname string
var uidvalue 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 _, ok := ftype.Tag.Lookup("uid"); ok {
if uidvalue != nil {
panic(fmt.Errorf("uid must unique, %s and %s", uidname, fname))
}
uidname = fname
uidvalue = field.Interface()
break
}
}
}
_, dberr := t.store.db.Exec("update "+t.name+" set operator = ?, error_msg = ? where ? = ?", 10000, sql.NullString{String: err.Error(), Valid: true}, uidname, uidvalue)
if dberr != nil {
// email tell owner to deal with
panic(err)
}
}
func assign(field reflect.Value, src interface{}) (bool, error) {
switch field.Kind() {
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
s := asString(src)
i64, err := strconv.ParseInt(s, 10, field.Type().Bits())
if err != nil {
err = strconvErr(err)
return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err)
}
field.SetInt(i64)
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
s := asString(src)
u64, err := strconv.ParseUint(s, 10, field.Type().Bits())
if err != nil {
err = strconvErr(err)
return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err)
}
field.SetUint(u64)
case reflect.Float32, reflect.Float64:
s := asString(src)
f64, err := strconv.ParseFloat(s, field.Type().Bits())
if err != nil {
err = strconvErr(err)
return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err)
}
field.SetFloat(f64)
case reflect.String:
field.SetString(string(src.([]byte)))
case reflect.Interface:
return true, nil
}
return false, nil
}
func convert(src interface{}, field reflect.Value, columntype *sql.ColumnType) error {
// log.Println("type:", field.Type(), ",kind:", field.Kind(), ",field:", field, "scanType:", columntype.ScanType(), "databaseType:", columntype.DatabaseTypeName())
if field.Kind() == reflect.Ptr {
fn := field.Type().Elem() // New 一个 field.Type().Elem() . 然后判断 columntype 转化 成 NullString Time
field.Set(reflect.New(fn))
field = field.Elem()
// log.Println("type:", fn.Name(), ",kind:", field.Kind(), ",fieldtype:", field.Type())
}
// log.Println(field.Kind(), field, reflect.TypeOf(field).Elem().Name(), columntype.ScanType().Kind())
if src == nil {
return fmt.Errorf("converting NULL to %s is unsupported", field.Kind())
}
switch columntype.DatabaseTypeName() {
case "TINYINT", "SMALLINT", "MEDIUMINT", "INT", "BIGINT":
isdefault, err := assign(field, src)
if err != nil {
return err
}
if isdefault {
s := asString(src)
i64, err := strconv.ParseInt(s, 10, 64)
if err != nil {
err = strconvErr(err)
return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err)
}
// reflect.New(reflect.TypeOf(i64))
field.Set(reflect.ValueOf(i64))
}
case "FLOAT", "DOUBLE", "DECIMAL":
isdefault, err := assign(field, src)
if err != nil {
return err
}
if isdefault {
s := asString(src)
f64, err := strconv.ParseFloat(s, 64)
if err != nil {
err = strconvErr(err)
return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err)
}
field.Set(reflect.ValueOf(f64))
}
case "BINARY", "VARBINARY", "TINYBLOB", "BLOB", "MEDIUMBLOB", "LONGBLOB", "JSON":
isdefault, err := assign(field, src)
if err != nil {
return err
}
if isdefault {
field.Set(reflect.ValueOf(src.([]byte)))
}
case "CHAR", "VARCHAR", "TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT":
isdefault, err := assign(field, src)
if err != nil {
return err
}
if isdefault {
field.Set(reflect.ValueOf(string(src.([]byte))))
}
case "BIT":
var bits []byte = make([]byte, 8)
copy(bits, src.([]byte))
switch field.Kind() {
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
field.SetInt(int64(binary.LittleEndian.Uint64(bits)))
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
field.SetUint(binary.LittleEndian.Uint64(bits))
case reflect.Interface:
field.Set(reflect.ValueOf(binary.LittleEndian.Uint64(bits)))
}
case "YEAR", "TIME", "DATE", "DATETIME", "TIMESTAMP":
s := src.(time.Time)
switch field.Interface().(type) {
case time.Time:
field.Set(reflect.ValueOf(src))
case string:
field.SetString(s.Format(time.RFC3339Nano))
case []byte:
field.SetBytes([]byte(s.Format(time.RFC3339Nano)))
default:
}
}
// log.Println(fv, columntype.ScanType().Kind())
if iscan, ok := field.Addr().Interface().(sql.Scanner); ok {
err := iscan.Scan(src)
if err != nil {
return err
}
}
return nil
}

57
autostore_test.go Normal file
View File

@ -0,0 +1,57 @@
package intimate
import (
"database/sql"
"encoding/json"
"testing"
"time"
)
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"})
// re, _ := queue.Pop()
// pstreamer := re.(*TSreamer)
// m := make(map[string]interface{})
// json.Unmarshal(pstreamer.Iface.([]byte), &m)
// spew.Println(re.(*TSreamer), m)
streamer := &TSreamer{}
streamer.Uid = 1
streamer.UserID = &sql.NullString{String: "xixi", Valid: true}
streamer.Name = "streamer"
streamer.Operator = 0
streamer.Bit = 0b11
// streamer.Ext = &sql.NullString{String: "ext", Valid: true}
tag := make(map[string]interface{})
tag["json"] = true
tag["name"] = "test"
btag, err := json.Marshal(tag)
if err != nil {
t.Error(err)
}
streamer.Iface = btag
now := time.Now()
streamer.UpdateTime = &now
err = store.Table("streamer").InsertOrUpdate(streamer, DUpdate{Field: "userid"})
if err != nil {
t.Error(err)
}
}
type TSreamer struct {
Uid int `field:"uid" uid:"auto"`
Name interface{} `field:"name"`
UserID *sql.NullString `field:"userid"`
Ext *sql.NullString `field:"ext"`
Iface interface{} `field:"tag"`
Bit uint64 `field:"bit"`
Operator int `field:"operator"`
UpdateTime *time.Time `field:"update_time"`
}

2
build.sh
View File

@ -20,4 +20,4 @@ do
projectworkspace=$src/bin/$projectname projectworkspace=$src/bin/$projectname
cd $path && mkdir $projectworkspace -p && go build -o $projectworkspace/$projectname cd $path && mkdir $projectworkspace -p && go build -o $projectworkspace/$projectname
cd $src cd $src
done done

12
config.go
View File

@ -18,6 +18,18 @@ func init() {
// storeOpenrec = NewStore() // storeOpenrec = NewStore()
log.SetFlags(log.Llongfile | log.Ltime) log.SetFlags(log.Llongfile | log.Ltime)
// StoreExtractorDB 全局的Extractor DB 库链接
StoreExtractorDB = NewStore(InitConfig.Database.ExtractorURI)
// TStreamer 全局的Streamer
TStreamer = StoreExtractorDB.Table("streamer")
// TClog 全局的Clog
TClog = StoreExtractorDB.Table("collect_log")
// TStreamerList 全局的streamer list 这个表存的url. 进去可以找到主播的列表. 便于动态更新
TStreamerList = StoreExtractorDB.Table("streamer_list")
} }
// Config 配置 // Config 配置

376
convert.go Normal file
View File

@ -0,0 +1,376 @@
// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Type conversions for Scan.
package intimate
import (
"database/sql/driver"
"errors"
"fmt"
"reflect"
"strconv"
)
var errNilPtr = errors.New("destination pointer is nil") // embedded in descriptive error
// convertAssignRows copies to dest the value in src, converting it if possible.
// An error is returned if the copy would result in loss of information.
// dest should be a pointer type. If rows is passed in, the rows will
// be used as the parent for any cursor values converted from a
// driver.Rows to a *Rows.
// func convertAssignRows(dest, src interface{}, rows *sql.Rows) error {
// // Common cases, without reflect.
// switch s := src.(type) {
// case string:
// switch d := dest.(type) {
// case *string:
// if d == nil {
// return errNilPtr
// }
// *d = s
// return nil
// case *[]byte:
// if d == nil {
// return errNilPtr
// }
// *d = []byte(s)
// return nil
// case *sql.RawBytes:
// if d == nil {
// return errNilPtr
// }
// *d = append((*d)[:0], s...)
// return nil
// }
// case []byte:
// switch d := dest.(type) {
// case *string:
// if d == nil {
// return errNilPtr
// }
// *d = string(s)
// return nil
// case *interface{}:
// if d == nil {
// return errNilPtr
// }
// *d = cloneBytes(s)
// return nil
// case *[]byte:
// if d == nil {
// return errNilPtr
// }
// *d = cloneBytes(s)
// return nil
// case *sql.RawBytes:
// if d == nil {
// return errNilPtr
// }
// *d = s
// return nil
// }
// case time.Time:
// switch d := dest.(type) {
// case *time.Time:
// *d = s
// return nil
// case *string:
// *d = s.Format(time.RFC3339Nano)
// return nil
// case *[]byte:
// if d == nil {
// return errNilPtr
// }
// *d = []byte(s.Format(time.RFC3339Nano))
// return nil
// case *sql.RawBytes:
// if d == nil {
// return errNilPtr
// }
// *d = s.AppendFormat((*d)[:0], time.RFC3339Nano)
// return nil
// }
// case decimalDecompose:
// switch d := dest.(type) {
// case decimalCompose:
// return d.Compose(s.Decompose(nil))
// }
// case nil:
// switch d := dest.(type) {
// case *interface{}:
// if d == nil {
// return errNilPtr
// }
// *d = nil
// return nil
// case *[]byte:
// if d == nil {
// return errNilPtr
// }
// *d = nil
// return nil
// case *sql.RawBytes:
// if d == nil {
// return errNilPtr
// }
// *d = nil
// return nil
// }
// // The driver is returning a cursor the client may iterate over.
// }
// var sv reflect.Value
// switch d := dest.(type) {
// case *string:
// sv = reflect.ValueOf(src)
// switch sv.Kind() {
// case reflect.Bool,
// reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,
// reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64,
// reflect.Float32, reflect.Float64:
// *d = asString(src)
// return nil
// }
// case *[]byte:
// sv = reflect.ValueOf(src)
// if b, ok := asBytes(nil, sv); ok {
// *d = b
// return nil
// }
// case *sql.RawBytes:
// sv = reflect.ValueOf(src)
// if b, ok := asBytes([]byte(*d)[:0], sv); ok {
// *d = sql.RawBytes(b)
// return nil
// }
// case *bool:
// bv, err := driver.Bool.ConvertValue(src)
// if err == nil {
// *d = bv.(bool)
// }
// return err
// case *interface{}:
// *d = src
// return nil
// }
// if scanner, ok := dest.(sql.Scanner); ok {
// return scanner.Scan(src)
// }
// dpv := reflect.ValueOf(dest)
// if dpv.Kind() != reflect.Ptr {
// return errors.New("destination not a pointer")
// }
// if dpv.IsNil() {
// return errNilPtr
// }
// if !sv.IsValid() {
// sv = reflect.ValueOf(src)
// }
// dv := reflect.Indirect(dpv)
// if sv.IsValid() && sv.Type().AssignableTo(dv.Type()) {
// switch b := src.(type) {
// case []byte:
// dv.Set(reflect.ValueOf(cloneBytes(b)))
// default:
// dv.Set(sv)
// }
// return nil
// }
// if dv.Kind() == sv.Kind() && sv.Type().ConvertibleTo(dv.Type()) {
// dv.Set(sv.Convert(dv.Type()))
// return nil
// }
// // The following conversions use a string value as an intermediate representation
// // to convert between various numeric types.
// //
// // This also allows scanning into user defined types such as "type Int int64".
// // For symmetry, also check for string destination types.
// switch dv.Kind() {
// case reflect.Ptr:
// if src == nil {
// dv.Set(reflect.Zero(dv.Type()))
// return nil
// }
// dv.Set(reflect.New(dv.Type().Elem()))
// return convertAssignRows(dv.Interface(), src, rows)
// case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
// if src == nil {
// return fmt.Errorf("converting NULL to %s is unsupported", dv.Kind())
// }
// s := asString(src)
// i64, err := strconv.ParseInt(s, 10, dv.Type().Bits())
// if err != nil {
// err = strconvErr(err)
// return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, dv.Kind(), err)
// }
// dv.SetInt(i64)
// return nil
// case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
// if src == nil {
// return fmt.Errorf("converting NULL to %s is unsupported", dv.Kind())
// }
// s := asString(src)
// u64, err := strconv.ParseUint(s, 10, dv.Type().Bits())
// if err != nil {
// err = strconvErr(err)
// return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, dv.Kind(), err)
// }
// dv.SetUint(u64)
// return nil
// case reflect.Float32, reflect.Float64:
// if src == nil {
// return fmt.Errorf("converting NULL to %s is unsupported", dv.Kind())
// }
// s := asString(src)
// f64, err := strconv.ParseFloat(s, dv.Type().Bits())
// if err != nil {
// err = strconvErr(err)
// return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, dv.Kind(), err)
// }
// dv.SetFloat(f64)
// return nil
// case reflect.String:
// if src == nil {
// return fmt.Errorf("converting NULL to %s is unsupported", dv.Kind())
// }
// switch v := src.(type) {
// case string:
// dv.SetString(v)
// return nil
// case []byte:
// dv.SetString(string(v))
// return nil
// }
// }
// return fmt.Errorf("unsupported Scan, storing driver.Value type %T into type %T", src, dest)
// }
func strconvErr(err error) error {
if ne, ok := err.(*strconv.NumError); ok {
return ne.Err
}
return err
}
func cloneBytes(b []byte) []byte {
if b == nil {
return nil
}
c := make([]byte, len(b))
copy(c, b)
return c
}
func asString(src interface{}) string {
switch v := src.(type) {
case string:
return v
case []byte:
return string(v)
}
rv := reflect.ValueOf(src)
switch rv.Kind() {
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
return strconv.FormatInt(rv.Int(), 10)
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
return strconv.FormatUint(rv.Uint(), 10)
case reflect.Float64:
return strconv.FormatFloat(rv.Float(), 'g', -1, 64)
case reflect.Float32:
return strconv.FormatFloat(rv.Float(), 'g', -1, 32)
case reflect.Bool:
return strconv.FormatBool(rv.Bool())
}
return fmt.Sprintf("%v", src)
}
func asBytes(buf []byte, rv reflect.Value) (b []byte, ok bool) {
switch rv.Kind() {
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
return strconv.AppendInt(buf, rv.Int(), 10), true
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
return strconv.AppendUint(buf, rv.Uint(), 10), true
case reflect.Float32:
return strconv.AppendFloat(buf, rv.Float(), 'g', -1, 32), true
case reflect.Float64:
return strconv.AppendFloat(buf, rv.Float(), 'g', -1, 64), true
case reflect.Bool:
return strconv.AppendBool(buf, rv.Bool()), true
case reflect.String:
s := rv.String()
return append(buf, s...), true
}
return
}
var valuerReflectType = reflect.TypeOf((*driver.Valuer)(nil)).Elem()
// callValuerValue returns vr.Value(), with one exception:
// If vr.Value is an auto-generated method on a pointer type and the
// pointer is nil, it would panic at runtime in the panicwrap
// method. Treat it like nil instead.
// Issue 8415.
//
// This is so people can implement driver.Value on value types and
// still use nil pointers to those types to mean nil/NULL, just like
// string/*string.
//
// This function is mirrored in the database/sql/driver package.
func callValuerValue(vr driver.Valuer) (v driver.Value, err error) {
if rv := reflect.ValueOf(vr); rv.Kind() == reflect.Ptr &&
rv.IsNil() &&
rv.Type().Elem().Implements(valuerReflectType) {
return nil, nil
}
return vr.Value()
}
// decimal composes or decomposes a decimal value to and from individual parts.
// There are four parts: a boolean negative flag, a form byte with three possible states
// (finite=0, infinite=1, NaN=2), a base-2 big-endian integer
// coefficient (also known as a significand) as a []byte, and an int32 exponent.
// These are composed into a final value as "decimal = (neg) (form=finite) coefficient * 10 ^ exponent".
// A zero length coefficient is a zero value.
// The big-endian integer coefficient stores the most significant byte first (at coefficient[0]).
// If the form is not finite the coefficient and exponent should be ignored.
// The negative parameter may be set to true for any form, although implementations are not required
// to respect the negative parameter in the non-finite form.
//
// Implementations may choose to set the negative parameter to true on a zero or NaN value,
// but implementations that do not differentiate between negative and positive
// zero or NaN values should ignore the negative parameter without error.
// If an implementation does not support Infinity it may be converted into a NaN without error.
// If a value is set that is larger than what is supported by an implementation,
// an error must be returned.
// Implementations must return an error if a NaN or Infinity is attempted to be set while neither
// are supported.
//
// NOTE(kardianos): This is an experimental interface. See https://golang.org/issue/30870
type decimal interface {
decimalDecompose
decimalCompose
}
type decimalDecompose interface {
// Decompose returns the internal decimal state in parts.
// If the provided buf has sufficient capacity, buf may be returned as the coefficient with
// the value set and length set as appropriate.
Decompose(buf []byte) (form byte, negative bool, coefficient []byte, exponent int32)
}
type decimalCompose interface {
// Compose sets the internal decimal value from parts. If the value cannot be
// represented then an error should be returned.
Compose(form byte, negative bool, coefficient []byte, exponent int32) error
}

BIN
crx/myblock.crx
View File

Binary file not shown.

52
crx/myblock.pem
View File

@ -1,28 +1,28 @@
-----BEGIN PRIVATE KEY----- -----BEGIN PRIVATE KEY-----
MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDSG09DSvB03TOe MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDA6O6owySvzaJ8
eOmQwfiCIf0wa2WRB31ewxa6i/PRgEKeJSUvIsIuaECUer2ss+J3rwSS2lDpGuiw iv6aXRMcuk0lnNKquPNVHeraadSkBVhtaJqPAT7saOVnn9g4s/21OBV6wGmWdl5W
FnsVyZqKI/+Rcuc83YJGYg6OAzVMz6UL8YCWhXu3huTJ+V+a5iNereIC69ZERRJt 5+8h+F4qXrlGO8N1Tl8mmBDetqg6T6RrI9vB2Th9hjlp2PCGCEg/lX+tx25qBiDG
nXlWqsq6HKya+6BP9sX9CI4GTHQrnWBysAxsswhdnnnRvu+GxglWafSIzuS6OizT 9uCB3pU4KMyfof1Z2jbo7aOfttgxZ1YdssenB//+31QjeZVw7wZACKaSzl6Z+WvY
1M1CmkZxNvDJhTSOR7SJlIYm2kM5/fIL53BdndF2IGAjfV1WV7AjwhTfun5cViEO pastMZ95YgTQvUq1AP93ny8AG+/bKUlE2vZpDygUJAXT8R7PGVD6T0/hUjTZk9HM
i8niQUIMY4L0AiO9grFD1g1xIYkeuVBoLxOUBzPxJwQmb64gseb9Dvt0BKLRGoou hrLShNjsBMxWWsrc7rdfW+R44pSuw5kHtIBrjRdrr8m1W0BCGTlGjq7Sk0OOZbsO
SIOyE+KVAgMBAAECggEAI4b6J2kR0VUBEDwmVHO0K38HUstqNHSVgrNO0dLt8sAz sh7x/1kvAgMBAAECggEAQYFXOTs6g/AB0uSn2+brDIA+nh0IdCxlhGuqSgxRxniC
I44o5DhGqPW4a9L4ZS5SrkWyKonPcic6buISRIwfPVoacjQBfVWAXJnil6lbtyYK NHSmtxSnN0PPUka291o3Noga0xnvQMZk9oG/c/uQjjaoFy9ICddY/BP69oVzzoko
ZMNcqLcgBRfCcpOgEq91DiKta6yIwekDFXVyCdFd78v+9ML1J+hUsLVkXJTLdP88 vbDpeIO4aPdhG44s7nZ1ghHTz0C/P4VhL+65AaCBbxZ7lWE+VHqC1smLHDrMYsmr
PGamRWVd6vGy3QMRjyM29GLPgS+/6Vrp1cptSuYNqYhlszohmu8lBvzjH9jbPh9d PfOxk5LZwdNE/YmZuOj1zX3nQGX+wtK0EMslRbyYjg3EwXUa9d5WNnhyFxt5MD9m
GFrrd8Bs7IRCdtKZig/3fbln4JEyyOYE+gcT2jplPksB6mR/5DBIdkVbeuFwGB0+ w9G1rJh5uDUIS60xHUB7ET4BKJMLj+pfy8Ahf3SDFfIzinjuvKg+z1St0nTSNqI7
h1/PKlprNQt7+Ei0HhHnTib7lZP8WGo4HkSi7PsAGQKBgQD1Ptho0wJiI2+6gL1O QP5KFNvfxqcrtYwf0wU6MGmjJ98FFiDoPY2QQfIlwQKBgQDj3eWDHA02Kr9DAde9
iNsEJVKIQ2Sxdx3wI/qudphM99t6xKCpPyVI2Nd9PBf2jbZjGAaz+P/KQYxEqb6i CuR0y4LixFhVDp7mYBnYxod3GMkPCGhSfuRcA/Tu7JgpACK/zsLi9gbzPkUAS2Y/
PRcQ+i99wCQoRfnRvUbKA4goEpKwRXmvn+499dm6D5pEuumOXGQYCmaFXuLTRN/I ly8lBFyCYlY96mRF0FgzOUCwP5Xjc9roDgLi1RVM3+Tnv1FcIa0CkoXPuWtmoVYS
BL6GNgLtoZAlLjUXaWtk8TszGQKBgQDbUf3p3HLpCjRvRDW/vA5xj+08t7xtF9uO W0S/Y9JKOrV2CEJvirvM3hysHwKBgQDYuipzsV+i3qx4STKXa7F0F8jGp+jr4Pgc
NilGK79uOA4VnxE2w3ioYqQ7t3I8J/0rAzGKq3tylg4QX6UpQ4b2koRr2B3cqoAk GqnMo5ysFePFFd4QZ1UHJ8HNmRlXsD+hAawzljVL8yCns8dZOE99xvSjgFzA/p/N
dsRdNWAHwCNepz8hTLsZyuihzbNv2nHmoqhzjK/FcrBHx5NAM+T6OBpLzQBnbUzk QToOgCRy6YZ5kZQA1ldZOhoVS5xTJRf7eavLwnV/8VGWxFNXdlvzdZecC8YvwEbh
3wIcqm223QKBgQDo/IRxyY0pGMtLXoT6ODACF0b6JzRhGG37tuKvngGAlbQQRP7w +gONR9Sw8QKBgQDiBwnSs9CnGl/+J6XIbzWOZRC/v+Oe+5HDEzCQRt241W0a7EwP
6wmL1F2cH1wQon7UU34CupqfVnhgvvZZgToJqfU2PTTcgeYc6Pl4b7SJhWOQTOCX CK+cVpBvL2D/ypKtkHrghwr93Za0ZBZYna62GU1qSAEkCYzNc40JwpDYQzp7FEDg
BZQ7jvYCulHv27aIxaNd53uQVx2cYoFKr58lN+i+QtADUoujq0YYxshb+QKBgQDW ToWuW491YeG9v5kEuqLTGmSBWHugogiuzLm031+JIhM9sZO08Pi4jtoCIQKBgEKC
ZOti7kZCeuBRGIu2V56C8uBFp5MBzf2polZsqx1iIFfcWPfZ4fGUIYFMgwKfvbOl XyTB+idxTlxbmjr9wadu5R6RyNkaYQm+5Pz3+MyWqx+AZnWCQIXBnagKo8xVOjGe
lWSbmxB9LiSnaugoU0OezBG43rYqXV4Qxy0jtKagTPoGcFWtNrX7+7e3XD8Zi6Am 4cmy/wgfxJWyhirLLRncRJcofU581W2X5sksagMZxzrnghP/sy6etgJzAbZCW+nA
hkFHW3MEAB5EvNq8Oz6OP8Os78SCVn2BimMlJJFF3QKBgQCF+aEAiBv+ivcmHUeP nQga1RQ+altqRGlC2VUNcUliE5/z0cznfM+Oi6cBAoGBAK68wjkperl8+ARQrqII
2eBq9nLltPFAfXJ/p31MMQ6Jgo36DBqUeoLeyq/WfIXvwqbVbP9fANZrKoTPbI97 Zg41PVrsFv6XmuLWv3bqnsx81BTbnHdqmPT4pScPLZvj+6c3iX8D8wzhzbgZtS0I
dilCHUoO33rafXJy6jtaggtpz14tt9soecTop0vM/rU7tGtfBe6NXg9LRl+oDJCU h+r1yMEdzyRZMGtvdi/f/+1TuhOTr/ZeoqYHWEduo8iRYLQBN9gGEkkKBQ4d14zi
37I3a9Is+2CLyAUXWCk9mLfFsQ== 8elaha7vrG/teCrFoeFcXIPr
-----END PRIVATE KEY----- -----END PRIVATE KEY-----

126
extractor/nimo_extractor/nimo_extractor.go Normal file
View File

@ -0,0 +1,126 @@
package main
import (
"database/sql"
"intimate"
"log"
"time"
"github.com/474420502/extractor"
"github.com/tebeka/selenium"
)
// sstore 源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
// var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STNimo))
// // estore 解析存储连接实例
// var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
func main() {
Execute()
}
type LiveInfo struct {
Followers int64 `exp:"//div[contains(@class,'nimo-rm_followers')]//span[@class='text c2']" mth:"r:ExtractNumber"`
Views int64 `exp:"//div[contains(@class,'nimo-rm_audience')]//span[@class='text c2']" mth:"r:ExtractNumber"`
Channel string `exp:"//div[contains(@class,'nimo-rm_type')]//span"`
Gratuity []int64 `exp:"//div[contains(@class,'rank-item-after3')]//span[contains(@class,'nimo-currency__count')]"`
}
func Execute() {
wd := intimate.GetChromeDriver(3030)
count := 0
countlimit := 200
waitfor := intimate.NewWaitFor(wd)
ps := intimate.NewPerfectShutdown()
queue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.PNimo))
for !ps.IsClose() {
istreamer, err := queue.Pop()
if err != nil {
log.Println(err)
intimate.TStreamer.UpdateError(istreamer, err)
continue
}
streamer := istreamer.(*intimate.Streamer)
wd.Get(streamer.LiveUrl.String)
// wd.Get("https://www.nimo.tv/live/1253835677")
waitfor.Default("//div[contains(@class,'nimo-rm_followers')]//span[@class='text c2' and text() != '']", nil)
waitfor.WaitWithTimeout("//div[contains(@class,'rank-item-top3')]", 7*time.Second, nil)
element, err := wd.FindElement(selenium.ByXPATH, "//div[contains(@class,'rank-item-top3')]")
if err != nil {
log.Println(streamer.Uid, err)
} else {
err = element.MoveTo(50, 50)
element.Click()
if err != nil {
log.Println(streamer.Uid, err)
}
}
waitfor.Default("//div[contains(@class,'nimo-rm_audience')]//span[@class='text c2']", nil)
var pagesource string
pagesource, _ = wd.PageSource()
etor := extractor.ExtractHtmlString(pagesource)
li := etor.GetObjectByTag(LiveInfo{}).(*LiveInfo)
// log.Printf("%#v", li)
utime := sql.NullTime{Time: time.Now(), Valid: true}
clog := &intimate.CollectLog{}
clog.Platform = intimate.PNimo
clog.Followers = &sql.NullInt64{Int64: li.Followers, Valid: true}
clog.Views = &sql.NullInt64{Int64: li.Views, Valid: true}
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}
cuid, err := intimate.TClog.InsertRetAutoID(clog)
if err != nil {
panic(err)
}
streamer.Channel = &sql.NullString{String: li.Channel, Valid: true}
streamer.LatestLogUid = cuid
streamer.UpdateTime = &utime
streamer.Operator = 0
switch {
case li.Followers <= 1000:
streamer.UpdateInterval = 720
case li.Followers <= 10000:
streamer.UpdateInterval = 360
case li.Followers <= 100000:
streamer.UpdateInterval = 180
case li.Followers <= 1000000:
streamer.UpdateInterval = 90
default:
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)
err = intimate.TStreamer.Update(streamer)
if err != nil {
panic(err)
}
count++
if count >= countlimit {
count = 0
wd.Close()
wd.Quit()
wd = intimate.GetChromeDriver(3030)
}
}
}

2
tasks/openrec/openrec_task2/task_openrec_test.go → extractor/nimo_extractor/nimo_extractor_test.go
View File

@ -5,5 +5,5 @@ import (
) )
func TestMain(t *testing.T) { func TestMain(t *testing.T) {
main() Execute()
} }

305
extractor/openrec_extractor/openrec_extractor.go
View File

@ -9,12 +9,11 @@ import (
"time" "time"
"github.com/474420502/extractor" "github.com/474420502/extractor"
"github.com/474420502/gcurl"
"github.com/474420502/requests"
"github.com/tidwall/gjson" "github.com/tidwall/gjson"
) )
var estore = intimate.NewStoreExtractor()
var sstore = intimate.NewStoreSource(string(intimate.STOpenrec))
//UserInfo 提取信息的结构体 //UserInfo 提取信息的结构体
type UserInfo struct { type UserInfo struct {
UserName string `exp:"//p[ contains(@class, 'c-global__user__profile__list__name__text')]"` UserName string `exp:"//p[ contains(@class, 'c-global__user__profile__list__name__text')]"`
@ -27,141 +26,237 @@ type UserLive struct {
Title string `exp:"//h1[contains(@class,'MovieTitle__Title')]"` Title string `exp:"//h1[contains(@class,'MovieTitle__Title')]"`
LiveStartTime string `exp:"//meta[@itemprop='uploadDate']/@content"` LiveStartTime string `exp:"//meta[@itemprop='uploadDate']/@content"`
LiveEndTime string `exp:"//meta[@itemprop='duration']/@content"` LiveEndTime string `exp:"//meta[@itemprop='duration']/@content"`
Tags []string `exp:"//a[contains(@class,'TagButton')]"` Tags []string `exp:"//div[contains(@class,'MovieMetaContent__TagContainer')]//a[@role ='button']"`
} }
// Execute 执行 // Execute 执行
func Execute() { func Execute() {
ps := intimate.NewPerfectShutdown() ps := intimate.NewPerfectShutdown()
ses := requests.NewSession()
squeue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.Popenrec))
var lasterr error = nil var lasterr error = nil
for !ps.IsClose() { for !ps.IsClose() {
var err error istreamer, err := squeue.Pop()
source, err := sstore.Pop(intimate.TOpenrecUser, 0) // streamer, err := estore.Pop(intimate.Popenrec) //队列里弹出一个streamer行. 进行解析
if err != nil {
if istreamer == nil || err != nil {
if err != lasterr { if err != lasterr {
log.Println(err, lasterr) log.Println(err, lasterr)
lasterr = err lasterr = err
} }
time.Sleep(time.Second * 5) time.Sleep(time.Second * 2)
continue continue
} }
lasterr = nil streamer := istreamer.(*intimate.Streamer)
sdata := source.Ext.([]byte) userId := *streamer.UserId
datamap := gjson.ParseBytes(sdata).Map()
source.Operator = int32(intimate.OperatorError) var updateUrl map[string]string
userId := datamap["var_user_id"].String() err = json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl) // 反序列化update_url, 里面存了需要采集的url
if err != nil {
log.Println(err)
continue
}
// Check Userid
streamer := &intimate.Streamer{} userUrl := updateUrl["user"]
streamer.UserId = userId log.Println(userUrl)
// streamer.Platform = intimate.Popenrec 不需要更新字段 tp := ses.Get(userUrl) // 获取user url页面数据
resp, err := tp.Execute()
streamer.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
htmlUser := datamap["html_user"] if err != nil {
log.Println(err)
userEtor := extractor.ExtractHtmlString(htmlUser.String()) intimate.TStreamer.UpdateError(streamer, err)
ui, ok1 := userEtor.GetObjectByTag(UserInfo{}).(*UserInfo)
htmlLive := datamap["html_live"]
liveEtor := extractor.ExtractHtmlString(htmlLive.String())
ul, ok2 := liveEtor.GetObjectByTag(UserLive{}).(*UserLive)
jsonSupporters := datamap["json_supporters"]
clog := &intimate.CollectLog{}
if ok1 {
clog.Followers = sql.NullInt64{Int64: ui.Followers, Valid: true}
clog.Views = sql.NullInt64{Int64: ui.Views, Valid: true}
if ui.Views != 0 {
clog.IsLiveStreaming = true
}
streamer.UserName = sql.NullString{String: ui.UserName, Valid: true}
giverjson := jsonSupporters
var givers []interface{}
var gratuity int64 = 0
for _, v := range giverjson.Array() {
giverSource := gjson.Parse(v.String())
for _, item := range giverSource.Get("data.items").Array() {
givers = append(givers, item.Map())
gratuity += item.Get("total_yells").Int()
}
}
giversbytes, err := json.Marshal(givers)
if err != nil {
log.Println(err)
clog.ErrorMsg = sql.NullString{String: err.Error(), Valid: true}
} else {
clog.Giver = giversbytes
}
clog.Gratuity = sql.NullInt64{Int64: gratuity, Valid: true}
} else {
log.Println("UserInfo may be not exists")
estore.UpdateError(streamer, errors.New("UserInfo may be not exists"))
continue continue
} }
//log.Println(ul) cookies := ses.GetCookies(tp.GetParsedURL())
if ok2 {
clog.LiveTitle = sql.NullString{String: ul.Title, Valid: true}
startTime, err := time.ParseInLocation("2006-01-02T15:04:05Z07:00", ul.LiveStartTime, time.Local) scurl := updateUrl["supporters"] //获取打赏者的数据
if err != nil { curl := gcurl.Parse(scurl)
log.Println(err) supportersSession := curl.CreateSession()
} else {
clog.LiveStartTime = sql.NullTime{Time: startTime.Local(), Valid: true} temporary := curl.CreateTemporary(supportersSession)
duration, err := intimate.ParseDuration(ul.LiveEndTime) supportersSession.SetCookies(temporary.GetParsedURL(), cookies)
if err != nil { var supporters []string
log.Println(err) for { // supporters 数据需要登录信息. 下面为赋值 supporters链接获取的uid token random码
} else {
endTime := startTime.Add(duration) supportersQuery := temporary.GetQuery()
clog.LiveStartTime = sql.NullTime{Time: endTime.Local(), Valid: true}
for _, cookie := range cookies {
if cookie.Name == "uuid" {
supportersQuery.Set("Uuid", cookie.Value)
continue
}
if cookie.Name == "token" {
supportersQuery.Set("Token", cookie.Value)
continue
}
if cookie.Name == "random" {
supportersQuery.Set("Random", cookie.Value)
continue
} }
} }
if tags, err := json.Marshal(ul.Tags); err == nil { supportersQuery.Set("identify_id", userId)
clog.Tags = tags temporary.SetQuery(supportersQuery)
} else {
log.Println("json error", ul.Tags, clog.Tags) resp, err := temporary.Execute()
if err != nil {
log.Println(err)
} }
supporterjson := gjson.ParseBytes(resp.Content())
supporterdata := supporterjson.Get("data") //解析supporters获取的json数据
if supporterdata.Type == gjson.Null {
break
}
supporters = append(supporters, string(resp.Content()))
temporary.QueryParam("page_number").IntAdd(1)
} }
streamer.Uid = source.StreamerId.Int64 // cookies := cxt.Session().GetCookies(wf.GetParsedURL())
streamer.UpdateTime = source.UpdateTime // ext := make(map[string]interface{})
streamer.Tags = clog.Tags
clog.Platform = intimate.Popenrec jsonSupporters := supporters
clog.UserId = userId htmlUser := string(resp.Content())
clog.UpdateTime = source.UpdateTime
clog.StreamerUid = streamer.Uid
logUid := estore.InsertClog(clog) liveUrl := updateUrl["live"]
tp = ses.Get(liveUrl)
resp, err = tp.Execute()
if err != nil {
log.Println(err)
intimate.TStreamer.UpdateError(streamer, err)
continue
}
htmlLive := string(resp.Content())
// ext["var_user_id"] = userId
LiveUrl := "https://www.openrec.tv/live/" + userId // streamer.Platform = intimate.Popenrec
streamer.UpdateInterval = 120
streamer.LiveUrl = sql.NullString{String: LiveUrl, Valid: true} streamer.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
streamer.LatestLogUid = logUid streamer.Operator = 0
// streamer.Operator = 0 Extractor(streamer, userId, htmlUser, htmlLive, jsonSupporters)
log.Println(streamer.UserId)
estore.Update(streamer,
"user_name", streamer.UserName,
"user_id", streamer.UserId,
"live_url", streamer.LiveUrl,
"latest_log_uid", streamer.LatestLogUid,
"update_time", streamer.UpdateTime,
"tags", streamer.Tags,
)
source.Operator = int32(intimate.OperatorExtractorOK)
sstore.UpdateOperator(source)
} }
} }
func Extractor(streamer *intimate.Streamer, userId string, htmlUser, htmlLive string, jsonSupporters []string) {
// sdata := source.Ext.([]byte)
// datamap := gjson.ParseBytes(sdata).Map()
// userId := datamap["var_user_id"].String()
// streamer := &intimate.Streamer{}
// streamer.UserId = &userId
// streamer.Platform = intimate.Popenrec 不需要更新字段
// htmlUser := datamap["html_user"]
userEtor := extractor.ExtractHtmlString(htmlUser)
ui, ok1 := userEtor.GetObjectByTag(UserInfo{}).(*UserInfo)
// htmlLive := datamap["html_live"]
liveEtor := extractor.ExtractHtmlString(htmlLive)
ul, ok2 := liveEtor.GetObjectByTag(UserLive{}).(*UserLive)
// jsonSupporters := datamap["json_supporters"]
clog := &intimate.CollectLog{}
if ok1 {
clog.Followers = &sql.NullInt64{Int64: ui.Followers, Valid: true}
clog.Views = &sql.NullInt64{Int64: ui.Views, Valid: true}
if ui.Views != 0 {
clog.IsLiveStreaming = true
}
streamer.UserName = &sql.NullString{String: ui.UserName, Valid: true}
// giverjson := jsonSupporters
var givers []interface{}
var gratuity int64 = 0
for _, v := range jsonSupporters {
giverSource := gjson.Parse(v)
for _, item := range giverSource.Get("data.items").Array() {
givers = append(givers, item.Map())
gratuity += item.Get("total_yells").Int()
}
}
giversbytes, err := json.Marshal(givers)
if err != nil {
log.Println(err)
clog.ErrorMsg = &sql.NullString{String: err.Error(), Valid: true}
} else {
clog.Giver = giversbytes
}
clog.Gratuity = &sql.NullInt64{Int64: gratuity, Valid: true}
} else {
log.Println("UserInfo may be not exists")
intimate.TStreamer.UpdateError(streamer, errors.New("UserInfo may be not exists"))
return
}
//log.Println(ul)
if ok2 {
clog.LiveTitle = &sql.NullString{String: ul.Title, Valid: true}
startTime, err := time.ParseInLocation("2006-01-02T15:04:05Z07:00", ul.LiveStartTime, time.Local)
if err != nil {
log.Println(err)
} else {
clog.LiveStartTime = &sql.NullTime{Time: startTime.Local(), Valid: true}
duration, err := intimate.ParseDuration(ul.LiveEndTime)
if err != nil {
log.Println(err)
} else {
endTime := startTime.Add(duration)
clog.LiveEndTime = &sql.NullTime{Time: endTime.Local(), Valid: true}
}
}
if tags, err := json.Marshal(ul.Tags); err == nil {
clog.Tags = tags
} else {
log.Println("json error", ul.Tags, clog.Tags)
}
}
// streamer.Uid = source.StreamerId.Int64
// streamer.UpdateTime = &source.UpdateTime
if clog.Tags != nil {
streamer.Tags = clog.Tags
}
clog.Platform = intimate.Popenrec
clog.UserId = userId
clog.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
clog.StreamerUid = streamer.Uid
logUid, err := intimate.TClog.InsertRetAutoID(clog)
if err != nil {
log.Println(err)
return
}
LiveUrl := "https://www.openrec.tv/live/" + userId
streamer.LiveUrl = &sql.NullString{String: LiveUrl, Valid: true}
streamer.LatestLogUid = logUid
// streamer.Operator = 0
// log.Println(*streamer.UserId)
intimate.TStreamer.Update(streamer)
// source.Operator = int32(intimate.OperatorExtractorOK)
// sstore.UpdateOperator(source)
}

163
extractor/twitcasting_extractor/twitcasting_extractor.go
View File

@ -2,9 +2,9 @@ package main
import ( import (
"database/sql" "database/sql"
"encoding/json"
"intimate" "intimate"
"log" "log"
"regexp"
"strconv" "strconv"
"strings" "strings"
"time" "time"
@ -21,8 +21,8 @@ var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
type LiveData struct { type LiveData struct {
UserName string `exp:"//span[@class='tw-live-author__info-username']" method:"Text"` UserName string `exp:"//span[@class='tw-live-author__info-username']" method:"Text"`
Follower string `exp:"(//span[@class='tw-user-nav-list-count'])[2]" method:"Text"` Follower int64 `exp:"(//span[@class='tw-user-nav-list-count'])[2]" method:"r:ExtractNumber"`
MaxViews string `exp:"//span[@id='max_viewer_count']" method:"Text"` MaxViews int64 `exp:"//span[@id='max_viewer_count']/text()" method:"r:ExtractNumber"`
LiveTitle string `exp:"//meta[@property='og:title']" method:"AttributeValue,content"` LiveTitle string `exp:"//meta[@property='og:title']" method:"AttributeValue,content"`
LiveStart string `exp:"//time[@data-kind='relative']" method:"AttributeValue,datetime"` LiveStart string `exp:"//time[@data-kind='relative']" method:"AttributeValue,datetime"`
LiveDuration string `exp:"//span[@id='updatetimer']" method:"AttributeValue,data-duration"` LiveDuration string `exp:"//span[@id='updatetimer']" method:"AttributeValue,data-duration"`
@ -33,25 +33,36 @@ func main() {
ps := intimate.NewPerfectShutdown() ps := intimate.NewPerfectShutdown()
ses := requests.NewSession() ses := requests.NewSession()
streamerQueue := intimate.TStreamer.Queue(intimate.Streamer{}, intimate.ConditionDefault(intimate.Ptwitcasting))
for !ps.IsClose() { for !ps.IsClose() {
streamer, err := estore.Pop(intimate.Ptwitcasting) // streamer, err := estore.Pop(intimate.Ptwitcasting)
isteamer, err := streamerQueue.Pop()
if err != nil { if err != nil {
log.Println(err, streamer) log.Println(err, isteamer)
continue
} }
streamer.LiveUrl = sql.NullString{String: "https://twitcasting.tv/" + streamer.UserId, Valid: true} streamer := isteamer.(*intimate.Streamer)
streamer.LiveUrl = &sql.NullString{String: "https://twitcasting.tv/" + *streamer.UserId, Valid: true}
resp, err := ses.Get(streamer.LiveUrl.String).Execute() resp, err := ses.Get(streamer.LiveUrl.String).Execute()
if err != nil { if err != nil {
estore.UpdateError(streamer, err) intimate.TStreamer.UpdateError(streamer, err)
log.Println(err, streamer.UserId) log.Println(err, *streamer.UserId)
continue continue
} }
var ldata *LiveData var ldata *LiveData
// f, _ := os.OpenFile("./twistcasting.html", os.O_CREATE|os.O_RDWR|os.O_TRUNC, os.ModePerm)
// f.Write(resp.Content())
etor := extractor.ExtractHtml(resp.Content()) etor := extractor.ExtractHtml(resp.Content())
ldata = etor.GetObjectByTag(LiveData{}).(*LiveData) ildata := etor.GetObjectByTag(LiveData{})
ldata.MaxViews = regexp.MustCompile("\\d+").FindString(ldata.MaxViews) if ildata == nil {
log.Println(streamer.LiveUrl.String)
continue
}
ldata = ildata.(*LiveData)
// ldata.MaxViews = regexp.MustCompile("\\d+").FindString(ldata.MaxViews)
coincount := 0 coincount := 0
for i := 0; ; i++ { for i := 0; ; i++ {
@ -59,14 +70,14 @@ func main() {
giverurl := streamer.LiveUrl.String + "/backers/" + strconv.Itoa(i) giverurl := streamer.LiveUrl.String + "/backers/" + strconv.Itoa(i)
resp, err = ses.Get(giverurl).Execute() resp, err = ses.Get(giverurl).Execute()
if err != nil { if err != nil {
estore.UpdateError(streamer, err) intimate.TStreamer.UpdateError(streamer, err)
log.Panic(err) log.Panic(err)
} }
etor := extractor.ExtractHtml(resp.Content()) etor := extractor.ExtractHtml(resp.Content())
xp, err := etor.XPaths("//td[@class='tw-memorial-table-recent-point']") xp, err := etor.XPaths("//td[@class='tw-memorial-table-recent-point']")
if err != nil { if err != nil {
estore.UpdateError(streamer, err) intimate.TStreamer.UpdateError(streamer, err)
log.Panic(err) log.Panic(err)
} }
@ -90,63 +101,81 @@ func main() {
} }
} }
streamer.Platform = intimate.Ptwitcasting var tags []byte
streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true} tags, err = json.Marshal(ldata.Tags)
streamer.UserName = sql.NullString{String: ldata.UserName, Valid: true} if err != nil {
streamer.Operator = 0
// streamer.UpdateInterval = 60
clog := &intimate.CollectLog{}
clog.UserId = streamer.UserId
clog.Gratuity = sql.NullInt64{Int64: int64(coincount), Valid: true}
clog.Platform = streamer.Platform
clog.UpdateTime = streamer.UpdateTime
clog.LiveTitle = sql.NullString{String: ldata.LiveTitle, Valid: true}
fl, err := intimate.ParseNumberEx(ldata.Follower)
if err == nil {
clog.Followers = sql.NullInt64{Int64: int64(fl), Valid: true}
switch {
case fl <= 100:
streamer.UpdateInterval = 360
case fl <= 1000:
streamer.UpdateInterval = 240
case fl <= 100:
streamer.UpdateInterval = 120
default:
streamer.UpdateInterval = 60
}
} else {
log.Println(err)
}
views, err := strconv.Atoi(ldata.MaxViews)
if err == nil {
clog.Views = sql.NullInt64{Int64: int64(views), Valid: true}
} else {
clog.Views = sql.NullInt64{Int64: int64(0), Valid: true}
// log.Println(err, streamer.UserId)
}
// st, err := strconv.Atoi(ldata.LiveStart)
st, err := time.Parse("Mon, 02 Jan 2006 15:04:05 -0700", ldata.LiveStart)
if err == nil {
startTime := st
clog.LiveStartTime = sql.NullTime{Time: startTime, Valid: true}
dt, err := strconv.Atoi(ldata.LiveDuration)
if time.Now().Sub(startTime) >= time.Hour*24*90 {
streamer.Operator = 5
}
if err == nil {
endTime := startTime.Add((time.Duration)(dt) * time.Millisecond)
clog.LiveEndTime = sql.NullTime{Time: endTime, Valid: true}
} else {
log.Println(err, streamer.UserId)
}
} else {
log.Println(err, streamer.UserId) log.Println(err, streamer.UserId)
} }
streamer.LatestLogUid = estore.InsertClog(clog) streamer.Platform = intimate.Ptwitcasting
estore.UpdateStreamer(streamer) streamer.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
streamer.UserName = &sql.NullString{String: ldata.UserName, Valid: true}
streamer.Operator = 0
streamer.Tags = tags
// streamer.UpdateInterval = 60
clog := &intimate.CollectLog{}
clog.UserId = *streamer.UserId
clog.Gratuity = &sql.NullInt64{Int64: int64(coincount), Valid: true}
clog.Platform = streamer.Platform
clog.UpdateTime = streamer.UpdateTime
clog.LiveTitle = &sql.NullString{String: ldata.LiveTitle, Valid: true}
clog.Tags = tags
clog.Followers = &sql.NullInt64{Int64: int64(ldata.Follower), Valid: true}
switch {
case ldata.Follower <= 100:
streamer.UpdateInterval = 720
case ldata.Follower <= 1000:
streamer.UpdateInterval = 320
case ldata.Follower <= 10000:
streamer.UpdateInterval = 240
default:
streamer.UpdateInterval = 120
}
clog.Views = &sql.NullInt64{Int64: ldata.MaxViews, Valid: true}
if ldata.LiveStart != "" {
st, err := time.Parse("Mon, 02 Jan 2006 15:04:05 -0700", ldata.LiveStart)
if err == nil {
startTime := st
clog.LiveStartTime = &sql.NullTime{Time: startTime, Valid: true}
dt, err := strconv.Atoi(ldata.LiveDuration)
liveduration := time.Now().Sub(startTime)
switch {
case liveduration >= time.Hour*24*240:
streamer.Operator = 5
case liveduration >= time.Hour*24*60:
streamer.UpdateInterval = 60 * 24 * 30
case liveduration >= time.Hour*24*30:
streamer.UpdateInterval = 60 * 24 * 15
case liveduration >= time.Hour*24*15:
streamer.UpdateInterval = 60 * 24 * 7
case liveduration >= time.Hour*24*7:
streamer.UpdateInterval = 60 * 24 * 3
}
if err == nil {
endTime := startTime.Add((time.Duration)(dt) * time.Millisecond)
clog.LiveEndTime = &sql.NullTime{Time: endTime, Valid: true}
} else {
log.Println(err, streamer.UserId)
}
} else {
log.Println(err, streamer.UserId)
}
}
clog.StreamerUid = streamer.Uid
uid, err := intimate.TClog.InsertRetAutoID(clog)
if err != nil {
log.Println(err)
continue
}
streamer.LatestLogUid = uid
intimate.TStreamer.Update(streamer)
// estore.UpdateStreamer(streamer)
log.Println(*streamer.UserId)
} }
} }

2
extractor/twitcasting_extractor/twitcasting_extractor_test.go
View File

@ -10,7 +10,7 @@ import (
// Follower string `exp:".//span[@class='tw-user-nav-list-count']" method:"Text"` // Follower string `exp:".//span[@class='tw-user-nav-list-count']" method:"Text"`
// } // }
func estMain(t *testing.T) { func TestMain(t *testing.T) {
main() main()
} }

344
extractor/twitch_extractor/tiwtch_extractor.go
View File

@ -3,6 +3,7 @@ package main
import ( import (
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"fmt"
"intimate" "intimate"
"log" "log"
"regexp" "regexp"
@ -12,136 +13,287 @@ import (
"github.com/tebeka/selenium" "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() { func main() {
wd := intimate.GetChromeDriver(3030) wd := intimate.GetChromeDriver(3040)
ps := intimate.NewPerfectShutdown() ps := intimate.NewPerfectShutdown()
counter := intimate.NewCounter() queue := intimate.TStreamerList.Queue(intimate.StreamerList{}, intimate.ConditionDefault(intimate.Ptwitch))
counter.SetMaxLimit(200) var count = 0
counter.SetMaxToDo(func(olist ...interface{}) error { var countlimt = 200
owd := olist[0].(*selenium.WebDriver)
(*owd).Close()
(*owd).Quit()
*owd = intimate.GetChromeDriver(3030)
return nil
}, &wd)
var lasterr error = nil // var lasterr error = nil
// var err error // var err error
for !ps.IsClose() { for !ps.IsClose() {
streamer, err := estore.Pop(intimate.Ptwitch, 0)
if streamer == nil || err != nil { // sourceChannel, err := sstore.Pop(intimate.TTwitchChannel)
if err != lasterr { isl, err := queue.Pop()
log.Println(err, lasterr) if err != nil {
lasterr = err panic(err)
}
time.Sleep(time.Second * 2)
continue
} }
streamerlist := isl.(*intimate.StreamerList)
var updateUrl map[string]string weburl := streamerlist.Url + "?sort=VIEWER_COUNT"
json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl) err = wd.Get(weburl)
liveUrl := updateUrl["live"]
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 { if err != nil {
log.Println(err) 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 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 { 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 if err = intimate.TStreamer.InsertOrUpdate(streamer,
case fl > 10000: intimate.DUpdate{Field: "tags"},
streamer.UpdateInterval = 240 intimate.DUpdate{Field: "update_time"},
case fl > 1000: ); err != nil {
streamer.UpdateInterval = 360 log.Println(err)
case fl > 100: }
streamer.UpdateInterval = 720 }
case fl > 0:
streamer.UpdateInterval = 1440 log.Println("streamer find", len(articles))
if len(articles) == 0 {
intimate.TStreamerList.UpdateError(streamerlist, fmt.Errorf(""))
}
count++
if count >= countlimt {
count = 0
wd = intimate.GetChromeDriver(3031)
} }
streamer.UpdateTime = clog.UpdateTime
estore.UpdateStreamer(streamer)
counter.AddWithReset(1)
} }
wd.Close() wd.Close()
wd.Quit() 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 := streamer.LiveUrl.String
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 {
log.Println(err)
intimate.TStreamer.UpdateError(streamer, err)
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
streamer.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
intimate.TStreamer.UpdateError(streamer, fmt.Errorf(""))
}
return
}
err = extractFollowers(wd, clog)
if err != nil {
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.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 * 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 { func extractUserName(wd selenium.WebDriver, streamer *intimate.Streamer) error {
return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) { return wd.WaitWithTimeout(func(web selenium.WebDriver) (bool, error) {
label, err := web.FindElement(selenium.ByXPATH, "//a[@class='tw-interactive']//h1") label, err := web.FindElement(selenium.ByXPATH, "//a[@class='tw-interactive']//h1")
if err == nil { if err == nil {
if ltxt, err := label.Text(); err == nil && ltxt != "" { if ltxt, err := label.Text(); err == nil && ltxt != "" {
// log.Println("label:", ltxt) // log.Println("label:", ltxt)
streamer.UserName = sql.NullString{String: ltxt, Valid: true} streamer.UserName = &sql.NullString{String: ltxt, Valid: true}
return true, nil return true, nil
} }
} }
@ -161,7 +313,7 @@ func extractFollowers(wd selenium.WebDriver, clog *intimate.CollectLog) error {
} }
followers = regexp.MustCompile(`[\d,]+`).FindString(followers) followers = regexp.MustCompile(`[\d,]+`).FindString(followers)
fint, _ := intimate.ParseNumber(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) // log.Println("followers: ", followers, fint)
return true, nil return true, nil
}, 4*time.Second) }, 4*time.Second)
@ -174,7 +326,7 @@ func extractViews(wd selenium.WebDriver, clog *intimate.CollectLog) error {
if txt, err := views.Text(); err == nil { if txt, err := views.Text(); err == nil {
vint, _ := intimate.ParseNumber(txt) vint, _ := intimate.ParseNumber(txt)
clog.Views = sql.NullInt64{Int64: vint, Valid: true} clog.Views = &sql.NullInt64{Int64: vint, Valid: true}
// log.Println("views:", txt) // log.Println("views:", txt)
views.Click() views.Click()
@ -194,7 +346,7 @@ func extractTitle(wd selenium.WebDriver, clog *intimate.CollectLog) error {
title, err := web.FindElement(selenium.ByXPATH, `//h2[@data-a-target='stream-title']`) title, err := web.FindElement(selenium.ByXPATH, `//h2[@data-a-target='stream-title']`)
if err == nil { if err == nil {
if txt, err := title.Text(); 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 return true, nil
} }
} }
@ -246,7 +398,7 @@ func extractGratuity(wd selenium.WebDriver, clog *intimate.CollectLog) error {
log.Println(err) log.Println(err)
} }
} }
clog.Gratuity = sql.NullInt64{Int64: gratuity, Valid: true} clog.Gratuity = &sql.NullInt64{Int64: gratuity, Valid: true}
} }
return true, nil return true, nil
} }

81
extractor_field.go
View File

@ -3,26 +3,25 @@ package intimate
import ( import (
"database/sql" "database/sql"
"reflect" "reflect"
"time"
) )
type GetSet struct { type GetSet struct {
} }
type StreamerList struct { type StreamerList struct {
UrlHash []byte // UrlHash string `field:"urlhash" uid:"true"` //
Platform Platform // Platform string `field:"platform" ` //
Url string // Url string `field:"url" ` //
Label sql.NullString // Label *sql.NullString `field:"label" ` //
Serialize interface{} Serialize interface{} `field:"serialize" `
UpdateInterval int32 UpdateInterval int32 `field:"update_interval" `
UpdateTime time.Time // UpdateTime *sql.NullTime `field:"update_time" ` //
ErrorMsg sql.NullString ErrorMsg *sql.NullString `field:"error_msg" ` //
Operator int32 Operator int32 `field:"operator" `
LastOperator int32 LastOperator int32
} }
@ -38,25 +37,25 @@ func (sl *StreamerList) Set(field string, value interface{}) {
} }
type Streamer struct { type Streamer struct {
Uid int64 // Uid int64 `field:"uid" uid:"auto"` //
Platform Platform // Platform Platform `field:"platform"` //
UserId string // UserId *string `field:"user_id"` //
UserName sql.NullString // UserName *sql.NullString `field:"user_name"` //
LiveUrl sql.NullString // LiveUrl *sql.NullString `field:"live_url"` //
Channel sql.NullString // Channel *sql.NullString `field:"channel"` //
Tags interface{} Tags interface{} `field:"tags"`
Ext interface{} // Ext interface{} `field:"ext"` //
IsUpdateStreamer bool // 更新上面的内容 IsUpdateStreamer bool // 更新上面的内容
IsUpdateUrl bool IsUpdateUrl bool
UpdateInterval int32 UpdateInterval int32 `field:"update_interval"`
UpdateUrl interface{} UpdateUrl interface{} `field:"update_url"` // TODO: nil
LatestLogUid int64 LatestLogUid int64 `field:"latest_log_uid"`
UpdateTime sql.NullTime // UpdateTime *sql.NullTime `field:"update_time"` //
ErrorMsg sql.NullString ErrorMsg *sql.NullString `field:"error_msg"`
Operator int32 Operator int32 `field:"operator"`
LastOperator int32 LastOperator int32
} }
@ -72,24 +71,24 @@ func (ai *Streamer) Set(field string, value interface{}) {
} }
type CollectLog struct { type CollectLog struct {
LogUid int64 // 日志id LogUid int64 `field:"log_uid"` // 日志id
StreamerUid int64 // StreamerId 表id与 StreamerUid int64 `field:"streamer_uid"` // StreamerId 表id与
Platform Platform // Platform Platform `field:"platform"` //
UserId string // 平台的UserId UserId string `field:"user_id"` // 平台的UserId
IsLiveStreaming bool // IsLiveStreaming bool `field:"is_live_streaming"` //
IsError bool // IsError bool `field:"is_error"` //
Followers sql.NullInt64 // Followers *sql.NullInt64 `field:"followers"` //
Views sql.NullInt64 // Views *sql.NullInt64 `field:"views"` //
Giver interface{} // Giver interface{} `field:"giver"` //
Gratuity sql.NullInt64 // Gratuity *sql.NullInt64 `field:"gratuity"` //
LiveTitle sql.NullString // LiveTitle *sql.NullString `field:"live_title"` //
LiveStartTime sql.NullTime // LiveStartTime *sql.NullTime `field:"live_start_time"` //
LiveEndTime sql.NullTime // LiveEndTime *sql.NullTime `field:"live_end_time"` //
UpdateTime sql.NullTime // UpdateTime *sql.NullTime `field:"update_time"` //
Tags interface{} Tags interface{} `field:"tags"`
Ext interface{} // Ext interface{} `field:"ext"` //
ErrorMsg sql.NullString // ErrorMsg *sql.NullString `field:"error_msg"` //
} }
// Get Simple Value // Get Simple Value

9
go.mod
View File

@ -3,15 +3,16 @@ module intimate
go 1.15 go 1.15
require ( require (
github.com/474420502/extractor v0.9.5 github.com/474420502/extractor v0.9.6
github.com/474420502/focus v0.12.0 github.com/474420502/focus v0.12.0
github.com/474420502/gcurl v0.2.0 github.com/474420502/gcurl v0.4.5
github.com/474420502/requests v1.7.0 github.com/474420502/requests v1.9.1
github.com/davecgh/go-spew v1.1.1
github.com/go-sql-driver/mysql v1.5.0 github.com/go-sql-driver/mysql v1.5.0
github.com/lestrrat-go/libxml2 v0.0.0-20200215080510-6483566f52cb github.com/lestrrat-go/libxml2 v0.0.0-20200215080510-6483566f52cb
github.com/tebeka/selenium v0.9.9 github.com/tebeka/selenium v0.9.9
github.com/tidwall/gjson v1.6.0 github.com/tidwall/gjson v1.6.0
github.com/tidwall/pretty v1.0.1 // indirect github.com/tidwall/pretty v1.0.1 // indirect
golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect golang.org/x/net v0.0.0-20200822124328-c89045814202 // indirect
gopkg.in/yaml.v2 v2.3.0 gopkg.in/yaml.v2 v2.3.0
) )

22
go.sum
View File

@ -2,20 +2,20 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU= cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU=
cloud.google.com/go v0.41.0/go.mod h1:OauMR7DV8fzvZIl2qg6rkaIhD/vmgk4iwEw/h6ercmg= cloud.google.com/go v0.41.0/go.mod h1:OauMR7DV8fzvZIl2qg6rkaIhD/vmgk4iwEw/h6ercmg=
github.com/474420502/extractor v0.9.5 h1:nM3/1tNL8BHS3PN9eXvm8Ve3hqTReKNB+ba8VpdL8bw= github.com/474420502/extractor v0.9.6 h1:mwwzwVeu/sZ4FV190Bl32ta4iVjERBWL6eGLXRDlAtg=
github.com/474420502/extractor v0.9.5/go.mod h1:vJnXWmvO5bJDW4Yag0GoE2GxtHRg03TAxp2oXN1DcSY= github.com/474420502/extractor v0.9.6/go.mod h1:vJnXWmvO5bJDW4Yag0GoE2GxtHRg03TAxp2oXN1DcSY=
github.com/474420502/focus v0.12.0 h1:+icbmj7IEOefvTegHt5EpcHt6WFbe2miIrceUJx2Evo= github.com/474420502/focus v0.12.0 h1:+icbmj7IEOefvTegHt5EpcHt6WFbe2miIrceUJx2Evo=
github.com/474420502/focus v0.12.0/go.mod h1:d0PMjtMxFz1a9HIhwyFPkWa+JF+0LgOrEUfd8iZka6s= 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.4.4 h1:ZILu7RRjDBGHpTGmuWGKf1NZZbZsC7AHPlI8RHqs9As=
github.com/474420502/gcurl v0.1.2/go.mod h1:hws5q/Ao64bXLLDnldz9VyTQUndTWc/i5DzdEazFfoM= github.com/474420502/gcurl v0.4.4/go.mod h1:7w4knyVJa1ia4I1xd0krG51fKLGwMmNn5sfG2zPWbqM=
github.com/474420502/gcurl v0.2.0 h1:m6+vw4NX4f5Tfp7c3nuaIgHUE/7zTX6K3xK+pTCBoCo= github.com/474420502/gcurl v0.4.5 h1:4y+NbbBGRIWDpfe/iojdSUzlShcZmnkHV4T4etiWQsw=
github.com/474420502/gcurl v0.2.0/go.mod h1:kJZDbgXn5wbAaR+hhBi4Sbw44P4igJ7qYXC6mejLuhQ= github.com/474420502/gcurl v0.4.5/go.mod h1:7w4knyVJa1ia4I1xd0krG51fKLGwMmNn5sfG2zPWbqM=
github.com/474420502/htmlquery v1.2.4-0.20200812072201-e871dd09247a h1:E1T6CYQKsUn7fMvNbeKfISjBLfOJjZX4KpWwStT20Kc= github.com/474420502/htmlquery v1.2.4-0.20200812072201-e871dd09247a h1:E1T6CYQKsUn7fMvNbeKfISjBLfOJjZX4KpWwStT20Kc=
github.com/474420502/htmlquery v1.2.4-0.20200812072201-e871dd09247a/go.mod h1:AoSN890esHwNKecV0tCs+W0ele1xgFL1Jqk6UcrdxgU= github.com/474420502/htmlquery v1.2.4-0.20200812072201-e871dd09247a/go.mod h1:AoSN890esHwNKecV0tCs+W0ele1xgFL1Jqk6UcrdxgU=
github.com/474420502/requests v1.6.0 h1:f4h4j40eT0P5whhg9LdkotD8CaKjtuDu/vz9iSUkCgY=
github.com/474420502/requests v1.6.0/go.mod h1:SLXrQ5dL9c7dkIeKNUCBAjOIt3J9KFCS2RQjWJecNwo=
github.com/474420502/requests v1.7.0 h1:oaBwVrxZ7yZ+hDOKwHm2NflYib2y1geIUxBxQ2U48mw= github.com/474420502/requests v1.7.0 h1:oaBwVrxZ7yZ+hDOKwHm2NflYib2y1geIUxBxQ2U48mw=
github.com/474420502/requests v1.7.0/go.mod h1:SLXrQ5dL9c7dkIeKNUCBAjOIt3J9KFCS2RQjWJecNwo= github.com/474420502/requests v1.7.0/go.mod h1:SLXrQ5dL9c7dkIeKNUCBAjOIt3J9KFCS2RQjWJecNwo=
github.com/474420502/requests v1.9.1 h1:gCDmBCW1ym8zOvKKBGjEG9wafMY7imYk2u28+Oy1WEc=
github.com/474420502/requests v1.9.1/go.mod h1:x2T9l+e40R6kxxMvNm+YSZ9D6BHAXUDak4kQElIPJ9A=
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ= github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802 h1:1BDTz0u9nC3//pOCMdNH+CiXJVYJh5UQNCOBG7jbELc= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802 h1:1BDTz0u9nC3//pOCMdNH+CiXJVYJh5UQNCOBG7jbELc=
@ -24,8 +24,6 @@ github.com/BurntSushi/xgbutil v0.0.0-20160919175755-f7c97cef3b4e h1:4ZrkT/RzpnRO
github.com/BurntSushi/xgbutil v0.0.0-20160919175755-f7c97cef3b4e/go.mod h1:uw9h2sd4WWHOPdJ13MQpwK5qYWKYDumDqxWWIknEQ+k= github.com/BurntSushi/xgbutil v0.0.0-20160919175755-f7c97cef3b4e/go.mod h1:uw9h2sd4WWHOPdJ13MQpwK5qYWKYDumDqxWWIknEQ+k=
github.com/Pallinder/go-randomdata v1.1.0 h1:gUubB1IEUliFmzjqjhf+bgkg1o6uoFIkRsP3VrhEcx8= github.com/Pallinder/go-randomdata v1.1.0 h1:gUubB1IEUliFmzjqjhf+bgkg1o6uoFIkRsP3VrhEcx8=
github.com/Pallinder/go-randomdata v1.1.0/go.mod h1:yHmJgulpD2Nfrm0cR9tI/+oAgRqCQQixsA8HyRZfV9Y= github.com/Pallinder/go-randomdata v1.1.0/go.mod h1:yHmJgulpD2Nfrm0cR9tI/+oAgRqCQQixsA8HyRZfV9Y=
github.com/Pallinder/go-randomdata v1.2.0 h1:DZ41wBchNRb/0GfsePLiSwb0PHZmT67XY00lCDlaYPg=
github.com/Pallinder/go-randomdata v1.2.0/go.mod h1:yHmJgulpD2Nfrm0cR9tI/+oAgRqCQQixsA8HyRZfV9Y=
github.com/antchfx/xpath v1.1.6 h1:6sVh6hB5T6phw1pFpHRQ+C4bd8sNI+O58flqtg7h0R0= github.com/antchfx/xpath v1.1.6 h1:6sVh6hB5T6phw1pFpHRQ+C4bd8sNI+O58flqtg7h0R0=
github.com/antchfx/xpath v1.1.6/go.mod h1:Yee4kTMuNiPYJ7nSNorELQMr1J33uOpXDMByNYhvtNk= github.com/antchfx/xpath v1.1.6/go.mod h1:Yee4kTMuNiPYJ7nSNorELQMr1J33uOpXDMByNYhvtNk=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio= github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
@ -116,8 +114,8 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e h1:3G+cUijn7XD+S4eJFddp53Pv7+slrESplyjG25HgL+k= golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e h1:3G+cUijn7XD+S4eJFddp53Pv7+slrESplyjG25HgL+k=
golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200421231249-e086a090c8fd/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= golang.org/x/net v0.0.0-20200421231249-e086a090c8fd/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU= golang.org/x/net v0.0.0-20200822124328-c89045814202 h1:VvcQYSHwXgi7W+TpUR6A9g6Up98WAHf3f/ulnJ62IyA=
golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=

3
platform_list.go
View File

@ -12,4 +12,7 @@ const (
// Ptwitcasting twitcasting 平台 // Ptwitcasting twitcasting 平台
Ptwitcasting Platform = "twitcasting" Ptwitcasting Platform = "twitcasting"
// PNimo PNimo 平台
PNimo Platform = "nimo"
) )

79
store.go
View File

@ -228,15 +228,14 @@ func NewStoreExtractor() *StoreExtractor {
return &StoreExtractor{db: db} return &StoreExtractor{db: db}
} }
// Pop 弹出一条未处理的数据 // PopNoWait 弹出一个不用按时间间隔更新的主播信息, 主要用来测试.
func (store *StoreExtractor) Pop(platform Platform, operators ...int32) (*Streamer, error) { func (store *StoreExtractor) PopNoWait(platform Platform, condition string, operators ...int32) (*Streamer, error) {
tx, err := store.db.Begin() tx, err := store.db.Begin()
if err != nil { if err != nil {
return nil, err return nil, err
} }
var args = []interface{}{string(platform)} 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` 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 { if len(operators) == 0 {
selectSQL += " and operator = ?" selectSQL += " and operator = ?"
args = append(args, 0) args = append(args, 0)
@ -264,7 +263,52 @@ func (store *StoreExtractor) Pop(platform Platform, operators ...int32) (*Stream
s := &Streamer{} s := &Streamer{}
// uid, url, target_type, source, ext, operator // uid, url, target_type, source, ext, operator
err = row.Scan(&s.Uid, &s.UpdateTime, &s.UserId, &s.UpdateUrl, &s.IsUpdateStreamer, &s.UpdateInterval) 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 { if err != nil {
return nil, err return nil, err
} }
@ -316,9 +360,9 @@ func (store *StoreExtractor) InsertStreamerList(streamerlist IGet) (isExists boo
} }
// InsertStreamer Streamer表, 插入数据 // InsertStreamer Streamer表, 插入数据
func (store *StoreExtractor) InsertStreamer(streamer IGet) (isExists bool) { func (store *StoreExtractor) InsertStreamer(streamer *Streamer) (isExists bool) {
// select uid from table where platform = ? and user_id = ? // select uid from table where platform = ? and user_id = ?
selectSQL := "SELECT is_update_url, uid FROM " + StreamerTable + " WHERE platform = ? AND user_id = ?" // selectSQL := "SELECT is_update_url, uid FROM " + StreamerTable + " WHERE platform = ? AND user_id = ?"
tx, err := store.db.Begin() tx, err := store.db.Begin()
if err != nil { if err != nil {
panic(err) panic(err)
@ -335,18 +379,17 @@ func (store *StoreExtractor) InsertStreamer(streamer IGet) (isExists bool) {
} }
}() }()
row := tx.QueryRow(selectSQL+` LIMIT 1 FOR UPDATE`, streamer.Get("Platform"), streamer.Get("UserId")) streamer.UpdateTime = &sql.NullTime{Time: time.Now().Add(-time.Hour * 100000), Valid: true}
var isUpdateUrl bool _, err = tx.Exec("INSERT IGNORE INTO "+StreamerTable+"(platform, user_id, user_name, live_url, update_url, tags, update_time) VALUES(?,?,?,?,?,?,?);",
var Uid int64 streamer.Platform,
if err = row.Scan(&isUpdateUrl, &Uid); err == nil { streamer.UserId,
if isUpdateUrl { streamer.UserName,
tx.Exec("UPDATE "+StreamerTable+" SET update_url = ?", streamer.Get("UpdateUrl")) streamer.LiveUrl,
} streamer.UpdateUrl,
streamer.(ISet).Set("Uid", Uid) streamer.Tags,
return true streamer.UpdateTime,
} )
_, err = tx.Exec("INSERT INTO "+StreamerTable+"(platform, user_id, update_url, tags, update_time) VALUES(?,?,?,?,?);", streamer.Get("Platform"), streamer.Get("UserId"), streamer.Get("UpdateUrl"), streamer.Get("Tags"), time.Now().Add(-time.Hour*100000))
if err != nil { if err != nil {
panic(err) panic(err)
} }

1
supervisor_conf/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
conf.d

7
supervisor_conf/clear_log.sh Normal file
View File

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

9
supervisor_conf/createln.sh
View File

@ -1 +1,8 @@
ln -sf `pwd`/*.conf /etc/supervisor/conf.d/
CURPATH=`pwd`
SUPPATH=$(dirname "$CURPATH")
mkdir -p $CURPATH/conf.d
cp *.conf $CURPATH/conf.d/
sed -i "s#MYPATH#$SUPPATH#g" $CURPATH/conf.d/*.conf
ln -sf $CURPATH/conf.d/*.conf /etc/supervisor/conf.d/

10
supervisor_conf/nimo_extractor.conf Normal file
View File

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

13
supervisor_conf/nimo_task1.conf Normal file
View File

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

6
supervisor_conf/openrec_extractor.conf
View File

@ -2,9 +2,9 @@
nodaemon=true nodaemon=true
[program:openrec_extractor] [program:openrec_extractor]
directory = /home/eson/test/intimate/bin/openrec_extractor/ directory = MYPATH/bin/openrec_extractor/
command= /home/eson/test/intimate/bin/openrec_extractor/openrec_extractor command= MYPATH/bin/openrec_extractor/openrec_extractor
autorestart=true autorestart=true
stderr_logfile=/home/eson/test/intimate/bin/openrec_extractor/log stderr_logfile=MYPATH/bin/openrec_extractor/log
stderr_logfile_maxbytes=0 stderr_logfile_maxbytes=0
stopsignal=QUIT stopsignal=QUIT

6
supervisor_conf/openrec_task1.conf
View File

@ -2,9 +2,9 @@
nodaemon=true nodaemon=true
[program:openrec_ranking] [program:openrec_ranking]
directory = /home/eson/test/intimate/bin/openrec_task1 directory = MYPATH/bin/openrec_task1
command= /home/eson/test/intimate/bin/openrec_task1/openrec_task1 command= MYPATH/bin/openrec_task1/openrec_task1
autorestart=true autorestart=true
stderr_logfile=/home/eson/test/intimate/bin/openrec_task1/log stderr_logfile=MYPATH/bin/openrec_task1/log
stderr_logfile_maxbytes=0 stderr_logfile_maxbytes=0
stopsignal=QUIT stopsignal=QUIT

12
supervisor_conf/openrec_task2.conf
View File

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

12
supervisor_conf/test.conf
View File

@ -1,12 +0,0 @@
# [supervisord]
# nodaemon=true
# [program:openrec_extractor]
# directory = /home/eson/test/intimate/extractor/openrec_extractor
# command= /home/eson/test/intimate/extractor/openrec_extractor/openrec_extractor
# autorestart=true
# # stdout_logfile=/home/eson/test/intimate/extractor/openrec/stdout
# # stdout_logfile_maxbytes=0
# stderr_logfile=/home/eson/test/intimate/extractor/openrec/log
# stderr_logfile_maxbytes=0
# stopsignal=QUIT

6
supervisor_conf/twitcasting_extractor.conf
View File

@ -3,12 +3,12 @@ nodaemon=false
[program:twitcasting_extractor] [program:twitcasting_extractor]
environment=DISPLAY=":99" environment=DISPLAY=":99"
directory = /home/eson/test/intimate/bin/twitcasting_extractor/ directory = MYPATH/bin/twitcasting_extractor/
command= /home/eson/test/intimate/bin/twitcasting_extractor/twitcasting_extractor command= MYPATH/bin/twitcasting_extractor/twitcasting_extractor
process_name=%(program_name)s_%(process_num)02d ;多进程名称 process_name=%(program_name)s_%(process_num)02d ;多进程名称
numprocs=3 ;启动多个进程 numprocs=3 ;启动多个进程
autorestart=true autorestart=true
stderr_logfile=/home/eson/test/intimate/bin/twitcasting_extractor/log stderr_logfile=MYPATH/bin/twitcasting_extractor/log
# stderr_logfile=%(supervisorctl.var.directory)s/log # stderr_logfile=%(supervisorctl.var.directory)s/log
stderr_logfile_maxbytes=0 stderr_logfile_maxbytes=0
stopsignal=QUIT stopsignal=QUIT

6
supervisor_conf/twitcasting_task1.conf
View File

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

8
supervisor_conf/twitch_extractor.conf
View File

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

6
supervisor_conf/twitch_extractor_p1.conf → supervisor_conf/twitch_extractor_p1.conf_eg
View File

@ -3,11 +3,11 @@ nodaemon=true
[program:twitch_extractor_p1] [program:twitch_extractor_p1]
environment=DISPLAY=":99",pac_proxy=http://localhost:1090/pac environment=DISPLAY=":99",pac_proxy=http://localhost:1090/pac
directory = /home/eson/test/intimate/bin/twitch_extractor directory = MYPATH/bin/twitch_extractor
command= /home/eson/test/intimate/bin/twitch_extractor/twitch_extractor command= MYPATH/bin/twitch_extractor/twitch_extractor
process_name=%(program_name)s_%(process_num)02d ;多进程名称 process_name=%(program_name)s_%(process_num)02d ;多进程名称
numprocs=2 ;启动多个进程 numprocs=2 ;启动多个进程
autorestart=true autorestart=true
stderr_logfile=/home/eson/test/intimate/bin/twitch_extractor/log stderr_logfile=MYPATH/bin/twitch_extractor/log
stderr_logfile_maxbytes=0 stderr_logfile_maxbytes=0
stopsignal=QUIT stopsignal=QUIT

13
supervisor_conf/twitch_extractor_p2.conf
View File

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

13
supervisor_conf/twitch_task1.conf Normal file
View File

@ -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

13
supervisor_conf/twitch_task2.conf
View File

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

7
supervisor_conf/xvfb.conf
View File

@ -3,11 +3,6 @@
nodaemon=true nodaemon=true
[program:xvfb-99] [program:xvfb-99]
# directory = /home/eson/test/intimate/extractor/openrec_extractor
command=/usr/bin/Xvfb :99 -screen 0 1280x720x24 -ac -nolisten tcp -dpi 96 +extension RANDR -nolisten tcp command=/usr/bin/Xvfb :99 -screen 0 1280x720x24 -ac -nolisten tcp -dpi 96 +extension RANDR -nolisten tcp
autorestart=true autorestart=true
# # stdout_logfile=/home/eson/test/intimate/extractor/openrec/stdout
# # stdout_logfile_maxbytes=0
# stderr_logfile=/home/eson/test/intimate/extractor/openrec/log
# stderr_logfile_maxbytes=0
# stopsignal=QUIT

3
table_list.go
View File

@ -12,4 +12,7 @@ const (
// STTwitcasting STTwitcasting源table名称 // STTwitcasting STTwitcasting源table名称
STTwitcasting SourceTable = "source_twitcasting" STTwitcasting SourceTable = "source_twitcasting"
// STNimo nimo源table名称
STNimo SourceTable = "source_nimo"
) )

0
tasks/openrec/openrec_task2/main.go → tasks/nimo/nimo_task1/main.go
View File

109
tasks/nimo/nimo_task1/nimo_task1.go Normal file
View File

@ -0,0 +1,109 @@
package main
import (
"database/sql"
"encoding/json"
"intimate"
"log"
"time"
"github.com/474420502/gcurl"
"github.com/tidwall/gjson"
)
// estore 解析存储连接实例
// var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
// Execute 执行
func Execute() {
nimoapi := `curl 'https://api.nimo.tv/oversea/nimo/api/v2/liveRoom/liveRoomPage-1-100-/HK/1028/1000' \
-H 'authority: api.nimo.tv' \
-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: multipart/form-data; boundary=----WebKitFormBoundary3bCA1lzvhj4kBR4Q' \
-H 'accept: */*' \
-H 'origin: https://www.nimo.tv' \
-H 'sec-fetch-site: same-site' \
-H 'sec-fetch-mode: cors' \
-H 'sec-fetch-dest: empty' \
-H 'referer: https://www.nimo.tv/lives' \
-H 'accept-language: zh-CN,zh;q=0.9' \
--data-binary $'------WebKitFormBoundary3bCA1lzvhj4kBR4Q\r\nContent-Disposition: form-data; name="keyType"\r\n\r\n0\r\n------WebKitFormBoundary3bCA1lzvhj4kBR4Q\r\nContent-Disposition: form-data; name="body"\r\n\r\n{"deviceType":7,"requestSource":"WEB","iNetType":5}\r\n------WebKitFormBoundary3bCA1lzvhj4kBR4Q--\r\n' \
--compressed`
curl := gcurl.Parse(nimoapi)
ses := curl.CreateSession()
tp := curl.CreateTemporary(ses)
param := tp.PathParam("liveRoomPage-(1)-")
ps := intimate.NewPerfectShutdown()
for !ps.IsClose() {
resp, err := tp.Execute()
if err != nil {
log.Println(err)
time.Sleep(time.Second)
continue
}
result := gjson.ParseBytes(resp.Content())
roomlist := result.Get("data.result.liveRoomViewList")
if !roomlist.IsArray() {
log.Println("json is error")
log.Println(string(resp.Content()))
break
}
rooms := roomlist.Array()
log.Println(tp.GetURLRawPath(), "rooms:", len(rooms))
if len(rooms) == 0 {
param.IntSet(1)
time.Sleep(time.Minute * 15)
continue
}
for _, room := range rooms {
streamer := &intimate.Streamer{}
streamer.Platform = intimate.PNimo
if userid := room.Get("id").String(); userid != "" {
streamer.UserId = &userid
streamer.LiveUrl = &sql.NullString{String: "https://www.nimo.tv/live/" + userid, Valid: true}
channel := room.Get("roomTypeName").String()
streamer.Channel = &sql.NullString{String: channel, Valid: channel != ""}
username := room.Get("anchorName").String()
streamer.UserName = &sql.NullString{String: username, Valid: username != ""}
if rtags := room.Get("anchorLabels"); rtags.IsArray() {
var tags []string
for _, r := range rtags.Array() {
tag := r.Get("labelName").String()
tags = append(tags, tag)
}
data, err := json.Marshal(tags)
if err != nil {
panic(err)
}
streamer.Tags = data
}
streamer.UpdateInterval = 120
err = intimate.TStreamer.Insert(streamer)
if err != nil {
panic(err)
}
} else {
log.Println("userid is null.", room.String())
}
}
param.IntAdd(1)
}
}

55
tasks/nimo/nimo_task1/nimo_task1_test.go Normal file
View File

@ -0,0 +1,55 @@
package main
import (
"fmt"
"io/ioutil"
"os"
"testing"
"github.com/474420502/gcurl"
)
func CrawlContent(args ...interface{}) []byte {
nimoapi := `curl 'https://api.nimo.tv/oversea/nimo/api/v2/liveRoom/liveRoomPage-%d-30-/HK/1028/1000' \
-H 'authority: api.nimo.tv' \
-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: multipart/form-data; boundary=----WebKitFormBoundary3bCA1lzvhj4kBR4Q' \
-H 'accept: */*' \
-H 'origin: https://www.nimo.tv' \
-H 'sec-fetch-site: same-site' \
-H 'sec-fetch-mode: cors' \
-H 'sec-fetch-dest: empty' \
-H 'referer: https://www.nimo.tv/lives' \
-H 'accept-language: zh-CN,zh;q=0.9' \
--data-binary $'------WebKitFormBoundary3bCA1lzvhj4kBR4Q\r\nContent-Disposition: form-data; name="keyType"\r\n\r\n0\r\n------WebKitFormBoundary3bCA1lzvhj4kBR4Q\r\nContent-Disposition: form-data; name="body"\r\n\r\n{"deviceType":7,"requestSource":"WEB","iNetType":5}\r\n------WebKitFormBoundary3bCA1lzvhj4kBR4Q--\r\n' \
--compressed`
curl := gcurl.Parse(fmt.Sprintf(nimoapi, 1))
tp := curl.CreateTemporary(nil)
resp, err := tp.Execute()
if err != nil {
panic(err)
}
return resp.Content()
}
func openTestFile(...interface{}) []byte {
f, err := os.Open("../../../testfile/nimo1.json")
if err != nil {
panic(err)
}
data, err := ioutil.ReadAll(f)
if err != nil {
panic(err)
}
return data
}
var Crawl func(...interface{}) []byte
func Test(t *testing.T) {
Execute()
}

17
tasks/openrec/openrec_task1/task_openrec.go
View File

@ -11,11 +11,11 @@ import (
"github.com/tidwall/gjson" "github.com/tidwall/gjson"
) )
// sstore 源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql // // sstore 源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec)) // var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec))
// estore 解析存储连接实例 // // estore 解析存储连接实例
var estore *intimate.StoreExtractor = intimate.NewStoreExtractor() // var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
// Execute 执行方法 // Execute 执行方法
func Execute() { func Execute() {
@ -34,7 +34,7 @@ func Execute() {
-H 'if-none-match: W/"25edb-aUYBdmLqZcr6DW4ZWKX9r2aqolg"' \ -H 'if-none-match: W/"25edb-aUYBdmLqZcr6DW4ZWKX9r2aqolg"' \
--compressed` --compressed`
g := gcurl.ParseRawCURL(turl) g := gcurl.Parse(turl)
tp := g.Temporary() tp := g.Temporary()
for !ps.IsClose() { for !ps.IsClose() {
@ -71,7 +71,7 @@ func Execute() {
userid := User.Get("channel.id").String() userid := User.Get("channel.id").String()
streamer := &intimate.Streamer{} streamer := &intimate.Streamer{}
streamer.UserId = userid streamer.UserId = &userid
streamer.Platform = intimate.Popenrec streamer.Platform = intimate.Popenrec
updateUrl := make(map[string]interface{}) updateUrl := make(map[string]interface{})
@ -83,15 +83,16 @@ func Execute() {
updateUrlBytes, err := json.Marshal(updateUrl) updateUrlBytes, err := json.Marshal(updateUrl)
if err != nil { if err != nil {
estore.UpdateError(streamer, err) intimate.TStreamer.UpdateError(streamer, err)
continue continue
} }
streamer.UpdateUrl = updateUrlBytes streamer.UpdateUrl = updateUrlBytes
estore.InsertStreamer(streamer) intimate.TStreamer.Insert(streamer)
} }
} }
log.Println("streamer count:", len(result.Array()), tp.ParsedURL.String())
// 修改url query 参数的page递增. 遍历所有页面 // 修改url query 参数的page递增. 遍历所有页面
tp.QueryParam("page").IntAdd(1) tp.QueryParam("page").IntAdd(1)
time.Sleep(time.Second * 1) time.Sleep(time.Second * 1)

2
tasks/openrec/openrec_task2/.gitignore vendored
View File

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

164
tasks/openrec/openrec_task2/task_openrec.go
View File

@ -1,164 +0,0 @@
package main
import (
"database/sql"
"encoding/json"
"intimate"
"log"
"time"
"github.com/474420502/gcurl"
"github.com/474420502/requests"
"github.com/tidwall/gjson"
)
// sstore 源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_source.sql
var sstore *intimate.StoreSource = intimate.NewStoreSource(string(intimate.STOpenrec))
// estore 源存储实例, 为存储源数据的实现. 表格具体参考sql/intimate_extractor.sql
var estore *intimate.StoreExtractor = intimate.NewStoreExtractor()
func init() {
}
// Execute 执行方法
func Execute() {
ps := intimate.NewPerfectShutdown()
ses := requests.NewSession()
var lasterr error = nil
for !ps.IsClose() {
streamer, err := estore.Pop(intimate.Popenrec) //队列里弹出一个streamer行. 进行解析
if streamer == nil || err != nil {
if err != lasterr {
log.Println(err, lasterr)
lasterr = err
}
time.Sleep(time.Second * 2)
continue
}
userId := streamer.UserId
var updateUrl map[string]string
err = json.Unmarshal(streamer.UpdateUrl.([]byte), &updateUrl) // 反序列化update_url, 里面存了需要采集的url
if err != nil {
log.Println(err)
continue
}
// Check Userid
userUrl := updateUrl["user"]
log.Println(userUrl)
tp := ses.Get(userUrl) // 获取user url页面数据
resp, err := tp.Execute()
streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
if err != nil {
log.Println(err)
estore.UpdateError(streamer, err)
continue
}
cookies := ses.GetCookies(tp.GetParsedURL())
scurl := updateUrl["supporters"] //获取打赏者的数据
curl := gcurl.ParseRawCURL(scurl)
supportersSession := curl.CreateSession()
temporary := curl.CreateTemporary(supportersSession)
supportersSession.SetCookies(temporary.GetParsedURL(), cookies)
var supporters []string
for { // supporters 数据需要登录信息. 下面为赋值 supporters链接获取的uid token random码
supportersQuery := temporary.GetQuery()
for _, cookie := range cookies {
if cookie.Name == "uuid" {
supportersQuery.Set("Uuid", cookie.Value)
continue
}
if cookie.Name == "token" {
supportersQuery.Set("Token", cookie.Value)
continue
}
if cookie.Name == "random" {
supportersQuery.Set("Random", cookie.Value)
continue
}
}
supportersQuery.Set("identify_id", userId)
temporary.SetQuery(supportersQuery)
resp, err := temporary.Execute()
if err != nil {
log.Println(err)
}
supporterjson := gjson.ParseBytes(resp.Content())
supporterdata := supporterjson.Get("data") //解析supporters获取的json数据
if supporterdata.Type == gjson.Null {
break
}
supporters = append(supporters, string(resp.Content()))
temporary.QueryParam("page_number").IntAdd(1)
// page := supportersQuery.Get("page_number") // page_number 加1
// pageint, err := strconv.Atoi(page)
// if err != nil {
// log.Println(err)
// break
// }
// pageint++
// page = strconv.Itoa(pageint)
// supportersQuery.Set("page_number", page)
// temporary.SetQuery(supportersQuery)
}
// cookies := cxt.Session().GetCookies(wf.GetParsedURL())
ext := make(map[string]interface{})
ext["json_supporters"] = supporters
ext["html_user"] = string(resp.Content())
liveUrl := updateUrl["live"]
tp = ses.Get(liveUrl)
resp, err = tp.Execute()
if err != nil {
log.Println(err)
estore.UpdateError(streamer, err)
continue
}
ext["html_live"] = string(resp.Content())
ext["var_user_id"] = userId
extJsonBytes, err := json.Marshal(ext)
if err != nil {
log.Println(err)
estore.UpdateError(streamer, err)
continue
}
// streamer.Platform = intimate.Popenrec
streamer.UpdateInterval = 120
streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true}
streamer.Operator = 0
source := &intimate.Source{}
source.Target = intimate.TOpenrecUser
source.Ext = string(extJsonBytes)
source.StreamerId = sql.NullInt64{Int64: streamer.Uid, Valid: true}
sstore.Insert(source)
estore.UpdateStreamer(streamer)
}
}

2
tasks/twitcasting/twitcasting_task1/main_test.go
View File

@ -21,5 +21,5 @@ func TestUpdateTime(t *testing.T) {
} }
func TestMain(t *testing.T) { func TestMain(t *testing.T) {
main()
} }

27
tasks/twitcasting/twitcasting_task1/twitcasting.go
View File

@ -70,13 +70,15 @@ func Execute() {
if ok := queuedict[wurl]; !ok { if ok := queuedict[wurl]; !ok {
log.Println(wurl) log.Println(wurl)
sl := &intimate.StreamerList{} sl := &intimate.StreamerList{}
sl.Platform = intimate.Ptwitcasting sl.Platform = string(intimate.Ptwitcasting)
sl.Url = wurl sl.Url = wurl
sl.Operator = 0 sl.Operator = 0
sl.UpdateInterval = 120 sl.UpdateInterval = 120
sl.UpdateTime = time.Now() sl.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
sl.UrlHash = intimate.GetUrlHash(sl.Url)
estore.InsertStreamerList(sl) intimate.TStreamerList.Insert(sl)
// estore.InsertStreamerList(sl)
queue.Put(wurl) queue.Put(wurl)
queuedict[wurl] = true queuedict[wurl] = true
@ -102,12 +104,13 @@ func Execute() {
sp.TagUrl[i] = wurl sp.TagUrl[i] = wurl
if ok := queuedict[wurl]; !ok { if ok := queuedict[wurl]; !ok {
sl := &intimate.StreamerList{} sl := &intimate.StreamerList{}
sl.Platform = intimate.Ptwitcasting sl.Platform = string(intimate.Ptwitcasting)
sl.Url = wurl sl.Url = wurl
sl.Operator = 0 sl.Operator = 0
sl.UpdateInterval = 120 sl.UpdateInterval = 120
sl.UpdateTime = time.Now() sl.UpdateTime = &sql.NullTime{Time: time.Now(), Valid: true}
estore.InsertStreamerList(sl) sl.UrlHash = intimate.GetUrlHash(sl.Url)
intimate.TStreamerList.Insert(sl)
queue.Put(wurl) queue.Put(wurl)
queuedict[wurl] = true queuedict[wurl] = true
@ -122,17 +125,19 @@ func Execute() {
// log.Println(sp) // log.Println(sp)
streamer := &intimate.Streamer{} streamer := &intimate.Streamer{}
streamer.Platform = intimate.Ptwitcasting streamer.Platform = intimate.Ptwitcasting
streamer.LiveUrl = sql.NullString{String: sp.LiveUrl, Valid: true} streamer.LiveUrl = &sql.NullString{String: sp.LiveUrl, Valid: true}
if btags, err := json.Marshal(sp.Tag); err != nil { if btags, err := json.Marshal(sp.Tag); err != nil {
log.Println(err) log.Println(err)
} else { } else {
streamer.Tags = btags streamer.Tags = btags
} }
streamer.UpdateInterval = 120 streamer.UpdateInterval = 120
streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true} streamer.UpdateTime = intimate.GetUpdateTimeNow()
streamer.UserName = sql.NullString{String: sp.UserName, Valid: true} streamer.UserName = &sql.NullString{String: sp.UserName, Valid: true}
streamer.UserId = sp.UserId streamer.UserId = &sp.UserId
estore.InsertStreamer(streamer) streamer.Operator = 0
// estore.InsertStreamer(streamer)
intimate.TStreamer.Insert(streamer)
} }
log.Println("finish remain", queue.Size()) log.Println("finish remain", queue.Size())

169
tasks/twitch/twitch_task1/task_twitch.go
View File

@ -1,7 +1,6 @@
package main package main
import ( import (
"database/sql"
"intimate" "intimate"
"log" "log"
"time" "time"
@ -9,100 +8,120 @@ import (
"github.com/tebeka/selenium" "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()
// 获取类型的所有频道链接 // 获取类型的所有频道链接
// Execute 执行任务 // Execute 执行任务
func Execute() { func Execute() {
var err error
wd := intimate.GetChromeDriver(3030)
ps := intimate.NewPerfectShutdown() ps := intimate.NewPerfectShutdown()
weburl := "https://www.twitch.tv/directory?sort=VIEWER_COUNT" for !ps.IsClose() {
err = wd.Get(weburl) var err error
if err != nil { wd := intimate.GetChromeDriver(3030)
panic(err)
}
cardCondition := func(wd selenium.WebDriver) (bool, error) { weburl := "https://www.twitch.tv/directory?sort=VIEWER_COUNT"
elements, err := wd.FindElements(selenium.ByXPATH, "//span/a[contains(@data-a-target,'card-') and @href]") err = wd.Get(weburl)
if err != nil { if err != nil {
return false, err panic(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
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(hrefs) == 0 { cardCondition := func(wd selenium.WebDriver) (bool, error) {
delayerror-- elements, err := wd.FindElements(selenium.ByXPATH, "//span/a[contains(@data-a-target,'card-') and @href]")
if delayerror <= 0 { 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 lasthreflen = 0
var hrefs map[string]bool = make(map[string]bool)
var delayerror = 5
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 break
} }
} else {
delayerror = 5
}
for ii := 0; ii < 10; ii++ { if len(hrefs) == lasthreflen {
for _, card := range cards { delayerror--
href, err := card.GetAttribute("href") if delayerror <= 0 {
if err != nil { break
log.Println(href, err)
continue
} else {
hrefs[href] = true
} }
} else {
delayerror = 7
} }
break lasthreflen = len(hrefs)
}
if ps.IsClose() { for ii := 0; ii < 10; ii++ {
break for _, card := range cards {
} href, err := card.GetAttribute("href")
if err != nil {
log.Println(href, err)
continue
} else {
hrefs[href] = true
}
}
break
}
if len(cards) > 10 { if ps.IsClose() {
log.Println(len(cards)) break
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); }
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) 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)
} }
time.Sleep(time.Millisecond * 200)
wd.KeyDown(selenium.EndKey) for href := range hrefs {
time.Sleep(time.Millisecond * 200)
wd.KeyUp(selenium.EndKey) sl := &intimate.StreamerList{}
time.Sleep(time.Millisecond * 2500) sl.Url = href
sl.UrlHash = intimate.GetUrlHash(sl.Url)
sl.Platform = string(intimate.Ptwitch)
sl.UpdateTime = intimate.GetUpdateTimeNow()
err := intimate.TStreamerList.Insert(sl)
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")
wd.Close()
wd.Quit()
time.Sleep(time.Minute * 30)
} }
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")
} }

2
tasks/twitch/twitch_task1/task_twitch_test.go
View File

@ -4,7 +4,7 @@ import (
"testing" "testing"
) )
func estCase1(t *testing.T) { func TestCase1(t *testing.T) {
Execute() Execute()
} }

2
tasks/twitch/twitch_task2/.gitignore vendored
View File

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

6
tasks/twitch/twitch_task2/main.go
View File

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

172
tasks/twitch/twitch_task2/task_twitch.go
View File

@ -1,172 +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 {
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)
}
}

7
tasks/twitch/twitch_task2/task_twitch_test.go
View File

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

10205
testfile/nimo1.json Normal file
View File

File diff suppressed because it is too large Load Diff

2015
testfile/nimolive.html Normal file
View File

File diff suppressed because one or more lines are too long

47
utils.go
View File

@ -1,6 +1,8 @@
package intimate package intimate
import ( import (
"crypto/md5"
"database/sql"
"fmt" "fmt"
"log" "log"
"os" "os"
@ -28,6 +30,15 @@ func init() {
} }
// GetUpdateTimeNow 获取马上更新时间. 与第一次连用
func GetUpdateTimeNow() *sql.NullTime {
return &sql.NullTime{Time: time.Now().Add(-time.Hour * 100000), Valid: true}
}
func GetUrlHash(urlstr string) string {
return fmt.Sprintf("%x", md5.Sum([]byte(urlstr)))
}
// ParseNumber 去逗号解析数字 // ParseNumber 去逗号解析数字
func ParseNumber(num string) (int64, error) { func ParseNumber(num string) (int64, error) {
num = strings.Trim(num, " ") num = strings.Trim(num, " ")
@ -87,9 +98,12 @@ func GetChromeDriver(port int) selenium.WebDriver {
caps := selenium.Capabilities{"browserName": "chrome"} caps := selenium.Capabilities{"browserName": "chrome"}
chromecaps := chrome.Capabilities{} chromecaps := chrome.Capabilities{}
// chromecaps.AddExtension("/home/eson/test/myblock.crx")
for _, epath := range []string{"../../../crx/myblock.crx", "../../crx/myblock.crx"} { for _, epath := range []string{"../../../crx/myblock.crx", "../../crx/myblock.crx"} {
_, err := os.Stat(epath) _, err := os.Stat(epath)
if err == nil { if err == nil {
err := chromecaps.AddExtension(epath) err := chromecaps.AddExtension(epath)
if err != nil { if err != nil {
panic(err) panic(err)
@ -258,3 +272,36 @@ func (c *Counter) Add(n int) error {
} }
return nil return nil
} }
type WaitFor struct {
WebDriver selenium.WebDriver
}
func NewWaitFor(wd selenium.WebDriver) *WaitFor {
return &WaitFor{WebDriver: wd}
}
func (wf *WaitFor) Default(xpath string, do func(elements ...selenium.WebElement) bool) error {
return wf.WaitWithTimeout(xpath, 15*time.Second, do)
}
func (wf *WaitFor) WaitWithTimeout(xpath string, timeout time.Duration, do func(elements ...selenium.WebElement) bool) error {
return wf.WebDriver.WaitWithTimeout(func(wd selenium.WebDriver) (bool, error) {
elements, err := wd.FindElements(selenium.ByXPATH, xpath)
if err != nil {
log.Println(err)
return false, err
}
if len(elements) > 0 {
if do == nil {
return true, nil
}
if do(elements...) {
return true, nil
}
}
return false, nil
}, timeout)
}