# This file tells Git which files shouldn't be added to source control


.gitignore

# First, specify the base Docker image.
# You can see the Docker images from Apify at https://hub.docker.com/r/apify/.
# You can also use any other image from Docker Hub.
FROM apify/actor-python:3.9

# Second, copy just requirements.txt into the actor image,
# since it should be the only file that affects "pip install" in the next step,
# in order to speed up the build
COPY requirements.txt ./

# Install the packages specified in requirements.txt,
# Print the installed Python version, pip version
# and all installed packages with their versions for debugging
RUN echo "Python version:" \
 && python --version \
 && echo "Pip version:" \
 && pip --version \
 && echo "Installing dependencies from requirements.txt:" \
 && pip install -r requirements.txt \
 && echo "All installed Python packages:" \
 && pip freeze

# Next, copy the remaining files and directories with the source code.
# Since we do this after installing the dependencies, quick build will be really fast
# for most source file changes.
COPY . ./

# Specify how to launch the source code of your actor.
# By default, the main.py file is run
CMD python3 main.py


Dockerfile

An Awesome scrapper for data in Wikipedia

Search Term or Url

Language to Search

Return Only the description of the article

Description only

Output Type array or key value

Combine paragraphs into one when output_type is key_value

Combine paragraphs

Wikipedia Scrapper

INPUT_SCHEMA.json

apify.json

__version__ = '0.1.0'
__author__ = 'github.com/kilerhg'

import os
import json
import requests
import re
import logging

from bs4 import BeautifulSoup
from parsel import Selector
import html2text

TEMPLATE_URL = 'https://{lang}.wikipedia.org'
TEMPLATE_URL_WIKI = '/wiki/{title}'

BASE_URL_SEARCH = '/w/index.php?search={search}'

REGEX_VERIFY_URL_WIKI = r'http(s)?\:\/\/([a-z\-]*)\.wikipedia.org\/wiki\/(.*)'

XPATH_CONTENT_PAGE = '//div[@class="mw-parser-output"]/p|//div[@class="mw-parser-output"]/h2|//div[@class="mw-parser-output"]/ul|//div[@class="mw-parser-output"]/ol'
XPATH_CONTENT_H = '//span[@class="mw-headline"]'
XPATH_SEARCH_PAGE = '//div[@class="mw-search-result-heading"]/a/@href'

SEARCH_CONTENT = 'mw-search-result-heading'
EMPTY_PAGE = 'noarticletext mw-content-ltr'

ALL_OUTPUT_TYPE = ['array', 'key_value']

HTTP_CODE = {
    200: 'OK',
    400: 'Bad Request',
}

ALL_LANGS_WIKI = [
    'af',
    'pl',
    'sk',
    'ar',
    'ast',
    'az',
    'bg',
    'nan',
    'bn',
    'be',
    'ca',
    'cs',
    'cy',
    'da',
    'de',
    'et',
    'el',
    'en',
    'es',
    'eo',
    'eu',
    'fa',
    'fr',
    'gl',
    'ko',
    'hy',
    'hi',
    'hr',
    'id',
    'it',
    'he',
    'ka',
    'la',
    'lv',
    'lt',
    'hu',
    'mk',
    'arz',
    'ms',
    'min',
    'nl',
    'ja',
    'no',
    'nn',
    'ce',
    'uz',
    'pt',
    'kk',
    'ro',
    'ru',
    'simple',
    'ceb',
    'sl',
    'sr',
    'sh',
    'fi',
    'sv',
    'ta',
    'tt',
    'th',
    'tg',
    'azb',
    'tr',
    'uk',
    'ur',
    'vi',
    'vo',
    'war',
    'zh-yue',
    'zh',
    'my'
]


def request_website(url):
    respose = requests.get(url).content
    website = BeautifulSoup(respose,'html.parser')
    return website

def get_text_html(html_content):
    h = html2text.HTML2Text()
    h.ignore_links = True
    h.ignore_emphasis = True
    h.ignore_images = True
    h.ignore_tables = True

    text_clear = h.handle(str(html_content))

    text_clear = text_clear.replace('\n', ' ')
    text_clear = re.sub(r'( )+', ' ', text_clear)
    text_clear = str(text_clear).strip()
    
    return text_clear

def verify_page_empty(response):
    is_page_empty = EMPTY_PAGE in str(response)
    return is_page_empty

def verify_lang_valid(lang : str) -> str:
    is_lang_valid = lang in ALL_LANGS_WIKI
    return is_lang_valid

def verify_url_wiki(url : str) -> bool:
    is_url_valid = not not re.match(REGEX_VERIFY_URL_WIKI, url) # not empty
    return is_url_valid

