# --- package --- # --- custom_module --- from custom_lib import Custom_Lib as custom # --- common --- import json import os def create_config_path(company_nm, prm_title_nm, prm_ver_nm): home_dir = os.path.expanduser('~') appdata_dir = os.path.join(home_dir, 'AppData', 'Local') new_dir = f'{company_nm}/{prm_title_nm}/{prm_ver_nm}' if not os.path.isdir(f'{appdata_dir}/{new_dir}'): new_dir_path = os.path.join(appdata_dir, new_dir) os.makedirs(new_dir_path) return f'{appdata_dir}/{new_dir}' elif os.path.exists(f'{appdata_dir}/{new_dir}'): return f'{appdata_dir}/{new_dir}' else: return 'Fail-AppData directory not found' def get_total_weather(target_url, target_city): weather_dt = custom.my_crawling.get_lxml(target_url) for location in weather_dt.select("location"): if target_city == location.select_one("city").string: total_dt = (weather_dt.select("location")[0]('data')) # weather for xml_string in total_dt: tmef = xml_string.find('tmef').string if not '12:00' in tmef: tmef_rep = tmef.replace(filter_word1, "") tmef_rep_2 = tmef_rep.replace(filter_word2, "") wf = xml_string.find('wf').string tmx = xml_string.find('tmx').string tmn = xml_string.find('tmn').string weather4week.append([tmef_rep_2, wf, tmx, tmn]) return weather4week def get_local_weather(target_url): weather_dt = custom.my_crawling.get_lxml(target_url) local_dt = (weather_dt.select('data')) # weather local_category = weather_dt.find('category').string # address local_pubdate = weather_dt.find('pubdate').string # update for xml_string in local_dt: local_pop = xml_string.find('pop').string # 강수확률 local_wfkor = xml_string.find('wfkor').string # 날씨 weather4etc.append([local_category, local_pop, local_wfkor, local_pubdate]) break for xml_string in local_dt: local_temp = str(round(float(xml_string.find('temp').string))) # 현재온도 local_reh = xml_string.find('reh').string # 습도 local_ws = str(round(float(xml_string.find('ws').string)) ) # 풍속(m/s) local_wden = str(xml_string.find('wden').string).lower() # 풍향_kor local_hour = str(xml_string.find('hour').string) # 시각 weather4day.append([local_temp, local_reh, local_ws, local_wden, local_hour]) if weather4day: del weather4day[8:] return weather4etc, weather4day def merge_dt(total_url, local_url, total_city, ems_url, ems_port, ems_id, ems_pw, device_id, template_cd): total_dt = get_total_weather(total_url, total_city) local_etc, local_dt = get_local_weather(local_url) for a in local_dt: temp1 = int(a[0]) if temp1 > 0: if temp1 % 2 == 0: a.insert(1, f'temp_{temp1}') else: a.insert(1, f'temp_{temp1 + 1}') else: if temp1 % 2 == 0: a.insert(1, f'temp_{temp1}') else: a.insert(1, f'temp_{temp1 - 1}') a.insert(3, f'reh_{a[2]}') a[2] = f'{a[2]}%' a[5] = f'ws_{a[5]}' a[4] = f'{a[4]} m/s' a[-1] = '{:02d}:00'.format(int(a[-1])) for b in total_dt: if b[1] == '맑음': b[1] = f'wf_clear' elif b[1] == '구름조금' or b[1] == '구름많음': b[1] = f'wf_cloudly' elif b[1] == '구름많고 비' or b[1] == '구름많고 비/눈': b[1] = f'wf_cloudly_rain' elif b[1] == '구름많고 눈/비' or b[1] == '구름많고 눈': b[1] = f'wf_cloudly_snow' elif b[1] == '흐림': b[1] = f'wf_mostly_cloudly' elif b[1] == '흐리고 비' or b[1] == '흐리고 비/눈': b[1] = f'wf_mostly_cloudly_rain' elif b[1] == '흐리고 눈/비' or b[1] == '흐리고 눈': b[1] = f'wf_mostly_cloudly_snow' else: b[1] = f'wf_{b[0]}' b[2] = f'{b[2]}°' b[3] = f'{b[3]}°' for c in local_etc: c[1] = f'{c[1]}%' aa = {} bb = {} cc = {} dd = {} i = 0 ii = 0 iii = 0 for sublist in local_dt: for j in range(len(sublist)): local_col = local_cols[i] value = sublist[j] aa[local_col] = value i += 1 for sublist in total_dt: for j in range(len(sublist)): total_col = total_cols[ii] value = sublist[j] bb[total_col] = value ii += 1 for sublist in local_etc: for j in range(len(sublist)): local_etc_col = local_etc_cols[iii] value = sublist[j] cc[local_etc_col] = value iii += 1 dd['site_code'] = '' dd['barcode'] = device_id dd['items_name'] = '' dd['tag_tpl_cd'] = template_cd aa.update(bb) aa.update(cc) aa.update(dd) # 20260508 tak if you want to use esl api, you should import esl api # json2ems(ems_url, ems_port, ems_id, ems_pw, aa) def json2ems(ems_url, ems_port, ems_id, ems_pw, json_dt): v2_login_var = aerocodez_esl_api.v2_login(ems_url, ems_port, ems_id, ems_pw) if v2_login_var: if 'success' in v2_login_var.values() or 'password_should_be_changed' in v2_login_var.values(): log.info('v2_login-success') v2_login_var_list = v2_login_var['data'] for login_var_list in v2_login_var_list: for key, value in login_var_list.items(): if key == 'token': v2_token = value v2_query_title_merchandise_var = aerocodez_esl_api.v2_query_title_merchandise(ems_url, ems_port, v2_token) if v2_query_title_merchandise_var: if 'success' in v2_query_title_merchandise_var.values(): log.info('v2_query_title_merchandise-success') v2_query_title_merchandise_var_list = v2_query_title_merchandise_var['data'] sort_dict_a = sorted(v2_query_title_merchandise_var_list, key=lambda x: x['index']) for sorted_aa in sort_dict_a: for key, value in sorted_aa.items(): if key == 'name': ems_columns.append(value) v2_merchandise_info_var = aerocodez_esl_api.v2_merchandise_info(ems_url, ems_port, v2_token, json_dt['barcode']) if v2_merchandise_info_var: sort_dict_b = [{k: dct[k] for k in ems_columns[:-5]} for dct in [json_dt]] json_string = json.dumps(sort_dict_b[0], ensure_ascii=False) raw_data = json.loads(json_string) if 'success' in v2_merchandise_info_var.values(): log.info('v2_merchandise_info_var-sucess') v2_edit_merchandise_var = aerocodez_esl_api.v2_edit_merchandise(ems_url, ems_port, v2_token, json_dt['barcode'], raw_data) if v2_edit_merchandise_var: if 'success' in v2_edit_merchandise_var.values(): log.info('v2_edit_merchandise_var-success') else: log.error('v2_edit_merchandise_var-errorx002') else: log.error('v2_edit_merchandise_var-errorx001') elif 'unregistered_merchandise_key' in v2_merchandise_info_var.values(): log.info('v2_merchandise_info_var-unregistered_merchandise_key') v2_add_merchandise_var = aerocodez_esl_api.v2_add_merchandise(ems_url, ems_port, v2_token, raw_data) if v2_add_merchandise_var: if 'success' in v2_add_merchandise_var.values(): log.info('v2_add_merchandise-success') else: log.error('v2_add_merchandise-errorx002') else: log.error('v2_add_merchandise-errorx001') else: log.error('v2_merchandise_info_var-errorx002') else: log.error('v2_merchandise_info_var-errorx001') else: log.error('v2_query_title_merchandise-errorx002') else: log.error('v2_query_title_merchandise-errorx001') else: log.error('v2_login-errorx002') else: log.error('v2_login-errorx001') # --- prm_info_variables --- company = 'AeroCodeZ' prm_title = 'Weather Crawling' prm_renew_ver = 1 prm_modify_ver = 0 prm_bug_ver = 0 prm_ver = f'{prm_renew_ver}.{prm_modify_ver}.{prm_bug_ver}' # --- list_variables --- weather4week = [] weather4day = [] weather4etc = [] # --- dict_variables --- sort_weather = {} # --- columns_variables --- ems_columns = [] bs_cols = ['site_code', 'barcode', 'items_name', 'tag_tpl_cd'] local_etc_cols = ['curr_adrs', 'curr_pop', 'curr_wf', 'bs_update'] local_cols = ['temp1', 'temp_img1', 'reh1', 'reh_img1', 'ws1', 'ws_img1', 'day1', 'temp2', 'temp_img2', 'reh2', 'reh_img2', 'ws2', 'ws_img2', 'day2', 'temp3', 'temp_img3', 'reh3', 'reh_img3', 'ws3', 'ws_img3', 'day3', 'temp4', 'temp_img4', 'reh4', 'reh_img4', 'ws4', 'ws_img4', 'day4', 'temp5', 'temp_img5', 'reh5', 'reh_img5', 'ws5', 'ws_img5', 'day5', 'temp6', 'temp_img6', 'reh6', 'reh_img6', 'ws6', 'ws_img6', 'day6', 'temp7', 'temp_img7', 'reh7', 'reh_img7', 'ws7', 'ws_img7', 'day7', 'temp8', 'temp_img8', 'reh8', 'reh_img8', 'ws8', 'ws_img8', 'day8'] total_cols = ['date1', 'wf_img1', 'tmx1', 'tmn1', 'date2', 'wf_img2', 'tmx2', 'tmn2', 'date3', 'wf_img3', 'tmx3', 'tmn3', 'date4', 'wf_img4', 'tmx4', 'tmn4', 'date5', 'wf_img5', 'tmx5', 'tmn5', 'date6', 'wf_img6', 'tmx6', 'tmn6', 'date7', 'wf_img7', 'tmx7', 'tmn7', 'date8', 'wf_img8', 'tmx8', 'tmn8'] # --- filter_variables --- filter_word1 = '2023-' filter_word2 = '00:00' ''' build command''' # pyinstaller -F --specpath=.\ESL\app\spec --distpath=.\ESL\app\dist\ --workpath=.\ESL\app\bulid\ C:\Users\ATECAP\PycharmProjects\pythonProject\ESL\cmd\private\weather_crawling\Weather_Crawling.py # pyinstaller --distpath=.\ESL\app\dist\ --workpath=.\ESL\app\bulid\ -F C:\Users\ATECAP\PycharmProjects\pythonProject\ESL\app\spec\Weather_Crawling.spec if __name__ == '__main__': custom.init() log = custom.my_logger.logger() try: config_path = create_config_path(company, prm_title, prm_ver) if not os.path.isfile(f'{config_path}/user.config'): a = custom.my_file_config.save_config() a['system'] = {} a['system']['ems_url'] = '127.0.0.1' a['system']['ems_port'] = '8080' a['system']['ems_id'] = 'admin' a['system']['ems_pw'] = 'esl' a['crawling'] = {} a['crawling']['total_url'] = 'http://www.kma.go.kr/weather/forecast/mid-term-rss3.jsp?stnId=109' a['crawling']['local_url'] = 'http://www.kma.go.kr/wid/queryDFSRSS.jsp?zone=4113565500' a['crawling']['local_city'] = '성남' a['etc'] = {} a['etc']['device_id'] = '18FC2600002795A2' a['etc']['template_cd'] = 'A' custom.my_file_config.write_config(a, config_path, f'{config_path}/user.config') b = custom.my_file_config.read_config(f'{config_path}/user.config') merge_dt(b['crawling']['total_url'], b['crawling']['local_url'], b['crawling']['local_city'], b['system']['ems_url'], b['system']['ems_port'], b['system']['ems_id'], b['system']['ems_pw'], b['etc']['device_id'], b['etc']['template_cd']) except Exception as e: log.error(e)