1.11 添加、修改提交

fullproject/src_v1/config.json

@@ -7,11 +7,11 @@
     "mongodbName": "extract_kf",
 	"hints":"publishtime_1",
     "extractColl": "ceshi_info",
-    "projectColl": "ceshi_project",
+    "projectColl": "jh_project",
     "backupFlag": false,
     "backupColl": "jh_project1",
     "siteColl": "site",
-
+    "thread": 4,
     "jkmail": {
         "to": "wangjianghan@topnet.net.cn",
         "api": "http://10.171.112.160:19281/_send/_mail"

fullproject/src_v1/deepcopy.go

@@ -0,0 +1,119 @@
+package main
+
+import (
+	"reflect"
+	"time"
+)
+
+// Interface for delegating copy process to type
+type Interface interface {
+	DeepCopy() interface{}
+}
+
+// Iface is an alias to Copy; this exists for backwards compatibility reasons.
+func Iface(iface interface{}) interface{} {
+	return Copy(iface)
+}
+
+// Copy creates a deep copy of whatever is passed to it and returns the copy
+// in an interface{}.  The returned value will need to be asserted to the
+// correct type.
+func Copy(src interface{}) interface{} {
+	if src == nil {
+		return nil
+	}
+
+	// Make the interface a reflect.Value
+	original := reflect.ValueOf(src)
+
+	// Make a copy of the same type as the original.
+	cpy := reflect.New(original.Type()).Elem()
+
+	// Recursively copy the original.
+	copyRecursive(original, cpy)
+
+	// Return the copy as an interface.
+	return cpy.Interface()
+}
+
+// copyRecursive does the actual copying of the interface. It currently has
+// limited support for what it can handle. Add as needed.
+func copyRecursive(original, cpy reflect.Value) {
+	// check for implement deepcopy.Interface
+	if original.CanInterface() {
+		if copier, ok := original.Interface().(Interface); ok {
+			cpy.Set(reflect.ValueOf(copier.DeepCopy()))
+			return
+		}
+	}
+
+	// handle according to original's Kind
+	switch original.Kind() {
+	case reflect.Ptr:
+		// Get the actual value being pointed to.
+		originalValue := original.Elem()
+
+		// if  it isn't valid, return.
+		if !originalValue.IsValid() {
+			return
+		}
+		cpy.Set(reflect.New(originalValue.Type()))
+		copyRecursive(originalValue, cpy.Elem())
+
+	case reflect.Interface:
+		// If this is a nil, don't do anything
+		if original.IsNil() {
+			return
+		}
+		// Get the value for the interface, not the pointer.
+		originalValue := original.Elem()
+
+		// Get the value by calling Elem().
+		copyValue := reflect.New(originalValue.Type()).Elem()
+		copyRecursive(originalValue, copyValue)
+		cpy.Set(copyValue)
+
+	case reflect.Struct:
+		t, ok := original.Interface().(time.Time)
+		if ok {
+			cpy.Set(reflect.ValueOf(t))
+			return
+		}
+		// Go through each field of the struct and copy it.
+		for i := 0; i < original.NumField(); i++ {
+			// The Type's StructField for a given field is checked to see if StructField.PkgPath
+			// is set to determine if the field is exported or not because CanSet() returns false
+			// for settable fields.  I'm not sure why.  -mohae
+			if original.Type().Field(i).PkgPath != "" {
+				continue
+			}
+			copyRecursive(original.Field(i), cpy.Field(i))
+		}
+
+	case reflect.Slice:
+		if original.IsNil() {
+			return
+		}
+		// Make a new slice and copy each element.
+		cpy.Set(reflect.MakeSlice(original.Type(), original.Len(), original.Cap()))
+		for i := 0; i < original.Len(); i++ {
+			copyRecursive(original.Index(i), cpy.Index(i))
+		}
+
+	case reflect.Map:
+		if original.IsNil() {
+			return
+		}
+		cpy.Set(reflect.MakeMap(original.Type()))
+		for _, key := range original.MapKeys() {
+			originalValue := original.MapIndex(key)
+			copyValue := reflect.New(originalValue.Type()).Elem()
+			copyRecursive(originalValue, copyValue)
+			copyKey := Copy(key.Interface())
+			cpy.SetMapIndex(reflect.ValueOf(copyKey), copyValue)
+		}
+
+	default:
+		cpy.Set(original)
+	}
+}

fullproject/src_v1/init.go

@@ -21,6 +21,7 @@ var (
 	Sysconfig                map[string]interface{} //读取配置文件
 	MongoTool                *MongodbSim            //mongodb连接
 	ExtractColl, ProjectColl, BackupColl, SiteColl string	//抽取表、项目表、项目快照表、站点表
+	Thread					 int				    //配置项线程数
 	//NextNode                 []interface{}
 )
 
@@ -64,6 +65,7 @@ func init() {
 	ProjectColl = Sysconfig["projectColl"].(string)
 	BackupColl = Sysconfig["backupColl"].(string)
 	SiteColl = Sysconfig["siteColl"].(string)
+	Thread = util.IntAll(Sysconfig["thread"])
 	//NextNode = Sysconfig["nextNode"].([]interface{})
 	udpport, _ := Sysconfig["udpport"].(string)
 	udpclient = mu.UdpClient{Local: udpport, BufSize: 1024}
@@ -298,11 +300,11 @@ type ProjectInfo struct {
 	score         int
 	comStr        string
 	resVal, pjVal int
-	InfoFiled	map[string]InfoFiled	`json:"infofiled"`		//逻辑处理需要的info字段
+	InfoFiled	map[string]InfoField	`json:"infofiled"`		//逻辑处理需要的info字段
 }
 
 //存储部分招标信息字段，业务逻辑处理需要
-type InfoFiled struct {
+type InfoField struct {
 	Budget			float64				`json:"budget"`
 	Bidamount		float64				`json:"bidamount"`
 	ContractCode	string				`json:"contractcode"`

fullproject/src_v1/load_data.go

@@ -84,6 +84,15 @@ func (p *ProjectTask) loadData(starttime int64) {
 				bys, _ := json.Marshal(result)
 				var tmp *ProjectInfo
 				_ = json.Unmarshal(bys, &tmp)
+				tmpMap := make(map[string]InfoField)
+				infoMap := result["infofield"].(map[string]interface{})
+				for _, v := range infoMap{
+					var field InfoField
+					b, _ := json.Marshal(v)
+					_ = json.Unmarshal(b, &field)
+					tmpMap[tmp.Id.Hex()] = field
+				}
+				tmp.InfoFiled = tmpMap
 				pool <- tmp
 			}(result)
 		} else {

fullproject/src_v1/main.go

@@ -2,12 +2,14 @@ package main
 
 import (
 	"encoding/json"
+	"flag"
 	"log"
 	mu "mfw/util"
 	"net"
 	"os"
 	"os/signal"
 	"qfw/util"
+	qu "qfw/util"
 	"syscall"
 	"time"
 )
@@ -17,6 +19,8 @@ var (
 	SingleThread = make(chan bool, 1)
 	toaddr       = []*net.UDPAddr{} //下节点对象
 	ChSign       = make(chan os.Signal)
+
+	stype, sid, eid string 	//测试使用
 )
 
 func init() {
@@ -69,10 +73,87 @@ func main() {
 		}
 	}
 	P_QL.loadSite()
-	//go checkMapJob()
+	go checkMapJob()
 	time.Sleep(99999 * time.Hour)
 }
 
+//测试组人员使用
+func mainT() {
+	//sid = "5dfbd43ce9d1f601e43fa402"
+	//eid = "5e0954b30cf41612e061d0c8"
+
+	flag.StringVar(&sid, "sid", "", "开始id")
+	flag.StringVar(&eid, "eid", "", "结束id")
+	flag.StringVar(&stype, "stype", "", "stype")
+	flag.Parse()
+
+	mapinfo := map[string]interface{}{}
+	if sid == "" || eid == "" {
+		log.Println("sid, eid参数不能为空")
+		os.Exit(0)
+	}
+	if stype == "" {
+		log.Println("stype参数不能为空")
+		os.Exit(0)
+	}
+	mapinfo["gtid"] = qu.StringTOBsonId(sid)
+	mapinfo["lteid"] = qu.StringTOBsonId(eid)
+	mapinfo["stype"] = stype
+	mapinfo["ip"] = "127.0.0.1"
+	mapinfo["port"] = Sysconfig["udpport"]
+	if Sysconfig["loadStart"] != nil {
+		loadStart := util.Int64All(Sysconfig["loadStart"])
+		if loadStart > -1 {
+			P_QL.loadData(loadStart)
+		}
+	}
+	P_QL.loadSite()
+	task(mapinfo)
+	time.Sleep(5 * time.Second)
+}
+
+func task(mapInfo map[string]interface{})  {
+	SingleThread <- true
+	tasktype, _ := mapInfo["stype"].(string)
+	log.Println("tasktype:", tasktype)
+	switch tasktype {
+	case "ql": //全量合并
+		go func() {
+			defer func() {
+				<-SingleThread
+			}()
+			P_QL.currentType = tasktype
+			P_QL.pici = time.Now().Unix()
+			P_QL.taskQl(mapInfo)
+		}()
+	case "project": //增量合并，未抽取到项目名称或项目编号的不合并  bidding中mergestatus 1已合并 2字段问题不合并 3历史待合并
+		//合同、验收公告在6个月内查询不到可扩展到两年
+		go func() {
+			defer func() {
+				<-SingleThread
+			}()
+			P_QL.currentType = tasktype
+			P_QL.pici = time.Now().Unix()
+			P_QL.taskZl(mapInfo)
+		}()
+	case "updateInfo":		//招标字段变更
+		go func() {
+			defer func() {
+				<-SingleThread
+			}()
+			P_QL.currentType = tasktype
+			P_QL.pici = time.Now().Unix()
+			P_QL.taskUpdateInfo(mapInfo)
+		}()
+	case "history": //历史数据合并,暂时不写
+		go func() {
+			defer func() {
+				<-SingleThread
+			}()
+		}()
+	}
+}
+
 //udp调用信号
 func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 	switch act {

fullproject/src_v1/project.go

@@ -534,8 +534,8 @@ func (p *ProjectTask) NewProject(tmp map[string]interface{}, thisinfo *Info) (st
 		set["multipackage"] = 0
 	}
 
-	p1.InfoFiled = make(map[string]InfoFiled)
-	infofiled := InfoFiled{
+	p1.InfoFiled = make(map[string]InfoField)
+	infofiled := InfoField{
 		Budget:       thisinfo.Budget,
 		Bidamount:    thisinfo.Bidamount,
 		ContractCode: thisinfo.ContractCode,
@@ -821,7 +821,7 @@ func (p *ProjectTask) UpdateProject(tmp map[string]interface{}, thisinfo *Info,
 	CountAmount(pInfo, thisinfo)
 	set["budget"] = pInfo.Budget
 	set["bidamount"] = pInfo.Bidamount
-	infofiled := InfoFiled{
+	infofiled := InfoField{
 		Budget:       thisinfo.Budget,
 		Bidamount:    thisinfo.Bidamount,
 		ContractCode: thisinfo.ContractCode,
@@ -829,12 +829,14 @@ func (p *ProjectTask) UpdateProject(tmp map[string]interface{}, thisinfo *Info,
 		ProjectCode:  thisinfo.ProjectCode,
 		Bidstatus:    bs,
 	}
-	pInfo.InfoFiled[thisinfo.Id] = infofiled
+	copyMap := Copy(pInfo.InfoFiled).(map[string]InfoField)
+	copyMap[thisinfo.Id] = infofiled
 	tmpMap := make(map[string]interface{})
-	for k, v := range pInfo.InfoFiled{
+	for k, v := range copyMap{
 		tmpMap[k] = StructToMap(v)
 	}
 	tmpMap[thisinfo.Id] = StructToMap(infofiled)
+	pInfo.InfoFiled = copyMap
 	set["infofield"] = tmpMap
 
 	set["mpn"] = pInfo.MPN
@@ -960,8 +962,12 @@ func packageEle(map1 map[string]interface{}, id string) map[string]interface{} {
 
 func PackageFormat(info *Info, project *ProjectInfo) map[string]interface{} {
 	p1 := map[string]interface{}{}
-	if project != nil && project.Package != nil && len(project.Package) > 0 {
-		p1 = project.Package
+	packageCopy := make(map[string]interface{})
+	if project != nil && project.Package != nil {
+		packageCopy = Copy(project.Package).(map[string]interface{})
+	}
+	if packageCopy != nil && len(packageCopy) > 0 {
+		p1 = packageCopy
 		for k, v := range info.Package {
 			if v1, ok := v.(map[string]interface{}); ok {
 				v2 := map[string]interface{}{}
@@ -986,7 +992,7 @@ func PackageFormat(info *Info, project *ProjectInfo) map[string]interface{} {
 			}
 		}
 	} else {
-		for k, v := range info.Package {
+		for k, v := range packageCopy {
 			v1, _ := v.(map[string]interface{})
 			p2 := map[string]interface{}{}
 			p2 = packageEle(v1, info.Id)
@@ -996,6 +1002,7 @@ func PackageFormat(info *Info, project *ProjectInfo) map[string]interface{} {
 			p1[k] = []map[string]interface{}{p2}
 		}
 	}
+	p1 = packageCopy
 	return p1
 }
 
@@ -1039,20 +1046,23 @@ func CountAmount(project *ProjectInfo, info *Info) {
 			}
 		}else {
 			//招标不是单包
-			flag := false
-			if project.InfoFiled != nil && len(project.InfoFiled) > 0 {
-				for _, res := range project.InfoFiled {
-					if res.ProjectName == info.ProjectName {
-						if res.Budget < info.Budget {
-							project.Budget = project.Budget - res.Budget + info.Budget
-						}
-						flag = true
-						break
-					}
-				}
-				if !flag {
-					project.Budget = project.Budget + info.Budget
-				}
+			//flag := false
+			//if project.InfoFiled != nil && len(project.InfoFiled) > 0 {
+			//	for _, res := range project.InfoFiled {
+			//		if res.ProjectName == info.ProjectName {
+			//			if res.Budget < info.Budget {
+			//				project.Budget = project.Budget - res.Budget + info.Budget
+			//			}
+			//			flag = true
+			//			break
+			//		}
+			//	}
+			//	if !flag {
+			//		project.Budget = project.Budget + info.Budget
+			//	}
+			//}
+			if project.Budget < info.Budget {
+				project.Budget = info.Budget
 			}
 		}
 	}
@@ -1095,30 +1105,55 @@ func CountAmount(project *ProjectInfo, info *Info) {
 				}
 			}else {
 				//招标不是单包
-				flag := false
-				if project.InfoFiled != nil && len(project.InfoFiled) > 0 {
-					for _, res := range project.InfoFiled {
-						if res.Bidstatus == "合同" && res.ContractCode != "" && info.SubType == "合同" && info.ContractCode != "" {
-							if res.ContractCode == info.ContractCode {
-								if res.Bidamount < info.Bidamount {
-									project.Bidamount = project.Bidamount - res.Bidamount + info.Bidamount
+				//flag := false
+				//if project.InfoFiled != nil && len(project.InfoFiled) > 0 {
+				//	for _, res := range project.InfoFiled {
+				//		if res.Bidstatus == "合同" && res.ContractCode != "" && info.SubType == "合同" && info.ContractCode != "" {
+				//			if res.ContractCode == info.ContractCode {
+				//				if res.Bidamount < info.Bidamount {
+				//					project.Bidamount = project.Bidamount - res.Bidamount + info.Bidamount
+				//				}
+				//				flag = true
+				//				break
+				//			}
+				//		}else {
+				//			if res.ProjectName == info.ProjectName {
+				//				if res.Bidamount < info.Bidamount {
+				//					project.Bidamount = project.Bidamount - res.Bidamount + info.Bidamount
+				//				}
+				//				flag = true
+				//				break
+				//			}
+				//		}
+				//	}
+				//	if !flag {
+				//		project.Bidamount = project.Bidamount + info.Bidamount
+				//	}
+				//}
+				if info.SubType == "中标" || info.SubType == "成交" {
+					if project.Bidamount < info.Bidamount {
+						project.Bidamount = info.Bidamount
+					}else {
+						flag := false
+						if project.InfoFiled != nil && len(project.InfoFiled) > 0 {
+							for _, res := range project.InfoFiled {
+								if res.ContractCode != "" && res.ContractCode == info.ContractCode {
+									flag = true
+									break
+								}
+								if res.Bidamount == project.Bidamount {
+									flag = true
+									break
 								}
-								flag = true
-								break
 							}
-						}else {
-							if res.ProjectName == info.ProjectName {
-								if res.Bidamount < info.Bidamount {
-									project.Bidamount = project.Bidamount - res.Bidamount + info.Bidamount
+							if !flag {
+								project.Bidamount = project.Bidamount + info.Bidamount
+							}else {
+								if project.Bidamount < info.Bidamount {
+									project.Bidamount = info.Bidamount
 								}
-								flag = true
-								break
 							}
 						}
-
-					}
-					if !flag {
-						project.Bidamount = project.Bidamount + info.Bidamount
 					}
 				}
 			}
@@ -1127,7 +1162,7 @@ func CountAmount(project *ProjectInfo, info *Info) {
 }
 
 //结构体转map
-func StructToMap(filed InfoFiled) map[string]interface{} {
+func StructToMap(filed InfoField) map[string]interface{} {
 	//先转json
 	result, err := json.Marshal(filed)
 	if err != nil {

fullproject/src_v1/task.go

@@ -237,7 +237,7 @@ func (p *ProjectTask) taskQl(udpInfo map[string]interface{}) {
 	if coll == "" {
 		coll = ExtractColl
 	}
-	thread := util.IntAllDef(udpInfo["thread"], 4)
+	thread := util.IntAllDef(Thread, 4)
 	if thread > 0 {
 		p.thread = thread
 	}
@@ -280,7 +280,7 @@ func (p *ProjectTask) taskZl(udpInfo map[string]interface{}) {
 	if coll == "" {
 		coll = ExtractColl
 	}
-	thread := util.IntAllDef(udpInfo["thread"], 4)
+	thread := util.IntAllDef(Thread, 4)
 	if thread > 0 {
 		p.thread = thread
 	}
@@ -324,7 +324,7 @@ func (p *ProjectTask) taskUpdateInfo(udpInfo map[string]interface{}) {
 	if coll == "" {
 		coll = ExtractColl
 	}
-	thread := util.IntAllDef(udpInfo["thread"], 4)
+	thread := util.IntAllDef(Thread, 4)
 	if thread > 0 {
 		p.thread = thread
 	}

fullproject/src_v1/update.go

@@ -15,7 +15,7 @@ func (p *ProjectTask) modifyUpdate(pInfoId string, index int, info *Info, tmp ma
 	infoList := []interface{}(tmpPro["list"].(primitive.A))
 	infoMap := infoList[index].(map[string]interface{})
 	infoList[index] = updateValue(infoMap, modifyMap)
-	infofiled := InfoFiled{
+	infofiled := InfoField{
 		Budget:       info.Budget,
 		Bidamount:    info.Bidamount,
 		ContractCode: info.ContractCode,
@@ -70,7 +70,7 @@ func (p *ProjectTask) mergeAndModify(pInfoId string, index int, info *Info, tmp
 				infoList := []interface{}(tmpPro["list"].(primitive.A))
 				infoMap := infoList[index].(map[string]interface{})
 				infoList[index] = updateValue(infoMap, modifyMap)
-				infofiled := InfoFiled{
+				infofiled := InfoField{
 					Budget:       info.Budget,
 					Bidamount:    info.Bidamount,
 					ContractCode: info.ContractCode,