sm/flask_driver/parser_driver.py - master - MaxIs/savemoney

sm/flask_driver/parser_driver.py
473 строки21 KB
zalma2006
добавлен flask для работы с selenium в отдельном сервисе, добавлены точки API, добавлена загрузка отчётов из источника, подробнее в HISTORY.md
15 мар 2026, 19:36
15 мар 2026, 19:36ae3ea09
import inspect
import json
import logging
import shutil
import time
import traceback
import urllib
from io import StringIO
from pathlib import Path
from urllib.parse import unquote

import numpy as np
import pandas as pd
from selenium import webdriver
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.remote.webelement import WebElement
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver import ActionChains
from sql_worker import sql_worker
from selenium.common.exceptions import TimeoutException
from cust_funcs import get_simple_json2resp

REPORT_TYPE = {
    2: "Годовая",
    3: "Бухгалтерская (финансовая)",
    4: "Консолидированная",
    5: "Отчеты эмитента",
}

REPORT_COLS_TYPE = {
    2: ['№', 'тип_документа', 'отч_год', 'дата_утверждения', 'дата_основания', 'дата_размещения', 'файл', 'no'],
    3: ['№', 'тип_документа', 'отч_период', 'дата_основания', 'дата_размещения', 'файл', 'no'],
    4: ['№', 'тип_документа', 'отч_период', 'дата_основания', 'дата_размещения', 'файл', 'no'],
    5: ['№', 'тип_документа', 'отч_период', 'дата_основания', 'дата_размещения', 'файл', 'no'],
}