def search_url(lang, search) -> str:
    
    search = str(search).strip().replace(' ', '+')
    template_url_lang = TEMPLATE_URL.format(lang=lang)
    template_url_lang_search = template_url_lang + BASE_URL_SEARCH.format(search=search)

    response = request_website(template_url_lang_search)

    if SEARCH_CONTENT in str(response):
        obj_sel = Selector(str(response))
        content = obj_sel.xpath(XPATH_SEARCH_PAGE)[0].get()
        redirect_url = f'{template_url_lang}{content}'
    else:
        redirect_url = re.findall(r'<link href="(.*)?" rel="canonical"\/>', str(response))[0]
    
    return redirect_url

def content_process_page_array(response, first_section_name='Description', only_first_section=False):
    obj_sel = Selector(str(response))
    content = obj_sel.xpath(XPATH_CONTENT_PAGE)
    list_all_values = []
    section_name = first_section_name
    for tag in content:
        html_content = tag.get()
        name_tag = re.findall(r'^<([a-z\d]*)', html_content)[0]
        if name_tag[0] == 'h':
            if only_first_section:
                break
            obj_sel1 = Selector(str(html_content))
            content_h_tag = obj_sel1.xpath(XPATH_CONTENT_H)
            html_content = content_h_tag[0].get()
            section_name = get_text_html(html_content)
            
        elif name_tag[0] == 'p' or name_tag in ('ul', 'ol'):
            clear_text = re.sub(r'<sup(.*?)<\/sup>', '', html_content)
            clear_text = re.sub(r'<abbr(.*?)<\/abbr>', '', clear_text)
            clear_text = get_text_html(clear_text)
            clear_text = str(clear_text).strip()
            if clear_text:
                dict_register = {
                    'section_name': section_name,
                    'text': clear_text
                }
                list_all_values.append(dict_register)
        else:
            print(f'Invalid TAG: {name_tag}')
    return list_all_values

def content_process_page_key_value(response, first_section_name='Description', only_first_section=False, paragraph_together=False):
    obj_sel = Selector(str(response))
    content = obj_sel.xpath(XPATH_CONTENT_PAGE)
    dict_all_values = {}
    section_name = first_section_name
    for tag in content:
        html_content = tag.get()
        name_tag = re.findall(r'^<([a-z\d]*)', html_content)[0]
        if name_tag[0] == 'h':
            if only_first_section:
                break
            obj_sel1 = Selector(str(html_content))
            content_h_tag = obj_sel1.xpath(XPATH_CONTENT_H)
            html_content = content_h_tag[0].get()
            section_name = get_text_html(html_content)
            
        elif name_tag[0] == 'p' or name_tag in ('ul', 'ol'):
            clear_text = re.sub(r'<sup(.*?)<\/sup>', '', html_content)
            clear_text = re.sub(r'<abbr(.*?)<\/abbr>', '', clear_text)
            clear_text = get_text_html(clear_text)
            clear_text = str(clear_text).strip()
            if clear_text:
                if section_name not in dict_all_values:
                    
                    if paragraph_together:
                        dict_all_values[section_name] = ''
                        dict_all_values[section_name] += clear_text
                    else:
                        dict_all_values[section_name] = []
                        dict_all_values[section_name].append(clear_text)
                else:
                    if paragraph_together:
                        dict_all_values[section_name] += clear_text
                    else:
                        dict_all_values[section_name].append(clear_text)
        else:
            print(f'Invalid TAG: {name_tag}')
    return dict_all_values


def verify_fields_input_required(actor_input, list_errors):
    is_lang_valid = is_value_valid = False
    if 'lang' in actor_input and 'input_value' in actor_input:
        lang_value = str(actor_input['lang']).strip().lower()
        is_lang_valid = verify_lang_valid(lang_value)
        # if not is_lang_valid:
        #     error_dict['lang'] = f'Invalid language: {lang_value}'
        list_errors.append(f'Invalid language: {lang_value}')
        input_value = actor_input['input_value']
        if type(input_value) == list or type(input_value) == str:
            is_value_valid = len(input_value) > 0
        else:
            # error_dict['input_value'] = f'type {type(input_value)} not valid should be list or str'
            list_errors.append(f'type {type(input_value)} not valid should be list or str')
            is_value_valid = False
    elif not 'lang' in actor_input:
        list_errors.append(f'key \'lang\' not found')
    elif not 'input_value' in actor_input:
        list_errors.append(f'key \'input_value\' not found')
            
    all_required = all([is_lang_valid, is_value_valid])
    return all_required

def process_input_value(actor_input):
    if type(actor_input['input_value']) == str:
        return [actor_input['input_value']]
    else:
        return actor_input['input_value']

def verify_output_type(actor_input):
    if 'output_type' in actor_input:
        output_type = str(actor_input['output_type']).strip().lower()
        if output_type == 'array':
            return 'array'
        elif output_type == 'key_value':
            return 'key_value'
        else:
            return 'key_value' # default
    else:
        return 'key_value' # default

def verify_paragraph_together(actor_input):
    if 'paragraph_together' in actor_input:
        paragraph_together = actor_input['paragraph_together']
        if str(paragraph_together).strip().lower() == 'true':
            return True
        else:
            return False # default
    else:
        return False # default

