modules/processing_data.py - master - susub/report

modules/processing_data.py
385 строк19 KB
Александр Куклин
add registry SQ
29 мар 2026, 14:22
29 мар 2026, 14:22151fbfb
import pandas as pd
import re
from pprint import pprint
import os

from modules.massage.massager import print_info
from modules.data.processing import selecting_sheets, s_control, find_trays, split_cable
from modules.data.processing import fact_tray_count, project_tray_count


def cable_processing(params: dict) -> pd.DataFrame:
    """
    Функция обработки данных о кабелях
    :param params: словарь с параметрами
    :return: DataFrame с данными о кабелях
    """

    cable = pd.DataFrame(columns=['Титул',
                                  'Марка',
                                  'Длина по РД',
                                  'Проложено',
                                  'Сдано по ЗНИ',
                                  'ЗНИ не подписано'])
    for title in params['title']:
        sheets = selecting_sheets(params=params,
                                  title=title,
                                  search_list=['кабел'])
        if not sheets:
            print_info(message_code='error processing cable',
                       logs=params['log'],
                       params_message=title)
        else:
            print_info(message_code='cable list',
                       logs=params['log'],
                       params_message=[title, sheets])
            for sheet in sheets:
                df = params['xl'][title].parse(sheet)
                for column in df.columns:
                    if (column.lower().find('длин') != -1 and
                            column.lower().find('факт') != -1):
                        f_cab = column
                        break
                for column in df.columns:
                    if (column.lower().find('марка') != -1 and
                            column.lower().find('кабеля') != -1):
                        m_cab = column
                        break
                for column in df.columns:
                    if (column.lower().find('строительный') != -1 and
                            column.lower().find('контроль') != -1):
                        s_cab = column
                        break
                for column in df.columns:
                    if (column.lower().find('длин') != -1 and
                            column.lower().find('рд') != -1):
                        r_cab = column
                        break

                df = df[['Раздел', m_cab, f_cab, r_cab, s_cab]]
                df = df.fillna(value=0)
                df = df[df['Раздел'] != 0]
                df = df.apply(lambda row: split_cable(row=row,
                                                      col=r_cab), axis=1)
                df = df.apply(lambda row: s_control(row=row,
                                                    inspection=s_cab,
                                                    fact=f_cab), axis=1)

                for chapter in df['Раздел'].unique():
                    cable = pd.concat([cable,
                                       pd.DataFrame(
                                           {'Титул': [title],
                                            'Марка': [chapter],
                                            'Длина по РД': [
                                                df[
                                                    df['Раздел'] == chapter
                                                ][r_cab].sum()],
                                            'Проложено': [
                                                df[
                                                    df['Раздел'] == chapter
                                                ][f_cab].sum()],
                                            'Сдано по ЗНИ': [
                                                df[
                                                    df['Раздел'] == chapter
                                                ]['Сдано по ЗНИ'].sum()],
                                            'ЗНИ не подписано': [
                                                df[
                                                    df['Раздел'] == chapter
                                                ]
                                                ['ЗНИ не подписано'].sum()
                                           ]})], ignore_index=True)
                print_info(message_code='success processing cable',
                           logs=params['log'],
                           params_message=title)

    cable = pd.concat([cable,
                       pd.DataFrame({'Титул': ['Итого'],
                                     'Марка': [None],
                                     'Длина по РД': [
                                         cable['Длина по РД'].sum()
                       ],
                           'Проложено': [cable['Проложено'].sum()],
                           'Сдано по ЗНИ': [
                                         cable['Сдано по ЗНИ'].sum()
                       ],
                           'ЗНИ не подписано': [
                                         cable['ЗНИ не подписано'].sum()
                       ]})],
                      ignore_index=True)

    return cable


