Пример импорта из Яндекс Метрика

public class ScheduledJobHandler : IScheduledJobHandler
{
    // Идентификатор счетчика Яндекс Метрики.
    // Значение нужно заменить на идентификатор вашего счетчика.
    private const int CounterId = 1111111;

    // OAuth-токен для доступа к API Яндекс Метрики.
    // В продуктивной среде токен лучше хранить в защищенных настройках.
    private const string OAuthToken = "***";

    // Модель атрибуции.
    // lastsign - последний значимый источник.
    private const string Attribution = "lastsign";

    public async Task Execute(ScheduledJobContext context)
    {
        // Получаем HTTP-клиент из контекста задания.
        var http = context.GetHttpClient();

        // Получаем сервис для работы со сделками.
        var dealService = context.GetEntityService<Deal>();

        await context.Log("YM enrichment job started");

        try
        {
            // Загружаем визиты за последние 20 дней.
            // Текущий день исключается, так как данные в Метрике могут быть еще неполными.
            var dateFrom = DateTime.UtcNow.Date.AddDays(-20);
            var dateTo = DateTime.UtcNow.Date.AddDays(-1);

            // Загружаем визиты из Яндекс Метрики и группируем их по clientID.
            var visitsByClientId = await LoadVisitsAsync(
                context,
                http,
                dateFrom,
                dateTo,
                CancellationToken.None);

            await context.Log($"YM visits loaded: {visitsByClientId.Count}");
            await context.Log($"YM client Ids: {string.Join(", ", visitsByClientId.Keys)}");

            if (visitsByClientId.Count == 0)
            {
                await context.Log("No YM visits found for period");
                return;
            }

            // Ищем сделки, у которых:
            // 1. заполнен clientID;
            // 2. еще не заполнен технический признак обработки.
            var deals = dealService
                .Get(x => x.StringValue1 != null && x.StringValue1 != "" && x.IntegerValue1 == null)
                .ToList();

            await context.Log($"Deals loaded: {deals.Count}");

            var enriched = 0;
            var skipped = 0;
            var failed = 0;

            foreach (var deal in deals)
            {
                try
                {
                    // clientID должен быть заранее записан в сделку.
                    // Например, он может попадать в сделку из формы заявки на сайте.
                    var clientId = deal.StringValue1?.Trim();

                    if (string.IsNullOrWhiteSpace(clientId))
                    {
                        skipped++;
                        continue;
                    }

                    // Если визит по clientID не найден, сделка пропускается.
                    if (!visitsByClientId.TryGetValue(clientId, out var visit))
                    {
                        skipped++;
                        await context.Log($"Visit not found for clientId={clientId}, dealId={deal.Id}");
                        continue;
                    }

                    // Переносим данные визита в дополнительные поля сделки.
                    ApplyVisitToDeal(deal, visit);

                    // Записываем технический признак обработки.
                    // В примере используется Unix-время обработки.
                    deal.IntegerValue1 = (int)DateTimeOffset.UtcNow.ToUnixTimeSeconds();

                    // Сохраняем обновленную сделку.
                    await dealService.UpdateAsync(deal);

                    enriched++;
                }
                catch (Exception ex)
                {
                    failed++;
                    await context.Log(
                        $"Deal enrichment failed. DealId={deal.Id}. Error={ex.Message}",
                        DynamicCodeLogLevel.Error);
                }
            }

            await context.Log($"YM enrichment job finished. Enriched={enriched}, Skipped={skipped}, Failed={failed}");
        }
        catch (Exception ex)
        {
            await context.Log($"YM enrichment job failed: {ex}", DynamicCodeLogLevel.Error);
            throw;
        }
    }

