45 Commits

Author SHA1 Message Date
Ilia Miheev 1ae870da3a Добавил асинхронную версию getStaticPage 2026-05-02 11:20:43 +03:00
Ilia Miheev d1b1fd8e0b Добавлены тесты для всех классов 2026-04-30 00:41:27 +03:00
Ilia Miheev 4db335f399 Добавлены аннотации типов для методов в классах. Исправлены некоторые ошибки в методах и улучшена читаемость кода. 2026-04-30 00:17:24 +03:00
Ilia Miheev 080b28f50f Промежуточный фикс 2026-01-10 21:02:13 +03:00
Ilia Miheev 7f7fb9d108 Изменено значение по умолчанию из-за которого метод getText работал с ошибкой 2026-01-06 12:16:56 +03:00
Ilia Miheev 47450451ee Версия приложения обновлена до 3.9.1 2025-11-14 18:59:48 +03:00
Ilia Miheev 1adb34331c Исправлено описание метода exists 2025-11-14 18:36:13 +03:00
Ilia Miheev 56be75e923 Добавлен метод exists в класс Pars
Теперь в классе Pars есть метод exists для проверки наличия файла или папки
2025-11-14 18:22:00 +03:00
Ilia Miheev 3fe9784414 Revert "Добавлен метод exists"
This reverts commit d4b0f02ec7.
2025-11-13 16:30:35 +03:00
Ilia Miheev 823981b33c Revert "Версия приложения обновлена до 3.9.0"
This reverts commit 3ed1db0035.
2025-11-13 16:27:52 +03:00
Ilia Miheev 3ed1db0035 Версия приложения обновлена до 3.9.0 2025-11-13 16:08:23 +03:00
Ilia Miheev d4b0f02ec7 Добавлен метод exists 2025-11-13 16:07:50 +03:00
Ilia Miheev c5bc0254eb Улучшен пример использования класса Pars 2025-11-13 15:37:27 +03:00
Ilia Miheev b8a6dfa2a1 Исправлена ошибка в валидации преводящая к остановке кода 2025-11-12 17:23:59 +03:00
Ilia Miheev 1154570557 Исправлена опечатка в слове self приводящая к ошибке 2025-11-12 17:21:46 +03:00
Ilia Miheev 7d92d3fa2a Улучшен readme 2025-11-11 22:21:24 +03:00
Ilia Miheev b72a1a00cf Версия приложения обновлена до 3.8.0 2025-11-11 21:47:15 +03:00
Ilia Miheev eb8255d54c Улучшена читаемость кода
Разделил код строками на отдельные смысловые части для лучшей читаемости и добавил комментариев
2025-11-11 21:45:49 +03:00
Ilia Miheev 1184bfe2a0 Добавлен новый метод — listdir 2025-11-11 21:28:49 +03:00
Ilia Miheev ecfc326b36 Подправлены настройки валидации
Для сокращения кода код проверки вынесен в отдельную функцию. В некоторых методах валидации не было, сейчас все функции валидируются
2025-11-11 21:26:41 +03:00
Ilia Miheev bb99e7296c Слегка изменены примеры использования методов 2025-11-11 16:00:08 +03:00
Ilia Miheev a2fe484446 Добавлены зависимости для ZipManager
Когда создал ZipManager, то не изменил файл requirements.txt из-за чего там не хватало двух зависимостей
2025-11-11 15:58:54 +03:00
Ilia Miheev b3753fd60c Закрыт баг #18 2025-11-10 20:32:06 +03:00
Ilia Miheev a132c686ae Добавлены новые мета-теги в описание
Добавлены дополнительны ключевые слова, ссылки проекта, зависимость (cerberus)
2025-10-29 22:00:29 +03:00
Ilia Miheev 838ae59d8b Добавлен файл changelog 2025-10-28 18:52:40 +03:00
Ilia Miheev d4d8525fa2 Отредактирован пример кода с CsvManager
В примере было не нужное создание переменной writer. Так же путь к файлу был вынесен в переменную
2025-10-28 17:02:55 +03:00
Ilia Miheev a998dfee32 Merge pull request #15 from mavxa/patch-1
Исправлена орфографическая ошибка
2025-10-26 21:37:28 +03:00
mavxa c2fc53a5a7 Update README.md 2025-10-26 21:01:05 +03:00
mavxa 1865e2d6fa Fix method name from getDinamicPage to getDynamicPage 2025-10-25 22:27:58 +03:00
Ilia Miheev 3b5b57838b Версия приложения обновлена до 3.7.1
Добавлены ключевые слова и ссылка на сайт-документацию
2025-10-24 20:31:19 +03:00
Ilia Miheev 7e1d5f9880 Добавлена ссылка на сайт-документацию 2025-10-24 20:29:59 +03:00
Ilia Miheev b95aef9178 Merge pull request #11 from IliaMiheev/make-zipManager
Make zip manager
2025-10-13 16:53:14 +03:00
Ilia Miheev ca73f6ede4 Версия библиотеки обновлена до 3.7.0 2025-10-13 16:49:55 +03:00
Ilia Miheev 2f11637b78 Описана работа с ZipManager 2025-10-13 16:47:58 +03:00
Ilia Miheev 036f189712 Добавлен класс для работы с zip 2025-10-13 16:47:15 +03:00
Ilia Miheev 0650202e9c zip файлы добавлены в игнор 2025-10-13 15:09:38 +03:00
Ilia Miheev 8c789b4fd0 Версия приложения обновлена до 3.6.0 2025-10-01 20:18:44 +03:00
Ilia Miheev c85c11d3db Навигация в документации подредактирована 2025-10-01 20:18:20 +03:00
Ilia Miheev 1718d7bcec Добавлен класс для замера времени выполнения скрипта 2025-10-01 20:17:04 +03:00
Ilia Miheev be77ec2f6c Версия библиотеки обновлена до 3.5.2 2025-09-30 21:10:35 +03:00
Ilia Miheev 5589e95fb2 Добавлена валидация в ParsManager 2025-09-30 21:03:11 +03:00
Ilia Miheev 0fd3d26645 Сделана валидация в ParsManager 2025-09-30 20:53:55 +03:00
Ilia Miheev 87038c279b Исправлен вывод ошибки на более корректный 2025-09-30 20:53:03 +03:00
Ilia Miheev eaa0feca68 Добавлена валидация в CsvManager, а так же новые зависимости
Библиотеку progress забыл добавить в прошлый раз, а библиотека cerberus добавлена для валидации
2025-09-30 19:32:46 +03:00
Ilia Miheev 88c43eb58f Добавлен файл зависимостей 2025-09-29 22:17:33 +03:00
19 changed files with 1146 additions and 141 deletions
+2 -1
View File
@@ -2,8 +2,9 @@ __pycache__
*.html
*.csv
*.json
*.zip
ideas.md
*copy*
test/
venv/
.idea/
+33
View File
@@ -0,0 +1,33 @@
# Changelog
## 3.9.3 (10 января, 2026 год)
- Исправил предыдущие исправления. В 3.9.2 установил значение false (как в js), а теперь установил False (как положено в python)
## 3.9.2 (6 января, 2026 год)
- Изменено значение по умолчанию из-за которого метод getText работал с ошибкой
## 3.8.1 - 3.8.3 (11 - 14 ноября, 2025 год)
- Добавлен метод exists для проверки существования файлов и папок
- Исправлены орфографические и логические ошибки в коде
- Сделаны косметические исправления в доке
- Примеры использования описанны более лучшим образом
## 3.8.0 (11 ноября, 2025 год)
- В класс Pars добавлен метод listdir для получения списка файлов в указаной директории
## 3.7.3 (10 ноября, 2025 год)
- Исправлен баг [#18](https://github.com/IliaMiheev/ipars/issues/18)
## В предыдущих версиях
- Создан класс для работы с bs4 и запросами
- Создан класс для работы с json
- Создан класс для работы с csv
- Создан класс для работы с zip архивами
- Создан класс для работы с прогресс-барами
- Создан класс для засечения времени
+213 -77
View File
@@ -1,30 +1,58 @@
# Библиотека для работы с файлами во время парсинга
## Библиотека для работы с файлами во время парсинга
Во время парсинга часто приходится скачивать html-страницы, работать с json- и csv-файлами. Эта библиотека призвана облегчить написание кода для такого рода задач. Библиотека создавалась для удобства работы во время парсинга, но ничто не мешает использовать её просто для работы с json, csv и прогресс-барами.
Во время парсинга часто приходится скачивать html-страницы, работать с json- и csv-файлами. Эта библиотека призвана облегчить написание кода для такого рода задач, а так же предоставляет ряд дополнительных возможностей.
В библиотеке есть три класса для отдельных работ: Pars для работы с запросами и bs4, JsonManager для работы с json и CsvManager для работы с csv. Так же есть один класс для улучшения пользовательского опыта — ProgressBarManager, который создаёт прогресс-бар
Более удобная версия документации на сайте ([ru](https://iliamiheev.github.io/ipars-doc/#/./home), [en](https://iliamiheev.github.io/ipars-doc/#/./pages/en/README_EN))
Установить библиотеку:
#### В библиотеке есть три класса для основных работ:
1. **Pars** для работы с запросами и bs4
2. **JsonManager** для работы с json
3. **CsvManager** для работы с csv
### Так же есть три вспомогательных класса
1. **ProgressBarManager** для создания прогресс-баров
2. **TimerManager** для засечения времени выполнения определённого кода
3. **ZipManager** для архивации файлов и папок
### Установить библиотеку:
```bash
pip install ipars
```
или
```bash
pip3 install ipars
```
## Навигация
- [Работа с Pars](#работа-с-pars)
- [Коротко о методах Pars](#коротко-о-методах-pars)
- [Пример парсера с использованием ipars](#пример-парсера-с-использованием-ipars)
- [Коротко о методах](#коротко-о-методах-pars)
- [Пример использования класса Pars](#пример-использования-класса-pars)
- [Пример использования методов getAttributes и getTexts](#пример-использования-методов-getattributes-и-gettexts)
- [Работа с JsonManager](#работа-с-jsonmanager)
- [Коротко о методах JsonManager](#коротко-о-методах-jsonmanager)
- [Пример использования JsonManager](#пример-использования-jsonmanager)
- [Коротко о методах](#коротко-о-методах-jsonmanager)
- [Пример использования](#пример-использования-jsonmanager)
- [Работа с CsvManager](#работа-с-csvmanager)
- [Коротко о методах CsvManager](#коротко-о-методах-csvmanager)
- [Пример использования CsvManager](#пример-использования-csvmanager)
- [Коротко о методах](#коротко-о-методах-csvmanager)
- [Пример использования](#пример-использования-csvmanager)
- [Работа с ProgressBarManager](#работа-с-progressbarmanager)
- [Коротко о методах ProgressBarManager](#коротко-о-методах-progressbarmanager)
- [Пример использования ProgressBarManager](#пример-использования-progressbarmanager)
- [Коротко о методах](#коротко-о-методах-progressbarmanager)
- [Пример использования](#пример-использования-progressbarmanager)
- [Работа с TimerManager](#работа-с-timermanager)
- [Коротко о методах](#коротко-о-методах-timermanager)
- [Пример использования](#пример-использования-timermanager)
- [Работа с ZipManager](#работа-с-zipmanager)
- [Коротко о методах](#коротко-о-методах-zipmanager)
- [Пример использования](#пример-использования-zipmanager)
## Работа с Pars
### Работа с Pars
Данный класс предназначен для работы с запросами и полученной html-страницей.
Класс Pars не принимает никаких данных для конструкторов.
@@ -36,59 +64,82 @@ p = Pars()
### Коротко о методах Pars:
1. Метод **getStaticPage** принимает url страницы, путь, по которому сохранится страница, метод записи и заголовки запроса. Метод записи «wb» используется для сохранения картинок, по умолчанию writeMethod установлен как «w», что используется для html-страниц. Если заголовки запросов не указаны, то будут использоваться заголовки со случайным user-agent. Метод возвращает статус ответа сайта, что должно использоваться для введения проверок
- Метод **getStaticPage** принимает url страницы, путь по которому сохранится страница, метод записи и заголовки запроса. Метод записи «wb» используется для сохранения картинок, по умолчанию writeMethod установлен как «w», что используется для html-страниц. Если заголовки запросов не указаны, то будут использоваться заголовки со случайным user-agent. Метод возвращает статус ответа сайта, его можно использовать для проверкок.
2. Метод **getDynamicPage** с помощью библиотеки Selenium получает динамически обновляемую страницу. Это помогает, когда контент на странице подгружается динамически. Принимает url страницы, путь сохранения, closeWindow и timeSleep. По умолчанию браузер Selenium открывается в фоновом режиме, и работу браузера не видно, но если closeWindow указать как False, то будет виден процесс выполнения кода. С помощью timeSleep можно увеличить время загрузки страницы если контент на ней долго подгружается
```python
for index, url in enumerate(urlList):
if p.getStaticPage(url, f'./page{index}') == 404:
print('Не удалось получить страницу: ', url)
```
3. Метод **gpsa** (get page semi-automatically) похож на метод getDynamicPage, но работает в полуавтоматическом режиме. Он открывает страницу сайта и ждёт пока не будет нажат Enter в терминале. В этот момент можно зарегистрироваться на сайте и/или перейти на нужную вкладку сайта, после чего нажать Enter и метод спарсит страницу. Подходит для лент соцсетей, где неавторизованным пользователям контент ограничен. Принисает такие же аргументы для таких же целей, что и getDynamicPage, за исключением closeWindow
- Метод **getDynamicPage** с помощью библиотеки Selenium получает динамически обновляемую страницу. Это помогает, когда контент на странице подгружается динамически. Принимает url страницы, путь сохранения, closeWindow и timeSleep. По умолчанию браузер Selenium открывается в фоновом режиме, и работу браузера не видно, но если closeWindow указать как False, то будет виден процесс выполнения кода. С помощью timeSleep можно увеличить время загрузки страницы если контент на ней долго подгружается
4. Метод **returnBs4Object** возвращает объект beautifulsoup4. Принимает путь до html-страницы, содержимое которой преобразует в объект beautifulsoup, кодировку открытия файла (по умолчанию UTF-8) и тип парсера (по умолчанию lxml).
- Метод **gpsa** (get page semi-automatically) похож на метод getDynamicPage, но работает в полуавтоматическом режиме. Он открывает страницу сайта и ждёт пока не будет нажат Enter в терминале. В этот момент можно зарегистрироваться на сайте и/или перейти на нужную вкладку сайта, после чего нажать Enter и метод спарсит страницу. Подходит для лент соцсетей, где неавторизованным пользователям контент ограничен. Принисает такие же аргументы для таких же целей, что и getDynamicPage, за исключением closeWindow
5. Метод **getAttributes** нужена чтобы получить список атрибутов из списка объектов bs4. Принимает список объектов bs4 и название атрибута который будет извлекаться из элементов списка
- Метод **returnBs4Object** возвращает объект beautifulsoup4. Принимает путь до html-страницы, содержимое которой преобразует в объект beautifulsoup, кодировку открытия файла (по умолчанию UTF-8) и тип парсера (по умолчанию lxml).
6. Метод **getTexts** нужена чтобы получить список текста из списка объектов bs4. Принимает список объектов bs4 и параметр needFix. Если этот параметр установлен как True, то из текста будут удалены \n, \t и пробелы с концов
- Метод **getAttributes** нужена чтобы получить список атрибутов из списка объектов bs4. Принимает список объектов bs4 и название атрибута который будет извлекаться из элементов списка
7. Метод **pprint** используется для вывода значений переменных у которых большая вложеность. Например, если у Вас есть массив объектов, где в качестве значения ключа используется другой массив объектов
- Метод **getTexts** нужена чтобы получить список текста из списка объектов bs4. Принимает список объектов bs4 и параметр needFix. Если этот параметр установлен как True, то из текста будут удалены \n, \t и пробелы с концов
8. Метод **mkdir** используется для создания папки с именем _nameDir_ если она ещё не существует
- Метод **pprint** используется для вывода значений переменных у которых большая вложеность. Например, если у Вас есть массив объектов, где в качестве значения ключа используется другой массив объектов
### Пример парсера с использованием ipars:
- Метод **mkdir** используется для создания папки с именем _nameDir_ если она ещё не существует
- Метод **listdir** используется для получения списка файлов и папок в указанной директории
- Метод **exists** используется для проверки наличия файла или папки
### Пример использования класса Pars:
```py
# О классе ProgressBarManager читай ниже
from ipars import Pars, ProgressBarManager
p = Pars()
nameFile = 'index.html'
# Получаем html страницу
p.getDinamicPage(nameFile, 'https://duckduckgo.com/?q=теплица+социальных+технологий+youtube&iar=videos&atb=v454-1', closeWindow=0)
# Получаем объект BautifullSoup
soup = p.returnBs4Object(nameFile)
# Находим все карточки ответов
# Первые результаты выдачи те что хотелось получить, а остальные нет. Поэтому нам желательно получить первые 84 элемента
allCards = soup.find_all(class_='b_NgmZrVnRtV8MZMEjLs')[:84]
# Получаем все изображения
allImg = [card.find('img') for card in allCards]
# Получаем все ссылки
allSrc = p.getAttributes(allImg, 'src')
# Создаём папку img если её ещё нет
nameFolder = 'img'
p.mkdir(nameFolder)
# Создаём объект ProgressBarManager
bar = ProgressBarManager(len(allSrc))
def main():
# Получаем html страницу
p.getDynamicPage(nameFile, 'https://duckduckgo.com/?q=теплица+социальных+технологий+youtube&iar=videos&atb=v454-1', closeWindow=False, timeSleep=5)
# Скачиваем картинки
for index, url in enumerate(allSrc):
# Получаем объект BautifullSoup
soup = p.returnBs4Object(nameFile)
# Находим все карточки ответов
# Первые результаты выдачи те что хотелось получить, а остальные нет. Поэтому нам желательно получить первые 84 элемента
locator = 'b_NgmZrVnRtV8MZMEjLs'
allCards = soup.find_all(class_=locator)[:84]
if len(allCards) == 0:
print(f'''Something went wrong. Here is a list of possible causes:
1) DuckDuckGo has changed the class for video cards. The code uses the locator "{locator}"
2) The site did not load. Try increasing the timeSleep parameter or make a request later.''')
return
# Получаем все изображения
allImg = [card.find('img') for card in allCards]
# Получаем все ссылки
allSrc = p.getAttributes(allImg, 'src')
# Создаём папку img если её ещё нет
p.mkdir(nameFolder)
# Создаём объект ProgressBarManager
bar = ProgressBarManager(len(allSrc))
# Скачиваем картинки
for index, url in enumerate(allSrc):
url = 'https:' + url
p.getStaticPage(f'./{nameFolder}/img{index}.png', url, writeMethod='wb')
bar.next()
bar.finish()
bar.finish()
# Смотрим что появилось в папке img
p.pprint(p.listdir(nameFolder))
if __name__=='__main__':
main()
```
### Пример использования методов _getAttributes_ и _getTexts_
@@ -96,9 +147,10 @@ bar.finish()
```py
from ipars import Pars
p = Pars()
nameFile = 'index.html'
p.getStaticPage('./index.html', 'https://google.com')
soup = p.returnBs4Object('./index.html')
p.getStaticPage(nameFile, 'https://google.com')
soup = p.returnBs4Object(nameFile)
allTegA = soup.find_all('a')
a1 = p.getTexts(allTegA, needFix=1)
@@ -108,92 +160,106 @@ a2 = p.getAttributes(allTegA, 'href')
p.pprint(a2)
```
## Работа с JsonManager
### Работа с JsonManager
JsonManager принимает принимает только один аргумент — кодировку в которой будут читаться файлы. По умолчанию это UTF-8
Данный класс предназначен для записи и извлечения информации из json-файлов.
JsonManager принимает принимает только encoding — кодировку в которой будут читаться файлы (по умолчанию UTF-8)
```py
from ipars import JsonManager
j = JsonManager()
j = JsonManager(encoding="UTF-8")
```
### Коротко о методах JsonManager
1. Метод **load** используется для получения данных из json-файла по указанному пути
- Метод **load** используется для получения данных из json-файла по указанному пути
2. Метод **dump** используется для записи данных в json-файл. Принимает путь до файла и данные для записи
- Метод **dump** используется для записи данных в json-файл. Принимает путь до файла и данные для записи
3. Метод **pprint** такой же как и у Pars
- Метод **pprint** такой же как и у Pars
### Пример использования JsonManager
```py
from ipars import JsonManager
j = JsonManager()
nameFile = 'data.json'
# Записываем данные
j.dump('./data.json', [1, 2, 3, 4, 5, 6, 7])
j.dump(nameFile, [1, 2, 3, 4, 5, 6, 7])
# Получаем данные
data = j.load('./data.json')
data = j.load(nameFile)
j.pprint(data) # [1, 2, 3, 4, 5, 6, 7]
```
## Работа с CsvManager
### Работа с CsvManager
Класс CsvManager принимает три аргумента: символ переноса на новую строку _newline_ (по умолчанию — это пустая строка), кодировку открываемых файлов _encoding_ (по умолчанию UTF-8) и разделитель который используется в csv файле _delimiter_ (по умолчанию ";")
Данный класс предназначен для записи и извлечения информации из csv-файлов.
Класс CsvManager принимает три аргумента:
- newline — символ переноса на новую строку (по умолчанию используется пустая строка)
- encoding — кодировка открываемых файлов (UTF-8)
- delimiter — разделитель который используется в csv файле (;)
```py
from ipars import CsvManager
c = CsvManager()
c = CsvManager(newline="", encoding="UTF-8", delimiter=";")
```
### Коротко о методах CsvManager
1. Метод **writerow** записывает строку с csv файл. Метод принимает путь до csv файла, метод записи и список данных которые будут записанн в строку файла
- Метод **writerow** записывает строку с csv файл. Метод принимает путь до csv файла, метод записи и список данных которые будут записанн в строку файла
2. Метод **writerows** принимает теже самые аргументы что и writerow, только row должен быть двойным списком с данными для записи. Разница между этими методами в том что writerow записывает одну, а writerows столько сколько есть в двойном списке
- Метод **writerows** принимает теже самые аргументы что и writerow, только row должен быть двойным списком с данными для записи. Разница между этими методами в том что writerow записывает одну строку, а writerows столько сколько есть в двойном списке
3. Метод **getRows** используется для получения списка строк в csv файле. Метод принимает путь до файла откуда будут получены строки
- Метод **getRows** используется для получения списка строк в csv файле. Метод принимает путь до файла откуда будут получены строки
4. Метод **pprint** такой же как и у Pars
- Метод **pprint** такой же как и у Pars
### Пример использования CsvManager
```py
from ipars import CsvManager
c = CsvManager()
nameFile = 'data.csv'
# записываем заголовки
writer = c.writerow('./data.csv', 'w', ['Количество', 'Цена', 'Итог'])
c.writerow(nameFile, 'w', ['Количество', 'Цена', 'Итог'])
# записываем данные
writer = c.writerows('./data.csv', 'a', [
c.writerows(nameFile, 'a', [
["5", "5", "25"],
["6", "6", "36"],
["7", "7", "49"],
])
# получаем строки из таблицы
rows = c.getRows('./data.csv')
rows = c.getRows(nameFile)
# выводим строки таблицы
c.pprint(rows)
```
## Работа с ProgressBarManager
### Работа с ProgressBarManager
Класс создаёт прогресс-бар для лучшей видимости выполнения кода. Принимает пять аргументов:
Класс создаёт прогресс-бар для для отображения процесса выполнения кода. Принимает пять аргументов:
1. **max**: обязательный параметр, который указывает максимальное значение итераций в прогресс-баре
- **max**: обязательный параметр, который указывает максимальное значение итераций в прогресс-баре
2. **message**: сообщение перед прогресс-баром
3. **color**: цвет прогресс-бара
4. **fill**: заполнитель для сделанной части
5. **width**: размер прогресс-бара в символах
- **message**: сообщение перед прогресс-баром
- **color**: цвет прогресс-бара
- **fill**: заполнитель для выполненой части
- **width**: размер прогресс-бара в символах
```py
from ipars import ProgressBarManager
@@ -203,9 +269,9 @@ bar = ProgressBarManager(100) # Здесь max установлен как 100
### Коротко о методах ProgressBarManager
1. Метод **next** запускает следущую итерацию прогресс-бара
- Метод **next** запускает следущую итерацию прогресс-бара
2. Метод **finish** завершает работу прогресс-бара
- Метод **finish** завершает работу прогресс-бара
### Пример использования ProgressBarManager
@@ -246,6 +312,76 @@ for _ in range(maxValue):
bar.finish()
```
Если ты дочитал(-а) документацию до конца, то получай пожизненный запас здоровья❤. Помни, оно у тебя одно.
### Работа с TimerManager
Класс TimerManager предназначен для отслеживания времени выполнения кода. Он позволяет получить общее время работы в различных форматах.
```py
from ipars import TimerManager
t = TimerManager()
```
### Коротко о методах TimerManager
- Метод **start** — точка начала отсчёта времени
- Метод **end** — конечная точка отсчёта времени
- Метод **getWorkTime** возвращает общее время работы в указанном формате. Поддерживаемые форматы: _seconds_, _minutes_, _hours_. Параметр _ndigits_ указывает количество знаков после запятой для округления, по умолчанию число не округляется.
### Пример использования TimerManager
```py
from time import sleep
from ipars import TimerManager
t = TimerManager()
# Засекаем время
t.start()
sleep(2) # имитация двухсекундной работы
t.end()
# Выводим результат в разных форматах
print(f"Время работы в секундах: {t.getWorkTime()}") # 2.0008320808410645
print(f"Время работы в минутах: {t.getWorkTime(ndigits=2, format='minutes')}") # 0.03
print(f"Время работы в часах: {t.getWorkTime(ndigits=4, format='hours')}") # 0.0006
```
### Работа с ZipManager
Класс ZipManager нужен для архивации папоки файлов. Он принимает принимает только один аргумент — один из возможных уровней сжатия:
- none — без сжатия
- normal — обычное сжатие
- hard — увеличенное сжатие
- maximum — максимальное сжатие
```py
from ipars import ZipManager
z = ZipManager()
```
### Коротко о методах ZipManager
- Метод **zip_file** используется для архивирования одного файла. Принимает путь до исходного файла и путь выходного файла
- Метод **zip_folder** используется для архивирования каталога. Принимает путь до исходного каталога и путь выходного файла
### Пример использования ZipManager
```py
from ipars import ZipManager
z = ZipManager(compression='maximum')
z.zip_file('./your_file.txt', 'file_archive_maximum.zip')
z.zip_folder('./your_folder/', 'folder_archive_maximum.zip')
```
Если ты дочитал(-а) документацию до конца, то получай пожизненный запас здоровья ❤. Помни, оно у тебя одно.
+999999 HP
+31 -2
View File
@@ -1,20 +1,31 @@
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 файле
encoding: кодировка открываемого файла
delimiter: разделитель данных в csv файле'''
schema = {
'newline': {'type': 'string'},
'encoding': {'type': 'string'},
'delimiter': {'type': 'string'},
}
v = Validator(schema)
if not v.validate({'newline': newline, 'encoding': encoding, 'delimiter': delimiter}):
raise ValueError(v.errors)
self.newline = newline
self.encoding = encoding
self.delimiter = delimiter
def pprint(self, data: any) -> None:
def pprint(self, data: Any) -> None:
'''Выводим данные в удобочитаемом виде
data: данные которые надо вывести'''
@@ -26,6 +37,15 @@ class CsvManager:
filePath: путь до csv файла
mode (w/a): метод открытия файла
row: список записываемых данных'''
schema = {
'filePath': {'type': 'string'},
'mode': {'type': 'string', 'allowed': ['w', 'a'] },
'row': {'type': 'list'},
}
v = Validator(schema)
if not v.validate({'filePath': filePath, 'mode':mode, 'row':row}):
raise ValueError(v.errors)
with open(filePath, mode=mode, newline=self.newline, encoding=self.encoding) as file:
writer = csv.writer(file, delimiter=self.delimiter)
writer.writerow(row)
@@ -36,6 +56,15 @@ class CsvManager:
filePath: путь до csv файла
mode (w/a): метод открытия файла
row: список записываемых данных'''
schema = {
'filePath': {'type': 'string'},
'mode': {'type': 'string', 'allowed': ['w', 'a'] },
'row': {'type': 'list'},
}
v = Validator(schema)
if not v.validate({'filePath': filePath, 'mode':mode, 'row':row}):
raise ValueError(v.errors)
with open(filePath, mode=mode, newline=self.newline, encoding=self.encoding) as file:
writer = csv.writer(file, delimiter=self.delimiter)
writer.writerows(row)
+29 -4
View File
@@ -1,33 +1,58 @@
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: кодировка открываемого файла'''
schema = {
'encoding': {'type': 'string'}
}
v = Validator(schema)
if not v.validate({'encoding': encoding}):
raise ValueError(v.errors)
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 файла'''
schema = {
'pathToJsonFile': {'type': 'string'}
}
v = Validator(schema)
if not v.validate({'pathToJsonFile': pathToJsonFile}):
raise ValueError(v.errors)
with open(pathToJsonFile, encoding=self.encoding) as jsonFile:
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 файла
data: данные которые надо записать'''
schema = {
'pathToJsonFile': {'type': 'string'}
}
v = Validator(schema)
if not v.validate({'pathToJsonFile': pathToJsonFile}):
raise ValueError(v.errors)
with open(pathToJsonFile, 'w', encoding=self.encoding) as jsonFile:
json.dump(data, jsonFile, indent=4, ensure_ascii=0)
+234 -77
View File
@@ -5,39 +5,97 @@ from selenium import webdriver
from bs4 import BeautifulSoup
from pprint import pprint
from time import sleep
from typing import Any, Optional
import asyncio
import aiohttp
import requests
from os import mkdir
from os import mkdir, listdir
from os.path import exists
from cerberus import Validator
class Pars:
'''Модуль для работы с запросами и bs4'''
def mkdir(self, nameDir: str):
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) -> bool:
'''Возвращает True если указанный файл или папка сущуствует, иначе — False'''
schema = {'path': {'type': 'string'}}
expected = {'path': path}
self.__validation(schema, expected)
return exists(path)
def listdir(self, path: str) -> list:
'''Возвращает список файлов в указанной директории'''
schema = {'path': {'type': 'string'}}
expected = {'path': path}
self.__validation(schema, expected)
return listdir(path)
def mkdir(self, nameDir: str) -> None:
'''Создаёт папку если её ещё нет
nameDir: название папки которая будет создана'''
schema = {'nameDir': {'type': 'string'}}
expected = {'nameDir': nameDir}
self.__validation(schema, expected)
if not exists(nameDir):
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 файла
encoding: кодировка открытия html файла
parser: парсер, который будет использоваться для работы'''
schema = {
'pathToFile': {'type': 'string'},
'encoding': {'type': 'string'},
'parser': {'type': 'string'},
}
expected = {'pathToFile': pathToFile, 'encoding': encoding, 'parser': parser}
self.__validation(schema, expected)
with open(pathToFile, encoding=encoding) as file:
src = file.read()
soup = BeautifulSoup(src, parser)
return soup
def getTexts(self, arr: list, needFix: bool = 0) -> list:
def getTexts(self, arr: list, needFix: bool = False) -> Optional[list]:
'''Возвращаем текст из элементов bs4
arr: список объектов bs4 из которых будет извлекаться текст
needFix (0/1): если этот параметр установлен как True, то из текста будут удалены \\n, \\t и пробелы с концов'''
needFix (False/true): если этот параметр установлен как True, то из текста будут удалены \\n, \\t и пробелы с концов'''
schema = {
'arr': {'type': 'list'},
'needFix': {'type': 'boolean'},
}
expected = {'arr': arr, 'needFix': needFix}
self.__validation(schema, expected)
result = []
for item in arr:
data = item.text
if needFix:
data = data.strip()
data = data.replace('\t', '')
@@ -51,11 +109,22 @@ class Pars:
return None
return result
def getAttributes(self, arr: list, att: str) -> list:
def getAttributes(self, arr: list, att: str) -> Optional[list]:
'''Возвращаем список значений атрибутов
arr: список объектов bs4 из которых будет извлекаться атрибут
att: строка с названием атрибута по которому будет осуществляться поиск'''
schema = {
'arr': {'type': 'list'},
'att': {'type': 'string'}
}
expected = {'arr': arr, 'att': att}
self.__validation(schema, expected)
result = []
for item in arr:
data = item.get(att)
@@ -66,22 +135,129 @@ class Pars:
return None
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: путь, куда сохранится полученный файл
url: ссылка на данные
writeMethod: метод записи данных в файл. "W" записывает текст, "WB" — байты
writeMethod: метод записи данных в файл. "w" записывает текст, "wb" — байты
headers: заголовки запроса к серверу'''
if headers is None:
# Получаем случайный user agent
headers = self.__makeHeaders()
schema = {
'pathToSaveFile': {'type': 'string'},
'url': {'type': 'string'},
'writeMethod': {'type': 'string', 'allowed': ['w', 'wb']},
'headers': {'type': 'dict'},
}
expected = {'pathToSaveFile': pathToSaveFile, 'url': url, 'writeMethod': writeMethod, 'headers': headers,}
self.__validation(schema, expected)
try:
req = requests.get(url, headers=headers)
req.raise_for_status()
if writeMethod == 'w':
src = req.text
with open(pathToSaveFile, 'w', encoding='utf-8') as file:
file.write(src)
elif writeMethod == 'wb':
src = req.content
with open(pathToSaveFile, 'wb') as file:
file.write(src)
else:
raise ValueError(f"Неподдерживаемый метод записи: {writeMethod}")
return req.status_code
except requests.exceptions.HTTPError as httpErr:
raise RuntimeError(f"HTTP ошибка: {httpErr}") from httpErr
except Exception as e:
raise RuntimeError(e) from e
def __scrollAndSave(self, driver: webdriver.Chrome, timeSleep: int, pathToSaveFile: str) -> None:
'''Прокручиваем страницу до самого низа'''
lastHeight = driver.execute_script("return document.body.scrollHeight")
while True:
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
sleep(timeSleep)
newHeight = driver.execute_script("return document.body.scrollHeight")
if newHeight == lastHeight:
break
lastHeight = newHeight
htmlContent = driver.page_source
with open(pathToSaveFile, "w", encoding="utf-8") as file:
file.write(htmlContent)
def getDynamicPage(self, pathToSaveFile: str, url: str, closeWindow: bool = True, timeSleep: int = 2) -> None:
'''Получаем динамическую страницу
pathToSaveFile: путь, куда сохранится полученный файл
url: ссылка на сайт
closeWindow (False/True): если указана единица, то браузер открывается в фоновом режиме, 0 — открывается как обычное приложение
timeSleep: время ожидания в секундах браузера перед тем как скролить страницу дальше'''
schema = {
'pathToSaveFile': {'type': 'string'},
'url': {'type': 'string'},
'closeWindow': {'type': 'boolean'},
'timeSleep': {'type': 'integer', 'min': 2},
}
expected = {'pathToSaveFile': pathToSaveFile, 'url': url, 'closeWindow': closeWindow, 'timeSleep': timeSleep}
self.__validation(schema, expected)
options = Options()
if closeWindow:
options.add_argument('--headless')
with webdriver.Chrome(options=options) as driver:
driver.get(url)
self.__scrollAndSave(driver, timeSleep, pathToSaveFile)
def gpsa(self, pathToSaveFile: str, url: str, timeSleep: int = 2) -> None:
'''Получаем страницу в полуавтоматическом режиме
gpsa - get page semi-automatically
pathToSaveFile: путь, куда сохранится полученный файл
url: ссылка на сайт
timeSleep: время ожидания браузера перед тем как скролить страницу дальше'''
schema = {
'pathToSaveFile': {'type': 'string'},
'url': {'type': 'string'},
'timeSleep': {'type': 'integer', 'min': 2},
}
expected = {'pathToSaveFile': pathToSaveFile, 'url': url, 'timeSleep': timeSleep}
self.__validation(schema, expected)
with webdriver.Chrome() as driver:
driver.get(url)
a = input('Нажми ENTER когда окажешься на нужной тебе странице')
self.__scrollAndSave(driver, timeSleep, pathToSaveFile)
def __makeHeaders(self) -> dict:
'''Генерирует случайный User-Agent для запроса'''
software_names = [
SoftwareName.FIREFOX.value,
SoftwareName.CHROME.value,
@@ -96,81 +272,62 @@ class Pars:
OperatingSystem.MACOS.value,
]
user_agent_rotator = UserAgent(software_names=software_names, operating_systems=operating_systems, limit=10)
user_agent = user_agent_rotator.get_random_user_agent()
headers = {
"Accept": "*/*",
"User-Agent": user_agent
return {
'Accept': '*/*',
'User-Agent': user_agent_rotator.get_random_user_agent()
}
async def getStaticPageAsync(
self,
pathToSaveFile: str,
url: str,
writeMethod: str = 'w',
headers: Optional[dict] = None,
session: Optional[aiohttp.ClientSession] = None,
) -> int:
'''Асинхронно сохраняет статическую страницу и возвращает статус ответа
pathToSaveFile: путь, куда сохранится полученный файл
url: ссылка на данные
writeMethod: метод записи. "w" — текст, "wb" — байты (для картинок)
headers: заголовки запроса. Если None — подставится случайный User-Agent
session: aiohttp.ClientSession для переиспользования соединений'''
if headers is None:
headers = self.__makeHeaders()
schema = {
'pathToSaveFile': {'type': 'string'},
'url': {'type': 'string'},
'writeMethod': {'type': 'string', 'allowed': ['w', 'wb']},
'headers': {'type': 'dict'},
}
self.__validation(schema, {'pathToSaveFile': pathToSaveFile, 'url': url, 'writeMethod': writeMethod, 'headers': headers})
close_session = session is None
if close_session:
session = aiohttp.ClientSession()
try:
# Отправляем запрос
req = requests.get(url, headers=headers)
async with session.get(url, headers=headers) as response:
response.raise_for_status()
# Записываем данные
if writeMethod == 'w':
src = req.text
text = await response.text(encoding='utf-8')
with open(pathToSaveFile, 'w', encoding='utf-8') as file:
file.write(src)
file.write(text)
elif writeMethod == 'wb':
src = req.content
content = await response.read()
with open(pathToSaveFile, 'wb') as file:
file.write(src)
else:
raise ValueError(f"Неподдерживаемый метод записи: {writeMethod}")
file.write(content)
return req.status_code # Возвращаем статус ответа от сервера
return response.status
except requests.exceptions.HTTPError as httpErr:
raise RuntimeError(f"HTTP ошибка: {httpErr}") from http_err
except aiohttp.ClientResponseError as httpErr:
raise RuntimeError(f'HTTP ошибка: {httpErr}') from httpErr
except Exception as e:
raise RuntimeError(e) from e
def __scrollAndSave(self, driver, timeSleep, pathToSaveFile):
# Прокручиваем страницу до самого низа
lastHeight = driver.execute_script("return document.body.scrollHeight")
while True:
# Прокручиваем до низа страницы
driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
# Ждем загрузки страницы
sleep(timeSleep)
# Вычисляем новую высоту страницы
newHeight = driver.execute_script("return document.body.scrollHeight")
if newHeight == lastHeight:
break
lastHeight = newHeight
htmlContent = driver.page_source
with open(pathToSaveFile, "w", encoding="utf-8") as file:
file.write(htmlContent)
def getDinamicPage(self, pathToSaveFile: str, url: str, closeWindow: bool = 1, timeSleep: int = 2) -> None:
'''Получаем динамическую страницу
pathToSaveFile: путь, куда сохранится полученный файл
url: ссылка на сайт
closeWindow (0/1): если указана единица, то браузер открывается в фоновом режиме, 0 — открывается как обычное приложение
timeSleep: время ожидания браузера перед тем как скролить страницу дальше'''
# Устанавливаем опции для Chrome WebDriver
options = Options()
if closeWindow:
options.add_argument('--headless')
# открываем браузер
with webdriver.Chrome(options=options) as driver:
driver.get(url)
self.__scrollAndSave(driver, timeSleep, pathToSaveFile)
def gpsa(self, pathToSaveFile: str, url: str, timeSleep=2):
'''Получаем страницу в полуавтоматическом режиме
gpsa - get page semi-automatically
pathToSaveFile: путь, куда сохранится полученный файл
url: ссылка на сайт
timeSleep: время ожидания браузера перед тем как скролить страницу дальше'''
with webdriver.Chrome() as driver:
driver.get(url)
a = input('Нажми ENTER когда закончишь')
self.__scrollAndSave(driver, timeSleep, pathToSaveFile)
finally:
if close_session:
await session.close()
+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()
+43
View File
@@ -0,0 +1,43 @@
import time
from typing import Optional
class TimerManager:
'''Класс для отслеживания времени работы'''
def __init__(self) -> None:
'''Инициализация'''
self.startTime = 0
self.endTime = 0
self.workTime = 0
def start(self) -> None:
'''Устанавливает начальное время'''
self.startTime = time.time()
def end(self) -> None:
'''Устанавливает конечное время'''
self.endTime = time.time()
self.workTime = self.endTime - self.startTime
def getWorkTime(self, format: str = 'seconds', ndigits: Optional[int] = None) -> float:
'''Возвращает время в нужном формате'''
time_units = {
'seconds': 1,
'minutes': 60,
'hours': 3600
}
# Обработка ошибок
if self.startTime == 0:
raise ValueError('Для замера времени обязательно надо использовать метод "start" ДО выполнения кода')
if self.endTime == 0:
raise ValueError('Для замера времени обязательно надо использовать метод "end" ПОСЛЕ выполнения кода')
if format not in time_units:
raise ValueError(f"Неподдерживаемый формат '{format}'. Должен быть один из {list(time_units.keys())}")
# Округление результата
if ndigits is None:
return self.workTime / time_units[format]
else:
return round(self.workTime / time_units[format], ndigits=ndigits)
+51
View File
@@ -0,0 +1,51 @@
import os
import zipfile
class ZipManager:
"""
Параметр:
- compression (str):
- 'none': Без сжатия
- 'normal': Обычное сжатие
- 'hard': Увеличенное сжатие
- 'maximum': Максимальное сжатие
"""
def __init__(self, compression: str = 'normal') -> None:
self.compression = self.setCompression(compression)
def setCompression(self, compressionStr: str) -> int:
if compressionStr == 'none':
return zipfile.ZIP_STORED
elif compressionStr == 'normal':
return zipfile.ZIP_DEFLATED
elif compressionStr == 'hard':
return zipfile.ZIP_BZIP2
elif compressionStr == 'maximum':
return zipfile.ZIP_LZMA
else:
raise ValueError("Некорректное значение compression. Допустимы значения none, normal, hard, maximun")
def zipFile(self, filePath: str, zipFilePath: str) -> None:
"""Архивируем один файл
filePath (str): Путь к файлу, который нужно заархивировать
zipFilePath (str): Путь к выходному ZIP-файлу
"""
with zipfile.ZipFile(zipFilePath, 'w', compression=self.compression) as zipf:
zipf.write(filePath, os.path.basename(filePath))
def zipFolder(self, folderPath: str, zipFilePath: str) -> None:
"""Архивируем папку
folderPath (str): Путь к папке, которую нужно заархивировать
zipFilePath (str): Путь к выходному ZIP-файлу
"""
with zipfile.ZipFile(zipFilePath, 'w', compression=self.compression) as zipf:
baseName = os.path.basename(os.path.normpath(folderPath))
for root, _, files in os.walk(folderPath):
for file in files:
filePath = os.path.join(root, file)
arcname = os.path.join(baseName, os.path.relpath(filePath, folderPath))
zipf.write(filePath, arcname)
+2
View File
@@ -2,3 +2,5 @@ from .ParsManagerCode import Pars
from .JsonManagerCode import JsonManager
from .CsvManagerCode import CsvManager
from .ProgressBarCode import ProgressBarManager
from .TimerManagerCode import TimerManager
from .ZipManagerCode import ZipManager
BIN
View File
Binary file not shown.
+18 -1
View File
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name='ipars',
version='3.5.1',
version='3.9.3',
description='Библиотека для работы с файлами во время парсинга',
long_description=open('README.md', encoding='utf-8').read(),
long_description_content_type='text/markdown',
@@ -10,12 +10,29 @@ setup(
packages=find_packages(),
author='Ilia Miheev',
license='MIT',
url='https://iliamiheev.github.io/ipars-doc/#/./home',
install_requires=[
'requests',
'aiohttp',
'selenium',
'lxml',
'bs4',
'progress',
'random_user_agent',
'cerberus'
],
keywords=[
'ipars',
'ипарс',
'парсинг',
'скрапинг',
'parsing',
'scraping',
'bs4',
'beautifulsoup4'
],
project_urls={
'Bug Reports': 'https://github.com/IliaMiheev/ipars/issues',
'Source': 'https://github.com/IliaMiheev/ipars',
},
)
View File
+72
View File
@@ -0,0 +1,72 @@
import pytest
from ipars import CsvManager
def test_writerow_and_getrows(tmp_path):
cm = CsvManager()
filepath = str(tmp_path / 'data.csv')
row = ['Alice', '30', 'Engineer']
cm.writerow(filepath, 'w', row)
result = cm.getRows(filepath)
assert result == [row]
def test_writerows_and_getrows(tmp_path):
cm = CsvManager()
filepath = str(tmp_path / 'data.csv')
rows = [['Alice', '30'], ['Bob', '25'], ['Carol', '28']]
cm.writerows(filepath, 'w', rows)
result = cm.getRows(filepath)
assert result == rows
def test_append_mode(tmp_path):
cm = CsvManager()
filepath = str(tmp_path / 'data.csv')
cm.writerow(filepath, 'w', ['first'])
cm.writerow(filepath, 'a', ['second'])
result = cm.getRows(filepath)
assert result == [['first'], ['second']]
def test_custom_delimiter(tmp_path):
cm = CsvManager(delimiter=',')
filepath = str(tmp_path / 'data.csv')
row = ['one', 'two', 'three']
cm.writerow(filepath, 'w', row)
result = cm.getRows(filepath)
assert result == [row]
def test_invalid_mode_raises(tmp_path):
cm = CsvManager()
filepath = str(tmp_path / 'data.csv')
with pytest.raises(ValueError):
cm.writerow(filepath, 'x', ['data'])
def test_invalid_row_type_raises(tmp_path):
cm = CsvManager()
filepath = str(tmp_path / 'data.csv')
with pytest.raises(ValueError):
cm.writerow(filepath, 'w', 'not a list')
def test_invalid_constructor_args():
with pytest.raises(ValueError):
CsvManager(delimiter=123)
def test_pprint_does_not_raise(capsys):
cm = CsvManager()
cm.pprint([['a', 'b'], ['c', 'd']])
captured = capsys.readouterr()
assert 'a' in captured.out
+65
View File
@@ -0,0 +1,65 @@
import pytest
from ipars import JsonManager
def test_dump_and_load(tmp_path):
jm = JsonManager()
filepath = str(tmp_path / 'data.json')
data = {'name': 'ipars', 'version': 3, 'active': True}
jm.dump(filepath, data)
result = jm.load(filepath)
assert result == data
def test_dump_and_load_list(tmp_path):
jm = JsonManager()
filepath = str(tmp_path / 'list.json')
data = [1, 2, 3, 'hello']
jm.dump(filepath, data)
result = jm.load(filepath)
assert result == data
def test_dump_and_load_with_cyrillic(tmp_path):
jm = JsonManager()
filepath = str(tmp_path / 'cyrillic.json')
data = {'текст': 'привет'}
jm.dump(filepath, data)
result = jm.load(filepath)
assert result == data
def test_load_nonexistent_file():
jm = JsonManager()
with pytest.raises(FileNotFoundError):
jm.load('nonexistent_file_xyz.json')
def test_invalid_encoding_raises():
with pytest.raises(ValueError):
JsonManager(encoding=123)
def test_load_invalid_path_type():
jm = JsonManager()
with pytest.raises(ValueError):
jm.load(123)
def test_dump_invalid_path_type(tmp_path):
jm = JsonManager()
with pytest.raises(ValueError):
jm.dump(123, {'key': 'value'})
def test_pprint_does_not_raise(capsys):
jm = JsonManager()
jm.pprint({'key': 'value'})
captured = capsys.readouterr()
assert 'key' in captured.out
+189
View File
@@ -0,0 +1,189 @@
import pytest
from unittest.mock import patch, MagicMock
from bs4 import BeautifulSoup
from ipars import Pars
# --- exists ---
def test_exists_true(tmp_path):
p = Pars()
assert p.exists(str(tmp_path)) is True
def test_exists_false():
p = Pars()
assert p.exists('this_path_does_not_exist_xyz') is False
def test_exists_invalid_type():
p = Pars()
with pytest.raises(ValueError):
p.exists(123)
# --- listdir ---
def test_listdir_returns_list(tmp_path):
p = Pars()
(tmp_path / 'a.txt').write_text('a')
(tmp_path / 'b.txt').write_text('b')
result = p.listdir(str(tmp_path))
assert isinstance(result, list)
assert 'a.txt' in result
assert 'b.txt' in result
def test_listdir_invalid_type():
p = Pars()
with pytest.raises(ValueError):
p.listdir(42)
# --- mkdir ---
def test_mkdir_creates_directory(tmp_path):
p = Pars()
new_dir = str(tmp_path / 'new_folder')
p.mkdir(new_dir)
assert (tmp_path / 'new_folder').exists()
def test_mkdir_does_not_raise_if_exists(tmp_path):
p = Pars()
existing = str(tmp_path)
p.mkdir(existing) # не должен бросать ошибку
def test_mkdir_invalid_type():
p = Pars()
with pytest.raises(ValueError):
p.mkdir(99)
# --- returnBs4Object ---
def test_returnBs4Object(tmp_path):
p = Pars()
html_file = tmp_path / 'page.html'
html_file.write_text('<html><body><h1>Hello</h1></body></html>', encoding='utf-8')
soup = p.returnBs4Object(str(html_file))
assert soup.find('h1').text == 'Hello'
def test_returnBs4Object_returns_beautifulsoup(tmp_path):
p = Pars()
html_file = tmp_path / 'page.html'
html_file.write_text('<p>Test</p>', encoding='utf-8')
result = p.returnBs4Object(str(html_file))
assert isinstance(result, BeautifulSoup)
# --- getTexts ---
def _make_soup_elements(html: str) -> list:
soup = BeautifulSoup(html, 'lxml')
return soup.find_all('div')
def test_getTexts_basic():
p = Pars()
elements = _make_soup_elements('<div>Hello</div><div>World</div>')
result = p.getTexts(elements)
assert result == ['Hello', 'World']
def test_getTexts_needfix_strips_whitespace():
p = Pars()
elements = _make_soup_elements('<div> Hello \n </div><div>\t World\t</div>')
result = p.getTexts(elements, needFix=True)
assert result == ['Hello', 'World']
def test_getTexts_returns_none_when_empty():
p = Pars()
elements = _make_soup_elements('<div></div>')
result = p.getTexts(elements)
assert result is None
def test_getTexts_invalid_type():
p = Pars()
with pytest.raises(ValueError):
p.getTexts('not a list')
# --- getAttributes ---
def test_getAttributes_basic():
p = Pars()
soup = BeautifulSoup('<a href="http://a.com">1</a><a href="http://b.com">2</a>', 'lxml')
elements = soup.find_all('a')
result = p.getAttributes(elements, 'href')
assert result == ['http://a.com', 'http://b.com']
def test_getAttributes_missing_attr():
p = Pars()
soup = BeautifulSoup('<div>no href</div>', 'lxml')
elements = soup.find_all('div')
result = p.getAttributes(elements, 'href')
assert result is None
def test_getAttributes_invalid_type():
p = Pars()
with pytest.raises(ValueError):
p.getAttributes('not a list', 'href')
# --- pprint ---
def test_pprint_does_not_raise(capsys):
p = Pars()
p.pprint({'key': 'value', 'list': [1, 2, 3]})
captured = capsys.readouterr()
assert 'key' in captured.out
# --- getStaticPage (с моком запроса) ---
def test_getStaticPage_saves_file(tmp_path):
p = Pars()
filepath = str(tmp_path / 'page.html')
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = '<html><body>Mocked</body></html>'
with patch('requests.get', return_value=mock_response):
status = p.getStaticPage(filepath, 'http://example.com')
assert status == 200
assert (tmp_path / 'page.html').read_text(encoding='utf-8') == mock_response.text
def test_getStaticPage_binary_mode(tmp_path):
p = Pars()
filepath = str(tmp_path / 'image.png')
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.content = b'\x89PNG\r\n'
with patch('requests.get', return_value=mock_response):
status = p.getStaticPage(filepath, 'http://example.com/img.png', writeMethod='wb')
assert status == 200
assert (tmp_path / 'image.png').read_bytes() == mock_response.content
def test_getStaticPage_invalid_write_method(tmp_path):
p = Pars()
filepath = str(tmp_path / 'page.html')
with pytest.raises(ValueError):
p.getStaticPage(filepath, 'http://example.com', writeMethod='r')
+29
View File
@@ -0,0 +1,29 @@
from ipars import ProgressBarManager
def test_full_cycle_does_not_raise():
bar = ProgressBarManager(max=3)
bar.next()
bar.next()
bar.next()
bar.finish()
def test_single_iteration():
bar = ProgressBarManager(max=1)
bar.next()
bar.finish()
def test_custom_message():
bar = ProgressBarManager(max=2, message='Загрузка')
bar.next()
bar.next()
bar.finish()
def test_custom_fill_and_width():
bar = ProgressBarManager(max=2, fill='=', width=20)
bar.next()
bar.next()
bar.finish()
+69
View File
@@ -0,0 +1,69 @@
import time
import pytest
from ipars import TimerManager
def test_getWorkTime_seconds():
timer = TimerManager()
timer.start()
time.sleep(0.1)
timer.end()
result = timer.getWorkTime('seconds')
assert 0.05 < result < 0.5
def test_getWorkTime_minutes():
timer = TimerManager()
timer.start()
time.sleep(0.1)
timer.end()
result = timer.getWorkTime('minutes')
assert result < 0.01
def test_getWorkTime_hours():
timer = TimerManager()
timer.start()
time.sleep(0.1)
timer.end()
result = timer.getWorkTime('hours')
assert result < 0.001
def test_getWorkTime_ndigits():
timer = TimerManager()
timer.start()
time.sleep(0.1)
timer.end()
result = timer.getWorkTime('seconds', ndigits=2)
assert isinstance(result, float)
assert len(str(result).split('.')[-1]) <= 2
def test_getWorkTime_returns_float():
timer = TimerManager()
timer.start()
timer.end()
result = timer.getWorkTime()
assert isinstance(result, float)
def test_raises_without_start():
timer = TimerManager()
with pytest.raises(ValueError, match='"start"'):
timer.getWorkTime()
def test_raises_without_end():
timer = TimerManager()
timer.start()
with pytest.raises(ValueError, match='"end"'):
timer.getWorkTime()
def test_raises_invalid_format():
timer = TimerManager()
timer.start()
timer.end()
with pytest.raises(ValueError, match='milliseconds'):
timer.getWorkTime('milliseconds')
+87
View File
@@ -0,0 +1,87 @@
import zipfile
import pytest
from ipars import ZipManager
def test_zipfile_creates_archive(tmp_path):
zm = ZipManager()
source = tmp_path / 'hello.txt'
source.write_text('hello world')
zip_path = str(tmp_path / 'output.zip')
zm.zipFile(str(source), zip_path)
assert (tmp_path / 'output.zip').exists()
with zipfile.ZipFile(zip_path, 'r') as zf:
assert 'hello.txt' in zf.namelist()
def test_zipfile_content_is_correct(tmp_path):
zm = ZipManager()
source = tmp_path / 'data.txt'
source.write_text('test content')
zip_path = str(tmp_path / 'output.zip')
zm.zipFile(str(source), zip_path)
with zipfile.ZipFile(zip_path, 'r') as zf:
content = zf.read('data.txt').decode('utf-8')
assert content == 'test content'
def test_zipfolder_creates_archive(tmp_path):
zm = ZipManager()
folder = tmp_path / 'myfolder'
folder.mkdir()
(folder / 'a.txt').write_text('aaa')
(folder / 'b.txt').write_text('bbb')
zip_path = str(tmp_path / 'folder.zip')
zm.zipFolder(str(folder), zip_path)
assert (tmp_path / 'folder.zip').exists()
with zipfile.ZipFile(zip_path, 'r') as zf:
names = zf.namelist()
assert any('a.txt' in n for n in names)
assert any('b.txt' in n for n in names)
def test_zipfolder_nested(tmp_path):
zm = ZipManager()
folder = tmp_path / 'root'
folder.mkdir()
sub = folder / 'sub'
sub.mkdir()
(sub / 'deep.txt').write_text('deep')
zip_path = str(tmp_path / 'nested.zip')
zm.zipFolder(str(folder), zip_path)
with zipfile.ZipFile(zip_path, 'r') as zf:
names = zf.namelist()
assert any('deep.txt' in n for n in names)
def test_set_compression_none():
zm = ZipManager('none')
assert zm.compression == zipfile.ZIP_STORED
def test_set_compression_normal():
zm = ZipManager('normal')
assert zm.compression == zipfile.ZIP_DEFLATED
def test_set_compression_hard():
zm = ZipManager('hard')
assert zm.compression == zipfile.ZIP_BZIP2
def test_set_compression_maximum():
zm = ZipManager('maximum')
assert zm.compression == zipfile.ZIP_LZMA
def test_invalid_compression_raises():
with pytest.raises(ValueError):
ZipManager('ultra')