def verify_only_description(actor_input):
    if 'only_description' in actor_input:
        only_description = actor_input['only_description']
        if str(only_description).strip().lower() == 'true':
            return True
        else:
            return False # default
    else:
        return False # default
        
def verify_and_process_fields_input(actor_input, list_errors):
    valid_actor_input = {}
    is_valid_fields = False
    if verify_fields_input_required(actor_input, list_errors):
        valid_actor_input['lang'] = str(actor_input['lang']).strip().lower()
        valid_actor_input['input_value'] = process_input_value(actor_input)
        valid_actor_input['output_type'] = verify_output_type(actor_input)
        valid_actor_input['paragraph_together'] = verify_paragraph_together(actor_input)
        valid_actor_input['only_description'] = verify_only_description(actor_input)
        is_valid_fields = True
    return is_valid_fields, valid_actor_input

def process_output(response, output_type, paragraph_together=False, first_section_name='Description', only_first_section=False):
    if output_type == 'array':
        result = content_process_page_array(response, first_section_name=first_section_name, only_first_section=only_first_section)
    elif output_type == 'key_value':
        result = content_process_page_key_value(response, first_section_name=first_section_name, only_first_section=only_first_section, paragraph_together=paragraph_together)
    else:
        result = []
        logging.warning(f'Invalid output_type: {output_type}')
    return result

def process_actor_main(actor_input):
    list_result_output = []
    list_errors = []
    list_warnings = []
    status_code = 200
    is_valid_fields, valid_actor_input = verify_and_process_fields_input(actor_input, list_errors)
    if not is_valid_fields:
        logging.warning(f'Invalid fields input')
        status_code = 400
        return list_result_output, list_errors, list_warnings, status_code
    for item in valid_actor_input['input_value']:
        if verify_url_wiki(item):
            response = request_website(item)
            if not verify_page_empty(response):
                result_url = process_output(response, output_type=valid_actor_input['output_type'], paragraph_together=valid_actor_input['paragraph_together'], only_first_section=valid_actor_input['only_description'])
                dict_item_url = {
                    'input_value': item,
                    'output_value': result_url
                }
                list_result_output.append(dict_item_url)
            else:
                list_warnings.append(f'Page invalid: {item}')
                logging.warning(f'Page invalid: {item}')
        else:
            redirect_url = search_url(valid_actor_input['lang'], item)
            response = request_website(redirect_url)
            result = process_output(response, output_type=valid_actor_input['output_type'], paragraph_together=valid_actor_input['paragraph_together'], only_first_section=valid_actor_input['only_description'])
            dict_item = {
                'input_value': item,
                'output_value': result
            }
            list_result_output.append(dict_item)
    dict_json = {
        'version': __version__,
        'input': actor_input,
        'output': list_result_output,
        #'warnings': list_warnings,
        #'errors': list_errors,
        'status_code': status_code,
        'status': HTTP_CODE[status_code],
    }
    # return list_result_output, list_errors, list_warnings, status_code
    return dict_json
    dict_json = process_actor_main(actor_input)

functions.py

import json
import os

import functions

from apify_client import ApifyClient


# Run the main function of the script, if the script is executed directly
if __name__ == '__main__':
    # Initialize the main ApifyClient instance
    client = ApifyClient(os.environ['APIFY_TOKEN'], api_url=os.environ['APIFY_API_BASE_URL'])

    # Get the resource subclient for working with the default key-value store of the actor
    default_kv_store_client = client.key_value_store(os.environ['APIFY_DEFAULT_KEY_VALUE_STORE_ID'])

    # Get the value of the actor input and print it
    actor_input = default_kv_store_client.get_record(os.environ['APIFY_INPUT_KEY'])['value']
    print('Actor input:')
    print(json.dumps(actor_input, indent=2))

    # Set the 'OUTPUT' key-value store record to the same value as the input
    default_kv_store_client.set_record('OUTPUT', actor_input)

    # Get the resource subclient for working with the default dataset of the actor
    default_dataset_client = client.dataset(os.environ['APIFY_DEFAULT_DATASET_ID'])

    dict_json = functions.process_actor_main(actor_input)

    # Push some dummy items to the default dataset
    default_dataset_client.push_items(dict_json)


main.py

# Add your dependencies here.
# See https://pip.pypa.io/en/latest/cli/pip_install/#requirements-file-format
# for how to format them

apify-client==0.5.0
beautifulsoup4==4.11.1
bs4==0.0.1
certifi==2021.10.8
charset-normalizer==2.0.12
cssselect==1.1.0
html2text==2020.1.16
idna==3.3
lxml==4.8.0
parsel==1.6.0
requests==2.26.0
six==1.16.0
soupsieve==2.3.2.post1
urllib3==1.26.9
w3lib==1.22.0

requirements.txt

Wikipedia Scraper, designed and developed to make your life easier on searching for answers.

Where next?