    private static void ApplyVisitToDeal(Deal deal, YmVisitDto visit)
    {
        // Основной идентификатор пользователя Метрики.
        deal.StringValue1 = visit.ClientId;

        // Источник трафика и рекламная система.
        deal.StringValue3 = visit.TrafficSource;
        deal.StringValue4 = visit.AdvEngine;

        // Страница-источник и первая страница визита.
        deal.StringValue5 = visit.Referer;
        deal.StringValue6 = visit.StartUrl;

        // UTM-метки.
        deal.StringValue7 = visit.UtmSource;
        deal.StringValue8 = visit.UtmMedium;
        deal.StringValue9 = visit.UtmCampaign;
        deal.StringValue10 = visit.UtmContent;
        deal.StringValue11 = visit.UtmTerm;

        // Данные Яндекс Директа.
        deal.StringValue12 = visit.DirectCampaignName;
        deal.StringValue13 = visit.DirectAdGroupName;
        deal.StringValue14 = visit.DirectPlatform;
        deal.StringValue15 = visit.DirectPlatformType;

        // Идентификаторы кампании и группы объявлений.
        deal.IntegerValue2 = SafeToInt32(visit.DirectCampaignId);
        deal.IntegerValue3 = SafeToInt32(visit.DirectAdGroupId);

        // Признак нового пользователя.
        deal.BooleanValue1 = visit.IsNewUser;
    }

    private static int? SafeToInt32(ulong? value)
    {
        // В Яндекс Метрике идентификаторы могут быть больше int.MaxValue.
        // Если значение не помещается в Int32, поле не заполняется.
        if (!value.HasValue || value.Value > int.MaxValue)
            return null;

        return (int)value.Value;
    }

    private async Task<Dictionary<string, YmVisitDto>> LoadVisitsAsync(
        ScheduledJobContext context,
        SafeHttpClient http,
        DateTime dateFromUtc,
        DateTime dateToUtc,
        CancellationToken ct)
    {
        // Список полей, которые будут выгружены из Logs API.
        var fields = string.Join(",",
            "ym:s:clientID",
            "ym:s:dateTimeUTC",
            "ym:s:isNewUser",
            "ym:s:startURL",
            "ym:s:referer",
            "ym:s:" + Attribution + "TrafficSource",
            "ym:s:" + Attribution + "AdvEngine",
            "ym:s:" + Attribution + "UTMSource",
            "ym:s:" + Attribution + "UTMMedium",
            "ym:s:" + Attribution + "UTMCampaign",
            "ym:s:" + Attribution + "UTMContent",
            "ym:s:" + Attribution + "UTMTerm",
            "ym:s:" + Attribution + "DirectClickOrder",
            "ym:s:" + Attribution + "DirectBannerGroup",
            "ym:s:" + Attribution + "DirectClickOrderName",
            "ym:s:" + Attribution + "ClickBannerGroupName",
            "ym:s:" + Attribution + "DirectPlatform",
            "ym:s:" + Attribution + "DirectPlatformType"
        );

        // Создаем запрос на подготовку выгрузки визитов.
        var createUrl =
            $"https://api-metrika.yandex.net/management/v1/counter/{CounterId}/logrequests" +
            $"?source=visits&date1={dateFromUtc:yyyy-MM-dd}&date2={dateToUtc:yyyy-MM-dd}" +
            $"&fields={Uri.EscapeDataString(fields)}";

        var createRequest = new HttpRequestMessage(HttpMethod.Post, createUrl);
        createRequest.Headers.Add("Authorization", "OAuth " + OAuthToken);

        long requestId = 0;

        try
        {
            // Отправляем запрос на создание выгрузки.
            var createResponse = await http.SendAsync(createRequest);
            createResponse.EnsureSuccessStatusCode();

            // Получаем идентификатор log request.
            requestId = ExtractRequestId(await createResponse.Content.ReadAsStringAsync(ct));

            await context.Log("YM logrequest created. RequestId=" + requestId);

            var pollDelay = TimeSpan.FromSeconds(5);
            var maxPollDelay = TimeSpan.FromSeconds(30);

            // Ожидаем, пока Яндекс Метрика подготовит выгрузку.
            while (true)
            {
                ct.ThrowIfCancellationRequested();

                var statusRequest = new HttpRequestMessage(
                    HttpMethod.Get,
                    $"https://api-metrika.yandex.net/management/v1/counter/{CounterId}/logrequest/{requestId}");

                statusRequest.Headers.Add("Authorization", "OAuth " + OAuthToken);

                try
                {
                    var statusResponse = await http.SendAsync(statusRequest);
                    statusResponse.EnsureSuccessStatusCode();

                    var status = ExtractStatus(await statusResponse.Content.ReadAsStringAsync(ct));

                    if (status == "processed")
                        break;

                    if (status == "created" || status == "processing")
                    {
                        await Task.Delay(pollDelay, ct);

                        // Увеличиваем интервал между проверками, но не больше 30 секунд.
                        var nextSeconds = Math.Min(pollDelay.TotalSeconds * 2, maxPollDelay.TotalSeconds);
                        pollDelay = TimeSpan.FromSeconds(nextSeconds);

                        continue;
                    }

                    throw new InvalidOperationException("Unexpected YM logrequest status: " + status);
                }
                finally
                {
                    statusRequest.Dispose();
                }
            }

            // Получаем информацию о частях подготовленной выгрузки.
            var infoRequest = new HttpRequestMessage(
                HttpMethod.Get,
                $"https://api-metrika.yandex.net/management/v1/counter/{CounterId}/logrequest/{requestId}");

            infoRequest.Headers.Add("Authorization", "OAuth " + OAuthToken);

            try
            {
                var infoResponse = await http.SendAsync(infoRequest);
                infoResponse.EnsureSuccessStatusCode();

                var partNumbers = ExtractPartNumbers(await infoResponse.Content.ReadAsStringAsync(ct));

                var allRows = new List<YmVisitDto>();

                // Скачиваем все части выгрузки.
                foreach (var partNumber in partNumbers)
                {
                    var downloadRequest = new HttpRequestMessage(
                        HttpMethod.Get,
                        $"https://api-metrika.yandex.net/management/v1/counter/{CounterId}/logrequest/{requestId}/part/{partNumber}/download");

                    downloadRequest.Headers.Add("Authorization", "OAuth " + OAuthToken);

                    try
                    {
                        var downloadResponse = await http.SendAsync(downloadRequest);
                        downloadResponse.EnsureSuccessStatusCode();

                        // Яндекс Метрика возвращает данные в TSV-формате.
                        allRows.AddRange(ParseVisitsTsv(await downloadResponse.Content.ReadAsStringAsync(ct)));
                    }
                    finally
                    {
                        downloadRequest.Dispose();
                    }
                }

                // Для каждого clientID оставляем последний визит по dateTimeUTC.
                return allRows
                    .Where(x => !string.IsNullOrWhiteSpace(x.ClientId))
                    .GroupBy(x => x.ClientId)
                    .Select(g => g.OrderByDescending(x => x.DateTimeUtc ?? DateTime.MinValue).First())
                    .ToDictionary(x => x.ClientId, x => x);
            }
            finally
            {
                infoRequest.Dispose();
            }
        }
        finally
        {
            createRequest.Dispose();

            // После скачивания данных очищаем log request в Яндекс Метрике.
            if (requestId != 0)
            {
                var cleanRequest = new HttpRequestMessage(
                    HttpMethod.Post,
                    $"https://api-metrika.yandex.net/management/v1/counter/{CounterId}/logrequest/{requestId}/clean");

                cleanRequest.Headers.Add("Authorization", "OAuth " + OAuthToken);

                try
                {
                    var cleanResponse = await http.SendAsync(cleanRequest);

                    if (!cleanResponse.IsSuccessStatusCode)
                    {
                        await context.Log(
                            $"YM logrequest clean failed. RequestId={requestId}, Status={cleanResponse.StatusCode}");
                    }
                }
                finally
                {
                    cleanRequest.Dispose();
                }
            }
        }
    }

