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; /// /// 方式乙:按 mdp_source API 段 + mdp_entity.source_api_path 拉取 JSON → target_table_name(贴源)。 /// 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 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(StringComparer.Ordinal); 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); // 贴源 source_table 优先用逻辑源表名(与 DB 路线/std 过滤一致);未配置时回落 API path var stagingTable = !string.IsNullOrWhiteSpace(entity.SourceTableName) ? entity.SourceTableName! : entity.SourceApiPath!; written += await _writer.UpsertAsync( source, entity, stagingTable, dict, rawJson, dedup, ctx); newCursor = dedup; } if (!string.IsNullOrEmpty(newCursor) && newCursor != entity.LastCursor) { await _db.Updateable() .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? cfg = null; if (!string.IsNullOrWhiteSpace(source.ApiAuthConfig)) { try { cfg = JsonSerializer.Deserialize>(source.ApiAuthConfig!); } catch { /* ignore */ } } cfg ??= new Dictionary(); 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 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 { el.Clone() }; return new List(); } 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 JsonElementToDict(JsonElement el) { var dict = new Dictionary(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; } }