Непрерывный сбор данных из торговых точек для визуализации в Yandex DataLens

Функция СтруктураВJSON(Значение)
	НастройкиСериализацииJSON = Новый НастройкиСериализацииJSON();
	НастройкиСериализацииJSON.ФорматСериализацииДаты = ФорматДатыJSON.ISO;	
		
	ПараметрыЗаписи = Новый ПараметрыЗаписиJSON(ПереносСтрокJSON.Нет);	
	
	ЗаписьJSON = Новый ЗаписьJSON();	
	ЗаписьJSON.УстановитьСтроку(ПараметрыЗаписи);

	ЗаписатьJSON(ЗаписьJSON, Значение, НастройкиСериализацииJSON);

	Возврат ЗаписьJSON.Закрыть();

КонецФункции

 Функция ПолучитьУзел()
    Выб = ПланыОбмена.expBI_Обмен.Выбрать();
    Пока Выб.Следующий() Цикл
        Если Не Выб.ЭтотУзел Тогда
            Возврат Выб.Ссылка;
        КонецЕсли;
    КонецЦикла;
    УстановитьПривилегированныйРежим(Истина);
    Нов =  ПланыОбмена.expBI_Обмен.СоздатьУзел();
    Нов.Наименование = "Облако"; 
    Нов.ОбменДанными.Загрузка=Истина;
    Нов.Код = Строка(Новый УникальныйИдентификатор());
    Нов.Записать();
    Возврат Нов.Ссылка;    
КонецФункции

 Функция ДанныеСправочника(Имя, СправочникОбъект)
    Результат = Новый Структура;
    Результат.Вставить("type", Имя);
    Результат.Вставить("guid", Строка(СправочникОбъект.Ссылка.УникальныйИдентификатор()));
    Результат.Вставить("name", СправочникОбъект.Наименование);
    Результат.Вставить("code", СправочникОбъект.Код);
    Возврат Результат;
КонецФункции 

 Функция ДанныеРегистраНакопления(Имя, Набор)
    Результат = Новый Массив; 
    Для каждого Стр из Набор Цикл
        Если Стр.Количество=0 и Стр.Сумма=0 Тогда
            Продолжить;
        КонецЕсли;
        Данные = Новый Структура;
        Данные.Вставить("type", Имя);
        Данные.Вставить("guid", Строка(Набор.Отбор.Регистратор.Значение.УникальныйИдентификатор()));
        Данные.Вставить("date", Формат(Стр.Период, "ДФ=yyyy-MM-ddTHH:mm:ss"));
        Данные.Вставить("id", Стр.НомерСтроки);
        Данные.Вставить("product", Строка(Стр.Номенклатура.УникальныйИдентификатор()));
        Данные.Вставить("unit", Строка(Стр.Склад.УникальныйИдентификатор()));
        Данные.Вставить("count",  Стр.Количество);
        Данные.Вставить("sum",  Стр.Сумма);
        Результат.Добавить(Данные);
    КонецЦикла;
    Возврат Результат;
КонецФункции

