[Практика сканирования] Пример сканирования китайских экстренных новостей

искусственный интеллект сбор данных

«Это 16-й день моего участия в ноябрьском испытании обновлений. Подробную информацию об этом событии см.:Вызов последнего обновления 2021 г."

Основное содержание: используйте selenium и bs4 для регулярного автоматического сканирования экстренных новостей Китая.

Тело кода:EventExtractor, ChinaEmergencyNewsExtractor, save_results, contentExtractor

EventExtractor

Используйте этот класс для управления поисковыми роботами для получения информации о веб-странице и обработки информации о веб-странице.

1.__init__(self)Настройте некоторые параметры по умолчанию в:

  • options.add_experimental_option("prefs", {"profile.managed_default_content_settings.images": 2}): закройте изображение браузера, чтобы повысить эффективность отклика веб-страницы.
  • chrome_options.add_argument('--headless'): скрыть всплывающие окна браузера.
  • webdriver.Chrome(): здесь используйте драйвер браузера Google.
  • url: просканировать URL главной страницы
  • html: сохранить просканированные HTML-данные.
  • save_dir: сохранить просканированный каталог файлов
# 爬取器
class EventExtractor():
    def __init__(self):
        options = webdriver.ChromeOptions()
        options.add_experimental_option("prefs", {"profile.managed_default_content_settings.images": 2})
        chrome_options = Options()
        chrome_options.add_argument('--headless')
        self.browser = webdriver.Chrome("chromedriver.exe", options=options,chrome_options=chrome_options)
        self.url = 'http://www.12379.cn/html/gzaq/fmytplz/index.shtml'
        self.html = None
        self.save_dir = './chinaEmergency/chinaEEP.xlsx'
    # 登录
    def login(self):
        self.browser.get(self.url)
        WebDriverWait(self.browser, 1000).until(
            EC.presence_of_element_located((By.CLASS_NAME, 'main'))
        )
    # 获取当前网页资源
    def get_html(self):
        self.html = self.browser.page_source
    # 跳转下一页
    def next_page(self,verbose=False):
        try:
            submit_page = self.browser.find_element(by=By.XPATH,value=r"//*[@class='next']") # find_element_by_xpath(r"//*[@class='next']")
            submit_page.click()
            WebDriverWait(self.browser, 1000).until(
                EC.presence_of_element_located((By.CLASS_NAME, 'main'))
            )
            curr_page = self.browser.find_element(by=By.XPATH,value=r"//*[@class='avter']")
            if verbose and curr_page != 0:
                print(int(curr_page) - 1)
        except:
            print("页面挑战异常结束")
    # 网页处理函数
    def ProcessHTML(self, verbose=False, withDuplicates=True):
        """
        html:html格式的输入
        verbose:控制爬取可视化打印的标志位,0表示不显示，1表示html打印，2表示详细打印，默认为1
        函数输入：html
        函数输出：具有新闻信息的pandas表格
        """
        duplicates_flag = False
        if not os.path.exists(self.save_dir):
            names = ['title', 'content', 'public_time', 'urls']
            history = pd.DataFrame([], columns=names)
            history.to_excel(self.save_dir, index=False)
            print("首次运行 爬取时间:{}".format(time.strftime('%Y.%m.%d', time.localtime(time.time()))))
        else:
            history = pd.read_excel(self.save_dir)
            history_key = history["public_time"].sort_values(ascending=False).values[0]
            print("历史爬取节点:{} 爬取时间:{}".format(history_key,time.strftime('%Y.%m.%d',time.localtime(time.time()))))
        soup = BeautifulSoup(self.html)
        tbody = soup.find("div", attrs={"class": "list_content"})
        for idx, tr in enumerate(tbody.find_all("li")):
            # 设置随机睡眠，防止封IP
            if random.randint(1, 2) == 2:
                time.sleep(random.randint(3, 6))
            a = tr.find("a")
            title = a.text
            href = "http://www.12379.cn" + a["href"]
            content = contentExtractor(href)
            public_time = tr.find("span").text
            results = [title, content, public_time, href]
            if verbose:
                print([public_time, title, content, href])
            # 查重
            if withDuplicates and public_time in history["public_time"].values and title in history["title"].values:
                duplicates_flag = True
                print("监测到重复爬取，已停止后续爬取行为")
                break
            else:
                save_results(results)
            # 查重
        return duplicates_flag
    # 关闭driver
    def close(self):
        self.browser.close()

2.login(self): URL-адрес доступа,untilСередина — это задержка ожидания, и следующий шаг будет достигнут только при обнаружении .

3.get_html(self): Получить html-данные текущей страницы.

4.next_page: Прыжки страницы

