#coding:UTF8 from utils.Utils import * from dataSource.source import getConnect_ots from utils.multiThread import MultiThreadHandler from queue import Queue import json from utils.hashUtil import aesCipher import inspect from tablestore import * set_columns = set() list_df_columns = [] from Crypto.Cipher import PKCS1_v1_5 as Cipher_pksc1_v1_5 from Crypto.PublicKey import RSA def rsa_encrpt(string, public_key): rsakey = RSA.importKey(public_key) # 读取公钥 cipher = Cipher_pksc1_v1_5.new(rsakey) # 因为encryptor.encrypt方法其内部就实现了加密再次Base64加密的过程，所以这里实际是通过下面的1和2完成了JSEncrypt的加密方法 encrypt_text = cipher.encrypt(string.encode()) # 1.对账号密码组成的字符串加密 return encrypt_text def getOneContact(contacts,tojson=True,mobile_first=True,mobile_only=True,high_level_first=True): mobile_person = "" mobile_no = '' phone_person = "" phone_no = '' if contacts is None: return "","" try: if isinstance(contacts,str): list_contacts = json.loads(contacts) else: list_contacts = contacts list_contacts.sort(key=lambda x:x.get("level",0),reverse=True) for _contact in list_contacts: _level = _contact.get("level") is_mobile = _contact.get("is_mobile",0) if is_mobile==1: _contact["mobile_no"] = _contact.get("phone_no") if _contact.get("mobile_no","")!="": mobile_person = _contact.get("contact_person","") mobile_no = _contact.get("mobile_no","") if _level==40: if mobile_person!="": mobile_person += "(法人)" if _contact.get("phone_no","")!="": phone_person = _contact.get("contact_person","") phone_no = _contact.get("phone_no","") if _level==40: if phone_person!="": phone_person += "(法人)" if mobile_first: if mobile_no!="" and mobile_person!="": return mobile_person,mobile_no else: if mobile_only: return mobile_person,mobile_no except Exception as e: pass return phone_person,phone_no def getMobiles(contacts,to_json=True): if to_json: list_contacts = json.loads(contacts) else: list_contacts = contacts list_result = [] for _c in list_contacts: if _c.get("mobile_no","")!="": list_result.append([_c.get("contact_person",""),_c.get("mobile_no")]) return list_result def set_dict_item(_dict,name,v): _dict[name] = getLegal_str(v) if name not in set_columns: set_columns.add(name) list_df_columns.append(getLegal_str(name)) def set_dict_item_columns(set_columns1,list_df_columns1,_dict,name,v): _dict[name] = getLegal_str(v) if name not in set_columns1: set_columns1.add(name) list_df_columns1.append(getLegal_str(name)) def getRowData_document(df_data,rows,set_line,list_keyword,dict_channel): # list_data = getRow_ots(rows) for row in rows: item = {} _dict = row set_dict_item(item,"docid",_dict.get("docid","")) set_dict_item(item,"公告标题",_dict.get("doctitle","")) set_dict_item(item,"公告内容",_dict.get("doctextcon","")) set_dict_item(item,"附件内容",_dict.get("attachmenttextcon","")) set_dict_item(item,"公告类别",dict_channel.get(_dict.get("docchannel",""),"")) set_dict_item(item,"关键词",",".join(list(set(re.findall("|".join([re.escape(str(a).replace("(","（").replace(")","）")) for a in list_keyword]),re.sub("\s","",str(row.get("doctitle","")+row.get("doctextcon","")+row.get("attachmenttextcon","")).replace("(","（").replace(")","）"))))))) set_dict_item(item,"产品",_dict.get("product","")) set_dict_item(item,"省份",_dict.get("province","")) # item["区域"] = "%s-%s-%s"%(_dict.get("province",""),_dict.get("city",""),_dict.get("district","")) set_dict_item(item,"城市",_dict.get("city","")) set_dict_item(item,"区县",_dict.get("district","")) set_dict_item(item,"发布时间",_dict.get("page_time","")) set_dict_item(item,"截标时间",_dict.get("time_bidclose","")) set_dict_item(item,"开标时间",_dict.get("time_bidopen","")) set_dict_item(item,"创建时间",_dict.get("crtime","")) set_dict_item(item,"招标方式",_dict.get("bidway","")) set_dict_item(item,"行业一级分类",_dict.get("industry","")) set_dict_item(item,"行业二级分类",_dict.get("info_type","")) set_dict_item(item,"uuid",_dict.get("uuid")) set_dict_item(item,"公告标题_refine",re.sub(r'工程|服务|询价|比价|谈判|竞争性|磋商|结果|中标|招标|采购|的|公示|公开|成交|公告|评标|候选人|交易|通知|废标|流标|终止|中止|一笔|预告|单一来源|询价|竞价|合同', '', _dict.get("doctitle",""))) set_dict_item(item,"项目编号",_dict.get("project_code","")) set_dict_item(item,"招标单位",_dict.get("tenderee","")) set_dict_item(item,"招标联系人",_dict.get("tenderee_contact","")) set_dict_item(item,"招标联系人电话",_dict.get("tenderee_phone","")) set_dict_item(item,"代理单位",_dict.get("agency","")) set_dict_item(item,"代理联系人",_dict.get("agency_contact","")) set_dict_item(item,"代理联系人电话",_dict.get("agency_phone","")) set_dict_item(item,"比地招标公告地址","http://www.bidizhaobiao.com/excel_detail.do?code=%s"%(str(aesCipher.encrypt('{"docid":%d}'%_dict.get("docid"))))) set_dict_item(item,"开标时间",_dict.get("time_bidopen","")) set_dict_item(item,"截标时间",_dict.get("time_bidclose","")) sub_docs_json = _dict.get("sub_docs_json") set_tenderer = set() if sub_docs_json is not None: for _doc in json.loads(sub_docs_json): if "win_tenderer" in _doc: set_dict_item(item,"中标单位",_doc["win_tenderer"]) if "second_tenderer" in _doc: set_tenderer.add(_doc.get("second_tenderer")) if "third_tenderer" in _doc: set_tenderer.add(_doc.get("third_tenderer")) if "win_tenderee_manager" in _doc: set_dict_item(item,"中标单位联系人",_doc["win_tenderee_manager"]) if "win_tenderee_phone" in _doc: set_dict_item(item,"中标单位联系电话",_doc["win_tenderee_phone"]) if "win_bid_price" in _doc and float(0 if _doc["win_bid_price"]=="" else _doc["win_bid_price"])>0: set_dict_item(item,"中标金额",_doc["win_bid_price"]) if "bidding_budget" in _doc and float(0 if _doc["bidding_budget"]=="" else _doc["bidding_budget"])>0: set_dict_item(item,"招标金额",_doc["bidding_budget"]) set_dict_item(item,"入围供应商",",".join(list(set_tenderer))) if "招标金额" not in item: set_dict_item(item,"招标金额","") if "中标金额" not in item: set_dict_item(item,"中标金额","") if "中标单位" not in item: set_dict_item(item,"中标单位","") if "中标单位联系人" not in item: set_dict_item(item,"中标单位联系人","") if "中标单位联系电话" not in item: set_dict_item(item,"中标单位联系电话","") # if item["中标单位"] not in set_enter: # continue _line = "%s-%s-%s-%s-%s-%s"%(item["省份"],item["城市"],item["项目编号"],item["招标单位"],item["中标单位"],str(item["中标金额"])) # if re.search("[大中小]学|幼儿园|医院|公司",item["招标单位"]) is not None: # continue # if _line in set_line: # continue # if _dict.get("docid","") in set_ig_docid: # continue # if item["招标金额"]=="": # continue # set_line.add(_line) for k,v in item.items(): if k not in df_data: df_data[k] = [] df_data[k].append(v) def getDictEnterprise(list_enterprise,columns_to_get = ["reg_capital","actual_capital","industry","estiblish_time","social_staff_num","zhong_biao_number","tou_biao_number","credit_code"]): task_queue = Queue() result_queue= Queue() for _enterprise in list_enterprise: task_queue.put(str(_enterprise)) def _handle(item,result_queue,ots_client): try: primary_key = [("name",item)] consumed,return_row,next_token = ots_client.get_row("enterprise",primary_key,columns_to_get,None,1) dict_data = getRow_ots_primary(return_row) if dict_data is not None: bool_q = BoolQuery(must_queries=[TermQuery("enterprise_name",item), TermQuery("status",1)]) rows,next_token,total_count,is_allowed = ots_client.search("enterprise_contact","enterprise_contact_index", SearchQuery(bool_q,limit=10), columns_to_get=ColumnsToGet(["contact_person","phone_no","is_mobile","level","is_legal_person","is_manager","is_shareholder"],ColumnReturnType.SPECIFIED)) list_contact = getRow_ots(rows) dict_data["contacts"] = list_contact result_queue.put({item:dict_data}) except Exception as e: traceback.print_exc() ots_client = getConnect_ots() mt = MultiThreadHandler(task_queue=task_queue,task_handler=_handle,result_queue=result_queue,thread_count=50,ots_client=ots_client) mt.run() dict_enterprise = {} while True: try: _dict = result_queue.get(False) for k,v in _dict.items(): dict_enterprise[k] = v except Exception as e: break return dict_enterprise def splitIntoList(_str,_splitkeys): list_words = [] for _word in re.split(_splitkeys,_str): if _word.strip()=="": continue list_words.append(_word) return list_words def fixContactPerson(df_data,list_df_columns,get_legal_person=False): set_enterprise = set() if len(df_data.keys())>0: for _tenderee,_agency,_win_tenderer in zip(df_data["招标单位"],df_data["代理单位"],df_data["中标单位"]): set_enterprise.add(_tenderee) set_enterprise.add(_agency) set_enterprise.add(_win_tenderer) if "" in set_enterprise: set_enterprise.remove("") if None in set_enterprise: set_enterprise.remove(None) dict_enterprise = getDictEnterprise(list(set_enterprise),columns_to_get = ["procurement_system","company_org_type","reg_capital","actual_capital","contacts","estiblish_time","social_staff_num","zhong_biao_number","tou_biao_number","credit_code","legal_person_name","phone_number"]) # conn = getConnection_oracle() # cursor = conn.cursor() if len(set_enterprise)>0: for _i in range(len(df_data["招标单位"])): _enterprise_name = df_data["招标单位"][_i] if df_data["招标联系人电话"][_i]=="": contacts = dict_enterprise.get(_enterprise_name,{}).get("contacts") if contacts is not None: _person,_phone = getOneContact(contacts) df_data["招标联系人"][_i] = _person df_data["招标联系人电话"][_i] = _phone if "信用代码" not in df_data: df_data["信用代码"] = [] df_data["信用代码"].append(dict_enterprise.get(_enterprise_name,{}).get("credit_code","")) if "招标人采购系统" not in df_data: df_data["招标人采购系统"] = [] df_data["招标人采购系统"].append(dict_enterprise.get(_enterprise_name,{}).get("procurement_system","")) if "招标人类型" not in df_data: df_data["招标人类型"] = [] df_data["招标人类型"].append(dict_enterprise.get(_enterprise_name,{}).get("company_org_type","")) # if "原网地址" not in df_data: # df_data["原网地址"] = [] # if df_data["公告类别"][_i]=="招标公告": # table_name = "bxkc.T_ZHAO_BIAO_GONG_GAO" # else: # table_name = "bxkc.T_ZHONG_BIAO_XIN_XI" # sql = "select detail_link from %s where id='%s' "%(table_name,df_data["uuid"][_i]) # cursor.execute(sql) # rows = cursor.fetchall() # if len(rows)>0: # df_data["原网地址"].append(rows[0][0]) # else: # df_data["原网地址"].append("") _enterprise_name = df_data["代理单位"][_i] if df_data["代理联系人电话"][_i]=="": contacts = dict_enterprise.get(_enterprise_name,{}).get("contacts") if contacts is not None: _person,_phone = getOneContact(contacts) df_data["代理联系人"][_i] = _person df_data["代理联系人电话"][_i] = _phone _enterprise_name = df_data["中标单位"][_i] if get_legal_person: _person = dict_enterprise.get(_enterprise_name,{}).get("legal_person_name","") _phone = dict_enterprise.get(_enterprise_name,{}).get("phone_number","") if len(_phone)==11 and _phone[0]=="1": df_data["中标单位联系人"][_i] = _person df_data["中标单位联系电话"][_i] = _phone else: if df_data["中标单位联系电话"][_i]=="": contacts = dict_enterprise.get(_enterprise_name,{}).get("contacts") if contacts is not None: _person,_phone = getOneContact(contacts,mobile_only=True) df_data["中标单位联系人"][_i] = _person df_data["中标单位联系电话"][_i] = _phone list_df_columns.extend(['信用代码','招标人采购系统','招标人类型']) def generateBoolShouldQuery(list_field,list_should_words,cls): list_should_q = [] assert isinstance(list_field,(list)) assert isinstance(list_should_words,(list)) assert cls in (TermQuery,MatchPhraseQuery,WildcardQuery,RangeQuery) for word in list_should_words: for field in list_field: if cls in (RangeQuery,): list_should_q.append(cls(field,*field)) if cls in (WildcardQuery,): list_should_q.append(cls(field,"*%s*"%word)) else: list_should_q.append(cls(field,word)) return BoolQuery(should_queries=list_should_q) # excel 数据处理库 import openpyxl # excel 数据样式设置类 from openpyxl.styles import Font, PatternFill, Border, Side, Alignment from openpyxl.styles import Border, Side, colors import pandas as pd from openpyxl.utils import get_column_letter from openpyxl.styles import Font, Alignment import os def my_border(t_border, b_border, l_border, r_border): border = Border(top=Side(border_style=t_border, color=colors.BLACK), bottom=Side(border_style=b_border, color=colors.BLACK), left=Side(border_style=l_border, color=colors.BLACK), right=Side(border_style=r_border, color=colors.BLACK)) return border #初始化制定区域边框为所有框线 def format_border(ws,s_column, s_index, e_column , e_index): for row in tuple(ws[s_column + str(s_index):e_column + str(e_index)]): for cell in row: cell.border = my_border('thin', 'thin', 'thin', 'thin') def adjust_excel(source,target,columns=["A","B","C","D"]): wb = openpyxl.load_workbook(source) for sheet in wb.sheetnames: ws = wb[sheet] df = pd.read_excel(source,sheet_name=sheet) # 把表头改到最后一行 df.loc[len(df)]=list(df.columns) list_row_height = [] for col in df.columns: # 获取列序号 index = list(df.columns).index(col) # 获取行字母表头 letter = get_column_letter(index+1) # 获取当前列最大宽度 collen = df[col].apply(lambda x :len(str(x).encode())).max() # 设置列宽为最大长度比例 _width = min(30,collen*0.9) ws.column_dimensions[letter].width = _width for i in df.index: # 设置单元格对齐方式 Alignment(horizontal=水平对齐模式,vertical=垂直对齐模式,text_rotation=旋转角度,wrap_text=是否自动换行) alignment = Alignment(horizontal='center', vertical='center', text_rotation=0, wrap_text=True) format_border(ws,columns[0], 0, columns[-1], len(df)) # 根据实际列数量修改 for j in columns: ws[j+str(i+1)].alignment = alignment list_width = [] for col in df.columns: # 获取列序号 index = list(df.columns).index(col) # 获取行字母表头 letter = get_column_letter(index+1) # 获取当前列最大宽度 list_width.append(len(str(df[col][i]).encode())) ws.row_dimensions[index].height = 20+max(list_width)//30*5 wb.save(target) def getDocument(list_query,columns,table_name="document",table_index="document_index",thread_count=30,sort_column="page_time"): task_queue = Queue() for _q in list_query: task_queue.put(_q) print("task_queue_size",task_queue.qsize()) result_queue = Queue() list_row = [] ots_client = getConnect_ots() def _handle(_dict,result_queue,ots_client): try: item = _dict.get("query") _limit = _dict.get("limit") keyword = _dict.get("keyword") l_rows = [] rows,next_token,total_count,is_all_succeed = ots_client.search(table_name,table_index, SearchQuery(item,sort=Sort(sorters=[FieldSort(sort_column,SortOrder.DESC)]),limit=100,get_total_count=True), ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED)) dict_row = getRow_ots(rows) if keyword is not None: for _row in dict_row: _row["keyword"] = keyword l_rows.extend(dict_row) log("total count:%d"%total_count) _count = len(dict_row) while next_token: if _limit and len(l_rows)>=_limit: break rows,next_token,total_count,is_all_succeed = ots_client.search(table_name,table_index, SearchQuery(item,next_token=next_token,limit=100,get_total_count=True), ColumnsToGet(columns,return_type=ColumnReturnType.SPECIFIED)) dict_row = getRow_ots(rows) if keyword is not None: for _row in dict_row: _row["keyword"] = keyword l_rows.extend(dict_row) _count += len(dict_row) print("%d/%d"%(_count,total_count)) result_queue.put(l_rows) except Exception as e: traceback.print_exc() mt = MultiThreadHandler(task_queue,_handle,result_queue,thread_count,ots_client=ots_client) mt.run() while 1: try: dict_row = result_queue.get(False) list_row.extend(dict_row) except Exception as e: break return list_row class ExportEntity(): def __init__(self,table,table_index,list_query,columns,just_get_totol_count=False): self.table = table self.table_index = table_index self.ots_client = getConnect_ots() self.task_queue = Queue() self.result_queue = Queue() for _q in list_query: self.task_queue.put(_q) def _handle(self,item,task_queue): self.ots_client.search(self.table,self.table_index, SearchQuery()) def export(self): pass class A(): pass def test(a,bc={"a":2}): pass if __name__ == '__main__': # print(dir(test)) # _sign = inspect.signature(test) # print(_sign.parameters) # for _o in _sign.parameters: # print(_sign.parameters[_o].default) a = A() print(str(a))