Функция ПодготовитьПакет() 
	
	РазмерПакета = 30;
	Выбрано=0;
	
	Результат = Новый Структура;
	
	Ссылки = Новый Массив;
	УдалитьРегистрацию = Новый Массив;
	ДанныеДляОтправки = Новый Массив;
	
	Узел = ПолучитьУзел();	
	ВыборкаИзменений = ПланыОбмена.ВыбратьИзменения(Узел, Узел.НомерОтправленного + 1);
	Пока ВыборкаИзменений.Следующий() Цикл
        Данные = ВыборкаИзменений.Получить();
		Если Данные = Неопределено Тогда
			Продолжить;
		КонецЕсли;             
		
		Если ТипЗнч(Данные) = Тип("УдалениеОбъекта") Тогда //Не учитываем для учебных целей            
			Продолжить	
			
		ИначеЕсли ТипЗнч(Данные) = Тип("СправочникОбъект.Номенклатура")Тогда			
			Ссылки.Добавить(Данные);   
			Если Данные.ЭтоГруппа Тогда
				ДанныеСправочника = ДанныеСправочника("groups", Данные)
			Иначе
				ДанныеСправочника = ДанныеСправочника("products", Данные);
				ДанныеСправочника.Вставить("group", Строка(Данные.Родитель.УникальныйИдентификатор()))
			КонецЕсли;
			ДанныеДляОтправки.Добавить(ДанныеСправочника);			
		
		ИначеЕсли ТипЗнч(Данные) = Тип("СправочникОбъект.СтруктурныеЕдиницы")Тогда			
			Ссылки.Добавить(Данные.Ссылка);
			ДанныеДляОтправки.Добавить(ДанныеСправочника("units", Данные));
			
		ИначеЕсли ТипЗнч(Данные) = Тип("РегистрНакопленияНаборЗаписей.Продажи") Тогда
			ДанныеРегистра = ДанныеРегистраНакопления("sales", Данные);
			Если ДанныеРегистра.Количество()>0 Тогда
				Ссылки.Добавить(Данные);
				ОбщегоНазначенияКлиентСервер.ДополнитьМассив(ДанныеДляОтправки, ДанныеРегистра);
				Выбрано = Выбрано + ДанныеРегистра.Количество();
			Иначе
				УдалитьРегистрацию.Добавить(Данные);
			КонецЕсли;
			
		КонецЕсли;
		Выбрано = Выбрано + 1;
		Если Выбрано>РазмерПакета Тогда
			Прервать
		КонецЕсли;
	КонецЦикла;
	
	
	Результат.Вставить("УдалитьРегистрацию", УдалитьРегистрацию);
	Результат.Вставить("Ссылки", Ссылки);
	Результат.Вставить("Данные", ДанныеДляОтправки);
	Возврат Результат;

КонецФункции

Функция ПолучитьСоединение(Адрес) Экспорт
    СтруктураURI=ОбщегоНазначенияКлиентСервер.СтруктураURI(Адрес);
    Прокси = ПолучениеФайловИзИнтернета.ПолучитьПрокси("https");
    ЗащищенноеСоединение = ?(НРег(СтруктураURI.Схема) = "http", Неопределено, ОбщегоНазначенияКлиентСервер.НовоеЗащищенноеСоединение());
    
    Попытка
        Возврат Новый HTTPСоединение(СтруктураURI.Хост, СтруктураURI.Порт, , , Прокси, 120, ЗащищенноеСоединение);        
    Исключение
        ВызватьИсключение ПодробноеПредставлениеОшибки(ИнформацияОбОшибке());
    КонецПопытки;
КонецФункции

Процедура ВыполнитьОбмен() Экспорт
	
	Пока Истина Цикл
		Пакет = ПодготовитьПакет();
		Узел = Неопределено;
		
		Если Пакет.УдалитьРегистрацию.Количество()>0 Тогда
			Узел = ПолучитьУзел();
			Для каждого Рег из Пакет.УдалитьРегистрацию Цикл
				ПланыОбмена.УдалитьРегистрациюИзменений(Узел, Рег); 
			КонецЦикла;
		КонецЕсли;

		
		Если Пакет.Данные.Количество()=0 Тогда
			Возврат
		КонецЕсли;
		
		Адрес = "https://*************************.apigw.yandexcloud.net";
		
		Соединение = ПолучитьСоединение(Адрес);
		
		Запрос = Новый HTTPЗапрос("/data");
		Запрос.Заголовки.Вставить("Content-Type", "application/json"); 	
		
		Запрос.УстановитьТелоИзСтроки(СтруктураВJSON(Пакет.Данные));
		//ПолучитьДвоичныеДанныеИзСтроки(Запрос.ПолучитьТелоКакСтроку()).Записать("C:\Users\sedaiko\test.json");
		
		Ответ = Соединение.ОтправитьДляОбработки(Запрос);

		Если Ответ.КодСостояния<>200 Тогда
			ВызватьИсключение  Ответ.ПолучитьТелоКакСтроку()
		КонецЕсли;
		
		Если Узел = Неопределено Тогда 
			Узел = ПолучитьУзел();
		КонецЕсли; 
		Для каждого Стр из Пакет.Ссылки Цикл
			ПланыОбмена.УдалитьРегистрациюИзменений(Узел, Стр); 
		КонецЦикла;
	КонецЦикла;
	
