Используем Python вместе с 1С для сложных вычислений и предиктивной аналитики

Рассмотрим два основных способа интеграции Python с 1С: через COM-объекты и через вызов внешних скриптов.

Способ 1: Интеграция через COM-объекты

1. Создание COM-сервера на Python Сначала создадим Python-скрипт, который будет выступать в роли COM-сервера:

# python_com_server.py
import pythoncom
import win32com.server.register
import win32com.server.exception
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
import json
import warnings
warnings.filterwarnings('ignore')

class PythonAnalytics:
    _reg_clsid_ = "{Ваш-GUID-здесь}"  # Сгенерируйте уникальный GUID
    _reg_progid_ = "Python.Analytics"
    _reg_desc_ = "Python Analytics COM Server"
    _public_methods_ = ['PredictLinearRegression', 'PredictRandomForest', 
                       'CalculateStatistics', 'DataPreprocessing', 'TestConnection']
    _public_attrs_ = ['last_error']
    
    def __init__(self):
        self.last_error = ""
        self.models = {}
    
    def TestConnection(self):
        """Тест соединения с Python"""
        try:
            return "Python COM Server is working! NumPy version: " + np.__version__
        except Exception as e:
            self.last_error = str(e)
            return f"Error: {str(e)}"
    
    def DataPreprocessing(self, json_data):
        """Предобработка данных из JSON"""
        try:
            data = json.loads(json_data)
            df = pd.DataFrame(data)
            
            # Простая предобработка
            numeric_columns = df.select_dtypes(include=[np.number]).columns
            df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].mean())
            
            result = {
                'processed_data': df.to_dict('records'),
                'statistics': {
                    'rows': len(df),
                    'columns': len(df.columns),
                    'numeric_columns': list(numeric_columns)
                }
            }
            return json.dumps(result)
            
        except Exception as e:
            self.last_error = str(e)
            return json.dumps({'error': str(e)})
    
    def PredictLinearRegression(self, json_data, target_column):
        """Прогнозирование с помощью линейной регрессии"""
        try:
            data = json.loads(json_data)
            df = pd.DataFrame(data)
            
            # Разделение на признаки и целевую переменную
            X = df.drop(columns=[target_column])
            y = df[target_column]
            
            # Обучение модели
            model = LinearRegression()
            model.fit(X, y)
            
            # Прогноз
            predictions = model.predict(X)
            
            result = {
                'predictions': predictions.tolist(),
                'coefficients': model.coef_.tolist(),
                'intercept': model.intercept_,
                'r_squared': model.score(X, y)
            }
            
            return json.dumps(result)
            
        except Exception as e:
            self.last_error = str(e)
            return json.dumps({'error': str(e)})
    
    def PredictRandomForest(self, json_data, target_column, n_estimators=100):
        """Прогнозирование с помощью случайного леса"""
        try:
            data = json.loads(json_data)
            df = pd.DataFrame(data)
            
            # Разделение на признаки и целевую переменную
            X = df.drop(columns=[target_column])
            y = df[target_column]
            
            # Обучение модели
            model = RandomForestRegressor(n_estimators=n_estimators, random_state=42)
            model.fit(X, y)
            
            # Прогноз
            predictions = model.predict(X)
            
            result = {
                'predictions': predictions.tolist(),
                'feature_importance': dict(zip(X.columns, model.feature_importances_)),
                'r_squared': model.score(X, y)
            }
            
            return json.dumps(result)
            
        except Exception as e:
            self.last_error = str(e)
            return json.dumps({'error': str(e)})
    
    def CalculateStatistics(self, json_data):
        """Расчет расширенной статистики"""
        try:
            data = json.loads(json_data)
            df = pd.DataFrame(data)
            
            stats = {}
            for column in df.columns:
                if pd.api.types.is_numeric_dtype(df[column]):
                    stats[column] = {
                        'mean': float(df[column].mean()),
                        'median': float(df[column].median()),
                        'std': float(df[column].std()),
                        'min': float(df[column].min()),
                        'max': float(df[column].max()),
                        'q25': float(df[column].quantile(0.25)),
                        'q75': float(df[column].quantile(0.75))
                    }
            
            return json.dumps(stats)
            
        except Exception as e:
            self.last_error = str(e)
            return json.dumps({'error': str(e)})

def register_com_server():
    """Регистрация COM-сервера"""
    try:
        print("Registering COM server...")
        win32com.server.register.UseCommandLine(PythonAnalytics)
        print("COM server registered successfully!")
    except Exception as e:
        print(f"Registration failed: {e}")

