From 2c557e3b42e9fb8c50475f46fde0cf323a5367b8 Mon Sep 17 00:00:00 2001 From: eson Date: Mon, 7 Sep 2020 18:12:18 +0800 Subject: [PATCH] =?UTF-8?q?todo:=20twitcasting=20=E6=B5=8B=E8=AF=95=20stor?= =?UTF-8?q?e.=202.=20=E4=BF=AE=E6=94=B9streamer.=20tag=20=E7=AC=A6?= =?UTF-8?q?=E5=90=88=E8=A6=81=E6=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- autostore.go | 18 +- config.go | 12 + convert.go | 452 +++++++++--------- tasks/nimo/nimo_task1/nimo_task1.go | 2 +- tasks/openrec/openrec_task2/task_openrec.go | 10 - .../twitcasting_task1/main_test.go | 2 +- .../twitcasting_task1/twitcasting.go | 4 +- 7 files changed, 256 insertions(+), 244 deletions(-) diff --git a/autostore.go b/autostore.go index a7a7e65..4f061be 100644 --- a/autostore.go +++ b/autostore.go @@ -10,6 +10,18 @@ import ( "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 { @@ -287,12 +299,12 @@ func assign(field reflect.Value, src interface{}) (bool, error) { 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()) + // 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("type:", fn.Name(), ",kind:", field.Kind(), ",fieldtype:", field.Type()) } // log.Println(field.Kind(), field, reflect.TypeOf(field).Elem().Name(), columntype.ScanType().Kind()) @@ -302,7 +314,6 @@ func convert(src interface{}, field reflect.Value, columntype *sql.ColumnType) e } switch columntype.DatabaseTypeName() { - case "TINYINT", "SMALLINT", "MEDIUMINT", "INT", "BIGINT": isdefault, err := assign(field, src) if err != nil { @@ -387,7 +398,6 @@ func convert(src interface{}, field reflect.Value, columntype *sql.ColumnType) e if err != nil { return err } - return nil } return nil diff --git a/config.go b/config.go index 369ca21..547c5dc 100644 --- a/config.go +++ b/config.go @@ -18,6 +18,18 @@ func init() { // storeOpenrec = NewStore() 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 配置 diff --git a/convert.go b/convert.go index 4e8d149..23bea55 100644 --- a/convert.go +++ b/convert.go @@ -7,13 +7,11 @@ package intimate import ( - "database/sql" "database/sql/driver" "errors" "fmt" "reflect" "strconv" - "time" ) var errNilPtr = errors.New("destination pointer is nil") // embedded in descriptive error @@ -23,240 +21,240 @@ var errNilPtr = errors.New("destination pointer is nil") // embedded in descript // 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. - } +// 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 +// 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 - } +// 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) - } +// 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 - } +// 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) - } +// 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 - } +// 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 - } +// 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 - } - } +// // 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) -} +// 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 { diff --git a/tasks/nimo/nimo_task1/nimo_task1.go b/tasks/nimo/nimo_task1/nimo_task1.go index 827e0bf..96281a3 100644 --- a/tasks/nimo/nimo_task1/nimo_task1.go +++ b/tasks/nimo/nimo_task1/nimo_task1.go @@ -95,8 +95,8 @@ func Execute() { } streamer.UpdateInterval = 120 - // spew.Println(streamer) estore.InsertStreamer(streamer) + } else { log.Println("userid is null.", room.String()) } diff --git a/tasks/openrec/openrec_task2/task_openrec.go b/tasks/openrec/openrec_task2/task_openrec.go index c88fa6f..abfad26 100644 --- a/tasks/openrec/openrec_task2/task_openrec.go +++ b/tasks/openrec/openrec_task2/task_openrec.go @@ -111,16 +111,6 @@ func Execute() { 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()) diff --git a/tasks/twitcasting/twitcasting_task1/main_test.go b/tasks/twitcasting/twitcasting_task1/main_test.go index 18bfb6a..f856c03 100644 --- a/tasks/twitcasting/twitcasting_task1/main_test.go +++ b/tasks/twitcasting/twitcasting_task1/main_test.go @@ -21,5 +21,5 @@ func TestUpdateTime(t *testing.T) { } func TestMain(t *testing.T) { - + main() } diff --git a/tasks/twitcasting/twitcasting_task1/twitcasting.go b/tasks/twitcasting/twitcasting_task1/twitcasting.go index a11e724..e8a133a 100644 --- a/tasks/twitcasting/twitcasting_task1/twitcasting.go +++ b/tasks/twitcasting/twitcasting_task1/twitcasting.go @@ -132,7 +132,9 @@ func Execute() { streamer.UpdateTime = sql.NullTime{Time: time.Now(), Valid: true} streamer.UserName = sql.NullString{String: sp.UserName, Valid: true} streamer.UserId = sp.UserId - estore.InsertStreamer(streamer) + streamer.Operator = 33 + // estore.InsertStreamer(streamer) + intimate.TStreamer.Insert(streamer) } log.Println("finish remain", queue.Size())