from faker import Faker import pandas as pd import re import sys import json import unicodedata from bs4 import BeautifulSoup #解析requests请求到的HTML页面 from urllib.parse import urljoin import cpca # 地点 city mapping from selenium import webdriver import helium as hm from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from keywordInfo import key_pat, zwlx_list # 北京市信息发布，考编公告~ A _1 sub_url ="http://www.beijing.gov.cn/gongkai/rsxx/gwyzk/202211/t20221120_2862819.html" sub_url ="https://www.js.msa.gov.cn/art/2023/2/24/art_11436_1391666.html" def getDriver(): # Set path Selenium uas = Faker() CHROMEDRIVER_PATH = '/usr/bin/chromedriver' CHROMEDRIVER_PATH = './chromedriver' s = Service(CHROMEDRIVER_PATH) WINDOW_SIZE = "1920,1080" # Options chrome_options = Options() chrome_options.add_argument("--headless") chrome_options.add_argument("--window-size=%s" % WINDOW_SIZE) chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-infobars') chrome_options.add_argument('--disable-gpu') # 增加一个参数设置 chrome_options.add_argument('--disable-blink-features=AutomationControlled') chrome_options.add_experimental_option('excludeSwitches', ['enable-automation']) chrome_options.add_argument(f'user-agent={uas.chrome()}') with open('./stealth.min.js') as f: js = f.read() driver = webdriver.Chrome(executable_path=CHROMEDRIVER_PATH, options=chrome_options) driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {"source": js }) return driver def content_with_date(lines): if len(lines) < 1: return [] date_pattern1 = r'\d{1,2}月\d{1,2}日' date_pattern2 = r'[上|下]午' date_pattern3 = r'\d{4}年\d{1,2}月\d{1,2}日' inx_ = len(lines)-1 for inx_, line in enumerate(lines): matches = re.findall(f'({date_pattern1}|{date_pattern3})', line) if len(matches)>0: break if len(matches)<1: return [] # new_ = "\n".join(lines[:inx_+1]) # year_pattern = r'\d{4}年' # matches = re.findall(f'({year_pattern})', old_title) return lines[:inx_+1] #👏👏👏 这里需要调整获取多少段落，不至于获得太多冗余信息 def find_key_paragrap(search_text, paragraphs): # Loop through the paragraphs and print the text content of those that contain the search text for inx_, paragraph in enumerate(paragraphs): text = paragraph.text if search_text in text: # Get the index of the matched line # index = text.index(search_text) index = inx_ # Print the matched line and 5 lines before and after it start = max(0, inx_) end = min(len(paragraphs), index + 7) target_paragrap = paragraphs[start:end] texts = [i.text for i in target_paragrap] dt_lines = content_with_date(texts) if len(dt_lines) >= 1: return texts return None def titleLocInfo(title): """get loction and year from title""" # print(title) # print(driver.current_url) # zwk_year year_pattern = r'\d{4}年' matches = re.findall(f'({year_pattern})', title) zwk_year = matches[0] if len(matches) else "2023" # zwk_sheng area_df = cpca.transform([title]) # 省份 zwk_sheng = list(area_df["省"])[0] if area_df.shape[0] > 0 else "" a_ = list(area_df["市"])[0] if area_df.shape[0] > 0 else "" b_ = list(area_df["区"])[0] if area_df.shape[0] > 0 else "" zwk_diqu = a_ # 唯一标识 zwk_zip = list(area_df["adcode"])[0] if area_df.shape[0] > 0 else "" zwlx = zwlx_list[0] # 默认类型公务员 for i in zwlx_list: if i in title: zwlx = i res = [zwk_year, zwk_sheng, zwk_diqu, zwk_zip, zwlx] # print("\t".join(res)) return res def extract_from_driver(driver): """ get result from url request BeautifulSoup(texts,'html.parser') return: doc_item ,time source info, and attach information """ title=driver.title #[zwk_year, zwk_sheng, zwk_diqu, zwk_zip, zwlx] title_info = titleLocInfo(title) items_ = driver.find_elements_by_xpath("//p") items_ = [i.text for i in items_ if i.text != ""] context_to_label = "\n".join(items_) paragraphs = driver.find_elements_by_tag_name("p") paragraphs = [i for i in paragraphs if i.text.strip() != ""] # extract keyword info # key_pat["报名"], key_pat["考试"], key_pat["缴费"], key_pat["准考证"], key_pat["all"] def get_key_info(pt:list): for item in pt: res_ = find_key_paragrap(item, paragraphs) if res_ is not None: return res_ return "" bm_sj = get_key_info(key_pat["报名"]) fee_sj = get_key_info(key_pat["缴费"]) ks_sj = get_key_info(key_pat["考试"]) zkz_sj = get_key_info(key_pat["准考证"]) # 附件 links ".doc" or ".xls" ".xlsx" links = driver.find_elements_by_tag_name("a") unique_link = {} for link in links: url_ = link.get_attribute("href") content_ = link.get_attribute("textContent") url_con = url_ and (url_.endswith(".doc") or url_.endswith(".xls") or url_.endswith(".xlsx")) name_con = content_ and (content_.endswith(".doc") or content_.endswith(".xls") or content_.endswith(".xlsx")) if url_con or name_con: unique_link[content_] = url_ name = ["title", "zwk_year", "zwk_sheng", "zwk_diqu", "zwk_zip", "zwlx", "bm_sj", "fee_sj", "ks_sj", "zkz_sj", "fn_list" , "tidy_bm_sj","tidy_fee_sj", "tidy_ks_sj", "tidy_zkz_sj" ] doc_item = [title] doc_item.extend(title_info) doc_item.extend([bm_sj, fee_sj, ks_sj, zkz_sj, unique_link]) td_bm_sj = content_with_date(bm_sj) td_fee_sj = content_with_date(fee_sj) td_ks_sj = content_with_date(ks_sj) td_zkz_sj = content_with_date(zkz_sj) doc_item.extend([td_bm_sj, td_fee_sj, td_ks_sj, td_zkz_sj]) doc_dc = {} for k_, v_ in zip(name, doc_item): doc_dc[k_] = v_ return doc_dc # 用于表格化记录 def table_record_doc(doc): fn_dc = doc["fn_list"] row_item = [ doc["title"], doc["zwk_year"], doc["zwk_sheng"], doc["zwk_diqu"], doc["zwk_zip"], doc["zwlx"], "\n".join(doc["bm_sj"]), "\n".join(doc["fee_sj"]), "\n".join(doc["ks_sj"]), "\n".join(doc["zkz_sj"]), "\n".join(doc["tidy_bm_sj"]), "\n".join(doc["tidy_fee_sj"]), "\n".join(doc["tidy_ks_sj"]), "\n".join(doc["tidy_zkz_sj"]), "\n".join([f"{k}:{v}" for k,v in fn_dc.items()]) ] name = ["title", "zwk_year", "zwk_sheng", "zwk_diqu", "zwk_zip", "zwlx", "bm_sj", "fee_sj", "ks_sj", "zkz_sj", "tidy_bm_sj","tidy_fee_sj", "tidy_ks_sj", "tidy_zkz_sj", "fn_list"] a = pd.DataFrame(data=[row_item], columns=name) return row_item if __name__ == '__main__': mydriver = getDriver() # hm.goto(url) url = "https://www.js.msa.gov.cn/art/2023/2/24/art_11436_1391666.html" if len(sys.argv) > 1: url = sys.argv[1] hm.set_driver(mydriver) # 给它一个selnuim driver hm.go_to(sub_url) # mydriver.driver. import time time.sleep(2) res = extract_from_driver(mydriver) print("-raw, mostly contains----------------------------") print(res) print("报名，缴费，考试，准考证最相关信息") bm_sj = doc["bm_sj"] bm_sj = content_with_date(bm_sj) fee_sj = content_with_date(doc["fee_sj"]) ks_sj = content_with_date(doc["ks_sj"]) zkz_sj = content_with_date(doc["zkz_sj"]) print(bm_sj) print(fee_sj) print(ks_sj) print(zkz_sj) mydriver.close()