admin/ipars
1
1
Fork 0
Code Issues 13 Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлены аннотации типов для методов в классах. Исправлены некоторые ошибки в методах и улучшена читаемость кода.

Browse Source
This commit is contained in:
Ilia Miheev
2026-04-30 00:17:24 +03:00
parent 080b28f50f
commit 4db335f399
6 changed files with 34 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ipars/CsvManagerCode.py
+3 -2
View File
@@ -1,11 +1,12 @@
import csv
from pprint import pprint
from typing import Any
from cerberus import Validator
class CsvManager:
'''Класс для работы с csv файлами во время парсинга'''
def __init__(self, newline: str = '', encoding: str = 'utf8', delimiter: str = ';'):
def __init__(self, newline: str = '', encoding: str = 'utf8', delimiter: str = ';') -> None:
'''Конструктор
newline: новая строка в csv файле
@@ -24,7 +25,7 @@ class CsvManager:
self.encoding = encoding
self.delimiter = delimiter
def pprint(self, data: any) -> None:
def pprint(self, data: Any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
ipars/JsonManagerCode.py
+5 -4
View File
@@ -1,11 +1,12 @@
import json
from pprint import pprint
from typing import Any
from cerberus import Validator
class JsonManager:
'''Класс для работы с json файлами во время парсинга'''
def __init__(self, encoding: str = 'utf8'):
def __init__(self, encoding: str = 'utf8') -> None:
'''Конструктор
encoding: кодировка открываемого файла'''
@@ -18,13 +19,13 @@ class JsonManager:
self.encoding = encoding
def pprint(self, data: any) -> None:
def pprint(self, data: Any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
pprint(data)
def load(self, pathToJsonFile: str) -> json:
def load(self, pathToJsonFile: str) -> Any:
'''Получаем данные из json файла
pathToJsonFile: путь до json файла'''
@@ -40,7 +41,7 @@ class JsonManager:
src = json.load(jsonFile)
return src
def dump(self, pathToJsonFile: str, data: any) -> None:
def dump(self, pathToJsonFile: str, data: Any) -> None:
'''Записываем данные в json файл
pathToJsonFile: путь до json файла
ipars/ParsManagerCode.py
+14 -12
View File
@@ -5,6 +5,7 @@ from selenium import webdriver
from bs4 import BeautifulSoup
from pprint import pprint
from time import sleep
from typing import Any, Optional
import requests
from os import mkdir, listdir
from os.path import exists
@@ -14,14 +15,14 @@ class Pars:
'''Модуль для работы с запросами и bs4'''
def __validation(self, schema, expected):
def __validation(self, schema: dict, expected: dict) -> None:
'''Валидация введённых данных для методов'''
v = Validator(schema)
if not v.validate(expected):
raise ValueError(v.errors)
def exists(self, path: str) -> dict:
def exists(self, path: str) -> bool:
'''Возвращает True если указанный файл или папка сущуствует, иначе — False'''
schema = {'path': {'type': 'string'}}
expected = {'path': path}
@@ -30,7 +31,7 @@ class Pars:
return exists(path)
def listdir(self, path: str) -> dict:
def listdir(self, path: str) -> list:
'''Возвращает список файлов в указанной директории'''
schema = {'path': {'type': 'string'}}
expected = {'path': path}
@@ -39,7 +40,7 @@ class Pars:
return listdir(path)
def mkdir(self, nameDir: str):
def mkdir(self, nameDir: str) -> None:
'''Создаёт папку если её ещё нет
nameDir: название папки которая будет создана'''
@@ -51,7 +52,7 @@ class Pars:
mkdir(nameDir)
def returnBs4Object(self, pathToFile: str, encoding: str = 'utf8', parser: str = 'lxml'):
def returnBs4Object(self, pathToFile: str, encoding: str = 'utf8', parser: str = 'lxml') -> BeautifulSoup:
'''Возвращаем объект beautifulsoup
pathToFile: путь до html файла
@@ -75,7 +76,7 @@ class Pars:
return soup
def getTexts(self, arr: list, needFix: bool = False) -> list:
def getTexts(self, arr: list, needFix: bool = False) -> Optional[list]:
'''Возвращаем текст из элементов bs4
arr: список объектов bs4 из которых будет извлекаться текст
@@ -109,7 +110,7 @@ class Pars:
return result
def getAttributes(self, arr: list, att: str) -> list:
def getAttributes(self, arr: list, att: str) -> Optional[list]:
'''Возвращаем список значений атрибутов
arr: список объектов bs4 из которых будет извлекаться атрибут
@@ -135,14 +136,14 @@ class Pars:
return result
def pprint(self, data: any) -> None:
def pprint(self, data: Any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
pprint(data)
def getStaticPage(self, pathToSaveFile: str, url: str, writeMethod: str = 'w', headers: dict = None) -> int:
def getStaticPage(self, pathToSaveFile: str, url: str, writeMethod: str = 'w', headers: Optional[dict] = None) -> int:
'''Сохраняем статическую страницу и возвращаем статус ответа от сервера
pathToSaveFile: путь, куда сохранится полученный файл
@@ -186,6 +187,7 @@ class Pars:
try:
# Отправляем запрос
req = requests.get(url, headers=headers)
req.raise_for_status()
# Записываем данные
if writeMethod == 'w':
@@ -203,12 +205,12 @@ class Pars:
return req.status_code # Возвращаем статус ответа от сервера
except requests.exceptions.HTTPError as httpErr:
raise RuntimeError(f"HTTP ошибка: {httpErr}") from http_err
raise RuntimeError(f"HTTP ошибка: {httpErr}") from httpErr
except Exception as e:
raise RuntimeError(e) from e
def __scrollAndSave(self, driver, timeSleep, pathToSaveFile):
def __scrollAndSave(self, driver: webdriver.Chrome, timeSleep: int, pathToSaveFile: str) -> None:
# Прокручиваем страницу до самого низа
lastHeight = driver.execute_script("return document.body.scrollHeight")
while True:
@@ -256,7 +258,7 @@ class Pars:
self.__scrollAndSave(driver, timeSleep, pathToSaveFile)
def gpsa(self, pathToSaveFile: str, url: str, timeSleep=2):
def gpsa(self, pathToSaveFile: str, url: str, timeSleep: int = 2) -> None:
'''Получаем страницу в полуавтоматическом режиме
gpsa - get page semi-automatically
ipars/ProgressBarCode.py
+3 -3
View File
@@ -8,13 +8,13 @@ class ProgressBarManager:
color: цвет прогресс-бара
fill: заполнитель для сделанной части
width: размер прогресс-бара в символах'''
def __init__(self, max, message='Процесс работы', color='green', fill='#', width=32):
def __init__(self, max: int, message: str = 'Процесс работы', color: str = 'green', fill: str = '#', width: int = 32) -> None:
self.bar = Bar(max=max, message=message, color=color, fill=fill, suffix='%(index)d/%(max)d (%(percent)d%%)', width=width)
def next(self):
def next(self) -> None:
'''Запускаем следущую итерацию прогресс-бара'''
self.bar.next()
def finish(self):
def finish(self) -> None:
'''Завершаем работу класса'''
self.bar.finish()
ipars/TimerManagerCode.py
+4 -3
View File
@@ -1,8 +1,9 @@
import time
from typing import Optional
class TimerManager:
'''Класс для отслеживания времени работы'''
def __init__(self):
def __init__(self) -> None:
'''Инициализация'''
self.startTime = 0
self.endTime = 0
@@ -17,7 +18,7 @@ class TimerManager:
self.endTime = time.time()
self.workTime = self.endTime - self.startTime
def getWorkTime(self, format :str = 'seconds', ndigits :int = None) -> int:
def getWorkTime(self, format: str = 'seconds', ndigits: Optional[int] = None) -> float:
'''Возвращает время в нужном формате'''
time_units = {
'seconds': 1,
@@ -30,7 +31,7 @@ class TimerManager:
raise ValueError('Для замера времени обязательно надо использовать метод "start" ДО выполнения кода')
if self.endTime == 0:
raise ValueError('Для замера времени обязательно надо использовать метод "stop" ПОСЛЕ выполнения кода')
raise ValueError('Для замера времени обязательно надо использовать метод "end" ПОСЛЕ выполнения кода')
if format not in time_units:
raise ValueError(f"Неподдерживаемый формат '{format}'. Должен быть один из {list(time_units.keys())}")
ipars/ZipManagerCode.py
+5 -5
View File
@@ -10,10 +10,10 @@ class ZipManager:
- 'hard': Увеличенное сжатие
- 'maximum': Максимальное сжатие
"""
def __init__(self, compression :str = 'normal'):
self.compression = self.setСompression(compression)
def __init__(self, compression: str = 'normal') -> None:
self.compression = self.setCompression(compression)
def setСompression(self, compressionStr):
def setCompression(self, compressionStr: str) -> int:
if compressionStr == 'none':
return zipfile.ZIP_STORED
elif compressionStr == 'normal':
@@ -25,7 +25,7 @@ class ZipManager:
else:
raise ValueError("Некорректное значение compression. Допустимы значения none, normal, hard, maximun")
def zipFile(self, filePath:str, zipFilePath:str):
def zipFile(self, filePath: str, zipFilePath: str) -> None:
"""Архивируем один файл
filePath (str): Путь к файлу, который нужно заархивировать
@@ -35,7 +35,7 @@ class ZipManager:
with zipfile.ZipFile(zipFilePath, 'w', compression=self.compression) as zipf:
zipf.write(filePath, os.path.basename(filePath))
def zipFolder(self, folderPath:str, zipFilePath:str):
def zipFolder(self, folderPath: str, zipFilePath: str) -> None:
"""Архивируем папку
folderPath (str): Путь к папке, которую нужно заархивировать