Для тестовой развертки в докер контейнер
Лев
parent
4f8bf95efa
commit
7e42f59215
205
main.py
205
main.py
|
|
@ -1,202 +1,5 @@
|
||||||
import requests
|
import working_database
|
||||||
from bs4 import BeautifulSoup
|
import parse_saby
|
||||||
import re
|
|
||||||
|
|
||||||
# Используется два раза, создал отдельную переменную
|
|
||||||
URL_FORMATS = 'https://formats.saby.ru'
|
|
||||||
|
|
||||||
def parse_html(url=str):
|
|
||||||
"""
|
|
||||||
Фунция принимает строку URL, выполняет запрос.
|
|
||||||
Создает и возращает объект BeautifulSoup(HTML).
|
|
||||||
"""
|
|
||||||
# Запрос страницы
|
|
||||||
response = requests.get(url)
|
|
||||||
# Проверка статуса в ответе на запрос
|
|
||||||
if not(200 <= response.status_code <= 299):
|
|
||||||
# Повторный запрос
|
|
||||||
response = requests.get(url)
|
|
||||||
print("Ошибка при запросе: ", response.status_code)
|
|
||||||
return response.status_code
|
|
||||||
# Создание обьекта BeautifulSoup(HTML страница)
|
|
||||||
soup = BeautifulSoup(response.text, 'html.parser')
|
|
||||||
return soup
|
|
||||||
|
|
||||||
def parse_date_report(url=str):
|
|
||||||
"""
|
|
||||||
Функия ожидает list из URL ввида:
|
|
||||||
url = https://formats.saby.ru/report/fns/128513
|
|
||||||
Заходит по URL, запоминает дату.
|
|
||||||
Возвращает две строки ввида:
|
|
||||||
'date = 01.07.22'
|
|
||||||
'id_date = 5.01'
|
|
||||||
"""
|
|
||||||
# HTML report
|
|
||||||
soup = parse_html(url)
|
|
||||||
# Если не удалось отправить запрос, ворзвращаем URL и код ошибки.
|
|
||||||
# Они будут занесы в файл.
|
|
||||||
if soup == int:
|
|
||||||
return url, soup
|
|
||||||
# Поиск в HTML строки ввида: 'Действующий формат (с 10.01.23) 5.01'
|
|
||||||
div_element = soup.find('div', class_='controls-Dropdown__text')
|
|
||||||
# Извлекаем текст из элемента
|
|
||||||
text = div_element.get_text()
|
|
||||||
# Парсим нужные данные
|
|
||||||
# Уловия ловит: text = 'Действующий формат 02-04-2025 _2025_001'
|
|
||||||
if '_' in text:
|
|
||||||
regex = r'\w+\s\w+\s(\S+) (\S+)'
|
|
||||||
date, id_date = re.search(regex, text).groups()
|
|
||||||
return date, id_date
|
|
||||||
# Все остальные
|
|
||||||
else:
|
|
||||||
regex = r'(\d{2}\D\d{2}\D\d{2})(?:(?:.+)?\))? #?(\d+(?:\D\d+)?)'
|
|
||||||
date, id_date = re.search(regex, text).groups()
|
|
||||||
return date, id_date
|
|
||||||
|
|
||||||
def parse_reports(soup=BeautifulSoup, # HTML объект
|
|
||||||
report_title = str, # строка ввида: 'report/fns'
|
|
||||||
url_formats = str, # Строка ввида: 'https://formats.saby.ru'
|
|
||||||
name_title = str): # имя тайтла: 'fns'
|
|
||||||
"""
|
|
||||||
Достаются все необходимые данные, возвращаются в ввиде словаря:
|
|
||||||
{106538: ('fns', 'НД по косвенным налогам', '01.08.23', '5.04')}
|
|
||||||
"""
|
|
||||||
|
|
||||||
result_dict_data = {}
|
|
||||||
# Перебарает все URL, ищутся по тегу 'a'
|
|
||||||
for link in soup.find_all('a'):
|
|
||||||
# Ищет по тегу: href
|
|
||||||
href = link.get('href')
|
|
||||||
if f'{report_title}/' in href:
|
|
||||||
# id report
|
|
||||||
id = href.rstrip('/').split('/')[-1]
|
|
||||||
#URL report
|
|
||||||
url_report = f'{url_formats}{href}'
|
|
||||||
link = soup.find('a', href=href)
|
|
||||||
# Name report
|
|
||||||
span = link.find('span', class_="ProxySbisRu__registry-BrowserItem_typeName")
|
|
||||||
# Данные получены из url после парсинга
|
|
||||||
date, id_date = parse_date_report(url_report)
|
|
||||||
# Добавление всех данных в итоговый словарь
|
|
||||||
result_dict_data.update({id: (name_title, span.text, date, id_date)})
|
|
||||||
|
|
||||||
return result_dict_data
|
|
||||||
|
|
||||||
def write_report_data(filename, dict_name = dict, name_title = str):
|
|
||||||
"""
|
|
||||||
Сохраняем запись, каждая запись с новой строки:
|
|
||||||
'ключ: значение'
|
|
||||||
'ключ: значение'
|
|
||||||
...
|
|
||||||
"""
|
|
||||||
#Блок для красивого форлмения файла
|
|
||||||
|
|
||||||
# Вычисляем количество подчёркиваний слева и справа
|
|
||||||
def center_text():
|
|
||||||
"""
|
|
||||||
Функия парсит сколько нужно подчеркивания,
|
|
||||||
что бы центролизовать текст в строке с этими подчеркиваниями
|
|
||||||
Возвращает две строки одна для начало блока записей тайтал, вторая конца блока.
|
|
||||||
"""
|
|
||||||
start_str = f'START_{name_title}'
|
|
||||||
end_str = f'END_{name_title}'
|
|
||||||
dash = 100
|
|
||||||
list_result = []
|
|
||||||
# сначал вернеться dash_start, потом end_start.
|
|
||||||
for text in start_str, end_str:
|
|
||||||
remaining_space = dash - len(text)
|
|
||||||
left = remaining_space // 2
|
|
||||||
right = remaining_space - left # Чтобы учесть нечётную разницу
|
|
||||||
list_result.append('_' * left + text + '_' * right)
|
|
||||||
return list_result[:2]
|
|
||||||
dash_start, dash_end = center_text()
|
|
||||||
|
|
||||||
#Конец блока
|
|
||||||
#Запись в файл с красивым офрмление в виде нижнего подчеркивания
|
|
||||||
with open(filename, 'a', encoding='utf-8') as f:
|
|
||||||
f.write(f'\n{dash_start}\n')
|
|
||||||
for key, value in dict_name.items():
|
|
||||||
str_k_v = f'{key}: {value}\n'
|
|
||||||
f.write(str_k_v)
|
|
||||||
f.write(f'{dash_end}\n')
|
|
||||||
|
|
||||||
def search_title():
|
|
||||||
"""
|
|
||||||
Функция ищет все тайтлы на странице formats.saby.ru/report.
|
|
||||||
|
|
||||||
Возвращает:
|
|
||||||
Список URL-путей, например: ['/report/fns', '/report/example', ...]
|
|
||||||
|
|
||||||
Исключения:
|
|
||||||
ValueError: Если запрос к странице завершился с ошибкой (неверный статус).
|
|
||||||
"""
|
|
||||||
# url по которому в котором будет происходить поиск,
|
|
||||||
# Используется только в этой функции
|
|
||||||
url_format_report = 'https://formats.saby.ru/report'
|
|
||||||
|
|
||||||
# Получаем HTML-cтраницу
|
|
||||||
html = parse_html(url_format_report)
|
|
||||||
|
|
||||||
# Проверяем, что html не является кодом ошибки (int)
|
|
||||||
if isinstance(html, int):
|
|
||||||
error_message = f'Ошибка при запросе {url_format_report}: {html}'
|
|
||||||
raise Exception(error_message)
|
|
||||||
|
|
||||||
# Лист в который будут заноситься тайтлы
|
|
||||||
report_urls = set()
|
|
||||||
|
|
||||||
# Ищем все ссылки <a> с href, соответствующие шаблону /report/{title}
|
|
||||||
for link in html.find_all('a', href=True):
|
|
||||||
# Ищем по тегу: href
|
|
||||||
href = link['href']
|
|
||||||
# Проверям что href содержит: /report/{title}
|
|
||||||
if re.search(r'\/report\/(\w+)$', href):
|
|
||||||
report_urls .add(href)
|
|
||||||
return report_urls
|
|
||||||
|
|
||||||
# Имя файла в который будет записан результат кода
|
|
||||||
filename_save = 'ReportData.txt'
|
|
||||||
def clear_report_data_file():
|
|
||||||
# Удаляем старые записи, что бы записать актульные
|
|
||||||
with open(filename_save, 'w') as f:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def process_reports_parse():
|
|
||||||
"""
|
|
||||||
Функция про бегается по каждому тайтлу.
|
|
||||||
Для всех записей(reports) выполняется запрос HTML страницы,
|
|
||||||
которая парситься в объект BeautifulSoup(HTML страница).
|
|
||||||
Из это обьекта достаются не боходимые данные,
|
|
||||||
которые записываются в текстовый файл.
|
|
||||||
"""
|
|
||||||
# Очищаем файл перед записью
|
|
||||||
clear_report_data_file()
|
|
||||||
|
|
||||||
# Лист имеет вид: ['/report/fns', '/report/sfr'...]
|
|
||||||
list_title = search_title()
|
|
||||||
|
|
||||||
for report_title in list_title:
|
|
||||||
try:
|
|
||||||
# Получаем имя тайтла через парсинг
|
|
||||||
name_title = report_title.rstrip('/').split('/')[-1]
|
|
||||||
|
|
||||||
# URL на конкретный title
|
|
||||||
url_title = f'{URL_FORMATS}{report_title}'
|
|
||||||
|
|
||||||
# Объект HTML, конкретного title
|
|
||||||
soup = parse_html(url_title)
|
|
||||||
|
|
||||||
# Словарь с нужными данными
|
|
||||||
dict_result = parse_reports(soup, report_title, URL_FORMATS, name_title)
|
|
||||||
|
|
||||||
# Запись данных в текстовый файл
|
|
||||||
write_report_data(filename_save, dict_result, name_title)
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Ошибка при обработке отчета {report_title}: {str(e)}")
|
|
||||||
continue
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
process_reports_parse()
|
|
||||||
|
|
||||||
|
|
||||||
|
working_database.connect_hvac()
|
||||||
|
parse_saby.process_reports_parse()
|
||||||
|
|
@ -0,0 +1,202 @@
|
||||||
|
import requests
|
||||||
|
from bs4 import BeautifulSoup
|
||||||
|
import re
|
||||||
|
|
||||||
|
def parse_html(url: str):
|
||||||
|
"""
|
||||||
|
Фунция принимает строку URL, выполняет запрос.
|
||||||
|
Создает и возращает объект BeautifulSoup(HTML).
|
||||||
|
"""
|
||||||
|
# Запрос страницы
|
||||||
|
response = requests.get(url)
|
||||||
|
# Проверка статуса в ответе на запрос
|
||||||
|
if not(200 <= response.status_code <= 299):
|
||||||
|
# Повторный запрос
|
||||||
|
response = requests.get(url)
|
||||||
|
print("Ошибка при запросе: ", response.status_code)
|
||||||
|
return response.status_code
|
||||||
|
# Создание обьекта BeautifulSoup(HTML страница)
|
||||||
|
soup = BeautifulSoup(response.text, 'html.parser')
|
||||||
|
return soup
|
||||||
|
|
||||||
|
def parse_date_report(url: str):
|
||||||
|
"""
|
||||||
|
Функия ожидает list из URL ввида:
|
||||||
|
url = https://formats.saby.ru/report/fns/128513
|
||||||
|
Заходит по URL, запоминает дату.
|
||||||
|
Возвращает две строки ввида:
|
||||||
|
'date = 01.07.22'
|
||||||
|
'id_date = 5.01'
|
||||||
|
"""
|
||||||
|
# HTML report
|
||||||
|
soup = parse_html(url)
|
||||||
|
# Если не удалось отправить запрос, ворзвращаем URL и код ошибки.
|
||||||
|
# Они будут занесы в файл.
|
||||||
|
if soup == int:
|
||||||
|
return url, soup
|
||||||
|
# Поиск в HTML строки ввида: 'Действующий формат (с 10.01.23) 5.01'
|
||||||
|
div_element = soup.find('div', class_='controls-Dropdown__text')
|
||||||
|
# Извлекаем текст из элемента
|
||||||
|
text = div_element.get_text()
|
||||||
|
# Парсим нужные данные
|
||||||
|
# Уловия ловит: text = 'Действующий формат 02-04-2025 _2025_001'
|
||||||
|
if '_' in text:
|
||||||
|
regex = r'\w+\s\w+\s(\S+) (\S+)'
|
||||||
|
date, id_date = re.search(regex, text).groups()
|
||||||
|
return date, id_date
|
||||||
|
# Все остальные
|
||||||
|
else:
|
||||||
|
regex = r'(\d{2}\D\d{2}\D\d{2})(?:(?:.+)?\))? #?(\d+(?:\D\d+)?)'
|
||||||
|
date, id_date = re.search(regex, text).groups()
|
||||||
|
return date, id_date
|
||||||
|
|
||||||
|
def parse_reports(soup=BeautifulSoup, # HTML объект
|
||||||
|
report_title = str, # строка ввида: 'report/fns'
|
||||||
|
url_formats = str, # Строка ввида: 'https://formats.saby.ru'
|
||||||
|
name_title = str): # имя тайтла: 'fns'
|
||||||
|
"""
|
||||||
|
Достаются все необходимые данные, возвращаются в ввиде словаря:
|
||||||
|
{106538: ('fns', 'НД по косвенным налогам', '01.08.23', '5.04')}
|
||||||
|
"""
|
||||||
|
|
||||||
|
result_dict_data = {}
|
||||||
|
# Перебарает все URL, ищутся по тегу 'a'
|
||||||
|
for link in soup.find_all('a'):
|
||||||
|
# Ищет по тегу: href
|
||||||
|
href = link.get('href')
|
||||||
|
if f'{report_title}/' in href:
|
||||||
|
# id report
|
||||||
|
id = href.rstrip('/').split('/')[-1]
|
||||||
|
#URL report
|
||||||
|
url_report = f'{url_formats}{href}'
|
||||||
|
link = soup.find('a', href=href)
|
||||||
|
# Name report
|
||||||
|
span = link.find('span', class_="ProxySbisRu__registry-BrowserItem_typeName")
|
||||||
|
# Данные получены из url после парсинга
|
||||||
|
date, id_date = parse_date_report(url_report)
|
||||||
|
# Добавление всех данных в итоговый словарь
|
||||||
|
result_dict_data.update({id: (name_title, span.text, date, id_date)})
|
||||||
|
|
||||||
|
return result_dict_data
|
||||||
|
|
||||||
|
def write_report_data(filename, dict_name = dict, name_title = str):
|
||||||
|
"""
|
||||||
|
Сохраняем запись, каждая запись с новой строки:
|
||||||
|
'ключ: значение'
|
||||||
|
'ключ: значение'
|
||||||
|
...
|
||||||
|
"""
|
||||||
|
#Блок для красивого офорлмения файла
|
||||||
|
|
||||||
|
# Вычисляем количество подчёркиваний слева и справа
|
||||||
|
def center_text():
|
||||||
|
"""
|
||||||
|
Функия парсит сколько нужно подчеркивания,
|
||||||
|
что бы центролизовать текст в строке с этими подчеркиваниями
|
||||||
|
Возвращает две строки одна для начало блока записей тайтал, вторая конца блока.
|
||||||
|
"""
|
||||||
|
start_str = f'START_{name_title}'
|
||||||
|
end_str = f'END_{name_title}'
|
||||||
|
dash = 100
|
||||||
|
list_result = []
|
||||||
|
# сначал вернеться dash_start, потом end_start.
|
||||||
|
for text in start_str, end_str:
|
||||||
|
remaining_space = dash - len(text)
|
||||||
|
left = remaining_space // 2
|
||||||
|
right = remaining_space - left # Чтобы учесть нечётную разницу
|
||||||
|
list_result.append('_' * left + text + '_' * right)
|
||||||
|
return list_result[:2]
|
||||||
|
dash_start, dash_end = center_text()
|
||||||
|
|
||||||
|
#Конец блока
|
||||||
|
#Запись в файл с красивым офрмление в виде нижнего подчеркивания
|
||||||
|
with open(filename, 'a', encoding='utf-8') as f:
|
||||||
|
f.write(f'\n{dash_start}\n')
|
||||||
|
for key, value in dict_name.items():
|
||||||
|
str_k_v = f'{key}: {value}\n'
|
||||||
|
f.write(str_k_v)
|
||||||
|
f.write(f'{dash_end}\n')
|
||||||
|
|
||||||
|
def search_title():
|
||||||
|
"""
|
||||||
|
Функция ищет все тайтлы на странице formats.saby.ru/report.
|
||||||
|
|
||||||
|
Возвращает:
|
||||||
|
Список URL-путей, например: ['/report/fns', '/report/example', ...]
|
||||||
|
|
||||||
|
Исключения:
|
||||||
|
ValueError: Если запрос к странице завершился с ошибкой (неверный статус).
|
||||||
|
"""
|
||||||
|
# url по которому в котором будет происходить поиск,
|
||||||
|
url_format_report = 'https://formats.saby.ru/report'
|
||||||
|
|
||||||
|
# Получаем HTML-cтраницу
|
||||||
|
html = parse_html(url_format_report)
|
||||||
|
|
||||||
|
# Проверяем, что html не является кодом ошибки (int)
|
||||||
|
if isinstance(html, int):
|
||||||
|
error_message = f'Ошибка при запросе {url_format_report}: {html}'
|
||||||
|
raise Exception(error_message)
|
||||||
|
|
||||||
|
# < в который будут заноситься тайтлы
|
||||||
|
report_urls = set()
|
||||||
|
|
||||||
|
# Ищем все ссылки <a> с href, соответствующие шаблону /report/{title}
|
||||||
|
for link in html.find_all('a', href=True):
|
||||||
|
# Ищем по тегу: href
|
||||||
|
href = link['href']
|
||||||
|
# Проверям что href содержит: /report/{title}
|
||||||
|
if re.search(r'\/report\/(\w+)$', href):
|
||||||
|
report_urls .add(href)
|
||||||
|
return report_urls
|
||||||
|
|
||||||
|
# Имя файла в который будет записан результат кода
|
||||||
|
filename_save = 'ReportData.txt'
|
||||||
|
def clear_report_data_file():
|
||||||
|
"""
|
||||||
|
Удаляем старые записи, что бы записать актульные
|
||||||
|
"""
|
||||||
|
with open(filename_save, 'w') as f:
|
||||||
|
pass
|
||||||
|
|
||||||
|
def process_reports_parse():
|
||||||
|
"""
|
||||||
|
Функция пробегается по каждому тайтлу.
|
||||||
|
Для всех записей(reports) выполняется запрос HTML страницы,
|
||||||
|
которая парситься в объект BeautifulSoup(HTML страница).
|
||||||
|
Из это обьекта достаются не обходимые данные,
|
||||||
|
которые записываются в текстовый файл.
|
||||||
|
"""
|
||||||
|
# Очищаем файл перед записью
|
||||||
|
clear_report_data_file()
|
||||||
|
|
||||||
|
# Лист имеет вид: ['/report/fns', '/report/sfr'...]
|
||||||
|
list_title = search_title()
|
||||||
|
|
||||||
|
for report_title in list_title:
|
||||||
|
try:
|
||||||
|
# используется для создание полных URL
|
||||||
|
url_formats = 'https://formats.saby.ru'
|
||||||
|
# Получаем имя тайтла через парсинг
|
||||||
|
name_title = report_title.rstrip('/').split('/')[-1]
|
||||||
|
|
||||||
|
# URL на конкретный title
|
||||||
|
url_title = f'{url_formats}{report_title}'
|
||||||
|
|
||||||
|
# Объект HTML, конкретного title
|
||||||
|
soup = parse_html(url_title)
|
||||||
|
|
||||||
|
# Словарь с нужными данными
|
||||||
|
dict_result = parse_reports(soup, report_title, url_formats, name_title)
|
||||||
|
|
||||||
|
# Запись данных в текстовый файл
|
||||||
|
write_report_data(filename_save, dict_result, name_title)
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Ошибка при обработке отчета {report_title}: {str(e)}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
process_reports_parse()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
beautifulsoup4==4.13.4
|
beautifulsoup4==4.13.4
|
||||||
certifi==2025.7.14
|
certifi==2025.7.14
|
||||||
charset-normalizer==3.4.2
|
charset-normalizer==3.4.2
|
||||||
|
hvac==2.3.0
|
||||||
idna==3.10
|
idna==3.10
|
||||||
requests==2.32.4
|
requests==2.32.4
|
||||||
soupsieve==2.7
|
soupsieve==2.7
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,17 @@
|
||||||
|
import os
|
||||||
|
import hvac
|
||||||
|
|
||||||
|
def connect_hvac():
|
||||||
|
client = hvac.Client(
|
||||||
|
url='https://vlt.dataekb.ru:8222',
|
||||||
|
token=os.environ.get('VAULT_TOKEN'),
|
||||||
|
)
|
||||||
|
read_response = client.secrets.kv.v2.read_secret_version(path='oracledb', mount_point='kv')
|
||||||
|
pw = read_response['data']['data']['password']
|
||||||
|
un = read_response['data']['data']['user']
|
||||||
|
cs = read_response['data']['data']['cs']
|
||||||
|
|
||||||
|
print(read_response)
|
||||||
|
print(pw)
|
||||||
|
print(un)
|
||||||
|
print(cs)
|
||||||
Loading…
Reference in New Issue