if __name__ == '__main__':
    register_com_server()

Запустите скрипт с правами администратора для регистрации COM-сервера:

python python_com_server.py --register

 Модуль 1С для работы с Python через COM

// Основная процедура тестирования соединения
Процедура ТестСоединенияСPython()
    
    Попытка
        // Создание COM-объекта
        PythonCOM = Новый COMОбъект("Python.Analytics");
        
        // Тест соединения
        Результат = PythonCOM.TestConnection();
        Сообщить("Результат теста: " + Результат);
        
    Исключение
        Сообщить("Ошибка при создании COM-объекта: " + ОписаниеОшибки());
    КонецПопытки;
    
КонецПроцедуры

// Функция для прогнозирования с помощью линейной регрессии
Функция ПрогнозЛинейнаяРегрессия(Данные, ЦелеваяКолонка)
    
    Попытка
        PythonCOM = Новый COMОбъект("Python.Analytics");
        
        // Преобразование данных в JSON
        JSONДанные = ПреобразоватьВJSON(Данные);
        
        // Вызов метода Python
        РезультатJSON = PythonCOM.PredictLinearRegression(JSONДанные, ЦелеваяКолонка);
        
        // Парсинг результата
        Результат = ПрочитатьJSON(РезультатJSON);
        
        Возврат Результат;
        
    Исключение
        Сообщить("Ошибка прогнозирования: " + ОписаниеОшибки());
        Возврат Неопределено;
    КонецПопытки;
    
КонецФункции

// Функция для прогнозирования с помощью случайного леса
Функция ПрогнозСлучайныйЛес(Данные, ЦелеваяКолонка, КоличествоДеревьев = 100)
    
    Попытка
        PythonCOM = Новый COMОбъект("Python.Analytics");
        
        // Преобразование данных в JSON
        JSONДанные = ПреобразоватьВJSON(Данные);
        
        // Вызов метода Python
        РезультатJSON = PythonCOM.PredictRandomForest(JSONДанные, ЦелеваяКолонка, КоличествоДеревьев);
        
        // Парсинг результата
        Результат = ПрочитатьJSON(РезультатJSON);
        
        Возврат Результат;
        
    Исключение
        Сообщить("Ошибка прогнозирования: " + ОписаниеОшибки());
        Возврат Неопределено;
    КонецПопытки;
    
КонецФункции

// Функция для расчета статистики
Функция РассчитатьСтатистику(Данные)
    
    Попытка
        PythonCOM = Новый COMОбъект("Python.Analytics");
        
        // Преобразование данных в JSON
        JSONДанные = ПреобразоватьВJSON(Данные);
        
        // Вызов метода Python
        РезультатJSON = PythonCOM.CalculateStatistics(JSONДанные);
        
        // Парсинг результата
        Результат = ПрочитатьJSON(РезультатJSON);
        
        Возврат Результат;
        
    Исключение
        Сообщить("Ошибка расчета статистики: " + ОписаниеОшибки());
        Возврат Неопределено;
    КонецПопытки;
    
КонецФункции

// Вспомогательная функция для преобразования таблицы значений в JSON
Функция ПреобразоватьВJSON(ТаблицаЗначений)
    
    МассивДанных = Новый Массив;
    
    Для каждого Строка Из ТаблицаЗначений Цикл
        Запись = Новый Соответствие;
        Для каждого Колонка Из ТаблицаЗначений.Колонки Цикл
            Запись.Вставить(Колонка.Имя, Строка[Колонка]);
        КонецЦикла;
        МассивДанных.Добавить(Запись);
    КонецЦикла;
    
    Возврат JSON.ЗаписатьJSON(МассивДанных);
    
КонецФункции

// Вспомогательная функция для чтения JSON
Функция ПрочитатьJSON(JSONТекст)
    
    ЧтениеJSON = Новый ЧтениеJSON;
    ЧтениеJSON.Поток = Новый ПотокВПамяти;
    ЧтениеJSON.Поток.Записать(JSONТекст);
    
    Возврат JSON.Прочитать(ЧтениеJSON);
    
КонецФункции

