modules/data/processing.py - master - susub/report

modules/data/processing.py
258 строк10 KB
Александр Куклин
fix percentage of input control
19 мар 2026, 05:49
19 мар 2026, 05:499e82703
import pandas as pd
import re
from pprint import pprint

from modules.massage.massager import print_info


def selecting_sheets(params: dict,
                     title: str,
                     search_list: list) -> object:
    """
    Функция выбора листов по названию
    :param params: словарь с параметрами
    :param title: название титула
    :param search_list: список слов для поиска
    :return: список листов
    """

    sheets = []
    try:
        for sheet_name in params['xl'][title].sheet_names:
            for word in search_list:
                if sheet_name.lower().find(word) != -1:
                    sheets.append(sheet_name)
    except KeyError:
        sheets = None

    return sheets


def s_control(row: pd.DataFrame,
              inspection: str,
              fact: str,
              note=None) -> pd.DataFrame:
    """
    Функция расчета сданных по ЗНИ
    :param row: строка DataFrame
    :param inspection: название столбца с инспекцией
    :param fact: название столбца с фактически выполненными объёмами
    :param note: название столбца с примечанием
    :return: строка DataFrame с новыми столбцами
    """

    if str(row[inspection]).find('-') != -1:
        row['Сдано по ЗНИ'] = row[fact]
        row['ЗНИ не подписано'] = 0
    elif row[inspection] != 0:
        row['Сдано по ЗНИ'] = 0
        row['ЗНИ не подписано'] = row[fact]
    else:
        row['Сдано по ЗНИ'] = 0
        row['ЗНИ не подписано'] = 0
    if note is not None:
        row[note] = ' '.join(str(row[note]).strip().split(' '))
        if row[note] == 'не наш монтаж':
            row['Смонтировано не нами'] = 1
        else:
            row['Смонтировано не нами'] = 0

    return row


def find_trays(row: pd.DataFrame,
               col: str) -> pd.DataFrame:
    """
    Функция поиска лотков в таблице
    :param row: строка DataFrame
    :param col: название столбца с данными о лотках
    :return: строка DataFrame с новым столбцом
    """

    if re.findall(r'[Лл]оток', row[col]):
        if re.findall(r'[BbВ]=\d{2,3}', row[col]) and \
                re.findall(r'[HhН]=\d{2,3}', row[col]):
            b = re.findall(r'[BbВ]=\d{2,3}', row[col])
            h = re.findall(r'[HhН]=\d{2,3}', row[col])
            row['Лоток'] = 'Лоток ' +\
                           re.findall(r'\d{2,3}', b[0])[0] +\
                           'x' +\
                           re.findall(r'\d{2,3}', h[0])[0]
        if re.findall(r'\d{2,3}×\d{2,3}', row[col]):
            row['Лоток'] = 'Лоток ' +\
                'x'.join(re.findall(r'\d{2,3}×\d{2,3}',
                                    row[col])[0].split('×'))
        if re.findall(r'\d{2,3}x\d{2,3}', row[col]):
            row['Лоток'] = 'Лоток ' +\
                'x'.join(re.findall(r'\d{2,3}x\d{2,3}',
                                    row[col])[0].split('x'))
        if re.findall(r'\d{2,3}х\d{2,3}', row[col]):
            row['Лоток'] = 'Лоток ' +\
                'x'.join(re.findall(r'\d{2,3}х\d{2,3}',
                                    row[col])[0].split('х'))
    else:
        row['Лоток'] = None

    return row


def preprocessing_merge_tray(row: pd.DataFrame):
    """
    Функция предобработки таблицы с лотками для последующего объединения
    :param row: строка DataFrame
    :return: строка DataFrame с и измененными данными о лотках
    """

    if re.findall(r'\d{2,3}x\d{2,3}', row['Лоток']):
        row['Лоток'] = 'Лоток ' +\
            'x'.join(re.findall(r'\d{2,3}х\d{2,3}',
                                row['Лоток'])[0].split('х'))
    if re.findall(r'\d{2,3}х\d{2,3}', row['Лоток']):
        row['Лоток'] = 'Лоток ' +\
            'x'.join(re.findall(r'\d{2,3}х\d{2,3}',
                                row['Лоток'])[0].split('х'))

    return row