class SaveMoneyDriver:
    # возможные методы поиска на странице
    MAYBE_SEARCH_TYPE = {By.XPATH}

    BASE_ELEMENTS = {
        "inn_input": {
            'xpath': '//*[@id="query"] | //*[@id="textfieldCompany"]',
            'element': None
        },
        "sendButton": {
            'xpath': '//*[@id="sendButton"]',
            'element': None
        },
    }

    def __init__(self, timeout: int = 10, headless: bool = True):
        self.curr_path = Path(__file__).parent
        self.driver_timeout = timeout
        self.sql_w = sql_worker  # для чтения скаченных файлов
        self.headless_driver = headless
        self.base_elements = self.BASE_ELEMENTS.copy()
        self.elements = self.BASE_ELEMENTS.copy()
        self.base_url = 'https://www.e-disclosure.ru'
        self.report_suffix = '/portal/company.aspx?id='
        self.report_url = self.base_url + self.report_suffix
        self.logger = logging.getLogger(self.__class__.__name__)
        self.current_url = self.base_url
        self.geckodriver_path = None
        self.options = None
        self.driver = None
        self.wait = None
        self.reports = dict()
        self.prefs = None  # настройки скачивания файлов

    def _set_geckodriver_path(self):
        """Найти драйвер"""
        gl = self.curr_path.absolute()
        geckodriver_path = [x for x in gl.glob('**/geckodriver*')]
        if len(geckodriver_path) == 0:
            raise FileNotFoundError('No geckodriver found! Download geckodriver first.')
        elif len(geckodriver_path) > 1:
            raise IndexError('More than one geckodriver found.')
        else:
            geckodriver_path = geckodriver_path[0]
        self.geckodriver_path = str(geckodriver_path)

    def find_element(self, by: str, search: str):
        el = self.wait.until(EC.visibility_of_element_located((by, search)))
        return el

    def update_base_elements(self, by: str):
        """Обновляем базовые элементы на странице"""
        self.current_url = self.driver.current_url
        if by not in self.MAYBE_SEARCH_TYPE:
            raise ValueError(f"{by} maybe once in {self.MAYBE_SEARCH_TYPE}")
        for n, info in self.base_elements.items():
            try:
                el = self.find_element(by, info[by])
                self.wait.until(EC.element_to_be_clickable(el))
                self.base_elements[n]['element'] = el
            except Exception as e:
                print(e)
                print(f"{n}, {by}, {info[by]} not found")

        self.update_elements_from_base()

    def update_elements_from_base(self):
        """Обновляем список элементов"""
        for n, info in self.base_elements.items():
            self.elements[n]['element'] = info['element']

    def add_new_elements(self, key, val, by):
        """Добавить элемент в self.elements"""
        try:
            el = self.find_element(by, val[by])
            val['element'] = el
            self.elements[key] = val
        except Exception as e:
            print(e)
            print(f"{key}, {by}, {val[by]} not found")
        self.elements[key] = val

    def get_element_obj(self, name) -> WebElement | None:
        """Получить элемент из self.elements[name]"""
        return self.elements[name]['element']

    def find_idx(self, inn: str):
        """Найдем индекс компании в отчётности"""
        self.update_base_elements(By.XPATH)
        inn_field, send_button = self.get_element_obj('inn_input'), self.get_element_obj('sendButton')
        inn_field.clear()
        inn_field.send_keys(str(inn))
        self.hide_page_loader()
        send_button.click()
        self.add_new_elements("searchResults", {'xpath': '//*[@id="searchResults"]', 'element': None},
                              By.XPATH)
        self.update_base_elements(By.XPATH)
        results = self.get_element_obj('searchResults')
        idx_company = None
        if results is not None:
            if results.text.lower().startswith('найдено сообщений'):
                el1 = results.find_element(By.XPATH, '//a[contains(@href, "/portal/company.aspx?id")]')
                href = el1.get_attribute('href')
                idx_company = href.split(self.report_suffix)[-1]
            elif results.text.lower().startswith('ничего не найдено'):
                idx_company = '-1'
        return idx_company

    def get_report_types(self, idx, types) -> list:
        """Сопоставляем типы отчётов"""
        elems = self.driver.find_elements(
            by=By.XPATH,
            value=f'//*[@id="cssmenu"]//a[contains(@href, "/otchyotnost-kompanii")]/'
                  f'following::a[contains(@href, "portal/files.aspx?id={idx}&type=")]')
        if len(elems) > 0:
            elems = [(int(x.get_attribute('href').split('type=')[1][0])
                      if x.get_attribute('href').split('type=')[1][0].isdigit() else -1,
                      x.get_attribute('href'), x) for x in elems]
            elems = [x for x in elems if x[0] in types]
        return elems

    def hide_page_loader(self):
        """Скроем мешающий элемент"""
        load_name = 'page-loader'
        loader = self.driver.find_element(by=By.ID, value=load_name)

        if loader:
            print(f"loader exists!")
            script_str = 'document.querySelectorAll(\'div[class^="page-loader"]\').forEach(e => {e.style.display = "none"});'
            self.driver.execute_script(script_str)
            WebDriverWait(self.driver, 5).until(EC.element_to_be_clickable(self.get_element_obj('sendButton')))

    def get_reports(self, inn, r_types: set, idx_company: int | None = None):
        """
        Получим ссылки на отчёты, элементы которые совпадают с types и данные по ним
        """
        if idx_company is None:
            idx_company = self.find_idx(inn)
        reports = []
        reports_tmp = dict()
        if idx_company is not None and idx_company != '-1':
            report_url = self.report_url + idx_company
            self.driver.get(report_url)
            self.add_new_elements("otchyotnostKompanii",
                                  {'xpath': '//*[@id="cssmenu"]//a[contains(@href, "/otchyotnost-kompanii")]',
                                   'element': None}, By.XPATH)
            el = self.get_element_obj('otchyotnostKompanii')
            if el is not None:
                # наводим курсор, чтобы увидеть элементы
                ActionChains(self.driver).move_to_element(el).perform()
                reports = self.get_report_types(idx_company, r_types)
        elif idx_company == '-1':
            reports_tmp = {'-1': 'у компании отсутствует отчётность в источнике'}
        if len(reports) > 0:
            reports_tmp = {x[0]: {'rtype': x[0], 'link': x[1]} for x in reports}
            for t, link, el in reports:
                self.driver.get(link)
                report = self.find_element(By.XPATH, '//table[contains(@class, "files-table")]')
                if report is not None:
                    files_link = report.find_elements(By.XPATH, '//a[@class="file-link" and '
                                                                'contains(@href, "portal/FileLoad.ashx?Fileid")]')
                    files_link = [x.get_attribute('href') for x in files_link]
                    df = pd.read_html(StringIO(str(report.get_attribute('outerHTML'))))[0]
                    df.rename(columns=dict(zip(df.columns, REPORT_COLS_TYPE[t])), inplace=True)
                    df = self.clear_df(df)
                    for idx, val in enumerate(df['файл']):
                        df.at[idx, 'файл'] = files_link[idx]
                    reports_tmp[t]['need_doc'] = self.get_need_doc(df, t)
                    reports_tmp[t]['df'] = self.convert_datetime2str(df).to_dict('list')
        self.driver.get(self.base_url)
        return reports_tmp

    @staticmethod
    def convert_datetime2str(df):
        df = df.copy()
        dcol = [col for col in df.columns if str(col).lower().startswith('дата_')]
        for col in dcol:
            df[col] = df[col].astype(str)
        return df

    @staticmethod
    def get_need_doc(df, r_type: int):
        """Получить данные по последнему отчёту"""
        tdf = df.copy()
        if r_type == 2:
            tdf = tdf[tdf['тип_документа'].str.lower().str.startswith('годовой')]
            tdf['отч_год'] = tdf['отч_год'].astype(int)
            tdf['отч_месяц'] = 'годовой отчёт'
            tdf.sort_values(by=['отч_год'], ascending=False, inplace=True)
        elif r_type > 2:
            if r_type == 3:
                tdf = tdf[tdf['тип_документа'].str.lower().str.contains('бухгалтерская отчетность')]
            elif r_type == 4:
                tdf = tdf[tdf['тип_документа'].str.lower().str.contains('консолидированная финансовая отчетность')]
            elif r_type == 5:
                tdf = tdf[tdf['тип_документа'].str.lower().str.contains('отчет эмитента')]
            # почистим отчётный период от запятых лишних пробелов и разделим на 2 столбца отч_год, отч_месяц для
            # сортировки
            tdf['отч_период'] = tdf['отч_период'].str.replace(',', '', regex=False).str.replace(' +', ' ',
                                                                                                regex=True).str.strip()
            time_df = tdf['отч_период'].copy().str.split(' ', expand=True)[[0, 1]]
            time_df.rename(columns={0: 'отч_год', 1: 'отч_месяц'}, inplace=True)
            time_df = time_df.convert_dtypes(convert_string=False, convert_integer=True, convert_boolean=False,
                                             convert_floating=False)
            tdf = pd.concat([tdf, time_df], axis=1)
            tdf.sort_values(by=['отч_год', 'отч_месяц'], ignore_index=True, ascending=False, inplace=True)
        tdf = SaveMoneyDriver.convert_datetime2str(tdf)
        need_doc = tdf.loc[0, :]
        need_doc = need_doc.to_dict()
        return need_doc

    @staticmethod
    def clear_df(df):
        """Очищаем таблицу и приводим в правильную форму"""

        def is_date(dval):
            return pd.to_datetime(dval, dayfirst=True, errors='coerce') is not pd.NaT

        df = df.copy()
        dropidx = set()
        for col in df.columns:
            if col.startswith('дата'):
                df[col] = df[col].apply(lambda x: x if is_date(x) else pd.NaT)
                df[col] = pd.to_datetime(df[col], dayfirst=True)
                dropidx.update(set(df[df[col].isna()].index))
            if col.startswith('отч_год'):
                df[col] = df['отч_год'].apply(lambda val: val if str(val).isdigit() else np.nan)
                df[col] = pd.to_numeric(df[col])
                dropidx.update(set(df[df[col].isna()].index))

        for idx, val in enumerate(df['тип_документа']):
            if pd.isna(val):
                dropidx.add(idx)
        df = df.drop(index=list(dropidx))
        df.reset_index(drop=True, inplace=True)
        return df

    def set_browser_prefs(self, download_dir=None):
        """Установим загрузку по умолчанию"""
        if download_dir is None:
            path = self.curr_path
            path = path / 'downloads'
            if not path.exists():
                path.mkdir(parents=True)
            download_dir = str(path.absolute())
        mime_app_download = ['application/octet-stream', 'application/pdf', 'text/csv', 'application/zip',
                             'application/x-gzip']
        mime_app_download = ','.join(mime_app_download)
        self.prefs = {
            "browser.download.folderList": 2,  # 2 означает, что будет использоваться custom path
            "browser.download.dir": download_dir,
            "browser.download.useDownloadDir": True,
            "browser.helperApps.neverAsk.saveToDisk": mime_app_download,
            "browser.download.manager.showWhenStarting": False,
            "pdfjs.disabled": True,  # Отключаем встроенный PDF-просмотрщик
            "browser.helperApps.alwaysAsk.force": False,
            "browser.download.manager.alertOnEXEOpen": False,
            "browser.download.manager.focusWhenStarting": False,
            "browser.download.manager.useWindow": False,
            "browser.download.manager.showAlertOnComplete": False,
            "browser.download.manager.closeWhenDone": False,
        }

    def get_download_filename(self, timeout: int = 30):
        """Получить название файла из скаченных файлов"""
        req = get_simple_json2resp()
        start_time = time.time()
        time.sleep(0.2)  # первоначальная задержка на запись в БД
        while time.time() - start_time < timeout:
            if time.time() - start_time > timeout:
                req['errors'].append('timeout download file')
                break
            df = self.sql_w.exec_query_from_table(
                """SELECT ma.place_id,
                          ma.anno_attribute_id,
                          ma.content,
                          ma.dateAdded,
                          a.name as destination
                   FROM (SELECT place_id, anno_attribute_id, content, dateAdded
                         FROM moz_annos
                         WHERE place_id = (SELECT MAX(place_id) FROM moz_annos)) as ma
                            LEFT JOIN moz_anno_attributes as a
                                      ON ma.anno_attribute_id = a.id;""")
            # файл скачивается
            if df.shape[0] == 1:
                time.sleep(0.3)
                continue
            # файл скачен
            elif df.shape[0] == 2:
                fname = df[df['destination'] == 'downloads/destinationFileURI']['content'].values[0].split('///')[-1]
                fname = Path(unquote(fname))
                state = json.loads(df[df['destination'] == 'downloads/metaData']['content'].values[0])
                state['download_time_sec'] = (state['endTime'] - (
                        df[df['destination'] == 'downloads/destinationFileURI']['dateAdded'].values[
                            0] / 1000)) / 1000
                req['message'] = 'success'
                req['filename'] = str(fname)
                req['download_time_sec'] = state['download_time_sec']
                break
            # файл не скачивается!
            else:
                req['errors'].append('Файл не скачивается!')
                break

        print(f"DEBUG get_download_filename req = {req}")
        if len(req['errors']) > 0:
            req['errors'] = '||'.join(req['errors'])
        return req

    def get_path_profile(self):
        """Получить путь где храняться temp файлы"""
        return Path(self.driver.capabilities['moz:profile'])

    def download_file_url(self, url, page_load_timeout: int = 1):
        """Скачаем файл по ссылке и вернём путь до файла"""
        req = get_simple_json2resp()
        self.driver.set_page_load_timeout(page_load_timeout)
        err = None
        try:
            res = self.driver.get(url)
            if hasattr(res, 'status_code'):
                status_code = res.status_code
                if status_code >= 400:
                    req['status_code'] = status_code
                    req['errors'].append(res.content[:500])
        except TimeoutException as e:
            err = e
        except Exception as e:
            err = e
            req['errors'].append(traceback.format_exc())
        finally:
            req['url'] = url
            req['message'] = 'success' if err is None or err != TimeoutException else err
            self.driver.set_page_load_timeout(self.driver_timeout)
        if len(req['errors']) == 0:
            req_df = self.get_download_filename(timeout=50)
            if len(req_df['errors']) == 0:
                req['filename'] = req_df['filename']
                req['download_time_sec'] = req_df['download_time_sec']
            else:
                req['errors'].append(req_df['errors'])
        print(f"DEBUG download_file_url req = {req}")
        if len(req['errors']) == 0:
            req['message'] = 'success'
        return req

    def set_options(self):
        """Установка опций драйвера"""
        self.options = Options()
        for k, v in self.prefs.items():
            self.options.set_preference(k, v)
        self.options.binary_location = self.geckodriver_path
        if self.headless_driver:
            self.options.add_argument("--headless")

    def set_driver(self):
        self.driver = webdriver.Firefox(options=self.options)
        self.driver.set_page_load_timeout(self.driver_timeout)
        self.wait = WebDriverWait(self.driver, self.driver_timeout)

    def set_sql_conn(self):
        path = Path(self.driver.capabilities['moz:profile'])
        place_sql_path = [x for x in path.glob('*places.sqlite')]
        if len(place_sql_path) == 1:
            place_sql_path = str(place_sql_path[0])
            self.sql_w.set_con_string(place_sql_path)
        else:
            err = f"не найдена база для чтения загрузок place_sql_path = {place_sql_path}"
            self.logger.exception(err)
            raise ValueError(err)

    def inicialize(self):
        """Установка драйвера и настроек к нему"""
        self._set_geckodriver_path()
        self.set_browser_prefs()
        self.set_options()
        self.set_driver()
        self.driver.get(self.base_url)
        self.set_sql_conn()

    def _clear_moz_profile(self):
        """Используется только при удалении драйвера, проводится очистка папки moz:profile"""
        frame = inspect.currentframe()
        caller = frame.f_back.f_code.co_name
        if caller == '__del__':
            dir_name = Path(self.driver.capabilities['moz:profile'])
            if dir_name.exists():
                shutil.rmtree(dir_name)

    def __del__(self):
        del self.sql_w
        self.driver.quit()
        self._clear_moz_profile()


class ReportData:
    """Класс отчёта для компании"""

    def __init__(self):
        self.rtype = None
        self.name = None
        self.df = None
        self.need_doc = None

    def get_name_byid(self):
        """Получить название отчёта по его id"""
        if isinstance(self.rtype, int):
            pass


class ReportsMetaData:
    def __init__(self):
        self.rtypes = REPORT_TYPE


class ReportsData(ReportsMetaData):
    """Класс отчётностей всех эмитентов, пока не знаю зачем, но может пригодится"""

    def __init__(self):
        super().__init__()
        self.report_data = dict()  # отчёты эмитента


class Company:
    """Класс компании по отчётности"""

    def __init__(self, inn: str):
        self.inn = inn
        self.idx_company = None
        self.reports = None

    def set_idx_company(self, driver: SaveMoneyDriver):
        if isinstance(driver, SaveMoneyDriver):
            self.idx_company = driver.find_idx(self.inn)