Добавил описание функций более полным, изменил названия переменных. аааааа

This commit is contained in:
Ilia Miheev
2025-05-20 19:47:39 +03:00
parent d39c76d56d
commit b411a1616a
+75 -37
View File
@@ -7,60 +7,86 @@ import requests
import json import json
import csv import csv
class CsvManager: class CsvManager:
'''Класс для работы с csv файлами во время парсинга''' '''Класс для работы с csv файлами во время парсинга'''
def __init__(self, newline: str = '', encoding: str = 'utf8', delimiter: str = ';'): def __init__(self, newline: str = '', encoding: str = 'utf8', delimiter: str = ';'):
'''Конструктор''' '''Конструктор
newline: новая строка в csv файле
encoding: кодировка открываемого файла
delimiter: разделитель данных в csv файле'''
self.newline = newline self.newline = newline
self.encoding = encoding self.encoding = encoding
self.delimiter = delimiter self.delimiter = delimiter
def pprint(self, data:any): def pprint(self, data: any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
pprint(data) pprint(data)
def writerow(self, file_path:str, mode:str, row:list): def writerow(self, filePath: str, mode: str, row: list) -> None:
'''Записываем строку в csv файл''' '''Записываем строку в csv файл
with open(file_path, mode=mode, newline=self.newline, encoding=self.encoding) as file:
filePath: путь до csv файла
mode (w/a): метод открытия файла
row: список записываемых данных'''
with open(filePath, mode=mode, newline=self.newline, encoding=self.encoding) as file:
writer = csv.writer(file, delimiter=self.delimiter) writer = csv.writer(file, delimiter=self.delimiter)
writer.writerow(row) writer.writerow(row)
def writerows(self, file_path:str, mode:str, row:list): def writerows(self, filePath: str, mode: str, row: list) -> None:
'''Записываем строки в csv файл''' '''Записываем строки в csv файл
with open(file_path, mode=mode, newline=self.newline, encoding=self.encoding) as file:
filePath: путь до csv файла
mode (w/a): метод открытия файла
row: список записываемых данных'''
with open(filePath, mode=mode, newline=self.newline, encoding=self.encoding) as file:
writer = csv.writer(file, delimiter=self.delimiter) writer = csv.writer(file, delimiter=self.delimiter)
writer.writerows(row) writer.writerows(row)
def getRows(self, file_path:str): def getRows(self, filePath: str) -> list:
'''Возвращает строки файла''' '''Возвращает строки файла
filePath: путь до csv файла'''
userRows = [] userRows = []
with open(filePath, mode='r', newline=self.newline, encoding=self.encoding) as csvfile:
with open(file_path, mode='r', newline=self.newline, encoding=self.encoding) as csvfile: csvReader = csv.reader(csvfile, delimiter=self.delimiter)
csv_reader = csv.reader(csvfile, delimiter=self.delimiter) for row in csvReader:
for row in csv_reader:
userRows.append(row) userRows.append(row)
return userRows return userRows
class JsonManager: class JsonManager:
'''Модуль для работы с json файлами''' '''Класс для работы с json файлами во время парсинга'''
def __init__(self, encoding: str = 'utf8'): def __init__(self, encoding: str = 'utf8'):
'''Конструктор
encoding: кодировка открываемого файла'''
self.encoding = encoding self.encoding = encoding
def pprint(self, data:any): def pprint(self, data: any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
pprint(data) pprint(data)
def load(self, pathToJsonFile:str): def load(self, pathToJsonFile: str) -> json:
'''Получаем данные из json файла''' '''Получаем данные из json файла
pathToJsonFile: путь до json файла'''
with open(pathToJsonFile, encoding=self.encoding) as jsonFile: with open(pathToJsonFile, encoding=self.encoding) as jsonFile:
src = json.load(jsonFile) src = json.load(jsonFile)
return src return src
def dump(self, pathToJsonFile: str, data: any) -> None:
'''Записываем данные в json файл
def dump(self, pathToJsonFile:str, data:any): pathToJsonFile: путь до json файла
'''Записываем данные в json файл''' data: данные которые надо записать'''
with open(pathToJsonFile, 'w', encoding=self.encoding) as jsonFile: with open(pathToJsonFile, 'w', encoding=self.encoding) as jsonFile:
json.dump(data, jsonFile, indent=4, ensure_ascii=0) json.dump(data, jsonFile, indent=4, ensure_ascii=0)
@@ -68,16 +94,24 @@ class JsonManager:
class Pars: class Pars:
'''Модуль для работы с запросами и HTML файлами во время парсинга''' '''Модуль для работы с запросами и HTML файлами во время парсинга'''
def returnBs4Object(self, pathToFile, myEncoding:str='utf8', parser:str='lxml'): def returnBs4Object(self, pathToFile: str, encoding: str = 'utf8', parser: str = 'lxml') -> bs4Object:
'''Возвращаем объект beautifulsoup''' '''Возвращаем объект beautifulsoup
with open(pathToFile, encoding=myEncoding) as file:
pathToFile: путь до html файла
encoding: кодировка открытия html файла
parser: парсер, который будет использоваться для работы'''
with open(pathToFile, encoding=encoding) as file:
src = file.read() src = file.read()
soup = BeautifulSoup(src, parser) soup = BeautifulSoup(src, parser)
return soup return soup
def getStaticPage(self, pathToSaveFile: str, url: str, writeMethod: str = 'w', headers: dict = None) -> int:
'''Получаем статическую страницу и возвращаем статус ответа от сервера
def getStaticPage(self, pathToSaveFile, url, writeMethod='w', headers: dict = None) -> int: pathToSaveFile: путь, куда сохранится полученный файл
'''Получаем статическую страницу и возвращаем статус ответа от сервера''' url: ссылка на данные
writeMethod: метод записи данных в файл. "W" записывает текст, "WB" — байты
headers: заголовки запроса к серверу'''
if headers is None: if headers is None:
headers = { headers = {
@@ -103,14 +137,18 @@ class Pars:
return req.status_code # Возвращаем статус ответа от сервера return req.status_code # Возвращаем статус ответа от сервера
except requests.exceptions.HTTPError as http_err: except requests.exceptions.HTTPError as httpErr:
raise RuntimeError(f"HTTP ошибка: {http_err}") from http_err raise RuntimeError(f"HTTP ошибка: {httpErr}") from http_err
except Exception as e: except Exception as e:
raise RuntimeError(e) from e raise RuntimeError(e) from e
def getDinamicPage(self, pathToSaveFile: str, url: str, closeWindow: bool = 1, timeSleep: int = 2) -> None:
'''Получаем динамическую страницу
def getDinamicPage(self, url, pathToSaveFile, closeWindow:bool=1, timeSleep:int=2) -> None: pathToSaveFile: путь, куда сохранится полученный файл
'''Получаем динамическую страницу''' url: ссылка на данные
closeWindow (0/1): если указана единица, то браузер открывается в фоновом режиме, 0 — открывается как обычное приложение
timeSleep: время ожидания браузера перед тем как скролить страницу дальше'''
# Устанавливаем опции для Chrome WebDriver # Устанавливаем опции для Chrome WebDriver
options = Options() options = Options()
@@ -121,19 +159,19 @@ class Pars:
with webdriver.Chrome(options=options) as driver: with webdriver.Chrome(options=options) as driver:
driver.get(url) driver.get(url)
# Прокручиваем страницу до самого низа # Прокручиваем страницу до самого низа
last_height = driver.execute_script("return document.body.scrollHeight") lastHeight = driver.execute_script("return document.body.scrollHeight")
while True: while True:
# Прокручиваем до низа страницы # Прокручиваем до низа страницы
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);") driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
# Ждем загрузки страницы # Ждем загрузки страницы
sleep(timeSleep) sleep(timeSleep)
# Вычисляем новую высоту страницы # Вычисляем новую высоту страницы
new_height = driver.execute_script("return document.body.scrollHeight") newHeight = driver.execute_script("return document.body.scrollHeight")
if new_height == last_height: if newHeight == lastHeight:
break break
last_height = new_height lastHeight = newHeight
# Получаем HTML-код страницы # Получаем HTML-код страницы
html_content = driver.page_source htmlContent = driver.page_source
# Сохраняем HTML-код в файл # Сохраняем HTML-код в файл
with open(pathToSaveFile, "w", encoding="utf-8") as file: with open(pathToSaveFile, "w", encoding="utf-8") as file:
file.write(html_content) file.write(htmlContent)