    private static long ExtractRequestId(string json)
    {
        // Извлекаем request_id из JSON-ответа Яндекс Метрики.
        using var doc = JsonDocument.Parse(json);
        return doc.RootElement.GetProperty("log_request").GetProperty("request_id").GetInt64();
    }

    private static string ExtractStatus(string json)
    {
        // Извлекаем статус подготовки выгрузки.
        using var doc = JsonDocument.Parse(json);
        return doc.RootElement.GetProperty("log_request").GetProperty("status").GetString() ?? string.Empty;
    }

    private static List<int> ExtractPartNumbers(string json)
    {
        // Извлекаем номера частей выгрузки.
        using var doc = JsonDocument.Parse(json);

        var result = new List<int>();

        if (!doc.RootElement.TryGetProperty("log_request", out var logRequest)) return result;
        if (!logRequest.TryGetProperty("parts", out var parts)) return result;

        foreach (var part in parts.EnumerateArray())
        {
            if (part.TryGetProperty("part_number", out var p))
                result.Add(p.GetInt32());
        }

        return result;
    }

    private static List<YmVisitDto> ParseVisitsTsv(string tsv)
    {
        // Разбираем TSV-ответ Яндекс Метрики в список объектов YmVisitDto.
        var result = new List<YmVisitDto>();

        using var reader = new StringReader(tsv);

        var headers = reader.ReadLine()?.Split('\t');
        if (headers == null) return result;

        string line;
        while ((line = reader.ReadLine()) != null)
        {
            if (string.IsNullOrWhiteSpace(line)) continue;

            var cols = line.Split('\t');
            var row = new Dictionary<string, string>(StringComparer.OrdinalIgnoreCase);

            for (var i = 0; i < headers.Length && i < cols.Length; i++)
                row[headers[i]] = string.IsNullOrWhiteSpace(cols[i]) ? null : cols[i];

            result.Add(new YmVisitDto
            {
                ClientId = GetString(row, "ym:s:clientID"),
                DateTimeUtc = GetDateTime(row, "ym:s:dateTimeUTC"),
                IsNewUser = GetBool(row, "ym:s:isNewUser"),
                StartUrl = GetString(row, "ym:s:startURL"),
                Referer = GetString(row, "ym:s:referer"),
                TrafficSource = GetString(row, "ym:s:" + Attribution + "TrafficSource"),
                AdvEngine = GetString(row, "ym:s:" + Attribution + "AdvEngine"),
                UtmSource = GetString(row, "ym:s:" + Attribution + "UTMSource"),
                UtmMedium = GetString(row, "ym:s:" + Attribution + "UTMMedium"),
                UtmCampaign = GetString(row, "ym:s:" + Attribution + "UTMCampaign"),
                UtmContent = GetString(row, "ym:s:" + Attribution + "UTMContent"),
                UtmTerm = GetString(row, "ym:s:" + Attribution + "UTMTerm"),
                DirectCampaignId = GetUInt64(row, "ym:s:" + Attribution + "DirectClickOrder"),
                DirectAdGroupId = GetUInt64(row, "ym:s:" + Attribution + "DirectBannerGroup"),
                DirectCampaignName = GetString(row, "ym:s:" + Attribution + "DirectClickOrderName"),
                DirectAdGroupName = GetString(row, "ym:s:" + Attribution + "ClickBannerGroupName"),
                DirectPlatform = GetString(row, "ym:s:" + Attribution + "DirectPlatform"),
                DirectPlatformType = GetString(row, "ym:s:" + Attribution + "DirectPlatformType")
            });
        }

        return result;
    }

