Добален планировщик заданий и README.md

README.md

@@ -0,0 +1,78 @@
+# About
+- Получаем и парсим данные с https://formats.saby.ru/
+- Выполняем валидацию данных для БД
+- Отправляем данные в БД oracle
+- Планировщик настроенный на определеное время
+
+# Usage 
+**Боевой** запуск через **main.py**  
+**Тестовый** запуск чере **test.py**. В test.py не выполняется подключение к БД,
+выполяется лишь подготовка данных. Благодоря этому **не имеет значение где запщуен код**. 
+Можно полностью отследить процесс парсинга и валидации. 
+
+# How it's works
+Запуск кода осущетсвляется через main.py или test.py  
+Важно: в test.py не выполняется 4 и 5 шаг.
+
+## 1. Получение данных с сайта.
+C перва запускается скрипт parse_saby.py  
+```
+result_dict_data = parse_saby.process_reports_parse()
+```
+Основным модулем для забора нужных данных является bs4 с классом BeautifulSoup.
+BeautifulSoup представляет объект html страницы, это позволяет обращаеться по тэгам,
+что бы достать нужные данные. Модуль возвращает cписок словарей. Пример:
+```
+result_dict_data = [{128513: ('fns', 'Уведомление о налогах для ЕНП', '01.07.22', None, '5.03')},
+{132526: ('sfr', 'АДВ-1 Анкета застрахованного лица', '09.01.23', '31.12.34', '2.0')},
+{...}]
+```
+
+## 2. parse_data_in_list()
+```
+list_data = parse_data_in_list(result_dict_data)
+```
+Легкий парсиннг, для преобразование, нужное в дальнейшем.
+
+## 3. validate_data() and validation.py
+```
+list_data_validated = validate_data(list_data)  
+```
+Тут из листа по листу передаем данные в класс DataValid из модуля validation.py.
+DataValid наследуется от класса BaseModul модуля pydantic.
+Сначало парсится дата, вторая дата в списке может быть None.  
+
+Потом проверяется соответсвие типов, и может выполняется явное преобразование. 
+Забираем валидные данные в новый лист. 
+Класс в рамках цикла пересоздается, для валидации следующего листа.
+
+## 4. Отправка данных в БД
+```
+working_database.SimpleDB().data_transfer_in_database(list_data_validated)
+```
+### 4.1. Инциализация модуля working_database.py, подключение к hvac
+_Сервер hvac настрое на работу в тихом режиме, реализуется с помощью: init_oracle_client()_
+
+Тут выполняется сначало подключение к hvac серверу, получение секретов,
+необохдимых для подключения к БД ```_create_db_pool_from_vault()```. Что бы подключиться к серверу hvac 
+используется секретный токен. Он забирается из переменной окружения ОС, 
+передается при создание контейнера(определенно в docker-compouse.yaml).
+
+### 4.2. Подключение и отправка данных в БД
+Метод класса ```data_transfer_in_database()``` получает данные для отправки в БД.
+Данные имеют структуру лист словарей. Выполняется подключение к БД используя пул секретов из шага 4.1.
+После чего передаются данные в процедуру P_RK_GOVERNMENT_REPORTS_INSERS. 
+
+## 5. Планировщик заданий. 
+```
+scheduler.launch_the_scheduler()
+```
+Планировщик работает в фоновом режиме, пока не наступит заданое время.
+Когда наступает заданое время запукает main.py. Время запуска по умолчанию 6 часов 0 минут.  
+Время можно изменить например на 9:30 следующим образом:  
+```
+scheduler.launch_the_scheduler(h=9, m=30)
+```
+
+
+

app/main.py

@@ -1,6 +1,7 @@
 import working_database
 import parse_saby
 from validation import DataValid
+import scheduler
 
 def parse_data_in_list(dict_data: dict) -> list:
     """
@@ -35,3 +36,4 @@ result_dict_data = parse_saby.process_reports_parse()
 list_data = parse_data_in_list(result_dict_data)
 list_data_validated = validate_data(list_data)                     
 working_database.SimpleDB().data_transfer_in_database(list_data_validated)
+scheduler.launch_the_scheduler()

app/parse_saby.py

@@ -1,7 +1,12 @@
+# pyright: reportOptionalMemberAccess=false, reportAttributeAccessIssue=false, reportOperatorIssue=false
+# pyright: reportArgumentType=false, reportIndexIssue=false, reportCallIssue=false, reportGeneralTypeIssues=false
+# Подсветка синтаксиса отключена, т.к. тип данных везде кооректены и обрабатывается if.
+
 import requests
 from bs4 import BeautifulSoup
 import re
 
+
 def parse_html(url: str):
     """
     Фунция принимает строку URL, выполняет запрос.