5.ProcessHTML(self, verbose=False, withDuplicates=True):обработка данных

  • При первом запуске будет создан новый файл Excel, а существующие исторические файлы будут дедуплицированы и просканированы на основе последней даты выпуска, а сканирование остановится, когда будут обнаружены заголовок и время.
  • Время сна устанавливается для сканирования каждой части информации, чтобы не нагружать сервер.
  • Сканирование другой информации относительно просто и не будет подробно объясняться.
  • Для извлечения текстового содержимого запускается другой селен для сканирования (обратите внимание, что срок действия некоторых ссылок истек, 404 сразу вернет текст как «Нет»), код выглядит следующим образом:
# 正文抽取函数
def contentExtractor(url):
    """
    函数输入：url
    函数输出：输出当前url下的新闻正文内容，若没有新闻内容则输出"无"
    """
    # 先检测是否404
    user_agent = {
        'User-Agent': 'Mozilla/5.0.html (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.html.2171.71 Safari/537.36'}
    r = requests.get(url, headers=user_agent, allow_redirects=False)
    if r.status_code == 404:
        return "无"
    else:
        # 没有404，则抽取网页正文内容
        chrome_options = Options()
        chrome_options.add_argument('--headless')
        driver = webdriver.Chrome("chromedriver.exe", chrome_options=chrome_options)
        driver.get(url)
        WebDriverWait(driver, 1000).until(
            EC.presence_of_element_located((By.CLASS_NAME, 'content_text'))
        )
        respond = driver.page_source
        driver.quit()
        soup = BeautifulSoup(respond)
        text = soup.find("div", attrs={"class": "content_text"}).text
        return text
  • будет вызываться, если дубликаты не обнаруженыsave_results(results)Запишите текущие данные, код выглядит следующим образом:
def save_results(results):
    results_path = './chinaEmergency/chinaEEP.xlsx'
    names = ['title', 'content', 'public_time', 'urls']
    results = {k: v for k, v in zip(names, results)}

    if not os.path.exists(results_path):
        df1 = pd.DataFrame([], columns=names)
        df1.to_excel(results_path, index=False)
    else:
        df1 = pd.read_excel(results_path)
        new = pd.DataFrame(results, index=[1])
        df1 = df1.append(new, ignore_index=True)
        df1.sort_values(by="public_time", ascending=True, inplace=True)
        df1.reset_index(drop=True, inplace=True)
        df1.to_excel(results_path, index=False)

ChinaEmergencyNewsExtractor

Заполните страницу, поворачивая гусенику через эту функцию

Event = EventExtractor()Создайте ранее написанный краулер

Event.login()Войдите, чтобы сканировать сайт

Проведите номер входной страницы через цикл for, номер страницы может быть введен сам по себе

Event.get_html()Получить текущие HTML-данные страницы

Event.ProcessHTML(verbose=True)Обрабатывать данные HTML и сохранять их в указанный файл Excel, а также возвращать логическое значение, указывающее, следует ли повторять сканирование. Если повторится, перестанет листать страницу, если не повторится, будет использован.Event.next_page()Переворачивайте страницы.

Event.close()закрыть селен браузер

# 可控制的事件抽取器
def ChinaEmergencyNewsExtractor(page_nums=3,verbose=1):
    Event = EventExtractor()
    Event.login()
    for i in range(0, page_nums):
        Event.get_html()
        page_break = Event.ProcessHTML(verbose=True)
        if page_break:
            break
        Event.next_page()
    # 关闭driver，防止后台进程残留
    Event.close()

Автоматическое сканирование и основная функция

Автоматическое сканирование очень просто定时爬取стратегия, используяretryБиблиотека умеет быстро выполнять запланированные задачи.Основной код выглядит следующим образом:

# 定时抽取
@retry(tries=30,delay=60*60*24)
def retry_data():
    ChinaEmergencyNewsExtractor(page_nums=3, verbose=1)
    raise

использоватьraiseЗапустите исключение и позвольте библиотеке повторных попыток перезапустить функцию

trys: указывает количество перезапусков, здесь установлено 30 раз

delay: указывает время задержки перезапуска в секундах в коде60*60*24всего один день задержки

То есть сканировать содержимое веб-страницы каждые 24 часа, всего 30 раз, чтобы завершить автоматическое получение данных за один месяц.

Основная функция:

# 主函数
if __name__ == '__main__':
    retry_data()

Показать результаты

Рис. 1 Интерфейс сканирования при положенной трубке

Рис. 2. Сканирование файлов

На данный момент завершен пример автоматического сканирования экстренных новостей, интеллектуальный анализ и анализ данных являются новыми, и есть ошибки или несовершенства, пожалуйста, критикуйте и исправьте! !

Категории