    private static string GetString(Dictionary<string, string> row, string key)
        => row.TryGetValue(key, out var value) ? value : null;

    private static ulong? GetUInt64(Dictionary<string, string> row, string key)
        => row.TryGetValue(key, out var value) &&
           ulong.TryParse(value, NumberStyles.Any, CultureInfo.InvariantCulture, out var number)
            ? number
            : null;

    private static bool? GetBool(Dictionary<string, string> row, string key)
    {
        if (!row.TryGetValue(key, out var value) || string.IsNullOrWhiteSpace(value))
            return null;

        return value == "1" ? true :
               value == "0" ? false :
               bool.TryParse(value, out var parsed) ? parsed : null;
    }

    private static DateTime? GetDateTime(Dictionary<string, string> row, string key)
    {
        if (!row.TryGetValue(key, out var value) || string.IsNullOrWhiteSpace(value))
            return null;

        return DateTime.TryParse(value, CultureInfo.InvariantCulture,
            DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal,
            out var dt) ? dt : null;
    }

    private sealed class YmVisitDto
    {
        public string ClientId { get; set; }
        public DateTime? DateTimeUtc { get; set; }
        public bool? IsNewUser { get; set; }

        public string TrafficSource { get; set; }
        public string AdvEngine { get; set; }
        public string Referer { get; set; }
        public string StartUrl { get; set; }

        public string UtmSource { get; set; }
        public string UtmMedium { get; set; }
        public string UtmCampaign { get; set; }
        public string UtmContent { get; set; }
        public string UtmTerm { get; set; }

        public ulong? DirectCampaignId { get; set; }
        public ulong? DirectAdGroupId { get; set; }
        public string DirectCampaignName { get; set; }
        public string DirectAdGroupName { get; set; }
        public string DirectPlatform { get; set; }
        public string DirectPlatformType { get; set; }
    }
}