КонецПроцедуры

Функция КлючРегламентногоЗадания() Экспорт
	возврат "expBI_РегламентноеЗаданиеОбмена.ВыполнитьСинхронзациюВФоне"
КонецФункции

Функция ДобавитьРегламентноеЗадание()
    КлючРЗ = КлючРегламентногоЗадания();
    РЗ = РегламентныеЗадания.СоздатьРегламентноеЗадание("ЗапускДополнительныхОбработок");
    РЗ.Наименование  = "Обмен с облаком";
    РЗ.Ключ             = КлючРЗ;
    РЗ.Использование = Истина;
	
    РЗ.Параметры.Добавить(Справочники.ДополнительныеОтчетыИОбработки.ПустаяСсылка());
    РЗ.Параметры.Добавить(КлючРЗ);
    РЗ.Записать();
    Возврат РЗ
КонецФункции

Процедура УстановитьИнтервалОбмена(Интервал) Экспорт
	КлючРЗ = КлючРегламентногоЗадания();
	РезультатПоиска = РегламентныеЗаданияСервер.НайтиЗадания(Новый Структура("Ключ", КлючРЗ));
	Если РезультатПоиска.Количество() = 0 Тогда
		РегламентноеЗадание = ДобавитьРегламентноеЗадание();
	Иначе
		РегламентноеЗадание = РезультатПоиска[0];
	КонецЕсли;
	
	Расписание = Новый РасписаниеРегламентногоЗадания;
	Расписание.ПериодПовтораДней=1;
	Расписание.ПериодПовтораВТечениеДня = Интервал*60;
	
	РегламентноеЗадание.Расписание = Расписание;	
	РегламентноеЗадание.Записать();
КонецПроцедуры

Функция ВыполнитьСинхронзациюВФоне(ИдентификаторКоманды, АдресРезультата = Неопределено) Экспорт 
    Отказ = Ложь;
    ТекстОшибки = ""; //сюда пишем подробности по ошибкам обработки
	
	expBI_ОбменДанными.ВыполнитьОбмен();
	 
   
    Если Не ПустаяСтрока(ТекстОшибки) Тогда
        ЗаписьЖурналаРегистрации(
            "Обмен с облаком", УровеньЖурналаРегистрации.Ошибка,,,ТекстОшибки);
    КонецЕсли;
    
    Результат = Новый Структура("Успешно, ТекстОшибки", Не Отказ, ТекстОшибки);
    Если АдресРезультата <> Неопределено Тогда
        ПоместитьВоВременноеХранилище(Результат, АдресРезультата);
    КонецЕсли;
    
    Возврат Результат;
КонецФункции

&Вместо("ВыполнитьКоманду")
Функция expBI_ВыполнитьКоманду(ПараметрыКоманды, АдресРезультата)
	Если Найти(ПараметрыКоманды.ИдентификаторКоманды, expBI_РегламентноеЗаданиеОбмена.КлючРегламентногоЗадания())=1 Тогда
		Возврат expBI_РегламентноеЗаданиеОбмена.ВыполнитьСинхронзациюВФоне(ПараметрыКоманды.ИдентификаторКоманды, АдресРезультата)
	КонецЕсли;
	
	
	// Вставить содержимое метода.
	Результат = ПродолжитьВызов(ПараметрыКоманды, АдресРезультата);
	Возврат Результат;

КонецФункции

&НаСервере
Процедура ПриСозданииНаСервере(Отказ, СтандартнаяОбработка)
	Если Интервал=0 Тогда
		Интервал=1;
		expBI_РегламентноеЗаданиеОбмена.УстановитьИнтервалОбмена(Интервал);
	КонецЕсли;
КонецПроцедуры

&НаСервере
Процедура ИнтервалПриИзмененииНаСервере()
	expBI_РегламентноеЗаданиеОбмена.УстановитьИнтервалОбмена(Интервал);
