AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Executors/MdpApiPullExecutor.cs

using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using SqlSugar;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Executors;

/// <summary>
/// 方式乙：按 mdp_source API 段 + mdp_entity.source_api_path 拉取 JSON → target_table_name（贴源）。
/// </summary>
public sealed class MdpApiPullExecutor : IMdpSourcePullExecutor, ITransient
{
    public string SupportedType => "API_PULL";

    private readonly IHttpClientFactory _httpClientFactory;
    private readonly ISqlSugarClient _db;
    private readonly MdpStagingWriter _writer;

    public MdpApiPullExecutor(IHttpClientFactory httpClientFactory, ISqlSugarClient db, MdpStagingWriter writer)
    {
        _httpClientFactory = httpClientFactory;
        _db = db;
        _writer = writer;
    }

    public async Task<MdpPullResult> PullAsync(MdpSource source, MdpEntity entity, MdpPullContext ctx, CancellationToken cancellationToken = default)
    {
        if (string.IsNullOrWhiteSpace(source.ApiBaseUrl))
            throw new InvalidOperationException($"源 {source.SourceCode} 未配置 api_base_url");
        if (string.IsNullOrWhiteSpace(entity.SourceApiPath))
            throw new InvalidOperationException($"实体 {entity.EntityCode} 未配置 source_api_path");
        if (string.IsNullOrWhiteSpace(entity.TargetTableName))
            throw new InvalidOperationException($"实体 {entity.EntityCode} 未配置 target_table_name");

        var client = _httpClientFactory.CreateClient("MdpApiPull");
        client.Timeout = TimeSpan.FromSeconds(120);

        var url = CombineUrl(source.ApiBaseUrl!, entity.SourceApiPath!);
        if (!string.IsNullOrWhiteSpace(entity.LastCursor) && !ctx.FullRefresh)
            url += (url.Contains('?') ? "&" : "?") + "cursor=" + Uri.EscapeDataString(entity.LastCursor);

        using var request = new HttpRequestMessage(HttpMethod.Get, url);
        ApplyAuth(request, source);

        using var response = await client.SendAsync(request, cancellationToken);
        var body = await response.Content.ReadAsStringAsync(cancellationToken);
        if (!response.IsSuccessStatusCode)
            throw new InvalidOperationException($"API_PULL 失败 HTTP {(int)response.StatusCode}: {Truncate(body, 500)}");

        using var doc = JsonDocument.Parse(string.IsNullOrWhiteSpace(body) ? "[]" : body);
        var items = ResolveArray(doc.RootElement, entity.ResponseDataPath);
        var written = 0;
        var now = DateTime.Now;
        string? newCursor = entity.LastCursor;
        var seen = new HashSet<string>(StringComparer.Ordinal);
        var pending = new List<(IDictionary<string, object?> Row, string RawJson, string SourceRowId)>();

        foreach (var item in items)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var dedup = ResolvePath(item, entity.DedupKeyPath) ?? Guid.NewGuid().ToString("N");
            if (!seen.Add(dedup)) continue;

            var rawJson = item.GetRawText();
            var dict = JsonElementToDict(item);
            pending.Add((dict, rawJson, dedup));
            newCursor = dedup;
        }

        if (pending.Count > 0)
        {
            // 贴源 source_table 优先用逻辑源表名（与 DB 路线/std 过滤一致）；未配置时回落 API path
            var stagingTable = !string.IsNullOrWhiteSpace(entity.SourceTableName)
                ? entity.SourceTableName!
                : entity.SourceApiPath!;
            var results = await _writer.UpsertBatchAsync(source, entity, stagingTable, pending, ctx);
            written += results.Sum();
        }

        if (!string.IsNullOrEmpty(newCursor) && newCursor != entity.LastCursor)
        {
            await _db.Updateable<MdpEntity>()
                .SetColumns(x => new MdpEntity
                {
                    LastCursor = newCursor,
                    LastSyncTo = now,
                    UpdateTime = now
                })
                .Where(x => x.Id == entity.Id)
                .ExecuteCommandAsync(cancellationToken);
        }