// Пример использования
Процедура ПримерИспользования()
    
    // Создание тестовых данных
    ТаблицаДанных = Новый ТаблицаЗначений;
    ТаблицаДанных.Колонки.Добавить("X1", Новый ОписаниеТипов("Число"));
    ТаблицаДанных.Колонки.Добавить("X2", Новый ОписаниеТипов("Число"));
    ТаблицаДанных.Колонки.Добавить("Y", Новый ОписаниеТипов("Число"));
    
    // Заполнение данными
    Для i = 1 По 10 Цикл
        Строка = ТаблицаДанных.Добавить();
        Строка.X1 = i;
        Строка.X2 = i * 2;
        Строка.Y = i * 3 + СлучайноеЧисло(1, 5);
    КонецЦикла;
    
    // Тест соединения
    ТестСоединенияСPython();
    
    // Расчет статистики
    Статистика = РассчитатьСтатистику(ТаблицаДанных);
    Если Статистика <> Неопределено Тогда
        Сообщить("Статистика: " + JSON.ЗаписатьJSON(Статистика));
    КонецЕсли;
    
    // Прогноз линейной регрессии
    ПрогнозЛР = ПрогнозЛинейнаяРегрессия(ТаблицаДанных, "Y");
    Если ПрогнозЛР <> Неопределено Тогда
        Сообщить("Прогноз линейной регрессии: " + JSON.ЗаписатьJSON(ПрогнозЛР));
    КонецЕсли;
    
    // Прогноз случайного леса
    ПрогнозСЛ = ПрогнозСлучайныйЛес(ТаблицаДанных, "Y", 50);
    Если ПрогнозСЛ <> Неопределено Тогда
        Сообщить("Прогноз случайного леса: " + JSON.ЗаписатьJSON(ПрогнозСЛ));
    КонецЕсли;
    
КонецПроцедуры

import json
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
import argparse

def main():
    parser = argparse.ArgumentParser(description='Python Analytics for 1C')
    parser.add_argument('--method', required=True, help='Method to execute')
    parser.add_argument('--input', required=True, help='Input JSON data')
    parser.add_argument('--output', required=True, help='Output file path')
    parser.add_argument('--target', help='Target column for prediction')
    parser.add_argument('--estimators', type=int, default=100, help='Number of estimators for Random Forest')
    
    args = parser.parse_args()
    
    try:
        # Чтение входных данных
        with open(args.input, 'r', encoding='utf-8') as f:
            input_data = json.load(f)
        
        result = {}
        
        if args.method == 'linear_regression':
            result = linear_regression_predict(input_data, args.target)
        elif args.method == 'random_forest':
            result = random_forest_predict(input_data, args.target, args.estimators)
        elif args.method == 'statistics':
            result = calculate_statistics(input_data)
        elif args.method == 'preprocessing':
            result = data_preprocessing(input_data)
        else:
            result = {'error': f'Unknown method: {args.method}'}
        
        # Запись результата
        with open(args.output, 'w', encoding='utf-8') as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
            
        print(f"Success: {args.method}")
        
    except Exception as e:
        error_result = {'error': str(e)}
        with open(args.output, 'w', encoding='utf-8') as f:
            json.dump(error_result, f, ensure_ascii=False, indent=2)
        print(f"Error: {str(e)}")

def linear_regression_predict(data, target_column):
    """Прогнозирование линейной регрессии"""
    df = pd.DataFrame(data)
    X = df.drop(columns=[target_column])
    y = df[target_column]
    
    model = LinearRegression()
    model.fit(X, y)
    predictions = model.predict(X)
    
    return {
        'predictions': predictions.tolist(),
        'coefficients': model.coef_.tolist(),
        'intercept': model.intercept_,
        'r_squared': model.score(X, y)
    }

def random_forest_predict(data, target_column, n_estimators):
    """Прогнозирование случайным лесом"""
    df = pd.DataFrame(data)
    X = df.drop(columns=[target_column])
    y = df[target_column]
    
    model = RandomForestRegressor(n_estimators=n_estimators, random_state=42)
    model.fit(X, y)
    predictions = model.predict(X)
    
    return {
        'predictions': predictions.tolist(),
        'feature_importance': dict(zip(X.columns, model.feature_importances_)),
        'r_squared': model.score(X, y)
    }

def calculate_statistics(data):
    """Расчет статистики"""
    df = pd.DataFrame(data)
    stats = {}
    
    for column in df.columns:
        if pd.api.types.is_numeric_dtype(df[column]):
            stats[column] = {
                'mean': float(df[column].mean()),
                'median': float(df[column].median()),
                'std': float(df[column].std()),
                'min': float(df[column].min()),
                'max': float(df[column].max()),
                'q25': float(df[column].quantile(0.25)),
                'q75': float(df[column].quantile(0.75))
            }
    
    return stats

