forked from iCON/CrawlerEngines
229 lines
8.3 KiB
Python
229 lines
8.3 KiB
Python
from bs4 import BeautifulSoup as soup
|
|
from selenium import webdriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.wait import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
import urllib.parse
|
|
import json
|
|
import time
|
|
import re
|
|
import os
|
|
from utils.logger import logger
|
|
|
|
# from utils.logger import logger
|
|
|
|
BASE_URL_EN = "https://english.news.cn"
|
|
BASE_URL_CN = "https://so.news.cn/"
|
|
XINHUA_OVERSEAS_REGIONS = [
|
|
"asiapacific",
|
|
"europe",
|
|
"africa",
|
|
"northamerica",
|
|
"german",
|
|
"20241016",
|
|
]
|
|
|
|
# The maximum number of blog posts to crawl
|
|
# C:\Users\MillionZhang\.cache\selenium\chromedriver\win64
|
|
MAX_BLOG_LIMIT = 10
|
|
|
|
|
|
class Crawler_NewsCN:
|
|
def __init__(self) -> None:
|
|
from selenium.webdriver.chrome.options import Options
|
|
|
|
chrome_options = Options()
|
|
chrome_options.add_argument("--headless")
|
|
chrome_options.add_argument("--no-sandbox")
|
|
chrome_options.add_argument("--disable-dev-shm-usage")
|
|
self.driver = webdriver.Chrome(options=chrome_options)
|
|
self.selected_blogs = {"titles": [], "urls": []}
|
|
|
|
def __search_topic(self, topic: str, lang):
|
|
if lang == "en":
|
|
self.driver.get(BASE_URL_EN)
|
|
WebDriverWait(self.driver, 10).until(EC.title_contains("Xinhua"))
|
|
|
|
# input topic to be searched in search bar
|
|
search_bar = self.driver.find_element(By.CLASS_NAME, "search-input")
|
|
search_bar.send_keys(topic)
|
|
|
|
# click search button
|
|
search_submit_button = self.driver.find_element(By.ID, "searchSubmit")
|
|
search_submit_button.click()
|
|
|
|
# close home window and switch to new window
|
|
self.driver.close()
|
|
self.driver.switch_to.window(self.driver.window_handles[0])
|
|
WebDriverWait(self.driver, 10).until(EC.title_contains("Xinhua research"))
|
|
else:
|
|
if lang == "cn":
|
|
self.driver.get(BASE_URL_CN)
|
|
WebDriverWait(self.driver, 10).until(EC.title_contains("新华搜索"))
|
|
|
|
# input topic to be searched in search bar
|
|
search_bar = self.driver.find_element(By.CLASS_NAME, "input")
|
|
search_bar.send_keys(topic)
|
|
|
|
# click search button
|
|
# search_submit_button = self.driver.find_element(By.ID, "searchSubmit")
|
|
search_submit_button = self.driver.find_element(
|
|
By.CLASS_NAME, "search-button"
|
|
)
|
|
search_submit_button.click()
|
|
self.driver.switch_to.window(self.driver.window_handles[0])
|
|
WebDriverWait(self.driver, 10).until(EC.title_contains("新华搜索"))
|
|
|
|
def __select_blog_in_search_page(self):
|
|
raw_blogs = self.driver.find_element(By.CLASS_NAME, "content").get_attribute(
|
|
"innerHTML"
|
|
)
|
|
raw_blogs = soup(raw_blogs, features="html.parser")
|
|
raw_blogs = raw_blogs.findAll("div", attrs={"class": "item"}, recursive=True)
|
|
|
|
if not raw_blogs:
|
|
return
|
|
|
|
for raw_blog in raw_blogs:
|
|
title = (
|
|
raw_blog.find("div", class_="title")
|
|
.text.replace("\n", "")
|
|
.replace(" ", "")
|
|
.lower()
|
|
)
|
|
# prevent crawl duplicate blog from different source
|
|
if not title in self.selected_blogs.get("titles"):
|
|
self.selected_blogs["titles"].append(title)
|
|
self.selected_blogs["urls"].append(raw_blog.find("a")["href"])
|
|
|
|
if len(self.selected_blogs["urls"]) < MAX_BLOG_LIMIT:
|
|
# go to next page
|
|
next_page_btn = self.driver.find_element(
|
|
By.CLASS_NAME, "ant-pagination-next"
|
|
)
|
|
if next_page_btn.get_attribute("aria-disabled") != "true":
|
|
try:
|
|
next_page_btn.click()
|
|
time.sleep(2)
|
|
self.__select_blog_in_search_page()
|
|
except:
|
|
import traceback
|
|
|
|
print(traceback.format_exc())
|
|
else:
|
|
self.selected_blogs["titles"] = self.selected_blogs["titles"][
|
|
:MAX_BLOG_LIMIT
|
|
]
|
|
self.selected_blogs["urls"] = self.selected_blogs["urls"][:MAX_BLOG_LIMIT]
|
|
|
|
# print(self.selected_blogs["urls"])
|
|
|
|
def __retrieve_overseas_blog(self) -> dict:
|
|
blog_container = self.driver.find_element(By.CLASS_NAME, "main.clearfix")
|
|
blog_title = blog_container.find_element(By.CLASS_NAME, "Btitle").text
|
|
blog_meta = blog_container.find_element(By.CLASS_NAME, "wzzy").text
|
|
blog_content = blog_container.find_element(By.ID, "detailContent").text
|
|
|
|
return {"title": blog_title, "meta": blog_meta, "content": blog_content}
|
|
|
|
def __retrieve_china_blog(self) -> dict:
|
|
blog_container = self.driver.find_element(By.CLASS_NAME, "conBox")
|
|
blog_title_meta_container = blog_container.find_element(By.CLASS_NAME, "conTop")
|
|
blog_title = blog_title_meta_container.find_element(By.TAG_NAME, "h1").text
|
|
blog_meta = blog_title_meta_container.find_element(
|
|
By.CLASS_NAME, "infoBox.clearfix"
|
|
).text
|
|
blog_content_container = blog_container.find_element(By.CLASS_NAME, "conLeft")
|
|
blog_content = blog_content_container.find_element(By.ID, "detailContent").text
|
|
|
|
return {"title": blog_title, "meta": blog_meta, "content": blog_content}
|
|
|
|
# 当lang=cn时
|
|
def __retrieve_cn_blog(self) -> dict:
|
|
blog_title = (
|
|
self.driver.find_element(By.CLASS_NAME, "head-line.clearfix")
|
|
.find_element(By.CLASS_NAME, "title")
|
|
.text
|
|
)
|
|
|
|
blog_meta = (
|
|
self.driver.find_element(By.CLASS_NAME, "header-cont.clearfix")
|
|
.find_element(By.CLASS_NAME, "source")
|
|
.text
|
|
)
|
|
|
|
blog_content = self.driver.find_element(By.ID, "detailContent").text
|
|
|
|
return {"title": blog_title, "meta": blog_meta, "content": blog_content}
|
|
|
|
def __get_and_save_blog(self, url: str, lang):
|
|
self.driver.get(url)
|
|
WebDriverWait(self.driver, 10).until(EC.title_contains("新华"))
|
|
|
|
if lang == "en":
|
|
# region_code = urllib.parse.urlparse(url).path.split("/")[1]
|
|
# if region_code in XINHUA_OVERSEAS_REGIONS:
|
|
# blog = self.__retrieve_overseas_blog()
|
|
# else:
|
|
# blog = self.__retrieve_china_blog()
|
|
|
|
blog = self.__retrieve_overseas_blog()
|
|
blog_title = blog.get("title", "")
|
|
else:
|
|
if lang == "cn":
|
|
blog = self.__retrieve_cn_blog()
|
|
blog_title = blog.get("title", "")
|
|
# print(blog_title)
|
|
|
|
# Remove invalid char in file_path_name on Windows
|
|
# invalid_chars_pattern = r'[\\/:*?"<>|]'
|
|
# blog_title = re.sub(invalid_chars_pattern, "", blog_title)
|
|
|
|
# file = open(os.path.join("", f"Xinhua_{blog_title}.json"), "w")
|
|
# json.dump(blog, file)
|
|
# file.close()
|
|
# time.sleep(2)
|
|
return blog
|
|
|
|
def search_and_save(self, topic: str, lang):
|
|
content_list = []
|
|
self.__search_topic(topic, lang)
|
|
time.sleep(1)
|
|
self.__select_blog_in_search_page()
|
|
# print(self.selected_blogs)
|
|
url_list = self.selected_blogs.get("urls", [])
|
|
|
|
idx = 1
|
|
for url in url_list:
|
|
logger.warning(f"{idx}/{len(url_list)} url:{url}")
|
|
content = self.__get_and_save_blog(url, lang)
|
|
content_list.append(content)
|
|
idx += 1
|
|
return content_list
|
|
|
|
def direct_save(self, url: str):
|
|
self.__get_and_save_blog(url)
|
|
|
|
def process(self, inputData):
|
|
# return {"full_name": "xxx", "date_of_birth": "1956-01-01"}
|
|
|
|
print("Crawler_NewsCN / inputData", inputData)
|
|
keyword = inputData["keyword"]
|
|
lang = inputData["lang"]
|
|
|
|
result = []
|
|
try:
|
|
result = self.search_and_save(keyword, lang)
|
|
# print("result", result)
|
|
except Exception as e:
|
|
import traceback
|
|
|
|
print(traceback.format_exc())
|
|
# finally:
|
|
# self.driver.quit()
|
|
|
|
logger.warning(
|
|
f"【Crawler_NewsCN process completed】, keyword={keyword}, result len={len(result)}"
|
|
)
|
|
return result
|