def split_cable(row: pd.DataFrame,
                col: str) -> pd.DataFrame:
    if str(row[col]).find('/') != -1:
        row[col] = float(row[col].split('/')[-1])
    else:
        row[col] = float(row[col])
    return row


def fact_tray_count(params: dict,
                    title: str) -> [pd.DataFrame | None]:
    """
    Функция подсчёта фактически смонтированных лотков
    :param params: словарь с параметрами
    :param title: название титула
    :return: DataFrame с данными о лотках
    """

    sheets = selecting_sheets(params=params,
                              title=title,
                              search_list=['лотки',
                                           'лоток'])
    if not sheets:
        print_info(message_code='error processing tray',
                   logs=params['log'],
                   params_message=title)
    else:
        tray = pd.DataFrame(columns=['Лоток',
                                     'Смонтировано',
                                     'Сдано по ЗНИ',
                                     'ЗНИ не подписано',
                                     'Марка',
                                     'Титул'])
        print_info(message_code='tray list',
                   logs=params['log'],
                   params_message=[title, sheets])
        for sheet in sheets:
            df = params['xl'][title].parse(sheet)
            df.drop(index=0, inplace=True)
            df = df[df['Дата'] != 'Сумма']
            columns_tray = []
            for column in df.columns:
                if column.lower().find('лотки') != -1 or \
                        column.lower().find('лоток') != -1:
                    columns_tray.append(column)

            s_cab = None
            for column in df.columns:
                if ((column.lower().find('строительный') != -1
                     and column.lower().find('контроль') != -1)
                        or column.lower().find('ск') != -1):
                    s_cab = column
                    break
            if not s_cab:
                print_info(message_code='error control tray',
                           logs=params['log'],
                           params_message=(title, sheet))
                control = {'Сдано по ЗНИ': [None],
                           'ЗНИ не подписано': [None]}
            else:
                control = {'Сдано по ЗНИ': [],
                           'ЗНИ не подписано': []}
                for column in columns_tray:
                    tmp = df.apply(lambda row: s_control(row=row,
                                                         inspection=s_cab,
                                                         fact=column),
                                   axis=1)
                    control['Сдано по ЗНИ'].append(
                        tmp['Сдано по ЗНИ'].sum()
                    )
                    control['ЗНИ не подписано'].append(
                        tmp['ЗНИ не подписано'].sum()
                    )
            df1 = df[columns_tray].sum()
            df1 = df1.reset_index()
            df1 = pd.concat([df1, pd.DataFrame(control)],
                            axis=1,
                            ignore_index=True)
            df1['Марка'] = sheet.split()[-1]
            df1['Титул'] = title
            df1.columns = ['Лоток',
                           'Смонтировано',
                           'Сдано по ЗНИ',
                           'ЗНИ не подписано',
                           'Марка',
                           'Титул']
            tray = pd.concat([tray,
                              df1],
                             ignore_index=True)

    try:
        if tray['Титул'].any():
            return tray
    except UnboundLocalError:
        return None
    except KeyError:
        return None


def project_tray_count(params: dict,
                       title: str) -> pd.DataFrame:
    """
    Функция подсчёта лотков в проекте
    :param params: словарь с параметрами
    :param title: название титула
    :return: DataFrame с данными о лотках
    """

    sheet = selecting_sheets(params=params,
                             title=title,
                             search_list=['специф'])
    trays = params['xl'][title].parse(sheet[0])
    for column in trays.columns:
        if str(column).lower().find('наим') != -1:
            name = column
        if str(column).lower().find('кол') != -1 and \
                (column.lower().find('рд') != -1 or
                    column.lower().find('рев') != -1):
            quantity = column

    trays = trays.fillna(value='0')
    trays = trays.apply(lambda row: find_trays(row=row,
                                               col=name),
                        axis=1)

    trays[quantity] = trays[quantity].fillna(value=0)
    trays['Титул'] = title

    trays = trays[(trays['Раздел'] != '0') &
                  (trays['Лоток'] != '0') &
                  (trays[
                      quantity] != '0')
                  ].groupby(
                      ['Титул',
                       'Раздел',
                       'Лоток']
    )[quantity].sum().reset_index()

    trays.columns = ['Титул', 'Марка', 'Лоток', 'по РД']

    return trays