jy-data-extract/fullproject/src_dev4/task.go

package main

import (
	"encoding/json"
	"log"
	"qfw/util"
	"regexp"
	"strings"
	"sync"
	//"strings"
	"time"

	"gopkg.in/mgo.v2/bson"
)

const (
	InitMinTime = int64(1325347200) //最小时间位置2012
)

//全量合并
func taskQl(udpInfo map[string]interface{}) {
	defer util.Catch()
	//1、检查pubilshtime索引
	db, _ := udpInfo["db"].(string)
	if db == "" {
		db = MongoTool.DbName
	}
	coll, _ := udpInfo["coll"].(string)
	if coll == "" {
		coll = ExtractColl
	}
	sess := MongoTool.GetMgoConn()
	bcon := false
	if sess.DB(db).C(coll).EnsureIndexKey("publishtime_1", "publishtime_-1") == nil {
		bcon = true
	} else {
		log.Println("publishtime_1索引不存在")
	}
	MongoTool.DestoryMongoConn(sess)
	thread := util.IntAllDef(udpInfo["thread"], 1)
	if bcon {
		//go SaveQueue()
		go updateQueue()
		go clearMem()
		//获取起始时间
		startTime, END := int64(0), int64(0)

		sts, bres := MongoTool.Find(ExtractColl, `{}`, "publishtime", `{"publishtime":1}`, true, 0, 1)
		if bres && sts != nil && len(*sts) == 1 {
			startTime = util.Int64All((*sts)[0]["publishtime"])
			sts, bres = MongoTool.Find(ExtractColl, `{}`, "-publishtime", `{"publishtime":1}`, true, 0, 1)
			if bres && sts != nil && len(*sts) == 1 {
				END = util.Int64All((*sts)[0]["publishtime"])
			}
			log.Println("查询到的起始时间", startTime, END)
		} else {
			return
		}
		startTime -= 1
		sum := 0
		if startTime < InitMinTime {
			q := map[string]interface{}{
				"publishtime": map[string]interface{}{
					"$gt":  startTime,
					"$lte": InitMinTime,
				},
			}
			sum = Mql(q, thread, db, coll, sum)
			startTime = InitMinTime
		}
		for {
			if startTime >= END {
				break
			}
			et := startTime + 50*86400
			if et >= END {
				et = END
			}
			q := map[string]interface{}{
				"publishtime": map[string]interface{}{
					"$gt":  startTime,
					"$lte": et,
				},
			}
			sum = Mql(q, thread, db, coll, sum)
			startTime = et
			time.Sleep(1 * time.Second)
		}
	}
	log.Println("task over!!!")
}

var wg = sync.WaitGroup{}

func Ids(udpInfo map[string]interface{}) {
	oid := []interface{}{}
	n1, _ := udpInfo["ids"].(string)
	idArr := strings.Split(n1, ",")
	for _, v := range idArr {
		oid = append(oid, util.StringTOBsonId(v))
	}
	thread := util.IntAllDef(udpInfo["thread"], 1)
	q := bson.M{"_id": bson.M{"$in": oid}}
	go updateQueue()
	go clearMem()
	Mql(q, thread, MongoTool.DbName, ExtractColl, 0)

}

func Mql(q map[string]interface{}, thread int, db, coll string, sum int) int {
	defer util.Catch()
	sess := MongoTool.GetMgoConn()
	defer MongoTool.DestoryMongoConn(sess)
	query := sess.DB(db).C(coll).Find(q).Sort("publishtime").Iter()
	pool := make(chan bool, thread)
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		info := ParseInfo(tmp)
		if info != nil && !((info.pnbval == 1 && info.Buyer != "") || info.pnbval == 0) {
			pool <- true
			go func(info *Info, tmp map[string]interface{}) {
				defer func() {
					currentTime = info.Publishtime
					<-pool
				}()
				startProjectMerge(info, tmp)
			}(info, tmp)
		} else {
			//log.Println("info err:", tmp["_id"], tmp["title"], tmp["buyer"])
		}
		if sum%1000 == 0 {
			log.Println("current", sum)
		}
		sum++
		tmp = make(map[string]interface{})
	}
	//阻塞
	for n := 0; n < thread; n++ {
		pool <- true
	}
	//完成
	log.Println("sontask over:", count, sum, q)
	return sum
}