def data_preprocessing(data):
    """Предобработка данных"""
    df = pd.DataFrame(data)
    
    numeric_columns = df.select_dtypes(include=[np.number]).columns
    df[numeric_columns] = df[numeric_columns].fillna(df[numeric_columns].mean())
    
    return {
        'processed_data': df.to_dict('records'),
        'statistics': {
            'rows': len(df),
            'columns': len(df.columns),
            'numeric_columns': list(numeric_columns)
        }
    }

if __name__ == '__main__':
    main()

// Модуль 1С для вызова Python скриптов

// Константы для настройки
Перем ПутьКPython;
Перем ПутьКСкрипту;

// Инициализация путей
Процедура Инициализация()
    ПутьКPython = "C:\Python39\python.exe";  // Укажите правильный путь
    ПутьКСкрипту = "C:\Scripts\python_script_runner.py";  // Укажите правильный путь
КонецПроцедуры

// Основная функция для вызова Python скрипта
Функция ВызватьPythonСкрипт(Метод, Данные, ДополнительныеПараметры = Неопределено)
    
    Инициализация();
    
    Попытка
        // Создание временных файлов
        ВходнойФайл = ПолучитьИмяВременногоФайла("json");
        ВыходнойФайл = ПолучитьИмяВременногоФайла("json");
        
        // Запись входных данных
        ЗаписатьJSONВФайл(Данные, ВходнойФайл);
        
        // Формирование команды
        Команда = СформироватьКоманду(Метод, ВходнойФайл, ВыходнойФайл, ДополнительныеПараметры);
        
        // Выполнение команды
        КодВозврата = ВыполнитьКоманду(Команда);
        
        Если КодВозврата = 0 Тогда
            // Чтение результата
            Результат = ПрочитатьJSONИзФайла(ВыходнойФайл);
            Возврат Результат;
        Иначе
            Сообщить("Ошибка выполнения Python скрипта. Код возврата: " + КодВозврата);
            Возврат Неопределено;
        КонецЕсли;
        
    Исключение
        Сообщить("Ошибка при вызове Python скрипта: " + ОписаниеОшибки());
        Возврат Неопределено;
    КонецПопытки;
    
КонецФункции

// Формирование командной строки
Функция СформироватьКоманду(Метод, ВходнойФайл, ВыходнойФайл, ДополнительныеПараметры)
    
    Команда = ОбернутьВКавычки(ПутьКPython) + " " + 
              ОбернутьВКавычки(ПутьКСкрипту) + " " +
              "--method " + Метод + " " +
              "--input " + ОбернутьВКавычки(ВходнойФайл) + " " +
              "--output " + ОбернутьВКавычки(ВыходнойФайл);
    
    Если ДополнительныеПараметры <> Неопределено Тогда
        Если ДополнительныеПараметры.Свойство("ЦелеваяКолонка") Тогда
            Команда = Команда + " --target " + ДополнительныеПараметры.ЦелеваяКолонка;
        КонецЕсли;
        
        Если ДополнительныеПараметры.Свойство("КоличествоДеревьев") Тогда
            Команда = Команда + " --estimators " + ДополнительныеПараметры.КоличествоДеревьев;
        КонецЕсли;
    КонецЕсли;
    
    Возврат Команда;
    
КонецФункции

// Выполнение команды и получение кода возврата
Функция ВыполнитьКоманду(Команда)
    
    ОбъектWScript = Новый COMОбъект("WScript.Shell");
    Процесс = ОбъектWScript.Exec(Команда);
    
    // Ожидание завершения
    Пока Процесс.Status = 0 Цикл
        // Процесс выполняется
        Ждать(100);
    КонецЦикла;
    
    Возврат Процесс.ExitCode;
    
КонецФункции

// Вспомогательные функции для работы с файлами
Процедура ЗаписатьJSONВФайл(Данные, ИмяФайла)
    
    JSONТекст = JSON.ЗаписатьJSON(Данные);
    ЗаписьТекста = Новый ЗаписьТекста;
    ЗаписьТекста.Открыть(ИмяФайла, "UTF-8");
    ЗаписьТекста.Записать(JSONТекст);
    ЗаписьТекста.Закрыть();
    
КонецПроцедуры

Функция ПрочитатьJSONИзФайла(ИмяФайла)
    
    ЧтениеТекста = Новый ЧтениеТекста;
    ЧтениеТекста.Открыть(ИмяФайла, "UTF-8");
    JSONТекст = ЧтениеТекста.Прочитать();
    ЧтениеТекста.Закрыть();
    
    Возврат JSON.ПрочитатьJSON(JSONТекст);
    
КонецФункции

