dataflow_mq.py 70 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558
  1. from BaseDataMaintenance.maintenance.dataflow import *
  2. from BaseDataMaintenance.common.activateMQUtils import *
  3. from BaseDataMaintenance.dataSource.source import getConnect_activateMQ,getConnection_postgres,getConnection_mysql,getConnection_oracle,getConnect_ots_capacity,getConnect_redis_doc
  4. from BaseDataMaintenance.dataSource.setttings import *
  5. from BaseDataMaintenance.model.postgres.attachment import Attachment_postgres
  6. import os
  7. from BaseDataMaintenance.common.ossUtils import *
  8. from BaseDataMaintenance.dataSource.pool import ConnectorPool
  9. from BaseDataMaintenance.model.ots.document import Document
  10. from BaseDataMaintenance.common.Utils import article_limit
  11. from BaseDataMaintenance.common.documentFingerprint import getFingerprint
  12. from BaseDataMaintenance.model.postgres.document_extract import *
  13. import sys
  14. sys.setrecursionlimit(1000000)
  15. class ActiveMQListener():
  16. def __init__(self,conn,_queue,*args,**kwargs):
  17. self.conn = conn
  18. self._queue = _queue
  19. def on_error(self, headers):
  20. log("===============")
  21. log('received an error %s' % str(headers.body))
  22. def on_message(self, headers):
  23. log("====message====")
  24. message_id = headers.headers["message-id"]
  25. body = headers.body
  26. self._queue.put({"frame":headers,"conn":self.conn},True)
  27. def __del__(self):
  28. self.conn.disconnect()
  29. class Dataflow_ActivteMQ_attachment(Dataflow_attachment):
  30. def __init__(self):
  31. Dataflow_attachment.__init__(self)
  32. self.mq_attachment = "/queue/dataflow_attachment"
  33. self.mq_attachment_failed = "/queue/dataflow_attachment_failed"
  34. self.mq_extract = "/queue/dataflow_extract"
  35. self.comsumer_count = 120
  36. self.retry_comsumer_count = 10
  37. self.retry_times = 5
  38. self.list_attachment_comsumer = []
  39. for _i in range(self.comsumer_count):
  40. listener_attachment = ActiveMQListener(getConnect_activateMQ(),self.queue_attachment)
  41. createComsumer(listener_attachment,self.mq_attachment)
  42. self.list_attachment_comsumer.append(listener_attachment)
  43. self.attach_pool = ConnectorPool(10,30,getConnection_postgres)
  44. self.conn_mq = getConnect_activateMQ()
  45. self.pool_mq = ConnectorPool(10,30,getConnect_activateMQ)
  46. self.session = requests.Session()
  47. def monitor_listener(self):
  48. for i in range(len(self.list_attachment_comsumer)):
  49. if self.list_attachment_comsumer[i].conn.is_connected():
  50. continue
  51. else:
  52. listener = ActiveMQListener(getConnect_activateMQ(),self.queue_attachment)
  53. createComsumer(listener,self.mq_attachment)
  54. self.list_attachment_comsumer[i] = listener
  55. def process_failed_attachment(self):
  56. from BaseDataMaintenance.java.MQInfo import getQueueSize
  57. attachment_size = getQueueSize("dataflow_attachment")
  58. failed_attachment_size = getQueueSize("dataflow_attachment_failed")
  59. if attachment_size<100 and failed_attachment_size>0:
  60. list_comsumer = []
  61. for _i in range(self.retry_comsumer_count):
  62. listener_attachment = ActiveMQListener(getConnect_activateMQ(),self.queue_attachment)
  63. list_comsumer.append(listener_attachment)
  64. createComsumer(listener_attachment,self.mq_attachment_failed)
  65. while 1:
  66. failed_attachment_size = getQueueSize("dataflow_attachment_failed")
  67. if failed_attachment_size==0:
  68. break
  69. time.sleep(10)
  70. for _c in list_comsumer:
  71. _c.conn.disconnect()
  72. def rec_attachments_by_interface(self,list_attach,_dochtmlcon,save=True):
  73. try:
  74. list_html = []
  75. swf_urls = []
  76. _not_failed = True
  77. for _attach in list_attach:
  78. #测试全跑
  79. _filemd5 = _attach.getProperties().get(attachment_filemd5)
  80. if _attach.getProperties().get(attachment_status) in (ATTACHMENT_PROCESSED,ATTACHMENT_TOOLARGE):
  81. log("%s has processed or toolarge"%(_filemd5))
  82. _html = _attach.getProperties().get(attachment_attachmenthtml,"")
  83. if _html is None:
  84. _html = ""
  85. list_html.append({attachment_filemd5:_filemd5,
  86. "html":_html})
  87. else:
  88. #has process_time then jump
  89. if len(str(_attach.getProperties().get(attachment_process_time,"")))>10 and _attach.getProperties().get(attachment_status)!=ATTACHMENT_INIT and not (_attach.getProperties().get(attachment_status)>=ATTACHMENT_MC_FAILED_FROM and _attach.getProperties().get(attachment_status)<=ATTACHMENT_MC_FAILED_TO):
  90. log("%s has process_time jump"%(_filemd5))
  91. _html = _attach.getProperties().get(attachment_attachmenthtml,"")
  92. if _html is None:
  93. _html = ""
  94. list_html.append({attachment_filemd5:_filemd5,
  95. "html":_html})
  96. else:
  97. log("%s requesting interface"%(_filemd5))
  98. _succeed = self.request_attachment_interface(_attach,_dochtmlcon)
  99. if not _succeed:
  100. _not_failed = False
  101. _html = _attach.getProperties().get(attachment_attachmenthtml,"")
  102. if _html is None:
  103. _html = ""
  104. list_html.append({attachment_filemd5:_filemd5,
  105. "html":_html})
  106. if _attach.getProperties().get(attachment_filetype)=="swf":
  107. swf_urls.extend(json.loads(_attach.getProperties().get(attachment_swfUrls,"[]")))
  108. if not _not_failed:
  109. return False,list_html,swf_urls
  110. return True,list_html,swf_urls
  111. except requests.ConnectionError as e1:
  112. raise e1
  113. except Exception as e:
  114. return False,list_html,swf_urls
  115. def attachment_recognize(self,_dict,result_queue):
  116. '''
  117. 识别附件内容
  118. :param _dict: 附件内容
  119. :param result_queue:
  120. :return:
  121. '''
  122. try:
  123. item = _dict.get("item")
  124. list_attach = _dict.get("list_attach")
  125. conn = _dict["conn"]
  126. message_id = _dict.get("message_id")
  127. _retry_times = item.get("retry_times",0)
  128. dhtml = Document_html({"partitionkey":item.get("partitionkey"),
  129. "docid":item.get("docid")})
  130. _dochtmlcon = item.get(document_tmp_dochtmlcon,"")
  131. dhtml.setValue(document_tmp_dochtmlcon,_dochtmlcon,True)
  132. dhtml.delete_bidi_a()
  133. dtmp = Document_tmp(item)
  134. start_time = time.time()
  135. #调用识别接口
  136. _succeed,list_html,swf_urls = self.rec_attachments_by_interface(list_attach,_dochtmlcon,save=True)
  137. _to_ack = False
  138. if not _succeed and _retry_times<self.retry_times:
  139. item[document_tmp_status] = random.randint(*flow_attachment_status_failed_to)
  140. item["retry_times"] = _retry_times+1
  141. #失败次数大于5次就放入失败队列,此队列的数据会在空闲时间重新处理一次
  142. if item["retry_times"]>=self.retry_times:
  143. send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment_failed)
  144. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment)
  145. #失败保存
  146. if _retry_times==0:
  147. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  148. dtmp.setValue(document_tmp_status,0,True)
  149. if not dtmp.exists_row(self.ots_client):
  150. dtmp.update_row(self.ots_client)
  151. dhtml.update_row(self.ots_client)
  152. if send_succeed:
  153. _to_ack = True
  154. else:
  155. try:
  156. log("docid:%d,retry:%d swf_urls:%s list_html:%s"%(dhtml.getProperties().get(document_docid),_retry_times,str(swf_urls),str(len(list_html))))
  157. dhtml.updateSWFImages(swf_urls)
  158. dhtml.updateAttachment(list_html)
  159. dtmp.setValue(document_tmp_attachment_extract_status,1,True)
  160. dtmp.setValue(document_tmp_dochtmlcon,dhtml.getProperties().get(document_tmp_dochtmlcon),True)
  161. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(dtmp.getProperties(),cls=MyEncoder),self.mq_extract)
  162. if send_succeed:
  163. _to_ack = True
  164. except Exception as e:
  165. traceback.print_exc()
  166. if _to_ack:
  167. ackMsg(conn,message_id)
  168. log("document:%d get attachments with result:%s %s retry_times:%d"%(item.get("docid"),str(_succeed),str(_to_ack),_retry_times))
  169. except Exception as e:
  170. traceback.print_exc()
  171. if time.time()-start_time<10:
  172. item["retry_times"] -= 1
  173. if send_msg_toacmq(self.pool_mq,json.dumps(item,cls=MyEncoder,ensure_ascii=False),self.mq_attachment):
  174. ackMsg(conn,message_id)
  175. def request_attachment_interface(self,attach,_dochtmlcon):
  176. filemd5 = attach.getProperties().get(attachment_filemd5)
  177. _status = attach.getProperties().get(attachment_status)
  178. _filetype = attach.getProperties().get(attachment_filetype)
  179. _path = attach.getProperties().get(attachment_path)
  180. _uuid = uuid4()
  181. objectPath = attach.getProperties().get(attachment_path)
  182. docids = attach.getProperties().get(attachment_docids)
  183. if objectPath is None:
  184. relative_path = "%s/%s"%(_uuid.hex[:4],_uuid.hex)
  185. else:
  186. relative_path = objectPath[5:].replace("//","/")
  187. localpath = "/FileInfo/%s"%(relative_path)
  188. if not os.path.exists(localpath):
  189. if not os.path.exists(os.path.dirname(localpath)):
  190. os.makedirs(os.path.dirname(localpath))
  191. local_exists = False
  192. else:
  193. local_exists = True
  194. _size = os.path.getsize(localpath)
  195. not_failed_flag = True
  196. try:
  197. d_start_time = time.time()
  198. if not local_exists:
  199. log("md5:%s path:%s not exists,start downloading"%(filemd5,objectPath))
  200. try:
  201. download_succeed = downloadFile(self.bucket,objectPath,localpath)
  202. except Exception as e:
  203. download_succeed = False
  204. else:
  205. log("md5:%s path:%s exists"%(filemd5,objectPath[5:]))
  206. if not (local_exists or download_succeed):
  207. _ots_attach = attachment(attach.getProperties_ots())
  208. _ots_exists = _ots_attach.fix_columns(self.ots_client,[attachment_attachmenthtml,attachment_classification,attachment_attachmentcon,attachment_path,attachment_status,attachment_filetype],True)
  209. log("md5:%s path:%s file not in local or oss,search ots.attachment"%(filemd5,objectPath))
  210. if _ots_attach.getProperties().get(attachment_attachmenthtml,"")!="" and str(_ots_attach.getProperties().get(attachment_status))!=str(ATTACHMENT_INIT):
  211. attach.setValue(attachment_attachmenthtml,_ots_attach.getProperties().get(attachment_attachmenthtml,""))
  212. attach.setValue(attachment_attachmentcon,_ots_attach.getProperties().get(attachment_attachmentcon,""))
  213. attach.setValue(attachment_status,_ots_attach.getProperties().get(attachment_status,""))
  214. attach.setValue(attachment_filetype,_ots_attach.getProperties().get(attachment_filetype,""))
  215. attach.setValue(attachment_classification,_ots_attach.getProperties().get(attachment_classification,""))
  216. if attach.exists(self.attach_pool):
  217. attach.update_row(self.attach_pool)
  218. else:
  219. attach.insert_row(self.attach_pool)
  220. try:
  221. if os.exists(localpath):
  222. os.remove(localpath)
  223. except Exception as e:
  224. pass
  225. return True
  226. if _ots_exists:
  227. objectPath = attach.getProperties().get(attachment_path)
  228. download_succeed = downloadFile(self.bucket,objectPath,localpath)
  229. if download_succeed:
  230. log("md5:%s path:%s download file from oss succeed"%(filemd5,objectPath))
  231. else:
  232. log("md5:%s path:%s download file from ots failed=="%(filemd5,objectPath))
  233. else:
  234. log("md5:%s path:%s not found in ots"%(filemd5,objectPath))
  235. if local_exists or download_succeed:
  236. _size = os.path.getsize(localpath)
  237. attach.setValue(attachment_size,_size,True)
  238. if _size>ATTACHMENT_LARGESIZE:
  239. attach.setValue(attachment_status, ATTACHMENT_TOOLARGE,True)
  240. log("attachment :%s of path:%s to large"%(filemd5,_path))
  241. _ots_attach = attachment(attach.getProperties_ots())
  242. _ots_attach.update_row(self.ots_client)
  243. #更新postgres
  244. if attach.exists(self.attach_pool):
  245. attach.update_row(self.attach_pool)
  246. else:
  247. attach.insert_row(self.attach_pool)
  248. if local_exists:
  249. upload_status = uploadFileByPath(self.bucket,localpath,objectPath)
  250. os.remove(localpath)
  251. return True
  252. time_download = time.time()-d_start_time
  253. #调用接口处理结果
  254. start_time = time.time()
  255. _filetype = attach.getProperties().get(attachment_filetype)
  256. # _data_base64 = base64.b64encode(open(localpath,"rb").read())
  257. # _success,_html,swf_images = getAttachDealInterface(_data_base64,_filetype)
  258. _success,_html,swf_images,classification = getAttachDealInterface(None,_filetype,path=localpath,session=self.session)
  259. log("process filemd5:%s %s of type:%s with size:%.3fM download:%ds recognize takes %ds,ret_size:%d"%(filemd5,str(_success),_filetype,round(_size/1024/1024,4),time_download,time.time()-start_time,len(_html)))
  260. if _success:
  261. if len(_html)<5:
  262. _html = ""
  263. else:
  264. if len(_html)>1:
  265. _html = "interface return error"
  266. else:
  267. sentMsgToDD("attach interface failed of docid:%s of filemd5:%s of type:%s size:%.3fM with result:%s"%(str(docids),filemd5,_filetype,round(_size/1024/1024,4),str(_html)))
  268. _html = ""
  269. return False
  270. swf_images = eval(swf_images)
  271. if attach.getProperties().get(attachment_filetype)=="swf" and len(swf_images)>0:
  272. swf_urls = json.loads(attach.getProperties().get(attachment_swfUrls,"[]"))
  273. if len(swf_urls)==0:
  274. objectPath = attach.getProperties().get(attachment_path,"")
  275. swf_dir = os.path.join(self.current_path,"swf_images",uuid4().hex)
  276. if not os.path.exists(swf_dir):
  277. os.mkdir(swf_dir)
  278. for _i in range(len(swf_images)):
  279. _base = swf_images[_i]
  280. _base = base64.b64decode(_base)
  281. filename = "swf_page_%d.png"%(_i)
  282. filepath = os.path.join(swf_dir,filename)
  283. with open(filepath,"wb") as f:
  284. f.write(_base)
  285. swf_urls = transformSWF(self.bucket,self.attachment_hub_url,objectPath,None,swf_dir)
  286. if os.path.exists(swf_dir):
  287. os.rmdir(swf_dir)
  288. attach.setValue(attachment_swfUrls,json.dumps(swf_urls,ensure_ascii=False),True)
  289. if re.search("<td",_html) is not None:
  290. attach.setValue(attachment_has_table,1,True)
  291. _file_title = self.getTitleFromHtml(filemd5,_dochtmlcon)
  292. filelink = self.getSourceLinkFromHtml(filemd5,_dochtmlcon)
  293. if _file_title!="":
  294. attach.setValue(attachment_file_title,_file_title,True)
  295. if filelink!="":
  296. attach.setValue(attachment_file_link,filelink,True)
  297. attach.setValue(attachment_attachmenthtml,_html,True)
  298. attach.setValue(attachment_attachmentcon,BeautifulSoup(_html,"lxml").get_text(),True)
  299. attach.setValue(attachment_status,ATTACHMENT_PROCESSED,True)
  300. attach.setValue(attachment_recsize,len(_html),True)
  301. attach.setValue(attachment_process_time,getCurrent_date(format="%Y-%m-%d %H:%M:%S"),True)
  302. attach.setValue(attachment_classification,classification,True)
  303. #更新ots
  304. _ots_attach = attachment(attach.getProperties_ots())
  305. _ots_attach.update_row(self.ots_client) #线上再开放更新
  306. #更新postgres
  307. if attach.exists(self.attach_pool):
  308. attach.update_row(self.attach_pool)
  309. else:
  310. attach.insert_row(self.attach_pool)
  311. if local_exists:
  312. upload_status = uploadFileByPath(self.bucket,localpath,objectPath)
  313. try:
  314. if upload_status and os.exists(localpath):
  315. os.remove(localpath)
  316. except Exception as e:
  317. pass
  318. return True
  319. else:
  320. return True
  321. except requests.ConnectionError as e1:
  322. raise e1
  323. except oss2.exceptions.NotFound as e:
  324. return True
  325. except Exception as e:
  326. traceback.print_exc()
  327. # def flow_attachment(self):
  328. # self.flow_attachment_producer()
  329. # self.flow_attachment_producer_comsumer()
  330. def getAttachPath(self,filemd5,_dochtmlcon):
  331. _soup = BeautifulSoup(_dochtmlcon,"lxml")
  332. _find = _soup.find("a",attrs={"data":filemd5})
  333. filelink = ""
  334. if _find is None:
  335. _find = _soup.find("img",attrs={"data":filemd5})
  336. if _find is not None:
  337. filelink = _find.attrs.get("src","")
  338. else:
  339. filelink = _find.attrs.get("href","")
  340. _path = filelink.split("/file")
  341. if len(_path)>1:
  342. return _path[1]
  343. def getAttachments(self,list_filemd5,_dochtmlcon):
  344. conn = self.attach_pool.getConnector()
  345. #搜索postgres
  346. try:
  347. to_find_md5 = []
  348. for _filemd5 in list_filemd5[:50]:
  349. if _filemd5 is not None:
  350. to_find_md5.append(_filemd5)
  351. conditions = ["filemd5 in ('%s')"%("','".join(to_find_md5))]
  352. list_attachment = Attachment_postgres.select_rows(conn,Attachment_postgres,"attachment",conditions)
  353. log("select localpath database %d/%d"%(len(list_attachment),len(to_find_md5)))
  354. set_md5 = set()
  355. for _attach in list_attachment:
  356. set_md5.add(_attach.getProperties().get(attachment_filemd5))
  357. list_not_in_md5 = []
  358. for _filemd5 in to_find_md5:
  359. if _filemd5 not in set_md5:
  360. list_not_in_md5.append(_filemd5)
  361. _path = self.getAttachPath(_filemd5,_dochtmlcon)
  362. log("getAttachments search in ots:%s"%(_filemd5))
  363. _attach = {attachment_filemd5:_filemd5}
  364. _attach_ots = attachment(_attach)
  365. if _attach_ots.fix_columns(self.ots_client,[attachment_status,attachment_path,attachment_attachmenthtml,attachment_attachmentcon,attachment_filetype,attachment_swfUrls,attachment_process_time],True):
  366. if _attach_ots.getProperties().get(attachment_status) is not None:
  367. log("getAttachments find in ots:%s"%(_filemd5))
  368. list_attachment.append(Attachment_postgres(_attach_ots.getProperties()))
  369. else:
  370. if _path[0]=="/":
  371. _path = _path[1:]
  372. _filetype = _path.split(".")[-1]
  373. _attach = {attachment_filemd5:_filemd5,
  374. attachment_filetype:_filetype,
  375. attachment_status:20,
  376. attachment_path:"%s/%s"%(_filemd5[:4],_path),
  377. attachment_crtime:getCurrent_date(format="%Y-%m-%d %H:%M:%S")}
  378. list_attachment.append(Attachment_postgres(_attach))
  379. return list_attachment
  380. except Exception as e:
  381. log("attachProcess comsumer error %s"%str(e))
  382. log(str(to_find_md5))
  383. traceback.print_exc()
  384. return []
  385. finally:
  386. self.attach_pool.putConnector(conn)
  387. def flow_attachment_producer(self,columns=[document_tmp_attachment_path,document_tmp_crtime]):
  388. q_size = self.queue_attachment.qsize()
  389. qsize_ocr = self.queue_attachment_ocr.qsize()
  390. qsize_not_ocr = self.queue_attachment_not_ocr.qsize()
  391. log("queue_attachment:%d,queue_attachment_ocr:%d,queue_attachment_not_ocr:%d"%(q_size,qsize_ocr,qsize_not_ocr))
  392. def flow_attachment_producer_comsumer(self):
  393. log("start flow_attachment comsumer")
  394. mt = MultiThreadHandler(self.queue_attachment,self.comsumer_handle,None,10,1,need_stop=False,restart=True)
  395. mt.run()
  396. def set_queue(self,_dict):
  397. list_attach = _dict.get("list_attach")
  398. to_ocr = False
  399. for attach in list_attach:
  400. if attach.getProperties().get(attachment_filetype) in ["bmp","jpeg","jpg","png","swf","pdf","tif"]:
  401. to_ocr = True
  402. break
  403. if to_ocr:
  404. self.queue_attachment_ocr.put(_dict,True)
  405. else:
  406. self.queue_attachment_not_ocr.put(_dict,True)
  407. def comsumer_handle(self,_dict,result_queue):
  408. try:
  409. frame = _dict["frame"]
  410. conn = _dict["conn"]
  411. message_id = frame.headers["message-id"]
  412. item = json.loads(frame.body)
  413. page_attachments = json.loads(item.get(document_tmp_attachment_path,"[]"))
  414. _dochtmlcon = item.get(document_tmp_dochtmlcon,"")
  415. if len(page_attachments)==0:
  416. self.set_queue({"item":item,"list_attach":[],"message_id":message_id,"conn":conn})
  417. else:
  418. list_fileMd5 = []
  419. for _atta in page_attachments:
  420. list_fileMd5.append(_atta.get(document_tmp_attachment_path_filemd5))
  421. list_attach = self.getAttachments(list_fileMd5,_dochtmlcon)
  422. self.set_queue({"item":item,"list_attach":list_attach,"message_id":message_id,"conn":conn})
  423. except Exception as e:
  424. traceback.print_exc()
  425. def remove_attachment_postgres(self):
  426. current_date = getCurrent_date(format="%Y-%m-%d")
  427. last_date = timeAdd(current_date,-2,format="%Y-%m-%d")
  428. sql = " delete from attachment where crtime<='%s 00:00:00' "%(last_date)
  429. conn = getConnection_postgres()
  430. cursor = conn.cursor()
  431. cursor.execute(sql)
  432. conn.commit()
  433. conn.close()
  434. def start_flow_attachment(self):
  435. schedule = BlockingScheduler()
  436. schedule.add_job(self.flow_attachment_process,"cron",second="*/20")
  437. # schedule.add_job(self.flow_attachment,"cron",second="*/10")
  438. schedule.add_job(self.flow_attachment_producer,"cron",second="*/10")
  439. schedule.add_job(self.flow_attachment_producer_comsumer,"cron",second="*/10")
  440. schedule.add_job(self.monitor_attachment_process,"cron",second="*/10")
  441. schedule.add_job(self.remove_attachment_postgres,"cron",hour="6")
  442. schedule.add_job(self.process_failed_attachment,"cron",minute="*/10")
  443. schedule.add_job(self.monitor_listener,"cron",minute="*/1")
  444. schedule.start()
  445. class Dataflow_ActivteMQ_extract(Dataflow_extract):
  446. class ExtractListener():
  447. def __init__(self,conn,_func,_idx,*args,**kwargs):
  448. self.conn = conn
  449. self._func = _func
  450. self._idx = _idx
  451. def on_message(self, headers):
  452. try:
  453. log("get message of idx:%d"%(self._idx))
  454. message_id = headers.headers["message-id"]
  455. body = headers.body
  456. log("get message %s crtime:%s"%(message_id,json.loads(body).get("crtime","")))
  457. self._func(_dict={"frame":headers,"conn":self.conn},result_queue=None)
  458. except Exception as e:
  459. traceback.print_exc()
  460. pass
  461. def on_error(self, headers):
  462. log('received an error %s' % str(headers.body))
  463. def __del__(self):
  464. self.conn.disconnect()
  465. def __init__(self):
  466. Dataflow_extract.__init__(self)
  467. self.industy_url = "http://127.0.0.1:15000/industry_extract"
  468. self.extract_interfaces = [["http://127.0.0.1:15030/content_extract",20],
  469. ["http://192.168.0.115:15030/content_extract",10]
  470. ]
  471. self.mq_extract = "/queue/dataflow_extract"
  472. self.mq_extract_failed = "/queue/dataflow_extract_failed"
  473. self.whole_weight = 0
  474. for _url,weight in self.extract_interfaces:
  475. self.whole_weight+= weight
  476. current_weight = 0
  477. for _i in range(len(self.extract_interfaces)):
  478. current_weight += self.extract_interfaces[_i][1]
  479. self.extract_interfaces[_i][1] = current_weight/self.whole_weight
  480. self.comsumer_count = 50
  481. self.pool_postgres = ConnectorPool(10,self.comsumer_count,getConnection_postgres)
  482. self.pool_redis_doc = ConnectorPool(10,self.comsumer_count,getConnect_redis_doc)
  483. self.conn_mq = getConnect_activateMQ()
  484. self.pool_mq = ConnectorPool(10,30,getConnect_activateMQ)
  485. self.block_url = RLock()
  486. self.url_count = 0
  487. self.session = requests.Session()
  488. self.list_extract_comsumer = []
  489. for _i in range(self.comsumer_count):
  490. listener_extract = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle,_i)
  491. createComsumer(listener_extract,self.mq_extract)
  492. self.list_extract_comsumer.append(listener_extract)
  493. def monitor_listener(self):
  494. for i in range(len(self.list_extract_comsumer)):
  495. if self.list_extract_comsumer[i].conn.is_connected():
  496. continue
  497. else:
  498. listener = self.ExtractListener(getConnect_activateMQ(),self.comsumer_handle)
  499. createComsumer(listener,self.mq_extract)
  500. self.list_extract_comsumer[i] = listener
  501. def getExtract_url(self):
  502. # _url_num = 0
  503. # with self.block_url:
  504. # self.url_count += 1
  505. # self.url_count %= self.whole_weight
  506. # _url_num = self.url_count
  507. _r = random.random()
  508. # _r = _url_num/self.whole_weight
  509. for _i in range(len(self.extract_interfaces)):
  510. if _r<=self.extract_interfaces[_i][1]:
  511. return self.extract_interfaces[_i][0]
  512. def request_extract_interface(self,json,headers):
  513. # _i = random.randint(0,len(self.extract_interfaces)-1)
  514. # _i = 0
  515. # _url = self.extract_interfaces[_i]
  516. _url = self.getExtract_url()
  517. log("extract_url:%s"%(str(_url)))
  518. resp = self.session.post(_url,json=json,headers=headers,timeout=10*60)
  519. return resp
  520. def request_industry_interface(self,json,headers):
  521. resp = requests.post(self.industy_url,json=json,headers=headers)
  522. return resp
  523. def flow_extract_producer(self,columns=[document_tmp_page_time,document_tmp_doctitle,document_tmp_docchannel,document_tmp_status,document_tmp_original_docchannel,document_tmp_web_source_no]):
  524. q_size = self.queue_extract.qsize()
  525. log("queue extract size:%d"%(q_size))
  526. def process_extract_failed(self):
  527. def _handle(_dict,result_queue):
  528. frame = _dict.get("frame")
  529. message_id = frame.headers["message-id"]
  530. subscription = frame.headers.setdefault('subscription', None)
  531. conn = _dict.get("conn")
  532. body = frame.body
  533. if body is not None:
  534. item = json.loads(body)
  535. item["extract_times"] = 10
  536. if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
  537. ackMsg(conn,message_id,subscription)
  538. from BaseDataMaintenance.java.MQInfo import getQueueSize
  539. extract_failed_size = getQueueSize("dataflow_extract_failed")
  540. extract_size = getQueueSize("dataflow_extract")
  541. log("extract_failed_size %s extract_size %s"%(str(extract_failed_size),str(extract_size)))
  542. if extract_failed_size>0 and extract_size<100:
  543. failed_listener = self.ExtractListener(getConnect_activateMQ(),_handle)
  544. createComsumer(failed_listener,self.mq_extract_failed)
  545. while 1:
  546. extract_failed_size = getQueueSize("dataflow_extract_failed")
  547. if extract_failed_size==0:
  548. break
  549. time.sleep(10)
  550. failed_listener.conn.disconnect()
  551. def flow_extract(self,):
  552. self.comsumer()
  553. def comsumer(self):
  554. mt = MultiThreadHandler(self.queue_extract,self.comsumer_handle,None,20,1,True)
  555. mt.run()
  556. def getExtract_json_fromDB(self,_fingerprint):
  557. conn = self.pool_postgres.getConnector()
  558. try:
  559. list_extract = Document_extract_postgres.select_rows(conn,Document_extract_postgres,"document_extract",[" fingerprint='%s'"%_fingerprint])
  560. if len(list_extract)>0:
  561. _extract = list_extract[0]
  562. return _extract.getProperties().get(document_extract_extract_json)
  563. except Exception as e:
  564. traceback.print_exc()
  565. finally:
  566. self.pool_postgres.putConnector(conn)
  567. return None
  568. def putExtract_json_toDB(self,fingerprint,docid,extract_json):
  569. _de = Document_extract_postgres({document_extract_fingerprint:fingerprint,
  570. document_extract_docid:docid,
  571. document_extract_extract_json:extract_json})
  572. _de.insert_row(self.pool_postgres,1)
  573. def getExtract_json_fromRedis(self,_fingerprint):
  574. db = self.pool_redis_doc.getConnector()
  575. try:
  576. _extract_json = db.get(_fingerprint)
  577. return _extract_json
  578. except Exception as e:
  579. log("getExtract_json_fromRedis error %s"%(str(e)))
  580. finally:
  581. try:
  582. if db.connection.check_health():
  583. self.pool_redis_doc.putConnector(db)
  584. except Exception as e:
  585. pass
  586. return None
  587. def putExtract_json_toRedis(self,fingerprint,extract_json):
  588. db = self.pool_redis_doc.getConnector()
  589. try:
  590. _extract_json = db.set(str(fingerprint),extract_json)
  591. db.expire(fingerprint,3600*2)
  592. return _extract_json
  593. except Exception as e:
  594. log("putExtract_json_toRedis error%s"%(str(e)))
  595. traceback.print_exc()
  596. finally:
  597. try:
  598. if db.connection.check_health():
  599. self.pool_redis_doc.putConnector(db)
  600. except Exception as e:
  601. pass
  602. def comsumer_handle(self,_dict,result_queue):
  603. try:
  604. log("start handle")
  605. frame = _dict["frame"]
  606. conn = _dict["conn"]
  607. message_id = frame.headers["message-id"]
  608. subscription = frame.headers.setdefault('subscription', None)
  609. item = json.loads(frame.body)
  610. dtmp = Document_tmp(item)
  611. dhtml = Document_html({"partitionkey":item.get("partitionkey"),
  612. "docid":item.get("docid")})
  613. extract_times = item.get("extract_times",0)+1
  614. item["extract_times"] = extract_times
  615. _dochtmlcon = item.get(document_tmp_dochtmlcon,"")
  616. html_len = len(_dochtmlcon)
  617. if html_len>50000:
  618. log("docid %s dochtmlcon too long len %d "%(str(item.get("docid")),html_len))
  619. try:
  620. _dochtmlcon = re.sub("<html>|</html>|<body>|</body>", "", _dochtmlcon)
  621. _soup = BeautifulSoup(_dochtmlcon,"lxml")
  622. if len(_dochtmlcon)>200000:
  623. _find = _soup.find("div",attrs={"class":"richTextFetch"})
  624. if _find is not None:
  625. _find.decompose()
  626. else:
  627. _soup = article_limit(_soup,50000)
  628. _dochtmlcon = str(_soup)
  629. except Exception as e:
  630. traceback.print_exc()
  631. ackMsg(conn,message_id,subscription)
  632. return
  633. log("docid %s len %d limit to %d"%(str(item.get("docid")),html_len,len(_dochtmlcon)))
  634. dhtml.setValue(document_tmp_dochtmlcon,_dochtmlcon,True)
  635. _extract = Document_extract({})
  636. _extract.setValue(document_extract2_partitionkey,item.get(document_partitionkey))
  637. _extract.setValue(document_extract2_docid,item.get(document_docid))
  638. all_done = 1
  639. data = {}
  640. for k,v in item.items():
  641. data[k] = v
  642. data["timeout"] = 440
  643. data["doc_id"] = data.get(document_tmp_docid,0)
  644. # if data["docid"]<298986054 and data["docid"]>0:
  645. # log("jump docid %s"%(str(data["docid"])))
  646. # ackMsg(conn,message_id,subscription)
  647. # return
  648. data["content"] = data.get(document_tmp_dochtmlcon,"")
  649. if document_tmp_dochtmlcon in data:
  650. data.pop(document_tmp_dochtmlcon)
  651. data["title"] = data.get(document_tmp_doctitle,"")
  652. data["web_source_no"] = item.get(document_tmp_web_source_no,"")
  653. data["web_source_name"] = item.get(document_tmp_web_source_name,"")
  654. data["original_docchannel"] = item.get(document_tmp_original_docchannel,"")
  655. _fingerprint = getFingerprint(str(data["title"])+str(data["content"]))
  656. if all_done>0:
  657. _time = time.time()
  658. # extract_json = self.getExtract_json_fromDB(_fingerprint)
  659. extract_json = self.getExtract_json_fromRedis(_fingerprint)
  660. log("get json from db takes %.4f"%(time.time()-_time))
  661. # extract_json = None
  662. _docid = int(data["doc_id"])
  663. if extract_json is not None:
  664. log("fingerprint %s exists docid:%s"%(_fingerprint,str(_docid)))
  665. _extract.setValue(document_extract2_extract_json,extract_json,True)
  666. else:
  667. resp = self.request_extract_interface(json=data,headers=self.header)
  668. if (resp.status_code >=200 and resp.status_code<=213):
  669. extract_json = resp.content.decode("utf8")
  670. _extract.setValue(document_extract2_extract_json,extract_json,True)
  671. _time = time.time()
  672. # self.putExtract_json_toDB(_fingerprint,_docid,extract_json)
  673. self.putExtract_json_toRedis(_fingerprint,extract_json)
  674. log("get json to db takes %.4f"%(time.time()-_time))
  675. else:
  676. log("%s--%s"%(str(resp.status_code),resp.content.decode("utf8")))
  677. all_done = -2
  678. # if all_done>0:
  679. # resp = self.request_industry_interface(json=data,headers=self.header)
  680. # if (resp.status_code >=200 and resp.status_code<=213):
  681. # _extract.setValue(document_extract2_industry_json,resp.content.decode("utf8"),True)
  682. # else:
  683. # log("%s--%s"%(str(resp.status_code),resp.content.decode("utf8")))
  684. # all_done = -3
  685. # _to_ack = False
  686. # if all_done>0 and len(_extract.getProperties().get(document_extract2_extract_json,""))<=2:
  687. # all_done = -4
  688. _extract.setValue(document_extract2_industry_json,"{}",True)
  689. try:
  690. if all_done!=1:
  691. sentMsgToDD("要素提取失败:docid:%d with result:%d"%(item.get(document_tmp_docid),all_done))
  692. if extract_times>=10:
  693. #process as succeed
  694. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  695. dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
  696. dtmp.update_row(self.ots_client)
  697. dhtml.update_row(self.ots_client)
  698. #replace as {}
  699. _extract.setValue(document_extract2_extract_json,"{}",True)
  700. _extract.setValue(document_extract2_industry_json,"{}",True)
  701. _extract.setValue(document_extract2_status,random.randint(1,50),True)
  702. _extract.update_row(self.ots_client)
  703. _to_ack = True
  704. elif extract_times>5:
  705. #transform to the extract_failed queue
  706. if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract_failed):
  707. #process as succeed
  708. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  709. dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
  710. dtmp.update_row(self.ots_client)
  711. dhtml.update_row(self.ots_client)
  712. #replace as {}
  713. _extract.setValue(document_extract2_extract_json,"{}",True)
  714. _extract.setValue(document_extract2_industry_json,"{}",True)
  715. _extract.setValue(document_extract2_status,random.randint(1,50),True)
  716. _extract.update_row(self.ots_client)
  717. _to_ack = True
  718. else:
  719. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract)
  720. #失败保存
  721. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  722. dtmp.setValue(document_tmp_status,60,True)
  723. if not dtmp.exists_row(self.ots_client):
  724. dtmp.update_row(self.ots_client)
  725. dhtml.update_row(self.ots_client)
  726. if send_succeed:
  727. _to_ack = True
  728. else:
  729. #process succeed
  730. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  731. dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
  732. if _docid==290816703:
  733. dtmp.setValue("test_json",extract_json,True)
  734. dtmp.update_row(self.ots_client)
  735. dhtml.update_row(self.ots_client)
  736. _extract.setValue(document_extract2_status,random.randint(1,50),True)
  737. _extract.update_row(self.ots_client)
  738. _to_ack = True
  739. except Exception:
  740. traceback.print_exc()
  741. if _to_ack:
  742. ackMsg(conn,message_id,subscription)
  743. log("process %s docid:%d %s"%(str(_to_ack),data["doc_id"],str(all_done)))
  744. except requests.ConnectionError as e1:
  745. item["extract_times"] -= 1
  746. if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
  747. ackMsg(conn,message_id,subscription)
  748. except Exception as e:
  749. traceback.print_exc()
  750. sentMsgToDD("要素提取失败:docid:%d with result:%s"%(item.get(document_tmp_docid),str(e)))
  751. log("process %s docid: failed message_id:%s"%(data["doc_id"],message_id))
  752. if extract_times>=10:
  753. #process as succeed
  754. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  755. dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
  756. dtmp.update_row(self.ots_client)
  757. dhtml.update_row(self.ots_client)
  758. #replace as {}
  759. _extract.setValue(document_extract2_extract_json,"{}",True)
  760. _extract.setValue(document_extract2_industry_json,"{}",True)
  761. _extract.setValue(document_extract2_status,random.randint(1,50),True)
  762. _extract.update_row(self.ots_client)
  763. ackMsg(conn,message_id,subscription)
  764. elif extract_times>5:
  765. #transform to the extract_failed queue
  766. if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract_failed):
  767. #process as succeed
  768. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  769. dtmp.setValue(document_tmp_status,random.randint(*flow_extract_status_succeed_to),True)
  770. dtmp.update_row(self.ots_client)
  771. dhtml.update_row(self.ots_client)
  772. #replace as {}
  773. _extract.setValue(document_extract2_extract_json,"{}",True)
  774. _extract.setValue(document_extract2_industry_json,"{}",True)
  775. _extract.setValue(document_extract2_status,random.randint(1,50),True)
  776. _extract.update_row(self.ots_client)
  777. ackMsg(conn,message_id,subscription)
  778. else:
  779. #transform to the extract queue
  780. #失败保存
  781. dtmp.setValue(document_tmp_dochtmlcon,"",False)
  782. dtmp.setValue(document_tmp_status,60,True)
  783. if not dtmp.exists_row(self.ots_client):
  784. dtmp.update_row(self.ots_client)
  785. dhtml.update_row(self.ots_client)
  786. if send_msg_toacmq(self.pool_mq,json.dumps(item,ensure_ascii=False),self.mq_extract):
  787. ackMsg(conn,message_id,subscription)
  788. def delete_document_extract(self,save_count=70*10000):
  789. conn = self.pool_postgres.getConnector()
  790. try:
  791. cursor = conn.cursor()
  792. sql = " select max(docid),min(docid) from document_extract "
  793. cursor.execute(sql)
  794. rows = cursor.fetchall()
  795. if len(rows)>0:
  796. maxdocid,mindocid = rows[0]
  797. d_mindocid = int(maxdocid)-save_count
  798. if mindocid<d_mindocid:
  799. sql = " delete from document_extract where docid<%d"%d_mindocid
  800. cursor.execute(sql)
  801. conn.commit()
  802. except Exception as e:
  803. traceback.print_exc()
  804. finally:
  805. self.pool_postgres.putConnector(conn)
  806. def start_flow_extract(self):
  807. schedule = BlockingScheduler()
  808. schedule.add_job(self.flow_extract_producer,"cron",second="*/20")
  809. schedule.add_job(self.process_extract_failed,"cron",minute="*/5")
  810. schedule.add_job(self.delete_document_extract,"cron",hour="*/5")
  811. schedule.add_job(self.monitor_listener,"cron",minute="*/5")
  812. schedule.start()
  813. from multiprocessing import RLock
  814. docid_lock = RLock()
  815. conn_mysql = None
  816. def generateRangeDocid(nums):
  817. global conn_mysql
  818. while 1:
  819. try:
  820. with docid_lock:
  821. if conn_mysql is None:
  822. conn_mysql = getConnection_mysql()
  823. cursor = conn_mysql.cursor()
  824. sql = "select serial_value from b2c_serial_no where serial_name='DocumentIdSerial'"
  825. cursor.execute(sql)
  826. rows = cursor.fetchall()
  827. current_docid = rows[0][0]
  828. next_docid = current_docid+1
  829. update_docid = current_docid+nums
  830. sql = " update b2c_serial_no set serial_value=%d where serial_name='DocumentIdSerial'"%(update_docid)
  831. cursor.execute(sql)
  832. conn_mysql.commit()
  833. return next_docid
  834. except Exception as e:
  835. conn_mysql = getConnection_mysql()
  836. # 自定义jsonEncoder
  837. class MyEncoder(json.JSONEncoder):
  838. def default(self, obj):
  839. if isinstance(obj, np.ndarray):
  840. return obj.tolist()
  841. elif isinstance(obj, bytes):
  842. return str(obj, encoding='utf-8')
  843. elif isinstance(obj, (np.float_, np.float16, np.float32,
  844. np.float64,Decimal)):
  845. return float(obj)
  846. elif isinstance(obj,str):
  847. return obj
  848. return json.JSONEncoder.default(self, obj)
  849. class Dataflow_init(Dataflow):
  850. class InitListener():
  851. def __init__(self,conn,*args,**kwargs):
  852. self.conn = conn
  853. self.get_count = 1000
  854. self.count = self.get_count
  855. self.begin_docid = None
  856. self.mq_init = "/queue/dataflow_init"
  857. self.mq_attachment = "/queue/dataflow_attachment"
  858. self.mq_extract = "/queue/dataflow_extract"
  859. self.pool_mq1 = ConnectorPool(1,4,getConnect_activateMQ)
  860. def on_error(self, headers):
  861. log('received an error %s' % headers.body)
  862. def getRangeDocid(self):
  863. begin_docid = generateRangeDocid(self.get_count)
  864. self.begin_docid = begin_docid
  865. self.count = 0
  866. def getNextDocid(self):
  867. if self.count>=self.get_count:
  868. self.getRangeDocid()
  869. next_docid = self.begin_docid+self.count
  870. self.count += 1
  871. return next_docid
  872. def on_message(self, headers):
  873. try:
  874. next_docid = int(self.getNextDocid())
  875. partitionkey = int(next_docid%500+1)
  876. message_id = headers.headers["message-id"]
  877. body = json.loads(headers.body)
  878. body[document_tmp_partitionkey] = partitionkey
  879. body[document_tmp_docid] = next_docid
  880. if body.get(document_original_docchannel) is None:
  881. body[document_original_docchannel] = body.get(document_docchannel)
  882. page_attachments = body.get(document_tmp_attachment_path,"[]")
  883. _uuid = body.get(document_tmp_uuid,"")
  884. if page_attachments!="[]":
  885. status = random.randint(1,10)
  886. body[document_tmp_status] = status
  887. if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_attachment):
  888. log("uuid:%s with docid:%s"%(str(_uuid),str(next_docid)))
  889. ackMsg(self.conn,message_id)
  890. else:
  891. log("send_msg_error on init listener")
  892. else:
  893. status = random.randint(11,50)
  894. body[document_tmp_status] = status
  895. if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_extract):
  896. log("uuid:%s with docid:%s"%(str(_uuid),str(next_docid)))
  897. ackMsg(self.conn,message_id)
  898. else:
  899. log("send_msg_error on init listener")
  900. except Exception as e:
  901. traceback.print_exc()
  902. if send_msg_toacmq(self.pool_mq1,json.dumps(body,cls=MyEncoder),self.mq_init):
  903. log("init error")
  904. ackMsg(self.conn,message_id)
  905. def __del__(self):
  906. self.conn.disconnect()
  907. del self.pool_mq1
  908. def __init__(self):
  909. Dataflow.__init__(self)
  910. self.mq_init = "/queue/dataflow_init"
  911. self.mq_attachment = "/queue/dataflow_attachment"
  912. self.mq_extract = "/queue/dataflow_extract"
  913. self.pool_oracle = ConnectorPool(10,15,getConnection_oracle)
  914. self.pool_mq = ConnectorPool(10,30,getConnect_activateMQ)
  915. self.ots_capacity = getConnect_ots_capacity()
  916. self.init_comsumer_counts = 2
  917. self.list_init_comsumer = []
  918. for i in range(self.init_comsumer_counts):
  919. listener = self.InitListener(getConnect_activateMQ())
  920. createComsumer(listener,self.mq_init)
  921. self.list_init_comsumer.append(listener)
  922. def monitor_listener(self):
  923. for i in range(len(self.list_init_comsumer)):
  924. if self.list_init_comsumer[i].conn.is_connected():
  925. continue
  926. else:
  927. listener = self.InitListener(getConnect_activateMQ())
  928. createComsumer(listener,self.mq_init)
  929. self.list_init_comsumer[i] = listener
  930. def temp2mq(self,object):
  931. conn_oracle = self.pool_oracle.getConnector()
  932. try:
  933. list_obj = object.select_rows(conn_oracle,type(object),object.table_name,[],limit=1000)
  934. for _obj in list_obj:
  935. ots_dict = _obj.getProperties_ots()
  936. if len(ots_dict.get("dochtmlcon",""))>500000:
  937. _obj.delete_row(conn_oracle)
  938. log("msg too long:%s,%d"%(ots_dict.get("uuid"),len(ots_dict.get("dochtmlcon",""))))
  939. continue
  940. if send_msg_toacmq(self.pool_mq,json.dumps(ots_dict,cls=MyEncoder),self.mq_init):
  941. #删除数据,上线放开
  942. _obj.delete_row(conn_oracle)
  943. else:
  944. log("send_msg_error111:%s,%d"%(ots_dict.get("uuid"),len(ots_dict.get("dochtmlcon",""))))
  945. self.pool_oracle.putConnector(conn_oracle)
  946. except Exception as e:
  947. traceback.print_exc()
  948. self.pool_oracle.decrease()
  949. def ots2mq(self):
  950. try:
  951. bool_query = BoolQuery(must_queries=[RangeQuery("status",1,51)])
  952. rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
  953. SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(document_docid)]),get_total_count=True,limit=100),
  954. ColumnsToGet(return_type=ColumnReturnType.ALL))
  955. list_data = getRow_ots(rows)
  956. for _data in list_data:
  957. _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
  958. document_tmp_docid:_data.get(document_tmp_docid),
  959. document_tmp_status:0}
  960. _document = Document(_d)
  961. page_attachments = _data.get(document_tmp_attachment_path,"[]")
  962. _document_html = Document(_data)
  963. _document_html.fix_columns(self.ots_capacity,[document_tmp_dochtmlcon],True)
  964. if page_attachments!="[]":
  965. status = random.randint(1,10)
  966. _data[document_tmp_status] = status
  967. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
  968. else:
  969. status = random.randint(11,50)
  970. _data[document_tmp_status] = status
  971. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
  972. if send_succeed:
  973. _document.update_row(self.ots_client)
  974. else:
  975. log("send_msg_error2222")
  976. while next_token:
  977. rows,next_token,total_count,is_all_succeed = self.ots_client.search("document","document_index",
  978. SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
  979. ColumnsToGet(return_type=ColumnReturnType.ALL))
  980. list_data = getRow_ots(rows)
  981. for _data in list_data:
  982. _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
  983. document_tmp_docid:_data.get(document_tmp_docid),
  984. document_tmp_status:0}
  985. _document = Document(_d)
  986. page_attachments = _data.get(document_tmp_attachment_path,"[]")
  987. _document_html = Document(_data)
  988. _document_html.fix_columns(self.ots_capacity,[document_tmp_dochtmlcon],True)
  989. if page_attachments!="[]":
  990. status = random.randint(1,10)
  991. _data[document_tmp_status] = status
  992. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
  993. else:
  994. status = random.randint(11,50)
  995. _data[document_tmp_status] = status
  996. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
  997. if send_succeed:
  998. _document.update_row(self.ots_client)
  999. else:
  1000. log("send_msg_error2222")
  1001. except Exception as e:
  1002. traceback.print_exc()
  1003. def otstmp2mq(self):
  1004. try:
  1005. bool_query = BoolQuery(must_queries=[TermQuery("status",0)])
  1006. rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
  1007. SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(document_docid)]),get_total_count=True,limit=100),
  1008. ColumnsToGet(return_type=ColumnReturnType.ALL))
  1009. list_data = getRow_ots(rows)
  1010. for _data in list_data:
  1011. _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
  1012. document_tmp_docid:_data.get(document_tmp_docid),
  1013. document_tmp_status:0}
  1014. _document = Document_tmp(_d)
  1015. page_attachments = _data.get(document_tmp_attachment_path,"[]")
  1016. _document_html = Document_html(_data)
  1017. _document_html.fix_columns(self.ots_client,[document_tmp_dochtmlcon],True)
  1018. if page_attachments!="[]":
  1019. status = random.randint(1,10)
  1020. _data[document_tmp_status] = status
  1021. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
  1022. else:
  1023. status = random.randint(11,50)
  1024. _data[document_tmp_status] = status
  1025. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
  1026. if send_succeed:
  1027. _document.setValue(document_tmp_status,1,True)
  1028. _document.update_row(self.ots_client)
  1029. else:
  1030. log("send_msg_error2222")
  1031. while next_token:
  1032. rows,next_token,total_count,is_all_succeed = self.ots_client.search("document_tmp","document_tmp_index",
  1033. SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
  1034. ColumnsToGet(return_type=ColumnReturnType.ALL))
  1035. list_data = getRow_ots(rows)
  1036. for _data in list_data:
  1037. _d = {document_tmp_partitionkey:_data.get(document_tmp_partitionkey),
  1038. document_tmp_docid:_data.get(document_tmp_docid),
  1039. document_tmp_status:0}
  1040. _document = Document_tmp(_d)
  1041. page_attachments = _data.get(document_tmp_attachment_path,"[]")
  1042. _document_html = Document_html(_data)
  1043. _document_html.fix_columns(self.ots_client,[document_tmp_dochtmlcon],True)
  1044. if page_attachments!="[]":
  1045. status = random.randint(1,10)
  1046. _data[document_tmp_status] = status
  1047. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_attachment)
  1048. else:
  1049. status = random.randint(11,50)
  1050. _data[document_tmp_status] = status
  1051. send_succeed = send_msg_toacmq(self.pool_mq,json.dumps(_document_html.getProperties(),cls=MyEncoder),self.mq_extract)
  1052. if send_succeed:
  1053. _document.setValue(document_tmp_status,1,True)
  1054. _document.update_row(self.ots_client)
  1055. else:
  1056. log("send_msg_error2222")
  1057. except Exception as e:
  1058. traceback.print_exc()
  1059. def test_dump_docid(self):
  1060. class TestDumpListener(ActiveMQListener):
  1061. def on_message(self, headers):
  1062. message_id = headers.headers["message-id"]
  1063. body = headers.body
  1064. self._queue.put(headers,True)
  1065. ackMsg(self.conn,message_id)
  1066. _queue = Queue()
  1067. listener1 = TestDumpListener(getConnect_activateMQ(),_queue)
  1068. listener2 = TestDumpListener(getConnect_activateMQ(),_queue)
  1069. createComsumer(listener1,"/queue/dataflow_attachment")
  1070. createComsumer(listener2,"/queue/dataflow_extract")
  1071. time.sleep(10)
  1072. list_item = []
  1073. list_docid = []
  1074. while 1:
  1075. try:
  1076. _item = _queue.get(timeout=2)
  1077. list_item.append(_item)
  1078. except Exception as e:
  1079. break
  1080. for item in list_item:
  1081. _item = json.loads(item.body)
  1082. list_docid.append(_item.get("docid"))
  1083. log(list_docid[:10])
  1084. log("len docid:%d set len:%d"%(len(list_docid),len(set(list_docid))))
  1085. def start_dataflow_init(self):
  1086. # self.test_dump_docid()
  1087. from BaseDataMaintenance.model.oracle.CaiGouYiXiangTemp import CaiGouYiXiangTemp
  1088. from BaseDataMaintenance.model.oracle.PaiMaiChuRangTemp import PaiMaiChuRangTemp
  1089. from BaseDataMaintenance.model.oracle.ZhaoBiaoGongGaoTemp import ZhaoBiaoGongGaoTemp
  1090. from BaseDataMaintenance.model.oracle.ZhaoBiaoYuGaoTemp import ZhaoBiaoYuGaoTemp
  1091. from BaseDataMaintenance.model.oracle.ZhongBiaoXinXiTemp import ZhongBiaoXinXiTemp
  1092. from BaseDataMaintenance.model.oracle.ZiShenJieGuoTemp import ZiShenJieGuoTemp
  1093. from BaseDataMaintenance.model.oracle.ChanQuanJiaoYiTemp import ChanQuanJiaoYiTemp
  1094. from BaseDataMaintenance.model.oracle.GongGaoBianGengTemp import GongGaoBianGeng
  1095. from BaseDataMaintenance.model.oracle.KongZhiJiaTemp import KongZhiJiaTemp
  1096. from BaseDataMaintenance.model.oracle.TuDiKuangChanTemp import TuDiKuangChanTemp
  1097. from BaseDataMaintenance.model.oracle.ZhaoBiaoDaYiTemp import ZhaoBiaoDaYiTemp
  1098. from BaseDataMaintenance.model.oracle.ZhaoBiaoWenJianTemp import ZhaoBiaoWenJianTemp
  1099. schedule = BlockingScheduler()
  1100. schedule.add_job(self.temp2mq,"cron",args=(CaiGouYiXiangTemp({}),),second="*/10")
  1101. schedule.add_job(self.temp2mq,"cron",args=(PaiMaiChuRangTemp({}),),second="*/10")
  1102. schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoGongGaoTemp({}),),second="*/10")
  1103. schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoYuGaoTemp({}),),second="*/10")
  1104. schedule.add_job(self.temp2mq,"cron",args=(ZhongBiaoXinXiTemp({}),),second="*/10")
  1105. schedule.add_job(self.temp2mq,"cron",args=(ZiShenJieGuoTemp({}),),second="*/10")
  1106. schedule.add_job(self.temp2mq,"cron",args=(ChanQuanJiaoYiTemp({}),),second="*/10")
  1107. schedule.add_job(self.temp2mq,"cron",args=(GongGaoBianGeng({}),),second="*/10")
  1108. schedule.add_job(self.temp2mq,"cron",args=(KongZhiJiaTemp({}),),second="*/10")
  1109. schedule.add_job(self.temp2mq,"cron",args=(TuDiKuangChanTemp({}),),second="*/10")
  1110. schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoDaYiTemp({}),),second="*/10")
  1111. schedule.add_job(self.temp2mq,"cron",args=(ZhaoBiaoWenJianTemp({}),),second="*/10")
  1112. schedule.add_job(self.ots2mq,"cron",second="*/10")
  1113. schedule.add_job(self.otstmp2mq,"cron",second="*/10")
  1114. schedule.add_job(self.monitor_listener,"cron",minute="*/1")
  1115. schedule.start()
  1116. def transform_attachment():
  1117. from BaseDataMaintenance.model.ots.attachment import attachment
  1118. from BaseDataMaintenance.model.postgres.attachment import Attachment_postgres
  1119. from BaseDataMaintenance.dataSource.source import getConnection_postgres,getConnect_ots
  1120. from threading import Thread
  1121. from queue import Queue
  1122. task_queue = Queue()
  1123. def comsumer(task_queue,pool_postgres):
  1124. while 1:
  1125. _dict = task_queue.get(True)
  1126. try:
  1127. attach = Attachment_postgres(_dict)
  1128. if not attach.exists(pool_postgres):
  1129. attach.insert_row(pool_postgres)
  1130. except Exception as e:
  1131. traceback.print_exc()
  1132. def start_comsumer(task_queue):
  1133. pool_postgres = ConnectorPool(10,30,getConnection_postgres)
  1134. comsumer_count = 30
  1135. list_thread = []
  1136. for i in range(comsumer_count):
  1137. _t = Thread(target=comsumer,args=(task_queue,pool_postgres))
  1138. list_thread.append(_t)
  1139. for _t in list_thread:
  1140. _t.start()
  1141. ots_client = getConnect_ots()
  1142. _thread = Thread(target=start_comsumer,args=(task_queue,))
  1143. _thread.start()
  1144. bool_query = BoolQuery(must_queries=[
  1145. RangeQuery(attachment_crtime,"2022-05-06"),
  1146. BoolQuery(should_queries=[TermQuery(attachment_status,10),
  1147. TermQuery(attachment_status,ATTACHMENT_TOOLARGE)])
  1148. ])
  1149. rows,next_token,total_count,is_all_succeed = ots_client.search("attachment","attachment_index",
  1150. SearchQuery(bool_query,sort=Sort(sorters=[FieldSort(attachment_crtime,sort_order=SortOrder.DESC)]),get_total_count=True,limit=100),
  1151. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))
  1152. list_dict = getRow_ots(rows)
  1153. for _dict in list_dict:
  1154. task_queue.put(_dict,True)
  1155. _count = len(list_dict)
  1156. while next_token:
  1157. rows,next_token,total_count,is_all_succeed = ots_client.search("attachment","attachment_index",
  1158. SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
  1159. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))
  1160. list_dict = getRow_ots(rows)
  1161. for _dict in list_dict:
  1162. task_queue.put(_dict,True)
  1163. _count += len(list_dict)
  1164. print("%d/%d,queue:%d"%(_count,total_count,task_queue.qsize()))
  1165. def del_test_doc():
  1166. ots_client = getConnect_ots()
  1167. bool_query = BoolQuery(must_queries=[RangeQuery("docid",range_to=0)])
  1168. list_data = []
  1169. rows,next_token,total_count,is_all_succeed = ots_client.search("document_tmp","document_tmp_index",
  1170. SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),get_total_count=True,limit=100),
  1171. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
  1172. print(total_count)
  1173. list_row = getRow_ots(rows)
  1174. list_data.extend(list_row)
  1175. while next_token:
  1176. rows,next_token,total_count,is_all_succeed = ots_client.search("document_tmp","document_tmp_index",
  1177. SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
  1178. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
  1179. list_row = getRow_ots(rows)
  1180. list_data.extend(list_row)
  1181. for _row in list_data:
  1182. _doc = Document_tmp(_row)
  1183. _doc.delete_row(ots_client)
  1184. _html = Document_html(_row)
  1185. if _html.exists_row(ots_client):
  1186. _html.delete_row(ots_client)
  1187. def fixDoc_to_queue_extract():
  1188. pool_mq = ConnectorPool(10,20,getConnect_activateMQ)
  1189. try:
  1190. ots_client = getConnect_ots()
  1191. ots_capacity = getConnect_ots_capacity()
  1192. bool_query = BoolQuery(must_queries=[
  1193. RangeQuery("crtime","2022-05-31"),
  1194. TermQuery("docchannel",114)
  1195. ])
  1196. list_data = []
  1197. rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
  1198. SearchQuery(bool_query,sort=Sort(sorters=[FieldSort("docid")]),get_total_count=True,limit=100),
  1199. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))
  1200. print(total_count)
  1201. list_row = getRow_ots(rows)
  1202. list_data.extend(list_row)
  1203. _count = len(list_row)
  1204. while next_token:
  1205. rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
  1206. SearchQuery(bool_query,next_token=next_token,get_total_count=True,limit=100),
  1207. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.ALL))
  1208. list_row = getRow_ots(rows)
  1209. list_data.extend(list_row)
  1210. _count = len(list_data)
  1211. print("%d/%d"%(_count,total_count))
  1212. task_queue = Queue()
  1213. for _row in list_data:
  1214. if "all_columns" in _row:
  1215. _row.pop("all_columns")
  1216. _html = Document(_row)
  1217. task_queue.put(_html)
  1218. def _handle(item,result_queue):
  1219. _html = item
  1220. _html.fix_columns(ots_capacity,["dochtmlcon"],True)
  1221. print(_html.getProperties().get(document_tmp_docid))
  1222. send_msg_toacmq(pool_mq,json.dumps(_html.getProperties()),"/queue/dataflow_extract")
  1223. mt = MultiThreadHandler(task_queue,_handle,None,30)
  1224. mt.run()
  1225. except Exception as e:
  1226. traceback.print_exc()
  1227. finally:
  1228. pool_mq.destory()
  1229. def check_data_synchronization():
  1230. # filepath = "C:\\Users\\Administrator\\Desktop\\to_check.log"
  1231. # list_uuid = []
  1232. # _regrex = "delete\s+(?P<tablename>[^\s]+)\s+.*ID='(?P<uuid>.+)'"
  1233. # with open(filepath,"r",encoding="utf8") as f:
  1234. # while 1:
  1235. # _line = f.readline()
  1236. # if not _line:
  1237. # break
  1238. # _match = re.search(_regrex,_line)
  1239. # if _match is not None:
  1240. # _uuid = _match.groupdict().get("uuid")
  1241. # tablename = _match.groupdict.get("tablename")
  1242. # if _uuid is not None:
  1243. # list_uuid.append({"uuid":_uuid,"tablename":tablename})
  1244. # print("total_count:",len(list_uuid))
  1245. import pandas as pd
  1246. from BaseDataMaintenance.common.Utils import load
  1247. task_queue = Queue()
  1248. list_data = []
  1249. df_data = load("uuid.pk")
  1250. # df_data = pd.read_excel("check.xlsx")
  1251. for uuid,tablename in zip(df_data["uuid"],df_data["tablename"]):
  1252. _dict = {"uuid":uuid,
  1253. "tablename":tablename}
  1254. list_data.append(_dict)
  1255. task_queue.put(_dict)
  1256. print("qsize:",task_queue.qsize())
  1257. ots_client = getConnect_ots()
  1258. def _handle(_item,result_queue):
  1259. bool_query = BoolQuery(must_queries=[TermQuery("uuid",_item.get("uuid"))])
  1260. rows,next_token,total_count,is_all_succeed = ots_client.search("document","document_index",
  1261. SearchQuery(bool_query,get_total_count=True),
  1262. columns_to_get=ColumnsToGet(return_type=ColumnReturnType.NONE))
  1263. _item["exists"] = total_count
  1264. mt = MultiThreadHandler(task_queue,_handle,None,30)
  1265. mt.run()
  1266. df_data = {"uuid":[],
  1267. "tablename":[],
  1268. "exists":[]}
  1269. for _data in list_data:
  1270. if _data["exists"]==0:
  1271. for k,v in df_data.items():
  1272. v.append(_data.get(k))
  1273. import pandas as pd
  1274. df2 = pd.DataFrame(df_data)
  1275. df2.to_excel("check1.xlsx")
  1276. current_path = os.path.abspath(os.path.dirname(__file__))
  1277. def fixDoc_to_queue_init(filename=""):
  1278. import pandas as pd
  1279. from BaseDataMaintenance.model.oracle.GongGaoTemp import dict_oracle2ots
  1280. if filename=="":
  1281. filename = os.path.join(current_path,"check.xlsx")
  1282. df = pd.read_excel(filename)
  1283. if "docchannel" in dict_oracle2ots:
  1284. dict_oracle2ots.pop("docchannel")
  1285. row_name = ",".join(list(dict_oracle2ots.keys()))
  1286. conn = getConnection_oracle()
  1287. cursor = conn.cursor()
  1288. _count = 0
  1289. for uuid,tablename,_exists,_toolong in zip(df["uuid"],df["tablename"],df["exists"],df["tolong"]):
  1290. if _exists==0 and _toolong==0:
  1291. _count += 1
  1292. _source = str(tablename).replace("_TEMP","")
  1293. sql = " insert into %s(%s) select %s from %s where id='%s' "%(tablename,row_name,row_name,_source,uuid)
  1294. cursor.execute(sql)
  1295. log("%d:%s"%(_count,sql))
  1296. conn.commit()
  1297. conn.close()
  1298. if __name__ == '__main__':
  1299. # di = Dataflow_init()
  1300. # di.start_dataflow_init()
  1301. # transform_attachment()
  1302. # del_test_doc()
  1303. # de = Dataflow_ActivteMQ_extract()
  1304. # de.start_flow_extract()
  1305. # fixDoc_to_queue_extract()
  1306. # check_data_synchronization()
  1307. # fixDoc_to_queue_init(filename="C:\\Users\\Administrator\\Desktop\\flow_init_2023-02-07.xlsx")
  1308. current_date = getCurrent_date(format="%Y-%m-%d")
  1309. last_date = timeAdd(current_date,-30,format="%Y-%m-%d")
  1310. sql = " delete from attachment where crtime<='%s 00:00:00' "%(last_date)
  1311. print(sql)