documentMerge.py 55 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281
  1. #coding:UTF8
  2. from odps.udf import annotate
  3. from odps.distcache import get_cache_archive
  4. from odps.distcache import get_cache_file
  5. from odps.udf import BaseUDTF,BaseUDAF
  6. import threading
  7. import logging
  8. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  9. import time
  10. import json
  11. def log(msg):
  12. logging.info(msg)
  13. # 配置pandas依赖包
  14. def include_package_path(res_name):
  15. import os, sys
  16. archive_files = get_cache_archive(res_name)
  17. dir_names = sorted([os.path.dirname(os.path.normpath(f.name)) for f in archive_files
  18. if '.dist_info' not in f.name], key=lambda v: len(v))
  19. _path = dir_names[0].split(".zip/files")[0]+".zip/files"
  20. log("add path:%s"%(_path))
  21. sys.path.append(_path)
  22. return os.path.dirname(dir_names[0])
  23. # 可能出现类似RuntimeError: xxx has been blocked by sandbox
  24. # 这是因为包含C的库,会被沙盘block,可设置set odps.isolation.session.enable = true
  25. def include_file(file_name):
  26. import os, sys
  27. so_file = get_cache_file(file_name)
  28. sys.path.append(os.path.dirname(os.path.abspath(so_file.name)))
  29. def include_so(file_name):
  30. import os, sys
  31. so_file = get_cache_file(file_name)
  32. with open(so_file.name, 'rb') as fp:
  33. content=fp.read()
  34. so = open(file_name, "wb")
  35. so.write(content)
  36. so.flush()
  37. so.close()
  38. #初始化业务数据包,由于上传限制,python版本以及archive解压包不统一等各种问题,需要手动导入
  39. def init_env(list_files,package_name):
  40. import os,sys
  41. if len(list_files)==1:
  42. so_file = get_cache_file(list_files[0])
  43. cmd_line = os.path.abspath(so_file.name)
  44. os.system("unzip -o %s -d %s"%(cmd_line,package_name))
  45. elif len(list_files)>1:
  46. cmd_line = "cat"
  47. for _file in list_files:
  48. so_file = get_cache_file(_file)
  49. cmd_line += " "+os.path.abspath(so_file.name)
  50. cmd_line += " > temp.zip"
  51. os.system(cmd_line)
  52. os.system("unzip -o temp.zip -d %s"%(package_name))
  53. # os.system("rm -rf %s/*.dist-info"%(package_name))
  54. # return os.listdir(os.path.abspath("local_package"))
  55. # os.system("echo export LD_LIBRARY_PATH=%s >> ~/.bashrc"%(os.path.abspath("local_package")))
  56. # os.system("source ~/.bashrc")
  57. sys.path.insert(0,os.path.abspath(package_name))
  58. # sys.path.append(os.path.join(os.path.abspath("local_package"),"interface_real"))
  59. import platform
  60. def getSet(list_dict,key):
  61. _set = set()
  62. for item in list_dict:
  63. if key in item:
  64. if item[key]!='' and item[key] is not None:
  65. if re.search("^[\d\.]+$",item[key]) is not None:
  66. _set.add(str(float(item[key])))
  67. else:
  68. _set.add(str(item[key]))
  69. return _set
  70. def split_with_time(list_dict,sort_key,timedelta=86400*120,more_than_one=True):
  71. group_num = 1
  72. if more_than_one:
  73. group_num = 2
  74. if len(list_dict)>0:
  75. if sort_key in list_dict[0]:
  76. list_dict.sort(key=lambda x:x[sort_key])
  77. list_group = []
  78. _begin = 0
  79. for i in range(len(list_dict)-1):
  80. if abs(list_dict[i][sort_key]-list_dict[i+1][sort_key])<=timedelta:
  81. continue
  82. else:
  83. _group = []
  84. for j in range(_begin,i+1):
  85. _group.append(list_dict[j])
  86. if len(_group)>1:
  87. list_group.append(_group)
  88. _begin = i + 1
  89. if len(list_dict)>=group_num:
  90. _group = []
  91. for j in range(_begin,len(list_dict)):
  92. _group.append(list_dict[j])
  93. if len(_group)>0:
  94. list_group.append(_group)
  95. return list_group
  96. return [list_dict]
  97. @annotate('bigint,bigint,string,string,string,string,string,string,bigint->string')
  98. class f_merge_rule_limit_num_contain_greater(BaseUDAF):
  99. '''
  100. 项目编号、中标单位、len(项目编号)>7、中标单位<> ""、合并后非空招标单位数<2、合并后同公告类型非空金额相同
  101. '''
  102. def __init__(self):
  103. import logging
  104. import json,re
  105. global json,logging,re
  106. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  107. def new_buffer(self):
  108. return [list()]
  109. def iterate(self, buffer,docid,page_time_stamp,set_limit_column1,set_limit_column2,set_limit_column3,set_limit_column4,contain_column,greater_column,MAX_NUM):
  110. buffer[0].append({"docid":docid,"page_time_stamp":page_time_stamp,"set_limit_column1":set_limit_column1,
  111. "set_limit_column2":set_limit_column2,"set_limit_column3":set_limit_column3,"set_limit_column4":set_limit_column4,
  112. "contain_column":contain_column,"greater_column":greater_column,"MAX_NUM":MAX_NUM})
  113. def merge(self, buffer, pbuffer):
  114. buffer[0].extend(pbuffer[0])
  115. def terminate(self, buffer):
  116. MAX_NUM = 5
  117. if len(buffer[0])>0:
  118. MAX_NUM = buffer[0][0]["MAX_NUM"]
  119. list_split = split_with_time(buffer[0],"page_time_stamp")
  120. list_group = []
  121. for _split in list_split:
  122. flag = True
  123. keys = ["set_limit_column1","set_limit_column2","set_limit_column3","set_limit_column4"]
  124. dict_set = {}
  125. for _key in keys:
  126. dict_set[_key] = set()
  127. if len(_split)>MAX_NUM:
  128. flag = False
  129. else:
  130. for _key in keys:
  131. logging.info(_key+str(getSet(_split,_key)))
  132. if len(getSet(_split,_key))>1:
  133. flag = False
  134. break
  135. MAX_CONTAIN_COLUMN = None
  136. #判断组内每条公告是否包含
  137. if flag:
  138. for _d in _split:
  139. contain_column = _d["contain_column"]
  140. if contain_column is not None and contain_column !="":
  141. if MAX_CONTAIN_COLUMN is None:
  142. MAX_CONTAIN_COLUMN = contain_column
  143. else:
  144. if len(MAX_CONTAIN_COLUMN)<len(contain_column):
  145. if contain_column.find(MAX_CONTAIN_COLUMN)==-1:
  146. flag = False
  147. break
  148. MAX_CONTAIN_COLUMN = contain_column
  149. else:
  150. if MAX_CONTAIN_COLUMN.find(contain_column)==-1:
  151. flag = False
  152. break
  153. if len(getSet(_split,"greater_column"))==1:
  154. flag = False
  155. break
  156. if flag:
  157. _set_docid = set()
  158. for item in _split:
  159. _set_docid.add(item["docid"])
  160. if len(_set_docid)>1:
  161. list_group.append(list(_set_docid))
  162. return json.dumps(list_group)
  163. def getDiffIndex(list_dict,key):
  164. _set = set()
  165. for _i in range(len(list_dict)):
  166. item = list_dict[_i]
  167. if key in item:
  168. if item[key]!='' and item[key] is not None:
  169. if re.search("^\d[\d\.]*$",item[key]) is not None:
  170. _set.add(str(float(item[key])))
  171. else:
  172. _set.add(str(item[key]))
  173. if len(_set)>1:
  174. return _i
  175. return len(list_dict)
  176. @annotate('bigint,bigint,string,string,string,string,string,string,string,bigint->string')
  177. class f_remege_limit_num_contain(BaseUDAF):
  178. '''
  179. 项目编号、中标单位、len(项目编号)>7、中标单位<> ""、合并后非空招标单位数<2、合并后同公告类型非空金额相同
  180. '''
  181. def __init__(self):
  182. import logging
  183. import json,re
  184. global json,logging,re
  185. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  186. def new_buffer(self):
  187. return [list()]
  188. def iterate(self, buffer,docid,page_time_stamp,set_limit_column1,set_limit_column2,set_limit_column3,set_limit_column4,contain_column1,contain_column2,notLike_column,confidence):
  189. buffer[0].append({"docid":docid,"page_time_stamp":page_time_stamp,"set_limit_column1":set_limit_column1,
  190. "set_limit_column2":set_limit_column2,"set_limit_column3":set_limit_column3,"set_limit_column4":set_limit_column4,
  191. "contain_column1":contain_column1,"contain_column2":contain_column2,"notLike_column":notLike_column,"confidence":confidence})
  192. def merge(self, buffer, pbuffer):
  193. buffer[0].extend(pbuffer[0])
  194. def getNotLikeSet(self,_dict,column_name):
  195. column_value = _dict.get(column_name,None)
  196. _set = set()
  197. if column_value is not None:
  198. for _i in range(1,len(column_value)):
  199. _set.add(column_value[_i-1:_i+1])
  200. _dict["notLike_set"] = _set
  201. def getSimilarity(self,_set1,_set2):
  202. _sum = max([1,min([len(_set1),len(_set2)])])
  203. return len(_set1&_set2)/_sum
  204. def terminate(self, buffer):
  205. list_group = []
  206. the_group = buffer[0]
  207. SIM_PROB = 0.6
  208. for _d in the_group:
  209. self.getNotLikeSet(_d,"notLike_column")
  210. #判断多个值与否
  211. keys = ["set_limit_column1","set_limit_column2","set_limit_column3","set_limit_column4"]
  212. re_merge = False
  213. for _key in keys:
  214. if len(getSet(the_group,_key))>1:
  215. re_merge = True
  216. break
  217. #判断是否相似而不相同
  218. re_merge_sim = False
  219. for _i1 in range(0,len(the_group)):
  220. for _j1 in range(_i1+1,len(the_group)):
  221. _set1 = the_group[_i1]["notLike_set"]
  222. _set2 = the_group[_j1]["notLike_set"]
  223. _sim = self.getSimilarity(_set1,_set2)
  224. if _sim>SIM_PROB and _sim<1:
  225. re_merge_sim = True
  226. break
  227. contain_keys = ["contain_column1","contain_column2"]
  228. logging.info(the_group)
  229. logging.info(str(re_merge)+str(re_merge_sim))
  230. if re_merge or re_merge_sim:
  231. the_group.sort(key=lambda x:x["confidence"],reverse=True)
  232. the_group.sort(key=lambda x:x["page_time_stamp"])
  233. #重新成组
  234. dict_docid_doc = {}
  235. for _doc in the_group:
  236. dict_docid_doc[_doc["docid"]] = _doc
  237. for _doc in the_group:
  238. merge_flag = False
  239. for _index in range(len(list_group)):
  240. _g = list_group[_index]
  241. hit_count = 0
  242. dict_temp = dict()
  243. #多个值的异常
  244. if re_merge:
  245. for _c_key in contain_keys:
  246. dict_temp[_c_key] = _g[_c_key]
  247. if _g[_c_key] is not None and _doc[_c_key] is not None:
  248. if len(_g[_c_key])>len(_doc[_c_key]):
  249. if str(_g[_c_key]).find(str(_doc[_c_key]))>=0:
  250. dict_temp[_c_key] = _g[_c_key]
  251. hit_count += 1
  252. else:
  253. if str(_doc[_c_key]).find(str(_g[_c_key]))>=0:
  254. dict_temp[_c_key] = _doc[_c_key]
  255. _g[_c_key] = _doc[_c_key]
  256. hit_count += 1
  257. else:
  258. hit_count = 1
  259. # if hit_count==len(contain_keys):
  260. if hit_count>0:
  261. _flag_sim = False
  262. #相似而不相同的异常
  263. if re_merge_sim:
  264. for _docid in _g["docid"]:
  265. tmp_d = dict_docid_doc[_docid]
  266. _sim = self.getSimilarity(tmp_d["notLike_set"],_doc["notLike_set"])
  267. if _sim>SIM_PROB and _sim<1:
  268. _flag_sim = True
  269. if not _flag_sim:
  270. for _c_key in dict_temp.keys():
  271. _g[_c_key] = dict_temp[_c_key]
  272. _g["docid"].append(_doc["docid"])
  273. merge_flag = True
  274. break
  275. if not merge_flag:
  276. _dict = dict()
  277. _dict["docid"] = [_doc["docid"]]
  278. for _c_key in contain_keys:
  279. _dict[_c_key] = _doc[_c_key]
  280. list_group.append(_dict)
  281. final_group = []
  282. #判断是否符合一个值
  283. for _group in list_group:
  284. _split = []
  285. for _docid in _group["docid"]:
  286. _split.append(dict_docid_doc[_docid])
  287. #通过置信度排序,尽可能保留组
  288. _split.sort(key=lambda x:x["confidence"],reverse=True)
  289. #置信度
  290. list_key_index = []
  291. for _k in keys:
  292. list_key_index.append(getDiffIndex(_split,_k))
  293. _index = min(list_key_index)
  294. final_group.append([_c["docid"] for _c in _split[:_index]])
  295. for _c in _split[_index:]:
  296. final_group.append([_c["docid"]])
  297. #若是找到两个以上,则全部单独成组,否则成一组
  298. # _flag = True
  299. # for _key in keys:
  300. # if len(getSet(_split,_key))>1:
  301. # _flag = False
  302. # break
  303. # if not _flag:
  304. # for _docid in _group["docid"]:
  305. # final_group.append([_docid])
  306. # else:
  307. # final_group.append(list(set(_group["docid"])))
  308. else:
  309. final_group = [list(set([item["docid"] for item in the_group]))]
  310. log(str(final_group))
  311. return json.dumps(final_group)
  312. def getCurrent_date(format="%Y-%m-%d %H:%M:%S"):
  313. _time = time.strftime(format,time.localtime())
  314. return _time
  315. @annotate('bigint->string')
  316. class f_get_single_merged_bychannel(BaseUDTF):
  317. def process(self,docid):
  318. _d = {"data":{str(docid):[]},"process_time":getCurrent_date()}
  319. self.forward(json.dumps(_d))
  320. @annotate('string->string')
  321. class f_get_single_merged_docids(object):
  322. def evaluate(self,_json):
  323. if _json!="" and _json is not None:
  324. _d = json.loads(_json)
  325. _keys = _d.get("data",{}).keys()
  326. return ",".join(list(_keys))
  327. return ""
  328. @annotate('bigint,bigint,bigint,string,string,string,string,string,string,string,bigint,bigint,string->string')
  329. class f_remege_limit_num_contain_bychannel(BaseUDAF):
  330. '''f_remege_limit_num_contain_bychannel
  331. 项目编号、中标单位、len(项目编号)>7、中标单位<> ""、合并后非空招标单位数<2、合并后同公告类型非空金额相同
  332. '''
  333. def __init__(self):
  334. import logging
  335. import json,re
  336. global json,logging,re
  337. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  338. def new_buffer(self):
  339. return [list()]
  340. def iterate(self, buffer,docid,docchannel,page_time_stamp,set_limit_column1,set_limit_column2,set_limit_column3,set_limit_column4,contain_column1,contain_column2,notLike_column,confidence,extract_count,json_dicttime):
  341. _dict = {"docid":docid,"docchannel":docchannel,"page_time_stamp":page_time_stamp,"set_limit_column1":set_limit_column1,
  342. "set_limit_column2":set_limit_column2,"set_limit_column3":set_limit_column3,"set_limit_column4":set_limit_column4,
  343. "contain_column1":contain_column1,"contain_column2":contain_column2,"notLike_column":notLike_column,"confidence":confidence,
  344. "extract_count":extract_count,"json_dicttime":json_dicttime}
  345. buffer[0].append(_dict)
  346. def merge(self, buffer, pbuffer):
  347. buffer[0].extend(pbuffer[0])
  348. def getNotLikeSet(self,_dict,column_name):
  349. column_value = _dict.get(column_name,None)
  350. _set = set()
  351. if column_value is not None:
  352. for _i in range(1,len(column_value)):
  353. _set.add(column_value[_i-1:_i+1])
  354. _dict["notLike_set"] = _set
  355. def getSimilarity(self,_set1,_set2):
  356. _sum = max([1,min([len(_set1),len(_set2)])])
  357. return len(_set1&_set2)/_sum
  358. def difftimecount(self,_dict1,_dict2):
  359. _count = 0
  360. for k,v in _dict1.items():
  361. if v is not None and v!="":
  362. v1 = _dict2.get(k)
  363. if v1 is not None and v1!="":
  364. if v!=v1:
  365. _count += 1
  366. return _count
  367. def splitByTimezone(self,list_dict,_key):
  368. cluster_docid = []
  369. dict_docid_key = {}
  370. dict_docid = {}
  371. for _dict in list_dict:
  372. if _dict.get(_key,"") is None or _dict.get(_key,"")=="":
  373. dict_docid_key[_dict.get("docid")] = {}
  374. else:
  375. dict_docid_key[_dict.get("docid")] = json.loads(_dict.get(_key))
  376. dict_docid[_dict.get("docid")] = _dict
  377. for _dict in list_dict:
  378. _find = False
  379. for _cl in cluster_docid:
  380. _legal = True
  381. for _c in _cl:
  382. if self.difftimecount(dict_docid_key.get(_c),dict_docid_key.get(_dict.get("docid")))>0:
  383. _legal = False
  384. break
  385. if _legal:
  386. _cl.append(_dict.get("docid"))
  387. _find = True
  388. if not _find:
  389. cluster_docid.append([_dict.get("docid")])
  390. _result = []
  391. for _cl in cluster_docid:
  392. _r = []
  393. for _c in _cl:
  394. _r.append(dict_docid.get(_c))
  395. _result.append(_r)
  396. return _result
  397. def terminate(self, buffer):
  398. list_group = []
  399. the_group = buffer[0]
  400. SIM_PROB = 0.6
  401. for _d in the_group:
  402. self.getNotLikeSet(_d,"notLike_column")
  403. #判断多个值与否
  404. keys = ["set_limit_column1","set_limit_column2","set_limit_column3","set_limit_column4"]
  405. re_merge = False
  406. for _key in keys:
  407. if len(getSet(the_group,_key))>1:
  408. log("has_more_than_one:%s"%str(getSet(the_group,_key)))
  409. re_merge = True
  410. break
  411. #判断是否相似而不相同
  412. re_merge_sim = False
  413. for _i1 in range(0,len(the_group)):
  414. for _j1 in range(_i1+1,len(the_group)):
  415. _set1 = the_group[_i1]["notLike_set"]
  416. _set2 = the_group[_j1]["notLike_set"]
  417. _sim = self.getSimilarity(_set1,_set2)
  418. if _sim>SIM_PROB and _sim<1:
  419. re_merge_sim = True
  420. break
  421. contain_keys = ["contain_column1","contain_column2"]
  422. logging.info(the_group)
  423. logging.info(str(re_merge)+str(re_merge_sim))
  424. #重新成组
  425. dict_docid_doc = {}
  426. for _doc in the_group:
  427. dict_docid_doc[_doc["docid"]] = _doc
  428. if re_merge or re_merge_sim:
  429. the_group.sort(key=lambda x:x["confidence"],reverse=True)
  430. the_group.sort(key=lambda x:x["page_time_stamp"])
  431. for _doc in the_group:
  432. merge_flag = False
  433. for _index in range(len(list_group)):
  434. _g = list_group[_index]
  435. hit_count = 0
  436. dict_temp = dict()
  437. #多个值的异常
  438. if re_merge:
  439. for _c_key in contain_keys:
  440. dict_temp[_c_key] = _g[_c_key]
  441. if _g[_c_key] is not None and _doc[_c_key] is not None:
  442. if len(_g[_c_key])>len(_doc[_c_key]):
  443. if str(_g[_c_key]).find(str(_doc[_c_key]))>=0:
  444. dict_temp[_c_key] = _g[_c_key]
  445. hit_count += 1
  446. else:
  447. if str(_doc[_c_key]).find(str(_g[_c_key]))>=0:
  448. dict_temp[_c_key] = _doc[_c_key]
  449. _g[_c_key] = _doc[_c_key]
  450. hit_count += 1
  451. else:
  452. hit_count = 1
  453. # if hit_count==len(contain_keys):
  454. if hit_count>0:
  455. _flag_sim = False
  456. #相似而不相同的异常
  457. if re_merge_sim:
  458. for _docid in _g["docid"]:
  459. tmp_d = dict_docid_doc[_docid]
  460. _sim = self.getSimilarity(tmp_d["notLike_set"],_doc["notLike_set"])
  461. if _sim>SIM_PROB and _sim<1:
  462. _flag_sim = True
  463. if not _flag_sim:
  464. for _c_key in dict_temp.keys():
  465. _g[_c_key] = dict_temp[_c_key]
  466. _g["docid"].append(_doc["docid"])
  467. merge_flag = True
  468. break
  469. if not merge_flag:
  470. _dict = dict()
  471. _dict["docid"] = [_doc["docid"]]
  472. for _c_key in contain_keys:
  473. _dict[_c_key] = _doc[_c_key]
  474. list_group.append(_dict)
  475. final_group = []
  476. #判断是否符合一个值
  477. for _group in list_group:
  478. _split = []
  479. for _docid in _group["docid"]:
  480. _split.append(dict_docid_doc[_docid])
  481. #通过置信度排序,尽可能保留组
  482. _split.sort(key=lambda x:x["confidence"],reverse=True)
  483. #置信度
  484. list_key_index = []
  485. for _k in keys:
  486. list_key_index.append(getDiffIndex(_split,_k))
  487. _index = min(list_key_index)
  488. final_group.append([_c["docid"] for _c in _split[:_index]])
  489. for _c in _split[_index:]:
  490. final_group.append([_c["docid"]])
  491. #若是找到两个以上,则全部单独成组,否则成一组
  492. # _flag = True
  493. # for _key in keys:
  494. # if len(getSet(_split,_key))>1:
  495. # _flag = False
  496. # break
  497. # if not _flag:
  498. # for _docid in _group["docid"]:
  499. # final_group.append([_docid])
  500. # else:
  501. # final_group.append(list(set(_group["docid"])))
  502. else:
  503. final_group = [list(set([item["docid"] for item in the_group]))]
  504. log("%s--%s"%("final_group",str(final_group)))
  505. #每个channel选择一篇公告
  506. final_group_channel = []
  507. for _group in final_group:
  508. dict_channel_id = {}
  509. otherChannel = 10000
  510. for _docid in _group:
  511. _channel = dict_docid_doc[_docid].get("docchannel")
  512. if _channel in [114,115,116,117]:
  513. otherChannel += 1
  514. _channel = otherChannel
  515. if _channel not in dict_channel_id:
  516. dict_channel_id[_channel] = []
  517. dict_channel_id[_channel].append({"docid":_docid,"page_time_stamp":dict_docid_doc[_docid].get("page_time_stamp"),
  518. "extract_count":dict_docid_doc[_docid].get("extract_count"),
  519. "json_dicttime":dict_docid_doc[_docid].get("json_dicttime")})
  520. #根据日期进行切分
  521. new_dict_channel_id = {}
  522. log("%s:%s"%("dict_channel_id",str(dict_channel_id)))
  523. for k,v in dict_channel_id.items():
  524. list_time_docids = split_with_time(v,"page_time_stamp",86400*6,more_than_one=False)
  525. log(list_time_docids)
  526. for _l in list_time_docids:
  527. list_t = self.splitByTimezone(_l,"json_dicttime")
  528. for _t in list_t:
  529. otherChannel += 1
  530. new_dict_channel_id[otherChannel] = _t
  531. log("%s:%s"%("new_dict_channel_id",str(new_dict_channel_id)))
  532. channel_dict = {}
  533. for k,v in new_dict_channel_id.items():
  534. v.sort(key=lambda x:x["docid"])
  535. v.sort(key=lambda x:x["extract_count"],reverse=True)
  536. channel_dict[v[0]["docid"]] = []
  537. for _docs in v[1:]:
  538. channel_dict[v[0]["docid"]].append(_docs["docid"])
  539. _d = {"data":channel_dict,"process_time":getCurrent_date()}
  540. final_group_channel.append(_d)
  541. return json.dumps(final_group_channel)
  542. @annotate('string -> string')
  543. class f_get_remerge_group_channel(BaseUDTF):
  544. '''
  545. 将多个组拆解成多条记录
  546. '''
  547. def __init__(self):
  548. import logging
  549. import json
  550. global json,logging
  551. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  552. def process(self,json_remerge):
  553. if json_remerge is not None:
  554. list_group = json.loads(json_remerge)
  555. for _group in list_group:
  556. self.forward(json.dumps(_group))
  557. @annotate('string -> string')
  558. class f_get_remerge_group(BaseUDTF):
  559. '''
  560. 将多个组拆解成多条记录
  561. '''
  562. def __init__(self):
  563. import logging
  564. import json
  565. global json,logging
  566. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  567. def process(self,json_remerge):
  568. if json_remerge is not None:
  569. list_group = json.loads(json_remerge)
  570. for _group in list_group:
  571. l_g = list(set(_group))
  572. l_g.sort(key=lambda x:x)
  573. list_docid = [str(_docid) for _docid in l_g]
  574. self.forward(",".join(list_docid))
  575. @annotate('bigint,bigint,string->string')
  576. class f_merge_probability(BaseUDAF):
  577. '''
  578. 合并组为一条记录
  579. '''
  580. def __init__(self):
  581. import json
  582. global json
  583. def new_buffer(self):
  584. return [[]]
  585. def iterate(self, buffer,docid,page_time_stamp,_type):
  586. buffer[0].append({"docid":docid,"page_time_stamp":page_time_stamp,"type":_type})
  587. def merge(self, buffer, pbuffer):
  588. buffer[0].extend(pbuffer[0])
  589. def terminate(self, buffer):
  590. list_dict = buffer[0]
  591. list_dict = list_dict[:10000]
  592. list_group = split_with_time(list_dict,sort_key="page_time_stamp",timedelta=86400*120)
  593. return json.dumps(list_group)
  594. @annotate('string -> bigint,bigint,bigint,bigint,string')
  595. class f_split_merge_probability(BaseUDTF):
  596. def __init__(self):
  597. import logging
  598. import json
  599. global logging,json
  600. logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  601. def process(self,list_group_str):
  602. logging.info("0")
  603. logging.info(list_group_str)
  604. if list_group_str is not None:
  605. logging.info("1")
  606. try:
  607. list_group = json.loads(list_group_str)
  608. logging.info("2")
  609. for _group in list_group:
  610. if len(_group)>0:
  611. _type = _group[0].get("type","")
  612. logging.info("3%d"%len(list_group))
  613. # _group.sort(key=lambda x:x["page_time_stamp"])
  614. _len = min(100,len(_group))
  615. for _index_i in range(_len):
  616. _count = 0
  617. for _index_j in range(_index_i+1,_len):
  618. if abs(_group[_index_j]["page_time_stamp"]-_group[_index_i]["page_time_stamp"])>86400*120:
  619. break
  620. _count += 1
  621. _docid1 = _group[_index_i]["docid"]
  622. _docid2 = _group[_index_j]["docid"]
  623. if _docid1<_docid2:
  624. self.forward(_docid1,_docid2,1,_len,_type)
  625. else:
  626. self.forward(_docid2,_docid1,1,_len,_type)
  627. except Exception as e:
  628. logging(str(e))
  629. @annotate('bigint,bigint,string->string')
  630. class f_merge_groupPairs(BaseUDAF):
  631. '''
  632. 合并组为一条记录
  633. '''
  634. def __init__(self):
  635. import json
  636. global json
  637. def new_buffer(self):
  638. return [[]]
  639. def iterate(self, buffer,is_exists,counts,_type):
  640. buffer[0].append({"is_exists":is_exists,"counts":counts,"_type":_type})
  641. def merge(self, buffer, pbuffer):
  642. buffer[0].extend(pbuffer[0])
  643. def terminate(self, buffer):
  644. list_dict = buffer[0]
  645. list_dict = list_dict[:10000]
  646. return json.dumps(list_dict)
  647. @annotate("string -> bigint,bigint,bigint")
  648. class f_merge_getLabel(BaseUDTF):
  649. def __init__(self):
  650. import logging
  651. import json
  652. global logging,json
  653. def process(self,str_docids):
  654. if str_docids is not None:
  655. list_docids = [int(i) for i in str_docids.split(",")]
  656. list_docids.sort(key=lambda x:x)
  657. _len = min(100,len(list_docids))
  658. for index_i in range(_len):
  659. docid_less = list_docids[index_i]
  660. for index_j in range(index_i+1,_len):
  661. docid_greater = list_docids[index_j]
  662. self.forward(docid_less,docid_greater,1)
  663. def getSimilarityOfString(str1,str2):
  664. _set1 = set()
  665. _set2 = set()
  666. if str1 is not None:
  667. for i in range(1,len(str1)):
  668. _set1.add(str1[i-1:i+1])
  669. if str2 is not None:
  670. for i in range(1,len(str2)):
  671. _set2.add(str2[i-1:i+1])
  672. _len = max(1,min(len(_set1),len(_set2)))
  673. return len(_set1&_set2)/_len
  674. def check_columns(tenderee_less,tenderee_greater,
  675. agency_less,agency_greater,project_code_less,project_code_greater,project_name_less,project_name_greater,
  676. win_tenderer_less,win_tenderer_greater,win_bid_price_less,win_bid_price_greater,
  677. bidding_budget_less,bidding_budget_greater,doctitle_refine_less,doctitle_refine_greater):
  678. flag = True
  679. _set_tenderee = set()
  680. if tenderee_less is not None and tenderee_less!="":
  681. _set_tenderee.add(tenderee_less)
  682. if tenderee_greater is not None and tenderee_greater!="":
  683. _set_tenderee.add(tenderee_greater)
  684. if len(_set_tenderee)>1:
  685. return False
  686. code_sim = getSimilarityOfString(project_code_less,project_code_greater)
  687. if code_sim>0.6 and code_sim<1:
  688. return False
  689. #同批次不同编号
  690. if getLength(project_code_less)>0 and getLength(project_code_greater)>0:
  691. _split_code_less = project_code_less.split("-")
  692. _split_code_greater = project_code_greater.split("-")
  693. if len(_split_code_less)>1 and len(_split_code_greater)>1:
  694. if _split_code_less[0]==_split_code_greater[0] and project_code_less!=project_code_greater:
  695. return False
  696. _set_win_tenderer = set()
  697. if win_tenderer_less is not None and win_tenderer_less!="":
  698. _set_win_tenderer.add(win_tenderer_less)
  699. if win_tenderer_greater is not None and win_tenderer_greater!="":
  700. _set_win_tenderer.add(win_tenderer_greater)
  701. if len(_set_win_tenderer)>1:
  702. return False
  703. _set_win_bid_price = set()
  704. if win_bid_price_less is not None and win_bid_price_less!="":
  705. _set_win_bid_price.add(float(win_bid_price_less))
  706. if win_bid_price_greater is not None and win_bid_price_greater!="":
  707. _set_win_bid_price.add(float(win_bid_price_greater))
  708. if len(_set_win_bid_price)>1:
  709. return False
  710. _set_bidding_budget = set()
  711. if bidding_budget_less is not None and bidding_budget_less!="":
  712. _set_bidding_budget.add(float(bidding_budget_less))
  713. if bidding_budget_greater is not None and bidding_budget_greater!="":
  714. _set_bidding_budget.add(float(bidding_budget_greater))
  715. if len(_set_bidding_budget)>1:
  716. return False
  717. return True
  718. def getSimLevel(str1,str2):
  719. str1_null = False
  720. str2_null = False
  721. _v = 0
  722. if str1 is None or str1=="":
  723. str1_null = True
  724. if str2 is None or str2=="":
  725. str2_null = True
  726. if str1_null and str2_null:
  727. _v = 2
  728. elif str1_null and not str2_null:
  729. _v = 4
  730. elif not str1_null and str2_null:
  731. _v = 6
  732. elif not str1_null and not str2_null:
  733. if str1==str2:
  734. _v = 10
  735. else:
  736. _v = 0
  737. return _v
  738. import math
  739. def featurnCount(_count,max_count=100):
  740. return max(0,min(1,_count))*(1/math.sqrt(max(1,_count-1)))
  741. def getLength(_str):
  742. return len(_str if _str is not None else "")
  743. @annotate("string->bigint")
  744. class f_get_min_counts(object):
  745. def evaluate(self,json_context):
  746. _context = json.loads(json_context)
  747. min_counts = 100
  748. for item in _context:
  749. if item["counts"]<min_counts:
  750. min_counts = item["counts"]
  751. return min_counts
  752. @annotate("string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string,string->string,double")
  753. class f_merge_featureMatrix(BaseUDTF):
  754. def __init__(self):
  755. import logging
  756. import json
  757. global logging,json
  758. def process(self,json_context,tenderee_less,tenderee_greater,
  759. agency_less,agency_greater,project_code_less,project_code_greater,project_name_less,project_name_greater,
  760. win_tenderer_less,win_tenderer_greater,win_bid_price_less,win_bid_price_greater,
  761. bidding_budget_less,bidding_budget_greater,doctitle_refine_less,doctitle_refine_greater):
  762. if not check_columns(tenderee_less,tenderee_greater,
  763. agency_less,agency_greater,project_code_less,project_code_greater,project_name_less,project_name_greater,
  764. win_tenderer_less,win_tenderer_greater,win_bid_price_less,win_bid_price_greater,
  765. bidding_budget_less,bidding_budget_greater,doctitle_refine_less,doctitle_refine_greater):
  766. return
  767. _context = json.loads(json_context)
  768. min_counts = 100
  769. dict_context = {}
  770. for item in _context:
  771. if item["counts"]<min_counts:
  772. min_counts = item["counts"]
  773. dict_context[item["_type"]] = [item["is_exists"],item["counts"]]
  774. context_key = ["tenderee","agency","project_code","project_name","win_tenderer","win_bid_price","bidding_budget","doctitle_refine"]
  775. list_matrix = []
  776. for index_i in range(len(context_key)):
  777. for index_j in range(index_i+1,len(context_key)):
  778. _key = "%s&%s"%(context_key[index_i],context_key[index_j])
  779. _v = featurnCount(dict_context.get(_key,[0,0])[1])
  780. list_matrix.append(_v)
  781. context3_key = ["tenderee","agency","win_tenderer","win_bid_price","bidding_budget"]
  782. for index_i in range(len(context3_key)):
  783. for index_j in range(index_i+1,len(context3_key)):
  784. for index_k in range(index_j+1,len(context3_key)):
  785. _key = "%s&%s&%s"%(context3_key[index_i],context3_key[index_j],context3_key[index_k])
  786. _v = featurnCount(dict_context.get(_key,[0,0])[1])
  787. list_matrix.append(_v)
  788. list_matrix.append(getSimLevel(tenderee_less,tenderee_greater)/10)
  789. list_matrix.append(getSimLevel(agency_less,agency_greater)/10)
  790. list_matrix.append(getSimilarityOfString(project_code_less,project_code_greater))
  791. list_matrix.append(getSimilarityOfString(project_name_less,project_name_greater))
  792. list_matrix.append(getSimLevel(win_tenderer_less,win_tenderer_greater)/10)
  793. list_matrix.append(getSimLevel(win_bid_price_less,win_bid_price_greater)/10)
  794. list_matrix.append(getSimLevel(bidding_budget_less,bidding_budget_greater)/10)
  795. list_matrix.append(getSimilarityOfString(doctitle_refine_less,doctitle_refine_greater))
  796. # set_tenderer = set()
  797. # if tenderee_less is not None and tenderee_less!="":
  798. # set_tenderer.add(tenderee_less)
  799. # if tenderee_greater is not None and tenderee_greater!="":
  800. # set_tenderer.add(tenderee_greater)
  801. #
  802. # set_win_tenderer = set()
  803. # if win_tenderer_less is not None and win_tenderer_less!="":
  804. # set_win_tenderer.add(win_tenderer_less)
  805. # if win_tenderer_greater is not None and win_tenderer_greater!="":
  806. # set_win_tenderer.add(win_tenderer_greater)
  807. #
  808. # set_bidding_budget = set()
  809. # if bidding_budget_less is not None and bidding_budget_less!="":
  810. # set_bidding_budget.add(bidding_budget_less)
  811. # if bidding_budget_greater is not None and bidding_budget_greater!="":
  812. # set_bidding_budget.add(bidding_budget_greater)
  813. #
  814. # set_win_bid_price = set()
  815. # if win_bid_price_less is not None and win_bid_price_less!="":
  816. # set_win_bid_price.add(win_bid_price_less)
  817. # if win_bid_price_greater is not None and win_bid_price_greater!="":
  818. # set_win_bid_price.add(win_bid_price_greater)
  819. json_matrix = json.dumps(list_matrix)
  820. same_project_code = False
  821. if project_code_less==project_code_greater and getLength(project_code_less)>0:
  822. same_project_code = True
  823. same_project_name = False
  824. if project_name_less==project_name_greater and getLength(project_name_less)>0:
  825. same_project_name = True
  826. same_doctitle_refine = False
  827. if doctitle_refine_less==doctitle_refine_greater and getLength(doctitle_refine_less)>0:
  828. same_doctitle_refine = True
  829. same_tenderee = False
  830. if tenderee_less==tenderee_greater and getLength(tenderee_less)>0:
  831. same_tenderee = True
  832. same_agency = False
  833. if agency_less==agency_greater and getLength(agency_less)>0:
  834. same_agency = True
  835. same_bidding_budget = False
  836. if bidding_budget_less==bidding_budget_greater and getLength(bidding_budget_less)>0:
  837. same_bidding_budget = True
  838. same_win_tenderer = False
  839. if win_tenderer_less==win_tenderer_greater and getLength(win_tenderer_less)>0:
  840. same_win_tenderer = True
  841. same_win_bid_price = False
  842. if win_bid_price_less==win_bid_price_greater and getLength(win_bid_price_less)>0:
  843. same_win_bid_price = True
  844. contain_doctitle = False
  845. if getLength(doctitle_refine_less)>0 and getLength(doctitle_refine_greater)>0 and (doctitle_refine_less in doctitle_refine_greater or doctitle_refine_greater in doctitle_refine_less):
  846. contain_doctitle = True
  847. contain_project_name = False
  848. if getLength(project_name_less)>0 and getLength(project_name_greater)>0 and (project_name_less in project_name_greater or project_name_greater in project_name_less):
  849. contain_project_name = True
  850. total_money_less = 0 if getLength(bidding_budget_less)==0 else float(bidding_budget_less)+0 if getLength(win_bid_price_less)==0 else float(win_bid_price_less)
  851. total_money_greater = 0 if getLength(bidding_budget_greater)==0 else float(bidding_budget_greater) +0 if getLength(win_bid_price_greater)==0 else float(win_bid_price_greater)
  852. if min_counts<10:
  853. _prob = 0.9
  854. if same_project_code and same_win_tenderer and same_tenderee:
  855. self.forward(json_matrix,_prob)
  856. return
  857. if same_tenderee and same_project_name and same_win_tenderer:
  858. self.forward(json_matrix,_prob)
  859. return
  860. if same_tenderee and same_doctitle_refine and same_win_tenderer:
  861. self.forward(json_matrix,_prob)
  862. return
  863. if same_tenderee and same_win_bid_price and same_win_tenderer:
  864. self.forward(json_matrix,_prob)
  865. return
  866. if same_project_code and same_win_bid_price and same_win_tenderer:
  867. self.forward(json_matrix,_prob)
  868. return
  869. if same_project_name and same_win_bid_price and same_win_tenderer:
  870. self.forward(json_matrix,_prob)
  871. return
  872. if same_doctitle_refine and same_win_bid_price and same_win_tenderer:
  873. self.forward(json_matrix,_prob)
  874. return
  875. if same_doctitle_refine and same_bidding_budget and same_win_tenderer:
  876. self.forward(json_matrix,_prob)
  877. return
  878. if same_tenderee and same_doctitle_refine and same_win_tenderer:
  879. self.forward(json_matrix,_prob)
  880. return
  881. if same_tenderee and same_project_code and same_project_name:
  882. self.forward(json_matrix,_prob)
  883. return
  884. if same_tenderee and same_project_code and same_doctitle_refine:
  885. self.forward(json_matrix,_prob)
  886. return
  887. if same_tenderee and same_bidding_budget and same_project_code:
  888. self.forward(json_matrix,_prob)
  889. return
  890. if same_tenderee and same_bidding_budget and same_doctitle_refine:
  891. self.forward(json_matrix,_prob)
  892. return
  893. if same_tenderee and same_bidding_budget and same_project_name:
  894. self.forward(json_matrix,_prob)
  895. return
  896. if same_doctitle_refine and same_project_code and same_project_name:
  897. self.forward(json_matrix,_prob)
  898. return
  899. if min_counts<=5:
  900. _prob = 0.8
  901. if same_project_code and same_tenderee:
  902. self.forward(json_matrix,_prob)
  903. return
  904. if same_project_code and same_win_tenderer:
  905. self.forward(json_matrix,_prob)
  906. return
  907. if same_project_name and same_project_code:
  908. self.forward(json_matrix,_prob)
  909. return
  910. if same_project_code and same_doctitle_refine:
  911. self.forward(json_matrix,_prob)
  912. return
  913. if total_money_less==total_money_greater and total_money_less>100000:
  914. if same_win_tenderer and (same_win_bid_price or same_bidding_budget):
  915. self.forward(json_matrix,_prob)
  916. return
  917. if same_project_code and same_bidding_budget:
  918. self.forward(json_matrix,_prob)
  919. return
  920. if same_project_code and same_win_bid_price:
  921. self.forward(json_matrix,_prob)
  922. return
  923. if same_bidding_budget and same_win_bid_price and (contain_project_name or contain_doctitle):
  924. self.forward(json_matrix,_prob)
  925. return
  926. if min_counts<=3:
  927. _prob = 0.7
  928. if same_project_name or same_project_code or same_doctitle_refine or contain_doctitle or contain_project_name:
  929. self.forward(json_matrix,_prob)
  930. return
  931. self.forward(json_matrix,0)
  932. class MergePredictor():
  933. def __init__(self):
  934. self.input_size = 46
  935. self.output_size = 2
  936. self.matrix = np.array([[-5.817399024963379, 3.367797374725342], [-18.3098201751709, 17.649206161499023], [-7.115952014923096, 9.236002922058105], [-5.054129123687744, 1.8316771984100342], [6.391637325286865, -7.57396125793457], [-2.8721542358398438, 6.826520919799805], [-5.426159858703613, 10.235260009765625], [-4.240962982177734, -0.32092899084091187], [-0.6378090381622314, 0.4834124445915222], [-1.7574478387832642, -0.17846578359603882], [4.325063228607178, -2.345501661300659], [0.6086963415145874, 0.8325914740562439], [2.5674285888671875, 1.8432368040084839], [-11.195490837097168, 17.4630184173584], [-11.334247589111328, 10.294097900390625], [2.639320135116577, -8.072785377502441], [-2.2689898014068604, -3.6194612979888916], [-11.129570960998535, 18.907018661499023], [4.526485919952393, 4.57423210144043], [-3.170452356338501, -1.3847776651382446], [-0.03280467540025711, -3.0471489429473877], [-6.601675510406494, -10.05613899230957], [-2.9116673469543457, 4.819308280944824], [1.4398306608200073, -0.6549674272537231], [7.091512203216553, -0.142232745885849], [-0.14478975534439087, 0.06628061085939407], [-6.775437831878662, 9.279582023620605], [-0.006781991105526686, 1.6472798585891724], [3.83730149269104, 1.4072834253311157], [1.2229349613189697, -2.1653425693511963], [1.445560336112976, -0.8397432565689087], [-11.325132369995117, 11.231744766235352], [2.3229124546051025, -4.623719215393066], [0.38562265038490295, -1.2645516395568848], [-1.3670002222061157, 2.4323790073394775], [-3.6994268894195557, 0.7515658736228943], [-0.11617227643728256, -0.820703387260437], [4.089913368225098, -4.693605422973633], [-0.4959050714969635, 1.5272167921066284], [-2.7135870456695557, -0.5120691657066345], [0.573157548904419, -1.9375460147857666], [-4.262857437133789, 0.6375582814216614], [-1.8825865983963013, 2.427532911300659], [-4.565115451812744, 4.0269083976745605], [-4.339804649353027, 6.754288196563721], [-4.31907320022583, 0.28193211555480957]])
  937. self.bias = np.array([16.79706382751465, -13.713337898254395])
  938. # self.model = load_model("model/merge.h5",custom_objects={"precision":precision,"recall":recall,"f1_score":f1_score})
  939. def activation(self,vec,_type):
  940. if _type=="relu":
  941. _vec = np.array(vec)
  942. return _vec*(_vec>0)
  943. if _type=="tanh":
  944. return np.tanh(vec)
  945. if _type=="softmax":
  946. _vec = np.array(vec)
  947. _exp = np.exp(_vec)
  948. return _exp/np.sum(_exp)
  949. def predict(self,input):
  950. _out = self.activation(self.activation(np.matmul(np.array(input).reshape(-1,self.input_size),self.matrix)+self.bias,"tanh"),"softmax")
  951. # print(self.model.predict(np.array(input).reshape(-1,46)))
  952. return _out
  953. @annotate('string,double -> double')
  954. class f_getMergeProb(BaseUDTF):
  955. def __init__(self):
  956. import json
  957. include_package_path("numpy-1.18.zip")
  958. import numpy as np
  959. global json,np
  960. self.mp = MergePredictor()
  961. def process(self,json_matrix,pre_prob):
  962. if not pre_prob>0.5:
  963. _matrix = json.loads(json_matrix)
  964. _prob = self.mp.predict(_matrix)[0][1]
  965. else:
  966. _prob = pre_prob
  967. if _prob>0.5:
  968. self.forward(float(_prob))
  969. @annotate('string -> bigint,bigint')
  970. class f_check_remerge_channel(BaseUDTF):
  971. '''
  972. 将多个组拆解成多条记录
  973. '''
  974. def __init__(self):
  975. import logging
  976. import json
  977. global json,logging
  978. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  979. def process(self,json_remerge):
  980. if json_remerge is not None:
  981. list_group = json.loads(json_remerge)
  982. for _group in list_group:
  983. _keys = _group.get("data").keys()
  984. if len(_keys)>0:
  985. main_docid = int(list(_keys)[0])
  986. for k,v in _group.get("data",{}).items():
  987. self.forward(main_docid,int(k))
  988. for _v in v:
  989. self.forward(main_docid,int(_v))
  990. @annotate('string -> bigint,bigint')
  991. class f_check_remerge(BaseUDTF):
  992. '''
  993. 将多个组拆解成多条记录
  994. '''
  995. def __init__(self):
  996. import logging
  997. import json
  998. global json,logging
  999. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  1000. def process(self,json_remerge):
  1001. if json_remerge is not None:
  1002. list_group = json.loads(json_remerge)
  1003. for _group in list_group:
  1004. for _docid in _group:
  1005. self.forward(_group[-1],_docid)
  1006. def getConfidence(rule_id):
  1007. if rule_id >=1 and rule_id <=20:
  1008. return 30
  1009. elif rule_id>=31 and rule_id<=50:
  1010. return 20
  1011. else:
  1012. return 10
  1013. @annotate('string,bigint -> bigint,bigint,bigint')
  1014. class f_arrange_group_single(BaseUDTF):
  1015. '''
  1016. 将多个组拆解成多条记录
  1017. '''
  1018. def __init__(self):
  1019. import logging
  1020. import json
  1021. global json,logging
  1022. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  1023. def process(self,json_set_docid,rule_id):
  1024. if json_set_docid is not None:
  1025. list_group = json.loads(json_set_docid)
  1026. for _group in list_group:
  1027. for index_i in range(len(_group)):
  1028. for index_j in range(len(_group)):
  1029. # if index_i!=index_j and _group[index_i]!=_group[index_j]:
  1030. if index_i!=index_j:
  1031. self.forward(_group[index_i],_group[index_j],getConfidence(rule_id))
  1032. @annotate('bigint,bigint->string')
  1033. class f_get_merge_docids(BaseUDAF):
  1034. '''
  1035. 合并组为一条记录
  1036. '''
  1037. def __init__(self):
  1038. import json
  1039. global json
  1040. def new_buffer(self):
  1041. return [set()]
  1042. def iterate(self, buffer,docid1,docid2):
  1043. buffer[0].add(docid1)
  1044. buffer[0].add(docid2)
  1045. def merge(self, buffer, pbuffer):
  1046. buffer[0] |= pbuffer[0]
  1047. def terminate(self, buffer):
  1048. set_docid = buffer[0]
  1049. list_docid = list(set_docid)
  1050. list_docid.sort(key=lambda x:x)
  1051. list_docid_str = []
  1052. for _docid in list_docid:
  1053. list_docid_str.append(str(_docid))
  1054. return ",".join(list_docid_str)
  1055. @annotate("string,string,string,string,string,string,string,string,string,string,string,string,string,string->string")
  1056. class f_encode_time(object):
  1057. def evaluate(self,time_bidclose,time_bidopen,time_bidstart,time_commencement,time_completion,time_earnest_money_end,time_earnest_money_start,time_get_file_end,time_get_file_start,time_publicity_end,time_publicity_start,time_registration_end,time_registration_start,time_release):
  1058. _dict = {"time_bidclose":time_bidclose,"time_bidopen":time_bidopen,"time_bidstart":time_bidstart,
  1059. "time_commencement":time_commencement,"time_completion":time_completion,"time_earnest_money_end":time_earnest_money_end,
  1060. "time_earnest_money_start":time_earnest_money_start,"time_get_file_end":time_get_file_end,"time_get_file_start":time_get_file_start,
  1061. "time_publicity_end":time_publicity_end,"time_publicity_start":time_publicity_start,"time_registration_end":time_registration_end,
  1062. "time_registration_start":time_registration_start,"time_release":time_release}
  1063. _encode = json.dumps(_dict)
  1064. return _encode
  1065. @annotate('string,string -> string,string')
  1066. class f_decode_ruwei(BaseUDTF):
  1067. def __init__(self):
  1068. import logging
  1069. import json
  1070. global json,logging
  1071. logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  1072. def process(self, page_time,sub_docs_json):
  1073. if sub_docs_json is not None:
  1074. for sub_docs in json.loads(sub_docs_json):
  1075. if sub_docs.get("win_tenderer","")!="":
  1076. self.forward(page_time,sub_docs.get("win_tenderer",""))
  1077. if sub_docs.get("second_tenderer","")!="":
  1078. self.forward(page_time,sub_docs.get("second_tenderer",""))
  1079. if sub_docs.get("third_tenderer","")!="":
  1080. self.forward(page_time,sub_docs.get("third_tenderer",""))
  1081. if __name__ == '__main__':
  1082. a = f_remege_limit_num_contain_bychannel()
  1083. buffer = a.new_buffer()
  1084. tmp_s = '''
  1085. 266523906 266539038 2022-09-08 1662566400 SDGP371525000202201000421_A 冠县第二实验小学平台教育信息化设备采购智慧屏 冠县第二实验小学平台教育信息化设备采购智慧屏成交公告 冠县第二实验小学平台教育信息化设备智慧屏 冠县第二实验小学 聊城市采购中心 山东润博网络有限公司 246890.0 101 0 12 "{"time_bidclose": "", "time_bidopen": "", "time_bidstart": "", "time_commencement": "", "time_completion": "", "time_earnest_money_end": "", "time_earnest_money_start": "", "time_get_file_end": "", "time_get_file_start": "", "time_publicity_end": "", "time_publicity_start": "", "time_registration_end": "", "time_registration_start": "", "time_release": ""}"
  1086. 266523906 266523906 2022-09-15 1663171200 SDGP371525000202201000421_A 冠县第二实验小学平台教育信息化设备采购智慧屏 冠县第二实验小学平台教育信息化设备采购智慧屏成交公告 冠县第二实验小学平台教育信息化设备智慧屏 冠县第二实验小学 聊城市采购中心 山东润博网络有限公司 246890.0 101 999 12 "{"time_bidclose": "", "time_bidopen": "", "time_bidstart": "", "time_commencement": "", "time_completion": "", "time_earnest_money_end": "", "time_earnest_money_start": "", "time_get_file_end": "", "time_get_file_start": "", "time_publicity_end": "", "time_publicity_start": "", "time_registration_end": "", "time_registration_start": "", "time_release": ""}"
  1087. '''
  1088. for _s in tmp_s.split("\n"):
  1089. ls = _s.split("\t")
  1090. if len(ls)!=17:
  1091. continue
  1092. _confid = 1 if ls[14] =="" else ls[14]
  1093. a.iterate(buffer,ls[1],ls[13],int(ls[3]),ls[8],ls[10],ls[11],ls[12],ls[7],ls[5],ls[4],_confid,ls[15],ls[16][1:-1])
  1094. # a.iterate(buffer,219957825,101,86400*4,"1","1","1","1","1","1","1",0,5,'{"time_bidclose": "", "time_bidopen": "2022-02-10", "time_bidstart": "", "time_commencement": "", "time_completion": "", "time_earnest_money_end": "", "time_earnest_money_start": "", "time_get_file_end": "", "time_get_file_start": "", "time_publicity_end": "2022-02-21", "time_publicity_start": "2022-02-11", "time_registration_end": "", "time_registration_start": "", "time_release": ""}')
  1095. # a.iterate(buffer,219957825,101,86400*4,"1","1","1","1","1","1","1",0,5,'{"time_bidclose": "", "time_bidopen": "2022-02-10", "time_bidstart": "", "time_commencement": "", "time_completion": "", "time_earnest_money_end": "", "time_earnest_money_start": "", "time_get_file_end": "", "time_get_file_start": "", "time_publicity_end": "2022-02-21", "time_publicity_start": "2022-02-11", "time_registration_end": "", "time_registration_start": "", "time_release": ""}')
  1096. # a.iterate(buffer,219957825,101,86400*4,"1","1","1","1","1","1","1",0,5,'{"time_bidclose": "", "time_bidopen": "2022-02-10", "time_bidstart": "", "time_commencement": "", "time_completion": "", "time_earnest_money_end": "", "time_earnest_money_start": "", "time_get_file_end": "", "time_get_file_start": "", "time_publicity_end": "2022-02-22", "time_publicity_start": "2022-02-11", "time_registration_end": "", "time_registration_start": "", "time_release": ""}')
  1097. print(a.terminate(buffer))
  1098. print(1)
  1099. print(getSimilarityOfString('37168100014015220220012_40785671','SDGP371681000202201000912'))