Простой Scraper_Downloader на основе Youtube-dl

Любая помощь с моим кодом. Я думаю, что это своего рода жесткий код, в котором я хочу извлекать ссылки со страницы и загружать их.

Он будет очищать некоторые ссылки с веб-сайтов (но Scraper функция, которую я написал, является жестким кодом) и загрузите видео, если они существуют, и я уверен, что это совсем не хорошо, и это скорее код запаха.

Код

import requests
import subprocess
import getpass
import time
from bs4 import BeautifulSoup as bs
# import argparse # Add Command Line Arguments to a Python Script (switch_flag)


def Login(url, login_route, username, password):

    headers = {
        'User-Agent': '', 
        'origin': url,
        'referer': url + login_route,
    }

    request_session = requests.session()

    csrf_token = request_session.get(url).cookies['csrftoken']

    login_payload = {
        'hidden_username': username,
        'password': password,
        'csrfmiddlewaretoken': csrf_token
    }

    login_request = request_session.post(
        url + login_route, headers=headers, data=login_payload)


    if login_request.status_code == 200:
        msg = f'nYou have logged in successfully {login_request.status_code}'
    else:
        msg = f'nError {login_request.status_code}'

    print(msg)


def get_user_input():

    url = input('URL: ')
    username = getpass.getpass('USERNAME: ')
    password = getpass.getpass('PASSWORD: ')
    login_route = input('LOGIN_ROUTE: ')

    return Login(url, login_route, username, password)


def Scraper(page_url):

    headers = {
        ''
    }

    page = requests.get(
        page_url,
        headers=headers,
    )
    soup = bs(page.text, "html.parser")

    URL_List = []
    link_count = 0
    for a_tag in soup.select('a[href^="/course/"]'):

        links = "https://maktabkhooneh.org" + a_tag["href"]

        URL_List.append(links)
        link_count += 1

    return URL_List


def Donwloader(url_list):

    URL_List = Scraper(url_list)
    download_count = 0
    
    try:
        for links in URL_List:
            command = f'youtube-dl  {links}'
            result = subprocess.call(command, shell=True)
            if result == 0:
                download_count += 1

    except KeyboardInterrupt:
        
        print('Paused ;)')
            

    return f'n{download_count} file(s) have been downloaded'


if __name__ == '__main__':

    page_url = input('Please enter the page URL: ')

    Login_permission = input('Login required Website [Y], [N]? ')
    if Login_permission == 'y' or Login_permission == 'Y':
        get_user_input()

    list_len = len(Scraper(page_url))
    Download_Permission = input(
        f'n{list_len} link(s) have been extracted. Do you want to DOWNLOAD them [Y], [N]? ')

    Scraper(page_url)

    if Download_Permission == 'y' or Download_Permission == 'Y':
        Donwloader(page_url)
    else:
        print('nProcess has been canceled ;)')

Несколько предложений.

Хорошо, что вы пользуетесь сессия но вы не используете его должным образом. Идея состоит в том, чтобы повторно использовать сеанс в вашем коде, чтобы повысить эффективность сканирования сайтов. Он используется в подпрограмме входа в систему, но в подпрограмме Scraper вы используете простой request.get (). Таким образом, вы на самом деле не пользуетесь request.session.

На самом деле вам следует установить пользовательский агент в ваших заголовках. Отсутствие одного или использование значения по умолчанию (что-то вроде: python-requests / 2.25.1) делает очевидным, что вы бот. Таким образом, у вашего скрипта могут возникнуть проблемы со сканированием некоторых сайтов. Я бы посоветовал вам подделать пользовательский агент, чтобы он соответствовал обычному браузеру. Это ничего не стоит.

Вы используете F-строки в своем коде, но иногда конкатенацию используют знак +. Для большей последовательности я предлагаю вам использовать только F-струны.

Я бы попытался реорганизовать ваш набор функций как класс. Ниже представлена ​​небольшая попытка, она не является полной и предназначена для демонстрационных целей. Я установил несколько заголовков, в частности, поддельный пользовательский агент (Firefox). Достаточно сделать это один раз на уровне сеанса, поскольку каждый запрос (get, post), использующий сеанс, наследует эти параметры. Так что вам не нужно повторяться. Но вы все равно можете передавать дополнительные заголовки в каждом конкретном случае, это то, что делается в маршруте входа в систему, где мы, например, хотим передать реферера.

Я попытался добавить немного абстракции, создав функцию для получения одной страницы. Затем я бы разработал другую функцию, которая фактически анализировала бы ее с помощью BS4 и т. Д.

Я также добавил некоторую базовую обработку ошибок на уровне приложения, а также некоторое ведение журнала на консоли, при желании вы можете легко добавить ведение журнала в файл (полезно для устранения неполадок).