def tray_processing(params) -> pd.DataFrame:
    """
    Функция обработки данных о лотках
    :param params: словарь с параметрами
    :return: DataFrame с данными о лотках
    """

    trays = pd.DataFrame(columns=['Титул',
                                  'Марка',
                                  'Лоток',
                                  'по РД',
                                  'Смонтировано',
                                  'ЗНИ не подписано',
                                  'Сдано по ЗНИ'])
    for title in params['title']:
        t_fact = fact_tray_count(params=params,
                                 title=title)
        t_project = project_tray_count(params=params,
                                       title=title)
        if t_fact is not None:
            t_project = t_project.merge(right=t_fact,
                                        on=['Титул', 'Марка', 'Лоток'],
                                        how='outer')
        else:
            for column in ['Смонтировано',
                           'Сдано по ЗНИ',
                           'ЗНИ не подписано']:
                t_project[column] = 0

        trays = pd.concat([trays,
                           t_project[['Титул',
                                      'Марка',
                                      'Лоток',
                                      'по РД',
                                      'Смонтировано',
                                      'ЗНИ не подписано',
                                      'Сдано по ЗНИ']]],
                          ignore_index=True)

        print_info(message_code='success processing tray',
                   logs=params['log'],
                   params_message=title)

    trays = trays.fillna(value=0)
    trays.columns = ['Титул',
                     'Марка',
                     'Лоток',
                     'по РД, шт',
                     'Смонтировано, шт',
                     'ЗНИ не подписано, шт',
                     'Сдано по ЗНИ, шт']
    trays['по РД, м'] = trays['по РД, шт'] * 3
    trays['Смонтировано, м'] = trays['Смонтировано, шт'] * 3
    trays['ЗНИ не подписано, м'] = trays['ЗНИ не подписано, шт'] * 3
    trays['Сдано по ЗНИ, м'] = trays['Сдано по ЗНИ, шт'] * 3

    trays = pd.concat([trays[['Титул',
                              'Марка',
                              'Лоток',
                              'по РД, шт',
                              'по РД, м',
                              'Смонтировано, шт',
                              'Смонтировано, м',
                              'Сдано по ЗНИ, шт',
                              'Сдано по ЗНИ, м',
                              'ЗНИ не подписано, шт',
                              'ЗНИ не подписано, м']],
                      pd.DataFrame({'Титул': ['Итого'],
                                    'Марка': [None],
                                    'Лоток': [None],
                                    'по РД, шт': [trays['по РД, шт'].sum()],
                                    'по РД, м': [trays['по РД, м'].sum()],
                                    'Смонтировано, шт': [
                                        trays['Смонтировано, шт'].sum()
                      ],
                          'Смонтировано, м': [
                                        trays['Смонтировано, м'].sum()
                      ],
                          'Сдано по ЗНИ, шт': [
                                        trays['Сдано по ЗНИ, шт'].sum()
                      ],
                          'Сдано по ЗНИ, м': [
                                        trays['Сдано по ЗНИ, м'].sum()
                      ],
                          'ЗНИ не подписано, шт': [
                                        trays['ЗНИ не подписано, шт'].sum()
                      ],
                          'ЗНИ не подписано, м': [
                                        trays['ЗНИ не подписано, м'].sum()
                      ]})], ignore_index=True)

    return trays


def equipment_processing(params) -> pd.DataFrame:
    """
    Функция обработки данных о оборудовании
    :param params: словарь с параметрами
    :return: DataFrame с данными о оборудовании
    """

    equipment = pd.DataFrame(columns=['Титул',
                                      'Марка',
                                      'Колличество по РД',
                                      'Смонтировано',
                                      'Сдано по ЗНИ',
                                      'ЗНИ не подписано',
                                      'Смонтировано не нами'])
    for title in params['title']:
        sheets = selecting_sheets(params=params,
                                  title=title,
                                  search_list=['оборуд'])
        if not sheets:
            print_info(message_code='error processing equipment',
                       logs=params['log'],
                       params_message=title)
        else:
            print_info(message_code='cable list',
                       logs=params['log'],
                       params_message=[title, sheets])
            for sheet in sheets:
                df = params['xl'][title].parse(sheet)
                for column in df.columns:
                    if column.lower().find('смонт') != -1:
                        fact = column
                    if column.lower().find('наим') != -1:
                        name = column
                    if (column.lower().find('строит') != -1 and
                            column.lower().find('контр') != -1):
                        s_cab = column
                    if column.lower().find('тэг') != -1:
                        tag = column
                    if (column.lower().find('марка') != -1 or
                            column.lower().find('титул') != -1):
                        section = column
                for column in df.columns:
                    if column.lower().find('прим') != -1:
                        note = column
                        break

                df = df[[section,
                         tag,
                         name,
                         fact,
                         s_cab,
                         note]]
                df = df.fillna(value=0)
                df = df[df[section] != 0]
                df = df.apply(lambda row: s_control(row=row,
                                                    inspection=s_cab,
                                                    fact=fact,
                                                    note=note), axis=1)

                for chapter in df[section].unique():
                    equipment = pd.concat(
                        [equipment,
                         pd.DataFrame(
                             {'Титул': [title],
                              'Марка': [chapter],
                              'Колличество по РД': df[
                                  df[section] == chapter][section].count(),
                              'Смонтировано': [df[df[section] == chapter][fact].sum()],
                              'Сдано по ЗНИ': [df[df[section] == chapter]['Сдано по ЗНИ'].sum()],
                              'ЗНИ не подписано': [
                                  df[
                                      df[section] == chapter
                                  ]['ЗНИ не подписано'].sum()],
                              'Смонтировано не нами': [
                                  df[df[section] == chapter][
                                      'Смонтировано не нами'
                                  ].sum()]
                              })], ignore_index=True)
                print_info(message_code='success processing equipment',
                           logs=params['log'],
                           params_message=title)
    equipment = pd.concat(
        [equipment,
         pd.DataFrame({'Титул': ['Итого'],
                       'Марка': [None],
                       'Колличество по РД': [
                           equipment['Колличество по РД'].sum()
         ],
             'Смонтировано': [df[df[section] == chapter][fact].sum()],
             'Сдано по ЗНИ': [df[df[section] == chapter]['Сдано по ЗНИ'].sum()],
             'ЗНИ не подписано': [
                           equipment['ЗНИ не подписано'].sum()
         ],
             'Смонтировано не нами': [
                           equipment['Смонтировано не нами'].sum()
         ]})], ignore_index=True)

    return equipment