КонецПроцедуры

&НаКлиенте
Процедура ИнтервалПриИзменении(Элемент)
	ИнтервалПриИзмененииНаСервере();
КонецПроцедуры

&НаСервереБезКонтекста
Процедура ВыполнитьОбменНаСервере()
	expBI_ОбменДанными.ВыполнитьОбмен();
КонецПроцедуры

&НаКлиенте
Процедура ВыполнитьОбмен(Команда)
	ВыполнитьОбменНаСервере();
КонецПроцедуры

CREATE TABLE `products`
(
    `guid` Utf8 NOT NULL,
    `code` Utf8,
    `group` Utf8,
    `name` Utf8,
    PRIMARY KEY (`guid`)
);

CREATE TABLE `units`
(
    `guid` Utf8 NOT NULL,
    `code` Utf8,
    `group` Utf8,
    `name` Utf8,
    PRIMARY KEY (`guid`)
);
CREATE TABLE `groups`
(
    `guid` Utf8 NOT NULL,
    `code` Utf8,
    `name` Utf8,
    PRIMARY KEY (`guid`)
);

CREATE TABLE `sales`
(
    `guid` Utf8 NOT NULL,
    `id` Int32,
    `date` Datetime,
    `unit` Utf8,
    `product` Utf8,
    `count` Float,
    `sum` Float,
    PRIMARY KEY (`guid`, `id`)
);

import re
from json import loads as json_loads
from datetime import datetime

from ydb import iam, aio as ydb_aio

credentials = iam.MetadataUrlCredentials()
# credentials=iam.ServiceAccountCredentials.from_file('key.json')

database = "/ru-central1/b1gi0b0h*******/etn8094o***********"
driver = ydb_aio.Driver(
    database=database,
    endpoint="grpcs://ydb.serverless.yandexcloud.net:2135",
    credentials=credentials,
)

pool = ydb_aio.SessionPool(driver, size=50)  # Max number of available session


async def execute_sql(session, query: str, parameters: dict):
    prepared_query = await session.prepare(query)
    await session.transaction().execute(prepared_query, parameters, commit_tx=True)


def get_type(val):
    if isinstance(val, bool):
        return 'Bool'
    elif isinstance(val, int):
        return 'Int32'
    elif isinstance(val, float):
        return 'Float'
    elif re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$', val) is not None:
        return 'DateTime'
    return 'Utf8'


def convert_val(value):
    if isinstance(value, str) and re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$', value) is not None:
        return int(datetime.strptime(value, "%Y-%m-%dT%H:%M:%S").timestamp())
    return value


async def load_data(messages):
    tables = []
    session = await pool.acquire()

    for msg in messages:
        table = msg['type']

        fields = [f for f in msg.keys() if f != 'type']
        values = [f"${f}" for f in fields]
        declare = [f'DECLARE ${f} AS {get_type(msg[f])}' for f in fields]
        parameters = {f"${f}": convert_val(msg[f]) for f in fields}

        query = f"{'; '.join(declare)}; UPSERT INTO {table} ({','.join(fields)}) VALUES ({','.join(values)})"
        await execute_sql(session, query, parameters)


async def handler(event, context):
    body = json_loads(event.get('body'))
    await load_data(body)

    return {
        'statusCode': 200,
        'body': dict(result=True),
    }

ydb

openapi: 3.0.0
info:
  title: Sample API
  version: 1.0.0
paths:
  /data:
    post:
      x-yc-apigateway-integration:
        type: cloud_functions
        function_id: d4euiv********
        service_account_id: ajegd***********

Выб = ПланыОбмена.expBI_Обмен.Выбрать();
Пока Выб.Следующий() Цикл
	Если Не Выб.ЭтотУзел Тогда
		Узел = Выб.Ссылка;
	КонецЕсли;
КонецЦикла;
ПланыОбмена.ЗарегистрироватьИзменения(Узел, Метаданные.Справочники.Номенклатура);
ПланыОбмена.ЗарегистрироватьИзменения(Узел, Метаданные.Справочники.СтруктурныеЕдиницы);