Функция ОбернутьВКавычки(Путь)
    Возврат """" + Путь + """";
КонецФункции

// Функции-обертки для конкретных методов
Функция ПрогнозЛинейнаяРегрессия(Данные, ЦелеваяКолонка)
    
    Параметры = Новый Структура;
    Параметры.Вставить("ЦелеваяКолонка", ЦелеваяКолонка);
    
    Возврат ВызватьPythonСкрипт("linear_regression", Данные, Параметры);
    
КонецФункции

Функция ПрогнозСлучайныйЛес(Данные, ЦелеваяКолонка, КоличествоДеревьев = 100)
    
    Параметры = Новый Структура;
    Параметры.Вставить("ЦелеваяКолонка", ЦелеваяКолонка);
    Параметры.Вставить("КоличествоДеревьев", КоличествоДеревьев);
    
    Возврат ВызватьPythonСкрипт("random_forest", Данные, Параметры);
    
КонецФункции

Функция РассчитатьСтатистику(Данные)
    Возврат ВызватьPythonСкрипт("statistics", Данные);
КонецФункции

Функция ПредобработкаДанных(Данные)
    Возврат ВызватьPythonСкрипт("preprocessing", Данные);
КонецФункции

// Пример использования
Процедура ПримерИспользованияСкриптов()
    
    // Создание тестовых данных
    ТаблицаДанных = СоздатьТестовыеДанные();
    
    // Преобразование в JSON-совместимый формат
    JSONДанные = ПреобразоватьТаблицуВМассив(ТаблицаДанных);
    
    // Расчет статистики
    Статистика = РассчитатьСтатистику(JSONДанные);
    Если Статистика <> Неопределено Тогда
        Сообщить("Статистика: " + JSON.ЗаписатьJSON(Статистика));
    КонецЕсли;
    
    // Прогноз линейной регрессии
    ПрогнозЛР = ПрогнозЛинейнаяРегрессия(JSONДанные, "Y");
    Если ПрогнозЛР <> Неопределено Тогда
        Сообщить("Прогноз линейной регрессии: " + JSON.ЗаписатьJSON(ПрогнозЛР));
    КонецЕсли;
    
КонецПроцедуры

// Вспомогательные функции
Функция СоздатьТестовыеДанные()
    
    ТаблицаДанных = Новый ТаблицаЗначений;
    ТаблицаДанных.Колонки.Добавить("X1", Новый ОписаниеТипов("Число"));
    ТаблицаДанных.Колонки.Добавить("X2", Новый ОписаниеТипов("Число"));
    ТаблицаДанных.Колонки.Добавить("Y", Новый ОписаниеТипов("Число"));
    
    Для i = 1 По 20 Цикл
        Строка = ТаблицаДанных.Добавить();
        Строка.X1 = i;
        Строка.X2 = i * 2;
        Строка.Y = i * 3 + СлучайноеЧисло(1, 10);
    КонецЦикла;
    
    Возврат ТаблицаДанных;
    
КонецФункции

Функция ПреобразоватьТаблицуВМассив(ТаблицаЗначений)
    
    МассивДанных = Новый Массив;
    
    Для каждого Строка Из ТаблицаЗначений Цикл
        Запись = Новый Соответствие;
        Для каждого Колонка Из ТаблицаЗначений.Колонки Цикл
            Запись.Вставить(Колонка.Имя, Строка[Колонка]);
        КонецЦикла;
        МассивДанных.Добавить(Запись);
    КонецЦикла;
    
    Возврат МассивДанных;
    
КонецФункции

Сравнение способов интеграции

COM-интеграция:

Плюсы:

• Высокая производительность

• Прямой вызов методов

• Простота отладки

• Минимальные накладные расходы

Минусы:

• Требует регистрации COM-сервера

• Менее гибкая при изменении Python-кода

• Возможны проблемы с совместимостью

Вызов внешних скриптов:

Плюсы:

• Большая гибкость

• Легкость обновления Python-кода

• Изоляция процессов

• Поддержка любой версии Python

Минусы:

• Большие накладные расходы

• Медленнее из-за запуска процесса

• Сложнее в отладке

Рекомендации по использованию

1. Для высоконагруженных систем используйте COM-интеграцию

2. Для гибкости и быстрого прототипирования используйте вызов скриптов

3. Обязательно обрабатывайте исключения в обоих случаях

4. Используйте временные файлы для передачи больших объемов данных

5. Настройте логирование для отладки интеграции

Этот подход позволяет эффективно использовать мощь Python для сложных вычислений и машинного обучения в рамках 1С-приложений.