@@ -9,7 +14,7 @@ def parse_html(url: str):
     """
     # Запрос страницы
     response = requests.get(url)
-    # Проверка статуса в ответе на запрос
+    # Проверка статуса
     if not(200 <= response.status_code <= 299):
         # Повторный запрос 
         response = requests.get(url)
@@ -32,8 +37,7 @@ def parse_date_report(url: str):
     # HTML report
     soup = parse_html(url)
     # Проверка на ошибку:
-    #TODO проверка на int, а ошибка про None
+    if soup == int: return None # Вызовит ошибку, что бы пропустить данный url
-    if soup == int: raise ValueError("Объект soup не должен быть None")
     # Поиск в HTML строки ввида: 'Действующий формат (с 10.01.23) 5.01'
     div_element = soup.find('div', class_='controls-Dropdown__text')
     # Извлекаем текст из элемента
@@ -75,21 +79,20 @@ def parse_reports(soup:BeautifulSoup, # HTML объект
                 link = soup.find('a', href=href)
                 # Name report
                 span = link.find('span', class_="ProxySbisRu__registry-BrowserItem_typeName")
+                try:
                     # Данные получены из url после парсинга
                     from_date, to_date, version = parse_date_report(url_report)
+                except Exception: continue # Может быть ошибка если url не доступен
                 # Добавление всех данных в итоговый словарь            
                 result_dict_data.update({id: (name_title, span.text, from_date, to_date, version)})
         except Exception as e:
-            print(f"Ошибка при обработке отчета {report_title}: {str(e)}")
+            print(f"Ошибка при обработке отчета {report_title}: ", e)
             continue
     return result_dict_data
 
-def write_report_data(dict_name:dict, name_title:str):
+def print_report_data(dict_name:dict, name_title:str):
     """
-    Сохраняем запись, каждая запись с новой строки:
+    Вывод на стандартный поток вывода итоговых данных
-    'ключ: значение'
-    'ключ: значение'
-    ...
     """
     #Блок для красивого офорлмения файла
     def center_text():
@@ -123,19 +126,17 @@ def search_title(url_format_report = 'https://formats.saby.ru/report'):
     """
     Функция ищет все тайтлы на странице formats.saby.ru/report.
     Парамметры функции:
-        url по которому в котором будет происходить поиск
+        url в котором будет происходить поиск
     Возвращает:
         Список URL-путей, например: ['/report/fns', '/report/example', ...]
     Исключения:
         ValueError: Если запрос к странице завершился с ошибкой (неверный статус).
     """  
-
     # Получаем HTML-cтраницу
     html = parse_html(url_format_report)
     # Проверяем, что html не является кодом ошибки (int)
     if isinstance(html, int):
-        error_message = f'Ошибка при запросе {url_format_report}: {html}'
+        raise Exception(f'Ошибка при запросе {url_format_report}: {html}')
-        raise Exception(error_message)
     
     # Множество в который будут заноситься тайтлы
     report_urls = set()
@@ -158,7 +159,6 @@ def process_reports_parse(url_formats = 'https://formats.saby.ru'):
     Параметр функции:
         url_formats - используется для создание полных URL
     """
-
     # Лист имеет вид: ['/report/fns', '/report/sfr'...]
     set_title = search_title()
     dict_result = {}
@@ -179,13 +179,14 @@ def process_reports_parse(url_formats = 'https://formats.saby.ru'):
             dict_result_title = parse_reports(soup, report_title, url_formats, name_title)
             dict_result.update(dict_result_title)
             # Вывод на стандратный поток вывода
-            write_report_data(dict_result, name_title)    
+            print_report_data(dict_result, name_title)    
         except Exception as e:
             print(f"Ошибка при обработке отчета {report_title}: {str(e)}")
             continue
     return dict_result
 
 if __name__ == '__main__':
+    # Можно запустить отдельно от всего проекта
     process_reports_parse()
 
 

app/scheduler.py

@@ -0,0 +1,22 @@
+from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.triggers.cron import CronTrigger
+import subprocess
+
+def run_parser(filename = 'main.py'):
+    print("Запуск main.py...")
+    subprocess.run(['python', filename])
+
+def launch_the_scheduler():
+    scheduler = BlockingScheduler() # Создание планировщика
+
+    # Каждый день в 6:00 утра запуск run_parser()
+    scheduler.add_job(run_parser, trigger=CronTrigger(hour=6, minute=00))
+
+    print("Планировщик запущен. Нажмите Ctrl+C для остановки.")
+
+    try:
+        scheduler.start()
+    except KeyboardInterrupt:
+        print("Планировщик остановлен")
+    except Exception as e:
+        print("Не непредвиденная ошибка: ", e)