kettle测试测试测试

# coding=utf-8 import re import pymysql import requests class ReQu(object): def __init__(self): self.db = pymysql.connect(host='192.168.0.227', user='chenva', password='chenva123', database='data_center', port=3306) self.cursor = self.db.cursor() def find_require_rule(self, rule_content_id): url = f"http://192.168.0.241:18000/lake/companySearch/findRequireRuleContentBasicInfo?ruleContentId={rule_content_id}" res = requests.get(url) print(res.json()) requireName = res.json().get('data').get('requireName') requireYear = res.json().get('data').get('requireYear') requireLevel = res.json().get('data').get('requireLevel') realRequireName = res.json().get('data').get('realRequireName') requireDesc = res.json().get('data').get('requireDesc') return requireName, requireYear, requireLevel, realRequireName, requireDesc def find_company_list(self, rule_content_id, scroll_id): url = f"http://192.168.0.241:18000/lake/companySearch/findCompanyListByRequireScRuleForDataCenter?ruleContentId={rule_content_id}&scrollId={scroll_id}" res = requests.get(url) data = res.json().get('data') if data is not None: scrollId = data.get('scrollId') companyInfoList = data.get("companyInfoList") return scrollId, companyInfoList return '', '' def write_mysql(self, rule_content_id, company_info_list): update_sql = f'update ods_lake_company_require_interface set delete_flag="1" where rule_content_id="{rule_content_id}"' self.cursor.execute(update_sql) self.db.commit() find_require_rule = self.find_require_rule(rule_content_id) requireid = find_require_rule[0] requireYear = find_require_rule[1] requireLevel = find_require_rule[2] realRequireName = find_require_rule[3] requireDesc = find_require_rule[4] batch_data = [] for i in company_info_list: data = tuple([rule_content_id, i, requireid, requireYear, requireLevel, realRequireName, requireDesc]) batch_data.append(data) # print(batch_data) insert_sql = f'''insert into ods_lake_company_require_interface(rule_content_id,company_info_id,require_id,require_year,require_level,require_name,require_desc) values (%s,%s,%s,%s,%s,%s,%s)''' self.cursor.executemany(insert_sql, batch_data) self.db.commit() def read_mysql(self): content = [] select_sql = f"select id from ods_swan_lake_company_require_generate where rule_type='1' and is_enable='1' and delete_flag='0' " self.cursor.execute(select_sql) batch_data = self.cursor.fetchall() # print(batch_data) for each_batch in batch_data: content_sql = f"select id from ods_swan_lake_company_require_generate_content where require_id='{each_batch[0]}' and delete_flag='0' " self.cursor.execute(content_sql) content_data = self.cursor.fetchall() for each_content in content_data: content.append(each_content[0]) return content def require(self, rule_content_id): print(rule_content_id) scroll_id = '' company_info_list = [] first_result = requ.find_company_list(rule_content_id, scroll_id) scroll_id = first_result[0] new_company_info_list = first_result[1] if len(new_company_info_list) > 0: for i in new_company_info_list: company_info_id = i.get('companyInfoId') company_info_list.append(company_info_id) while True: if scroll_id != '': next_result = requ.find_company_list(rule_content_id, scroll_id) scroll_id = next_result[0] next_company_info_list = next_result[1] if len(next_company_info_list) > 0: for j in next_company_info_list: company_info_id = j.get('companyInfoId') company_info_list.append(company_info_id) else: break self.write_mysql(rule_content_id, company_info_list) if __name__ == '__main__': requ = ReQu() # requ.find_require_rule('020a91aec23348959fcee1fc8ed4f16d') # requ.find_require_rule('9750cd8122384ad397e8e56ae38a8e2c') # requ.require('2d59eddd82b64c0fb5b661498c8bb26b') # requ.find_company_list('17bbe20a60464625bb7075832b0a863c', '') content_id_list = requ.read_mysql() # 从接口抓取之前先将原数据逻辑删除 requ.cursor.execute('update ods_lake_company_require_interface set delete_flag="1"') for content_id in content_id_list: requ.require(content_id)