source/gov.py

"""Module to crawl the website 'https://www.gov.cn' to fetch and process articles."""
import time
import urllib.request
from datetime import datetime, timedelta

from lxml import etree
from prefect import task, get_run_logger

from controllers.utils import crawl_by_url

@task(name = "Data Collection - gov", log_prints = True)
def crawl(delta):
    """
    Crawls the government website for policy interpretation and latest news articles.

    Args:
        delta (int): The number of days in the past to consider for crawling.

    Returns:
        None
    """
    logger = get_run_logger()
    logger.info("gov.cn")
    i = 0
    while i > -1:
        if i == 0:
            category_url = "https://www.gov.cn/zhengce/jiedu/home.htm"
        else:
            category_url = f"https://www.gov.cn/zhengce/jiedu/home_{i}.htm"
        i = i + 1
        req = urllib.request.urlopen(category_url, timeout=60)
        text = req.read()
        html_text = text.decode("utf-8")
        page = etree.HTML(html_text)
        articlelist = page.xpath("//div[contains(@class, 'news_box')]//h4")
        for article in articlelist:
            if isinstance(article, etree._Element):
                subelement = etree.tostring(article).decode()
                subpage = etree.HTML(subelement)
                date = subpage.xpath("//span/text()")[0]
                parsed_datetime = datetime.strptime(
                        time.strftime("%Y-%m-%d", time.strptime(date, "%Y-%m-%d")),
                        "%Y-%m-%d")
                if parsed_datetime < (datetime.today() - timedelta(days=delta)):
                    i = -1
                else:
                    urls = subpage.xpath("//a[contains(@target, '_blank')]/@href")
                    for url in urls:
                        try:
                            article = {}
                            url = url.replace('../', 'https://www.gov.cn/zhengce/')
                            if "https://www.gov.cn" in url:
                                article['category'] = "Policy Interpretation"
                                crawl_by_url(url, article)
                        except (urllib.error.URLError, etree.XMLSyntaxError) as error:
                            logger.error(error)
    i = 0
    while i > -1:
        if i == 0:
            category_url = "https://www.gov.cn/zhengce/zuixin/home.htm"
        else:
            category_url = f"https://www.gov.cn/zhengce/zuixin/home_{i}.htm"
        i = i + 1
        req = urllib.request.urlopen(category_url, timeout=60)
        text = req.read()
        html_text = text.decode("utf-8")
        page = etree.HTML(html_text)
        articlelist = page.xpath("//div[contains(@class, 'news_box')]//h4")
        for article in articlelist:
            if isinstance(article, etree._Element):
                subelement = etree.tostring(article).decode()
                subpage = etree.HTML(subelement)
                date = subpage.xpath("//span/text()")[0]
                parsed_datetime = datetime.strptime(
                        time.strftime("%Y-%m-%d", time.strptime(date, "%Y-%m-%d")),
                        "%Y-%m-%d")
                if parsed_datetime < (datetime.today() - timedelta(days=delta)):
                    i = -1
                else:
                    urls = subpage.xpath("//a[contains(@target, '_blank')]/@href")
                    for url in urls:
                        try:
                            article = {}
                            url = url.replace('../', 'https://www.gov.cn/zhengce/')
                            if "https://www.gov.cn" in url:
                                article['category'] = "Policy Interpretation"
                                crawl_by_url(url, article)
                        except (urllib.error.URLError, etree.XMLSyntaxError) as error:
                            logger.error(error)