var (
	titleGetPc  = regexp.MustCompile("^([-0-9a-zA-Z第号采招政询电审竞#]{8,}[-0-9a-zA-Z#]+)")
	titleGetPc1 = regexp.MustCompile("[\\[【(（](.{0,6}(编号|编码|项号|包号|代码|标段?号)[:：为])?([-0-9a-zA-Z第号采招政询电审竞#]{5,}([\\[\\]()（）][-0-9a-zA-Z第号采招审竞#]+[\\[\\]()（）][-0-9a-zA-Z第号采招审竞#]+)?)[\\]】)）]")
	titleGetPc2 = regexp.MustCompile("([-0-9a-zA-Z第号采政招询电审竞#]{8,}[-0-9a-zA-Z#]+)(.{0,5}公告)?$")
	pcReplace   = regexp.MustCompile("([\\[【(（〖〔《｛﹝{](重|第?[二三四再]次.{0,4})[\\]】)）〗〕》｝﹞}])$|[\\[\\]【】()（）〖〗〔〕《》｛｝﹝﹞－{}– 　]+|(号|重|第?[二三四五再]次(招标)?)$|[ _＿]+")
	StrOrNum    = regexp.MustCompile("^[0-9_-]{1,4}$|^[a-zA-Z_-]{1,4}$")
	StrOrNum2   = regexp.MustCompile("^[0-9_-]+$|^[a-zA-Z_-]+$")
)

func ParseInfo(tmp map[string]interface{}) (info *Info) {
	bys, _ := json.Marshal(tmp)
	var thisinfo *Info
	json.Unmarshal(bys, &thisinfo)
	if thisinfo == nil {
		return nil
	}
	if len(thisinfo.Topscopeclass) == 0 {
		thisinfo.Topscopeclass = []string{}
	}
	if len(thisinfo.Subscopeclass) == 0 {
		thisinfo.Subscopeclass = []string{}
	}

	//从标题中查找项目编号
	res := titleGetPc.FindStringSubmatch(thisinfo.Title)
	if len(res) > 1 && len(res[1]) > 6 && thisinfo.ProjectCode != res[1] && !numCheckPc.MatchString(res[1]) && !_zimureg1.MatchString(res[1]) {
		thisinfo.PTC = res[1]
	} else {
		res = titleGetPc1.FindStringSubmatch(thisinfo.Title)
		if len(res) > 3 && len(res[3]) > 6 && thisinfo.ProjectCode != res[3] && !numCheckPc.MatchString(res[3]) && !_zimureg1.MatchString(res[3]) {
			thisinfo.PTC = res[3]
		} else {
			res = titleGetPc2.FindStringSubmatch(thisinfo.Title)
			if len(res) > 1 && len(res[1]) > 6 && thisinfo.ProjectCode != res[1] && !numCheckPc.MatchString(res[1]) && !_zimureg1.MatchString(res[1]) {
				thisinfo.PTC = res[1]
			}
		}
	}

	if thisinfo.ProjectName != "" && len([]rune(thisinfo.ProjectName)) > 0 {
		thisinfo.ProjectName = pcReplace.ReplaceAllString(thisinfo.ProjectName, "")
		if thisinfo.ProjectName != "" {
			thisinfo.pnbval++
		}
	}

	if thisinfo.ProjectCode != "" || thisinfo.PTC != "" {
		if thisinfo.ProjectCode != "" {
			thisinfo.ProjectCode = pcReplace.ReplaceAllString(thisinfo.ProjectCode, "")
			if thisinfo.pnbval == 0 && len([]rune(thisinfo.ProjectCode)) < 5 {
				thisinfo.ProjectCode = StrOrNum.ReplaceAllString(thisinfo.ProjectCode, "")
			}
		} else {
			thisinfo.PTC = pcReplace.ReplaceAllString(thisinfo.PTC, "")
			if thisinfo.pnbval == 0 && len([]rune(thisinfo.PTC)) < 5 {
				thisinfo.PTC = StrOrNum.ReplaceAllString(thisinfo.PTC, "")
			}
		}
		if thisinfo.ProjectCode != "" || thisinfo.PTC != "" {
			thisinfo.pnbval++
		}
	}
	if thisinfo.ProjectCode == thisinfo.PTC || strings.Index(thisinfo.ProjectCode, thisinfo.PTC) > -1 {
		thisinfo.PTC = ""
	}

	if thisinfo.Buyer != "" && len([]rune(thisinfo.Buyer)) > 2 {
		thisinfo.pnbval++
	} else {
		thisinfo.Buyer = ""
	}

	thisinfo.LenPC = len([]rune(thisinfo.ProjectCode))
	thisinfo.LenPTC = len([]rune(thisinfo.PTC))
	thisinfo.LenPN = len([]rune(thisinfo.ProjectName))
	return thisinfo
}