Возможно, вы захотите использовать файл конфигурации для хранения URL-адресов и других параметров, чтобы избежать хлопот с вводом значений. Использование getpass подходит, если вы не хотите хранить учетные данные или другую конфиденциальную информацию.

PS: обсуждение использования shell=True с подпроцессом: Фактическое значение ‘shell = True’ в подпроцессе. Короче, я не думаю, что вам это нужно, это может быть даже вредно с точки зрения безопасности.

Поскольку вы удаляете сторонний сайт, HTML-код следует рассматривать как небезопасный, незащищенный и ненадежный пользовательский ввод.

И если вызов не удался (код возврата! = 0), не молчите, а бревно это событие. Все, что вам нужно сделать, это воспользоваться уже доступным регистратором и сделать что-то вроде:

    if result == 0:
        download_count += 1
    else:
        self.logger.warning(f"Failed to execute command: {command} - Return code: {result}")

Тогда, по крайней мере, у вас есть следы того, что идет не так во время процесса.

Возможно, в вашем сценарии не хватает этого: прослеживаемости. Хорошо распечатать несколько сообщений здесь и там, чтобы показать, что делает скрипт и на каком этапе он находится в данный момент. Особенно когда ожидается, что выполнение займет некоторое время, вы хотите иметь возможность отслеживать прогресс. Модуль регистрации создан для вас, используйте его и злоупотребляйте им.

Разное: в Scraper вы увеличиваете link_count но ты мог бы просто использовать len(URL_List) вместо. Нет необходимости в другой переменной.

PS: имеет смысл переместить класс в отдельный файл, чтобы его можно было повторно использовать и было лучшее разделение. Просто импортируйте его как другой модуль. Тогда основной скрипт будет меньше по размеру и более управляемым.

Скелет класса

import sys
import logging
import requests
import getpass

from bs4 import BeautifulSoup as bs


class Downloader:

    def __init__(self, logger=None):
        # set up logging
        self.logger = logger or logging.getLogger(__name__)
        self.logger.debug("Initialize class Downloader")

        # init session
        self.session = requests.session()

        # add some standard headers and spoof user agent
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:89.0) Gecko/20100101 Firefox/89.0",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.5"
        })
        self.logger.debug(f"Session headers: {self.session.headers}")

    def fetch_page(self, url: str, headers: None):
        """Fetch a single page and return request results - some extra headers may be provided
        """
        self.logger.debug(f"Fetching page: {url}")
        return self.session.get(url=url, headers=headers)

    def login(self, url: str, csrf_token: str, username: str, password: str, headers: None):
        """Log in to website - some extra headers may be provided
        """
        self.logger.debug(f"Logging in with user: {username}")
        login_payload = {
            'hidden_username': username,
            'password': password,
            'csrfmiddlewaretoken': csrf_token
        }
        return self.session.post(url=url, data=login_payload, headers=headers)


def main():

    try:
        # set up simple logging to console
        logging.getLogger().setLevel(logging.NOTSET)

        logger = logging.getLogger(__name__)

        # Add stdout handler, with level DEBUG
        console = logging.StreamHandler(sys.stdout)
        console.setLevel(logging.DEBUG)
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s',
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        console.setFormatter(formatter)
        logger.addHandler(console)

        # get URL and credentials
        url = input('URL: ')
        login_route = input('LOGIN_ROUTE: ')
        username = getpass.getpass('USERNAME: ')
        password = getpass.getpass('PASSWORD: ')

        # start crawling
        downloader = Downloader(logger=logger)
        page = downloader.fetch_page(url=url)
        if page.status_code == 200:
            # fetch the CSRF token - return None if not found
            csrf_token = page.cookies.get('csrftoken')
            if csrf_token is None:
                raise Exception("Failed to obtain the CSRF token")
            else:
                logger.debug(f"Got CSRF token: {csrf_token}")

            # proceed to login - add some extra headers on top of the session headers
            login_headers = {
                'origin': url,
                'referer': url + login_route,
            }
            login_request = downloader.login(
                url=login_route, csrf_token=csrf_token, username=username, password=password,
                headers=login_headers
            )
            if login_request.status_code == 200:
                logger.debug(f"You have logged in successfully - status code: {page.status_code}")
            else:
                raise Exception(f"Failed to login - status code: {login_request.status_code}")

            # TODO: loop on links and retrieve videos
        else:
            raise Exception(f"Failed to fetch URL: {url} - status code: {page.status_code}")


    # Ctrl-C
    except KeyboardInterrupt:
        logger.warning("Shutdown requested (KeyboardInterrupt)...")
        sys.exit(0)

    except Exception:
        logger.error("Exception occured", exc_info=True)
        sys.exit(1)
    finally:
        logger.debug("Application shutting down")


if __name__ == '__main__':
    main()