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;
}
}