def vk_processing(params: dict) -> pd.DataFrame:
    """
    Функция обработки данных о прохождении входного контроля
    :param params: словарь с параметрами
    :return: DataFrame с данными о прохождении входного контроля
    """
    vk = pd.DataFrame(columns=['Титул',
                               'Марка',
                               'Колличество по РД',
                               'Получено',
                               'Прошло ВК',
                               'Процент ВК'])
    for title in params['title']:
        for sheet_name in params['xl'][title].sheet_names:
            if re.search('[Сс]пециф', sheet_name):
                sheet = sheet_name
        df = params['xl'][title].parse(sheet)
        s = 0
        p = 0
        for column in df.columns:
            if re.search('[Нн]аимен', str(column)) and \
                    re.search('[Рр][Дд]', str(column)) and \
                    p == 0:
                project = column
                p += 1
            if str(column).lower().find('факт') != -1:
                fact = column
            if str(column) == "ВК":
                vk_control = column
            if (str(column).lower().find('марка') != -1 or
                str(column).lower().find('раздел') != -1) and \
                    s == 0:
                section = column
                s += 1
        for mark in df[section].dropna().unique():
            vk = pd.concat([vk,
                            pd.DataFrame({"Титул": [title],
                                          'Марка': [mark],
                                          'Колличество по РД': [len(df[project].dropna().unique())],
                                          'Получено': [df[fact].dropna().count()],
                                          'Прошло ВК': [df[vk_control].dropna().count()],
                                          'Процент ВК': [format((df[vk_control].dropna().count() / len(df[project].dropna().unique())) * 100, ".2f")]})],
                           ignore_index=True)
    vk = pd.concat([vk, pd.DataFrame({'Титул': ['Итого'],
                                      'Марка': [None],
                                      'Колличество по РД': [vk['Колличество по РД'].sum()],
                                      'Получено': [vk['Получено'].sum()],
                                      'Прошло ВК': [vk['Прошло ВК'].sum()],
                                      'Процент ВК': [format((vk['Прошло ВК'].sum() /
                                                            vk['Колличество по РД'].sum()) * 100, ".2f")]})],
                   ignore_index=True)
    return vk


def sq_processing(params: dict) -> pd.DataFrame:
    df = pd.DataFrame()
    xl = {}

    for title in params['title']:
        folder = os.sep.join(
            (params['path'] + os.sep + params['config']['title'][title]).split(os.sep)[:-1]) + os.sep
        for file in os.listdir(folder):
            if file.lower().find('реестр') != -1:
                xl[title] = pd.ExcelFile(folder + file).parse('Лист1')
                data = xl[title].dropna(
                    how='all', subset=xl[title].columns[1:])
                data['Титул'] = title
                df = pd.concat([df, data], ignore_index=True)
    df = df[['Титул',
             'Папка хранения',
             '№ п/п',
             '№ SQ',
             'Текст SQ',
             'Дата отправки',
             'Дата регистрации',
             'Дата ответа',
             'Согласовано/не согласовано',
             'Примечание']]
    return df