package extract import ( "encoding/json" "jy/clear" db "jy/mongodbutil" "jy/pretreated" ju "jy/util" "log" qu "qfw/util" "strings" "sync" "time" "gopkg.in/mgo.v2/bson" ) var ( lock sync.RWMutex cut = ju.NewCut() //获取正文并清理 ExtLogs map[*TaskInfo][]map[string]interface{} //抽取日志 TaskList map[string]*ExtractTask //任务列表 saveLimit = 200 //抽取日志批量保存 Fields = `{"title":1,"detail":1,"contenthtml":1,"href":1,"spidercode":1,"toptype":1,"area":1,"city":1}` ) //启动抽取 func StartExtractTaskId(taskId string) bool { ext := TaskList[taskId] if ext == nil { ext = &ExtractTask{} ext.Id = taskId ext.IsRun = true ext.InitTaskInfo() ext.TaskInfo.DB = db.MgoFactory(1, 3, 120, ext.TaskInfo.FromDbAddr, ext.TaskInfo.FromDB) ext.InitRulePres() ext.InitRuleBacks() ext.InitRuleCore() ext.InitTag() ext.InitClearFn() //只启动一次taskId go RunExtractTask(ext) } ext.IsRun = true TaskList[taskId] = ext return true } //停止抽取 func StopExtractTaskId(taskId string) bool { ext := TaskList[taskId] if ext != nil { ext.IsRun = false TaskList[taskId] = ext } //更新task.s_extlastid db.Mgo.UpdateById("task", taskId, `{"$set":{"s_extlastid":"`+ext.TaskInfo.LastExtId+`"}}`) return true } //开始抽取 func RunExtractTask(ext *ExtractTask) { if !ext.IsRun { return } query := bson.M{"_id": bson.M{"$gt": bson.ObjectIdHex(ext.TaskInfo.LastExtId)}} list, _ := ext.TaskInfo.DB.Find(ext.TaskInfo.FromColl, query, nil, Fields, false, -1, -1) for _, v := range *list { if !ext.IsRun { break } j := PreInfo(v) ext.TaskInfo.ProcessPool <- true go ext.ExtractProcess(j) ext.TaskInfo.LastExtId = qu.BsonIdToSId(v["_id"]) time.Sleep(1 * time.Second) } //更新task.s_extlastid db.Mgo.UpdateById("task", ext.Id, `{"$set":{"s_extlastid":"`+ext.TaskInfo.LastExtId+`"}}`) time.AfterFunc(30*time.Minute, func() { RunExtractTask(ext) }) } //信息预处理 func PreInfo(doc map[string]interface{}) *ju.Job { detail := "" d1, _ := doc["detail"].(string) d2, _ := doc["contenthtml"].(string) if len(d1) >= len(d2) || d2 == "" { detail = d1 } else { detail = d2 } detail = ju.CutLableStr(detail) detail = cut.ClearHtml(detail) doc["detail"] = detail href := qu.ObjToString(doc["href"]) if strings.HasPrefix(href, "http://") { href = href[7:] } else if strings.HasPrefix(href, "https://") { href = href[8:] } pos := strings.Index(href, "/") if pos > 0 { href = href[:pos] } doc["domain"] = href toptype := qu.ObjToString(doc["toptype"]) if qu.ObjToString(doc["type"]) == "bid" { toptype = "结果" } if toptype == "" { toptype = "*" } j := &ju.Job{ SourceMid: qu.BsonIdToSId(doc["_id"]), Category: toptype, Content: qu.ObjToString(doc["detail"]), SpiderCode: qu.ObjToString(doc["spidercode"]), Domain: qu.ObjToString(doc["domain"]), Href: qu.ObjToString(doc["href"]), Title: qu.ObjToString(doc["title"]), Data: &doc, City: qu.ObjToString(doc["city"]), Province: qu.ObjToString(doc["area"]), Result: map[string][]*ju.ExtField{}, } pretreated.AnalyStart(j) return j } //抽取 func (e *ExtractTask) ExtractProcess(j *ju.Job) { qu.Catch() qu.Try(func() { doc := *j.Data //全局前置规则,结果覆盖doc属性 for _, v := range e.RulePres { doc = ExtRegPre(doc, j, v, e.TaskInfo) } //log.Println("全局前置规则", doc) //抽取规则 for _, vc := range e.RuleCores { tmp := ju.DeepCopy(doc).(map[string]interface{}) //是否进入逻辑 if !ju.Logic(vc.LuaLogic, tmp) { continue } //抽取-前置规则 for _, v := range vc.RulePres { tmp = ExtRegPre(tmp, j, v, e.TaskInfo) } //log.Println("抽取-前置规则", tmp) //抽取-规则 for _, v := range vc.RuleCores { ExtRegCore(vc.ExtFrom, tmp, j, v, e) } //log.Println("抽取-规则", tmp) //抽取-后置规则 for _, v := range vc.RuleBacks { ExtRegBack(j, v, e.TaskInfo) } //log.Println("抽取-后置规则", tmp) //全局后置规则 for _, v := range e.RuleBacks { ExtRegBack(j, v, e.TaskInfo) } } //函数清理 for key, val := range j.Result { for _, v := range val { data := clear.DoClearFn(e.ClearFn[key], []interface{}{v.Value, j.Content}) v.Value = data[0] } } bs, _ := json.Marshal(j.Result) log.Println("抽取结果", j.SourceMid, string(bs)) //分析抽取结果并保存 todo AnalysisSaveResult(j.Data, j.Result, e.TaskInfo) }, func(err interface{}) { log.Println(err) <-e.TaskInfo.ProcessPool }) <-e.TaskInfo.ProcessPool } //前置过滤 func ExtRegPre(doc map[string]interface{}, j *ju.Job, in *RegLuaInfo, t *TaskInfo) map[string]interface{} { before := ju.DeepCopy(doc).(map[string]interface{}) extinfo := map[string]interface{}{} if in.IsLua { lua := ju.LuaScript{Code: in.Code, Name: in.Name, Doc: doc, Script: in.RuleText} if j != nil { lua.Block = j.Block } extinfo = lua.RunScript("pre") for k, v := range extinfo { //结果覆盖原doc doc[k] = v } AddExtLog("prereplace", j.SourceMid, before, extinfo, in, t) //抽取日志 } else { key := qu.If(in.Field == "", "detail", in.Field).(string) text := qu.ObjToString(doc[key]) extinfo[key] = in.RegPreBac.Reg.ReplaceAllString(text, "") doc[key] = extinfo[key] //结果覆盖原doc AddExtLog("prereplace", j.SourceMid, before, extinfo, in, t) //抽取日志 } return doc } //抽取-规则 func ExtRegCore(extfrom string, doc map[string]interface{}, j *ju.Job, in *RegLuaInfo, et *ExtractTask) { if in.IsLua { lua := ju.LuaScript{Code: in.Code, Name: in.Name, Doc: doc, Script: in.RuleText} if in.IsHasFields { //lua脚本配置有属性字段 lua.KvMap = getKvByLuaFields(j, in, et.Tag) } else { lua.KvMap = map[string][]map[string]interface{}{} } lua.Block = j.Block extinfo := lua.RunScript("core") for k, v := range extinfo { if j.Result[k] == nil { j.Result[k] = [](*ju.ExtField){} } if tmps, ok := v.([]map[string]interface{}); ok { for _, tmp := range tmps { j.Result[k] = append(j.Result[k], &ju.ExtField{k, qu.ObjToString(tmp["key"]), qu.ObjToString(tmp["type"]), qu.ObjToString(tmp["matchtype"]), extfrom, tmp["value"]}) } } } if len(extinfo) > 0 { AddExtLog("extract", j.SourceMid, nil, extinfo, in, et.TaskInfo) //抽取日志 } } else { //全文正则 text := qu.ObjToString(doc[extfrom]) if in.Field != "" { extinfo := extRegCoreToResult(extfrom, text, j, in) if len(extinfo) > 0 { AddExtLog("extract", j.SourceMid, nil, extinfo, in, et.TaskInfo) //抽取日志 } } } } //lua脚本根据属性设置提取kv值 func getKvByLuaFields(j *ju.Job, in *RegLuaInfo, t map[string][]*Tag) map[string][]map[string]interface{} { kvmap := map[string][]map[string]interface{}{} for _, vv := range in.LFields { field := qu.ObjToString(vv) tags := t[qu.ObjToString(vv)] //获取对应标签库 for _, bl := range j.Block { //冒号kv if bl.ColonKV != nil { kvs := bl.ColonKV.Kvs kvs2 := bl.ColonKV.Kvs_2 for _, tag := range tags { for _, kv := range kvs { if tag.Type == "string" { if kv.Key == tag.Key { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "colon1", "field": field, "key": tag.Key, "matchtype": "tag_string", }) } break } } else if tag.Type == "regexp" { if tag.Reg.MatchString(kv.Key) { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "colon1", "field": field, "key": tag.Key, "matchtype": "tag_regexp", }) } break } } } for _, kv := range kvs2 { if tag.Type == "string" { if kv.Key == tag.Key { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "colon2", "field": field, "key": tag.Key, "matchtype": "tag_string", }) } break } } else if tag.Type == "regexp" { if tag.Reg.MatchString(kv.Key) { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "colon2", "field": field, "key": tag.Key, "matchtype": "tag_regexp", }) } break } } } } } //空格kv if bl.SpaceKV != nil { kvs := bl.SpaceKV.Kvs for _, tag := range tags { for _, kv := range kvs { if tag.Type == "string" { if kv.Key == tag.Key { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "space", "field": field, "key": tag.Key, "matchtype": "tag_string", }) } break } } else if tag.Type == "regexp" { if tag.Reg.MatchString(kv.Key) { text := ju.TrimLRSpace(kv.Value, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "space", "field": field, "key": tag.Key, "matchtype": "tag_regexp", }) } break } } } } } //表格kv if bl.TableKV != nil { kv := bl.TableKV.Kv for _, tag := range tags { for k, val := range kv { if tag.Type == "string" { if k == tag.Key { text := ju.TrimLRSpace(val, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "table", "field": field, "key": tag.Key, "matchtype": "tag_string", }) } break } } else if tag.Type == "regexp" { if tag.Reg.MatchString(k) { text := ju.TrimLRSpace(val, "") if text != "" { kvmap[field] = append(kvmap[field], map[string]interface{}{ "value": text, "type": "table", "field": field, "key": tag.Key, "matchtype": "tag_regexp", }) } break } } } } } } } return kvmap } //正则提取结果 func extRegCoreToResult(extfrom, text string, j *ju.Job, v *RegLuaInfo) map[string]interface{} { extinfo := map[string]interface{}{} if v.RegCore.Bextract { //正则是两部分的,可以直接抽取的(含下划线) apos := v.RegCore.Reg.FindAllStringSubmatchIndex(text, -1) if len(apos) > 0 { pos := apos[0] for k, p := range v.RegCore.ExtractPos { if len(pos) > p { if pos[p] == -1 || pos[p+1] == -1 { continue } val := text[pos[p]:pos[p+1]] extinfo[k] = map[string]interface{}{ "field": v.Field, "key": v.Code, "type": "regexp", "matchtype": "regcontent", "extfrom": extfrom, "value": val, } if val != "" { if j.Result[v.Field] == nil { j.Result[k] = [](*ju.ExtField){} } j.Result[k] = append(j.Result[k], &ju.ExtField{k, v.Code, "regexp", "regcontent", extfrom, val}) } } } } } else { val := v.RegCore.Reg.ReplaceAllString(text, "") if val != "" { extinfo[v.Field] = map[string]interface{}{ "field": v.Field, "key": v.Code, "type": "regexp", "matchtype": "regcontent", "extfrom": extfrom, "value": val, } if j.Result[v.Field] == nil { j.Result[v.Field] = [](*ju.ExtField){} } j.Result[v.Field] = append(j.Result[v.Field], &ju.ExtField{v.Field, v.Code, "regexp", "regcontent", extfrom, val}) } } return extinfo } //后置过滤 func ExtRegBack(j *ju.Job, in *RegLuaInfo, t *TaskInfo) { if in.IsLua { result := getResultMapForLua(j) lua := ju.LuaScript{Code: in.Code, Name: in.Name, Result: result, Script: in.RuleText} if j != nil { lua.Block = j.Block } extinfo := lua.RunScript("back") for k, v := range extinfo { if tmps, ok := v.([]map[string]interface{}); ok { j.Result[k] = [](*ju.ExtField){} for _, tmp := range tmps { j.Result[k] = append(j.Result[k], &ju.ExtField{k, qu.ObjToString(tmp["key"]), qu.ObjToString(tmp["type"]), qu.ObjToString(tmp["matchtype"]), qu.ObjToString(tmp["extfrom"]), tmp["value"]}) } } } if len(extinfo) > 0 { AddExtLog("clear", j.SourceMid, result, extinfo, in, t) //抽取日志 } } else { extinfo := map[string]interface{}{} if in.Field != "" && j.Result[in.Field] != nil { tmp := j.Result[in.Field] exts := []interface{}{} for k, v := range tmp { text := qu.ObjToString(v.Value) if text != "" { text = in.RegPreBac.Reg.ReplaceAllString(text, in.RegPreBac.Replace) } j.Result[in.Field][k].Value = text exts = append(exts, map[string]interface{}{ "field": v.Field, "key": v.Key, "type": v.Type, "matchtype": v.MatchType, "extfrom": v.ExtFrom, "value": text, }) } extinfo[in.Field] = exts if len(extinfo) > 0 { AddExtLog("clear", j.SourceMid, tmp, extinfo, in, t) //抽取日志 } } else { for key, tmp := range j.Result { exts := []interface{}{} for k, v := range tmp { text := qu.ObjToString(v.Value) if text != "" { text = in.RegPreBac.Reg.ReplaceAllString(text, in.RegPreBac.Replace) } j.Result[key][k].Value = text exts = append(exts, map[string]interface{}{ "field": v.Field, "key": v.Key, "type": v.Type, "matchtype": v.MatchType, "extfrom": v.ExtFrom, "value": text, }) } extinfo[key] = exts } if len(extinfo) > 0 { AddExtLog("clear", j.SourceMid, j.Result, extinfo, in, t) //抽取日志 } } } } //获取抽取结果map[string][]interface{},lua脚本使用 func getResultMapForLua(j *ju.Job) map[string][]map[string]interface{} { result := map[string][]map[string]interface{}{} for key, val := range j.Result { if result[key] == nil { result[key] = []map[string]interface{}{} } for _, v := range val { tmp := map[string]interface{}{ "field": v.Field, "value": v.Value, "type": v.Type, "matchtype": v.MatchType, "key": v.Key, "extfrom": v.ExtFrom, } result[key] = append(result[key], tmp) } } return result } //抽取日志 func AddExtLog(ftype, sid string, before interface{}, extinfo interface{}, v *RegLuaInfo, t *TaskInfo) { if !t.IsEtxLog { return } logdata := map[string]interface{}{ "code": v.Code, "name": v.Name, "type": ftype, "ruletext": v.RuleText, "islua": v.IsLua, "field": v.Field, "version": t.Version, "taskname": t.Name, "before": before, "extinfo": extinfo, "sid": sid, "comeintime": time.Now().Unix(), } lock.Lock() ExtLogs[t] = append(ExtLogs[t], logdata) lock.Unlock() } //保存抽取日志 func SaveExtLog() { tmpLogs := map[*TaskInfo][]map[string]interface{}{} lock.Lock() tmpLogs = ExtLogs ExtLogs = map[*TaskInfo][]map[string]interface{}{} lock.Unlock() for k, v := range tmpLogs { if len(v) < saveLimit { db.Mgo.SaveBulk(k.TrackColl, v...) } else { for { if len(v) > saveLimit { tmp := v[:saveLimit] db.Mgo.SaveBulk(k.TrackColl, tmp...) v = v[saveLimit:] } else { db.Mgo.SaveBulk(k.TrackColl, v...) break } } } } time.AfterFunc(10*time.Second, SaveExtLog) } type FieldValue struct { Value interface{} Count int } //分析抽取结果并保存 func AnalysisSaveResult(doc *map[string]interface{}, result map[string][]*ju.ExtField, task *TaskInfo) { _id := qu.BsonIdToSId((*doc)["_id"]) //结果排序 values := map[string][]*ju.SortObject{} for key, val := range result { fieldValue := map[string]int{} for _, v := range val { value := qu.ObjToString(v.Value) fieldValue[value] += 1 } objects := []*ju.SortObject{} for k, v := range fieldValue { tmp := &ju.SortObject{ Key: k, Value: v, } objects = append(objects, tmp) } values[key] = ju.ExtSort(objects) } //从排序结果中取值 tmp := map[string]interface{}{} for key, val := range values { for _, v := range val { //取第一个 if v.Key != "" { tmp[key] = v.Key break } } } //保存抽取结果 task.DB.Update(task.SaveColl, `{"_id":"`+_id+`"}`, doc, true, false) //保存抽取详情 tmp["result"] = result db.Mgo.Update("extract_result", `{"_id":"`+_id+`"}`, tmp, true, false) }