        return new MdpPullResult
        {
            RowsPulled = items.Count,
            RowsWritten = written,
            NewCursor = newCursor,
            Message = "OK"
        };
    }

    private static void ApplyAuth(HttpRequestMessage request, MdpSource source)
    {
        var authType = (source.ApiAuthType ?? "NONE").Trim().ToUpperInvariant();
        if (authType is "NONE" or "") return;

        Dictionary<string, string>? cfg = null;
        if (!string.IsNullOrWhiteSpace(source.ApiAuthConfig))
        {
            try
            {
                cfg = JsonSerializer.Deserialize<Dictionary<string, string>>(source.ApiAuthConfig!);
            }
            catch { /* ignore */ }
        }
        cfg ??= new Dictionary<string, string>();

        switch (authType)
        {
            case "BEARER":
            case "TOKEN":
                if (cfg.TryGetValue("token", out var token) || cfg.TryGetValue("access_token", out token))
                    request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
                break;
            case "BASIC":
                if (cfg.TryGetValue("username", out var user) && cfg.TryGetValue("password", out var pwd))
                {
                    var bytes = Encoding.UTF8.GetBytes($"{user}:{pwd}");
                    request.Headers.Authorization = new AuthenticationHeaderValue("Basic", Convert.ToBase64String(bytes));
                }
                break;
            case "APIKEY":
                var header = cfg.GetValueOrDefault("header") ?? "X-API-Key";
                if (cfg.TryGetValue("apiKey", out var key) || cfg.TryGetValue("key", out key))
                    request.Headers.TryAddWithoutValidation(header, key);
                break;
            case "OAUTH2":
                if (cfg.TryGetValue("access_token", out var oauth))
                    request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", oauth);
                break;
        }
    }

    private static List<JsonElement> ResolveArray(JsonElement root, string? path)
    {
        var el = string.IsNullOrWhiteSpace(path) ? root : ResolveElement(root, path!) ?? root;
        if (el.ValueKind == JsonValueKind.Array)
            return el.EnumerateArray().Select(x => x.Clone()).ToList();
        if (el.ValueKind == JsonValueKind.Object)
            return new List<JsonElement> { el.Clone() };
        return new List<JsonElement>();
    }

    private static JsonElement? ResolveElement(JsonElement root, string path)
    {
        var cur = root;
        foreach (var part in path.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
        {
            if (cur.ValueKind != JsonValueKind.Object || !cur.TryGetProperty(part, out cur))
                return null;
        }
        return cur;
    }

    private static string? ResolvePath(JsonElement el, string? path)
    {
        if (string.IsNullOrWhiteSpace(path)) return null;
        var found = ResolveElement(el, path);
        if (found == null) return null;
        return found.Value.ValueKind switch
        {
            JsonValueKind.String => found.Value.GetString(),
            JsonValueKind.Number => found.Value.ToString(),
            JsonValueKind.True => "true",
            JsonValueKind.False => "false",
            _ => found.Value.ToString()
        };
    }

    private static string CombineUrl(string baseUrl, string path)
    {
        baseUrl = baseUrl.TrimEnd('/');
        path = path.StartsWith('/') ? path : "/" + path;
        return baseUrl + path;
    }

    private static string Truncate(string s, int max) =>
        string.IsNullOrEmpty(s) ? "" : (s.Length <= max ? s : s[..max]);

    private static Dictionary<string, object?> JsonElementToDict(JsonElement el)
    {
        var dict = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
        if (el.ValueKind != JsonValueKind.Object) return dict;
        foreach (var prop in el.EnumerateObject())
        {
            dict[prop.Name] = prop.Value.ValueKind switch
            {
                JsonValueKind.Null => null,
                JsonValueKind.String => prop.Value.GetString(),
                JsonValueKind.Number => prop.Value.TryGetInt64(out var l) ? l
                    : prop.Value.TryGetDecimal(out var d) ? d
                    : prop.Value.GetDouble(),
                JsonValueKind.True => true,
                JsonValueKind.False => false,
                _ => prop.Value.GetRawText()
            };
        }
        return dict;
    }
}