Files
jinko.yikatong/Src/Dlhis.ScheduledTask/Jobs/SupplementCostUploadJob.cs

377 lines
21 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

using System.Diagnostics;
using Dlhis.Entity.Canteen.Entities;
using Dlhis.Entity.Canteen.Enums;
using Dlhis.Jip.Api;
using Dlhis.Jip.Api.Requests;
using Dlhis.Jip.Api.Responses;
using Dlhis.ScheduledTask.Attributes;
using Newtonsoft.Json;
using SqlSugar;
namespace Dlhis.ScheduledTask.Jobs;
/// <summary>
/// 【临时·一次性】2026-02 消费费用补传(2086)
///
/// 背景:2026-03-08 00:00 的 <see cref="CostUploadJob"/> 在 00:20 前后进程异常终止,
/// 2 月仅成功上传 7,893 人 / 应有 20,972 人,缺失 13,078 人 / 5,333,866.15 元。
/// 排查与名单见 `_运维笔记/人脸平台/问题与解决汇总/20260913 *`(已三方核对一致)。
///
/// 与 <see cref="CostUploadJob"/> 的关系与差异:
/// 相同:取数口径(Type='Out'、Status='Valid'、CostDate∈[月初,次月初)、Source 排除
/// EMealVoucher/Birthday、合计≠0)、批大小 100、SUBTY=2086、BEGDA=月末、
/// BETRG=|合计|、只按 MSGTY=='S' 回写 UserBalanceUploadRecords;
/// 不同:① 目标年月写死为 2026-02(原任务固定"运行当月-1",无法补历史月份);
/// ② 先一次 GROUP BY 聚合成"工号→金额",不再逐人查询(原任务 N+1 查询是耗时主因);
/// ③ **逐批 try/catch**,单批失败不中断后续(原任务缺此保护,是本次事故的根因);
/// ④ 回写单独 try/catch,与"JIP 请求失败"分开记,避免把"上传成功但回写失败"误报成请求异常;
/// ⑤ 响应工号按"去前导零"二次匹配,匹配不到的人进告警清单而不是被静默跳过;
/// ⑥ 数量闸门 + 时长预算 + 演练模式(见下方常量),避免误发整月数据或超 2 小时被 Hangfire 重入队;
/// ⑦ 结束输出分类汇总(成功/失败/未返回/请求异常/回写失败/未处理)。
///
/// 幂等性(可安全重复执行):
/// 1. 只在 UserBalanceUploadRecords 中**没有**(2026,2)记录的人里取数 → 已成功的 7,893 人自动跳过;
/// 2. 补传成功后写入 (UserCode,2026,2,Cost) → 再次执行时这些人自动被排除。
/// ⚠️ 因此**必须**只在 MSGTY=='S' 时回写;绝不可按"缺失集合"全量回写,
/// 否则失败者会被标记为"已上传"而永久排除、永不补传。
/// ⚠️ 写库用的工号必须是**本地的请求工号**(不是 JIP 回显的工号),否则与复核 SQL 关联不上。
///
/// 风险与约束(务必遵守):
/// 1. HCM_0015_INSERT 对同一 (ZWORKCODE,SUBTY,BEGDA) 是"覆盖"还是"累加"**尚未与 JIP 确认**,
/// 因此对请求异常/超时的批次**不做任何自动重试**(JipClientHandler 也未配置重试策略)。
/// 超时/未返回结果的批次必须先人工核对 JIP 侧是否已落库,再决定是否重跑本任务。
/// 2. 手工已补传但未回写上传记录的工号必须排除,见 <see cref="ManuallyUploadedUserCodes"/>。
/// 3. 上报金额沿用原任务写法(<c>Math.Abs(合计)</c>,未做 ROUND),与现网行为一致,勿擅自改成保留 2 位。
///
/// 执行方式(本任务 cron 设成"每年 1/1 仅兜底",不会误跑):
/// - 第 1 步:保持 <see cref="DryRun"/> = true 发版,触发一次,核对日志数字是否 = 13,078 人 / 5,333,866.15 元;
/// - 第 2 步:核对无误后把 <see cref="DryRun"/> 改为 false,重新发版,再触发一次(真实上传);
/// - 触发:Hangfire 面板(http://10.27.21.11:31005/hangfire,admin/dlhis123)
/// → Recurring Jobs → "【临时】2026-02消费费用补传" → Trigger now;
/// 若不用面板:把下方 cron 临时改成"只命中一次"的具体日期(如 "0 30 4 1 10 ?")随发版生效。
/// - 本地调试需要注册时,给特性加 RunOnDevelopment = true(默认 false,避免开发环境误跑真接口)。
///
/// 补传完成后:**删除本文件**并从 scheduled 重新发版(即方案中的"用完即拆")。
/// </summary>
[RecurringJob(nameof(SupplementCostUploadJob), "0 0 0 1 1 ?", Name = "【临时】2026-02消费费用补传")]
public class SupplementCostUploadJob : BaseJob
{
/// <summary>目标数据所属年月(写死:只补 2026 年 2 月)</summary>
private const int TargetYear = 2026;
private const int TargetMonth = 2;
/// <summary>单批条数(与 <see cref="CostUploadJob"/> 一致)</summary>
private const int BatchSize = 100;
/// <summary>SUBTY:2086 餐卡消费金额</summary>
private const string SubTypeMealCardCost = "2086";
/// <summary>日志中最多打印多少个工号,避免刷爆日志</summary>
private const int MaxLogUserCodes = 500;
/// <summary>
/// 【安全闸门】待补传人数上限:预期 13,078 人(该月有消费 20,972 − 已上传 7,893 − 手工补传 1)。
/// 超过此上限说明取数口径异常(例如区间写错、IgnoreSources 失效),直接中止、不发起任何上传。
/// </summary>
private const int MaxAllowedUserCount = 15000;
/// <summary>
/// 【安全闸门】总时长预算(分钟)。Hangfire 的 InvisibilityTimeout = 2 小时
/// (见 <see cref="Packs.HangfirePack"/>),超时任务会被判定"丢失"并重新入队;
/// 这里主动提前停止后续批次,避免越界,同时保证已处理部分已正确回写。
/// </summary>
private const int MaxRunMinutes = 90;
/// <summary>
/// 【演练模式】true = 只取数、只打日志,**不调用 JIP 接口、不写任何数据**。
/// 首次必须先用 true 跑一遍,确认日志为 13,078 人 / 5,333,866.15 元 / 131 批,再改为 false 执行真实补传。
/// 注意:故意声明为 static readonly(非 const),否则编译器会把 switch 之后的代码判为"不可达"而报 CS0162;
/// 改动本值后**必须重新编译并重新发版**才生效。
/// </summary>
private static readonly bool DryRun = true;
/// <summary>
/// 已通过其它方式(Postman 手工)补传过、但**未回写** UserBalanceUploadRecords 的工号,必须排除,否则会重复提交。
/// 依据:`20260913 2月费用上传缺失排查与补传方案.md` 第 6.1 节 —— 30011368 已于 2026-03 手工补传 388 元。
/// </summary>
private static readonly List<string> ManuallyUploadedUserCodes = new() { "30011368" };
public SupplementCostUploadJob(IServiceProvider serviceProvider) : base(serviceProvider)
{
}
protected override async Task RunAsync(IServiceProvider serviceProvider)
{
var jipApi = serviceProvider.GetService<IJinkoJipApi>();
var stopwatch = Stopwatch.StartNew();
// 目标区间:2026-02-01 ~ 2026-03-01(必须单月,跨月会得到多个月相加)
var s = new DateTime(TargetYear, TargetMonth, 1);
var e = s.AddMonths(1);
var endDayOfMonth = e.AddDays(-1); // 2026-02-28 → BEGDA
var beginDate = endDayOfMonth.ToString("yyyy-MM-dd");
// 1) 该月已有上传记录的人(成功过一次就跳过,这是幂等的唯一依据)
var uploadedUserCodes = await DbContext.Queryable<UserBalanceUploadRecordEntity>()
.Where(x => x.Year == TargetYear && x.Month == TargetMonth)
.Select(x => x.UserCode)
.ToListAsync();
var uploadedSet = uploadedUserCodes.Where(x => !string.IsNullOrEmpty(x)).ToHashSet();
// 2) 一次聚合出"该月消费合计≠0"的工号与金额(口径与 CostUploadJob / 排查 SQL 完全一致)
var amounts = await DbContext.Queryable<BalanceRecordEntity>()
.Where(x => x.Status == BalanceRecordStatusEnumType.Valid)
.Where(x => x.Type == BalanceRecordTypeEnumType.Out)
.Where(x => x.CostDate >= s && x.CostDate < e)
.Where(x => !BalanceRecordSourceConstants.IgnoreSources.Contains(x.Source))
.GroupBy(x => x.UserCode)
.Select(x => new
{
x.UserCode,
Amount = SqlFunc.AggregateSum(x.Amount)
})
.ToListAsync();
// 3) 过滤:排除已上传、已手工补传、金额为 0 的人
var toUpload = amounts
.Where(x => !string.IsNullOrEmpty(x.UserCode))
.Where(x => x.Amount != 0)
.Where(x => !uploadedSet.Contains(x.UserCode))
.Where(x => !ManuallyUploadedUserCodes.Contains(x.UserCode))
.OrderBy(x => x.UserCode, StringComparer.Ordinal)
.ToList();
var totalAmount = toUpload.Sum(x => Math.Abs(x.Amount));
var batchTotal = (toUpload.Count + BatchSize - 1) / BatchSize;
Logger.LogInformation(
"【补传】开始(演练模式={dryRun}):区间[{s},{e}) BEGDA={begda} 该月有消费人数={all} 已上传跳过={uploaded} 本次待补传={count}人/合计{amount}元 共{batch}批",
DryRun, s.ToString("yyyy-MM-dd"), e.ToString("yyyy-MM-dd"), beginDate,
amounts.Count(x => x.Amount != 0), uploadedSet.Count, toUpload.Count, Math.Round(totalAmount, 2), batchTotal);
if (!toUpload.Any())
{
Logger.LogInformation("【补传】无待补传人员(若属预期,说明 2 月已补全或已回写完毕)");
return;
}
// 4) 数量闸门:口径异常时宁可不动,也不要误发整月数据
if (toUpload.Count > MaxAllowedUserCount)
{
Logger.LogError(
"【补传】待补传人数{count}超过安全上限{max},已中止(未上传任何数据)。请先核对取数口径与 UserBalanceUploadRecords 现状!",
toUpload.Count, MaxAllowedUserCount);
return;
}
// 5) 演练模式:只输出统计,不调 JIP、不写库
if (DryRun)
{
Logger.LogWarning(
"【补传·演练模式】未调用任何 JIP 接口、未写入任何数据!期望值:13078人/5333866.15元/131批;实际:{count}人/{amount}元/{batch}批;样例(前10):{sample}",
toUpload.Count, Math.Round(totalAmount, 2), batchTotal,
string.Join(',', toUpload.Take(10).Select(x => $"{x.UserCode}={Math.Abs(x.Amount)}")));
Logger.LogWarning("【补传·演练模式】核对无误后,请把 DryRun 改为 false 并重新发版,再触发一次执行真实补传");
return;
}
var processedCount = 0; // 已处理(进入循环)的人数
var writtenCount = 0; // 已成功上传并回写记录的人数
var writtenAmount = 0m;
var batchNo = 0;
var failedUserCodes = new List<string>(); // MSGTY != 'S'
var unreportedUserCodes = new List<string>(); // JIP 未返回结果(未确认)
var unmatchedUserCodes = new List<string>(); // 响应工号无法对应到本批请求(未确认)
var emptyResponseBatches = new List<int>(); // 响应为空(未确认)
var errorBatches = new List<string>(); // JIP 请求异常/超时(未确认,禁止盲目重发)
var writeBackFailed = new List<string>(); // JIP 成功但回写失败(重跑前必须核对)
foreach (var items in toUpload.Chunk(BatchSize))
{
// 6) 时长预算:避免单次运行超过 Hangfire 的 InvisibilityTimeout(2h) 被重新入队
if (stopwatch.Elapsed.TotalMinutes > MaxRunMinutes)
{
Logger.LogWarning(
"【补传】已运行{elapsed}分钟,超过预算{max}分钟,主动停止后续批次:剩余{rest}人未处理(已成功者已回写,重跑本任务会自动跳过)",
Math.Round(stopwatch.Elapsed.TotalMinutes), MaxRunMinutes, toUpload.Count - processedCount);
break;
}
batchNo++;
processedCount += items.Length;
var request = new UploadCostRequest
{
Data = items.Select(x => new UploadCostItemRequest
{
WorkCode = x.UserCode,
Code = SubTypeMealCardCost,
BeginDate = beginDate,
Charge = Math.Abs(x.Amount)
}).ToList()
};
UploadCostResponse response;
try
{
response = await jipApi.UploadCostAsync(request);
}
catch (Exception ex)
{
// 单批异常不中断后续批次;且不自动重试(幂等语义未确认)
errorBatches.Add($"#{batchNo}({items.First().UserCode}~{items.Last().UserCode})");
Logger.LogError(ex,
"【补传】第{batch}/{total}批 JIP请求异常,未回写(该批{count}人;若为超时,请先核对JIP侧是否已落库,再决定是否重跑):{request}",
batchNo, batchTotal, request.Data.Count, JsonConvert.SerializeObject(request));
continue;
}
var responseItems = response?.Data ?? new List<UploadCostItemResponse>();
if (responseItems.Count == 0)
{
emptyResponseBatches.Add(batchNo);
Logger.LogWarning("【补传】第{batch}/{total}批 响应为空,未回写(该批{count}人需人工确认):{request}",
batchNo, batchTotal, request.Data.Count, JsonConvert.SerializeObject(request));
continue;
}
// 7) 匹配:按"去前导零"二次匹配,容忍 JIP 回显补零(接口样例:请求 343 → 响应 00000343)
var successKeys = new HashSet<string>();
foreach (var resp in responseItems.Where(x => x.Success == "S"))
{
successKeys.Add(NormalizeUserCode(resp.WorkCode));
}
var failItems = responseItems.Where(x => x.Success != "S").ToList();
failedUserCodes.AddRange(failItems.Select(x => $"{x.WorkCode}:{x.Message}"));
var respondedKeys = responseItems.Select(x => NormalizeUserCode(x.WorkCode)).ToHashSet();
var requestKeys = items.Select(x => NormalizeUserCode(x.UserCode)).ToHashSet();
// 响应里出现的、不属于本批请求的工号 → 告警(不回写,避免张冠李戴)
unmatchedUserCodes.AddRange(responseItems
.Select(x => x.WorkCode)
.Where(x => !requestKeys.Contains(NormalizeUserCode(x)))
.Distinct());
// 本批请求了、但 JIP 既没给 S 也没给 E 的人 → 未确认
unreportedUserCodes.AddRange(items
.Where(x => !respondedKeys.Contains(NormalizeUserCode(x.UserCode)))
.Select(x => x.UserCode));
// 只回写"本批请求里被判定成功"的人,且**用本地请求工号**写库(保证与复核 SQL 可关联)
var writeItems = items.Where(x => successKeys.Contains(NormalizeUserCode(x.UserCode))).ToList();
if (unreportedUserCodes.Any() || unmatchedUserCodes.Any())
{
Logger.LogWarning("【补传】第{batch}/{total}批 结果不完整:未返回{no}人、响应异常工号{un}个(已记入告警清单,未回写)",
batchNo, batchTotal, unreportedUserCodes.Count, unmatchedUserCodes.Count);
}
if (!writeItems.Any())
{
Logger.LogWarning("【补传】第{batch}/{total}批 无成功记录,未回写;失败{count}项:{fail}",
batchNo, batchTotal, failItems.Count,
string.Join(',', failItems.Take(MaxLogUserCodes).Select(x => $"{x.WorkCode}:{x.Message}")));
continue;
}
// 8) 回写单独 try/catch:把"JIP 已成功但回写失败"与"JIP 请求失败"区分开
try
{
var workCodes = writeItems.Select(x => x.UserCode).ToList();
var existRecords = await DbContext.Queryable<UserBalanceUploadRecordEntity>()
.Where(x => workCodes.Contains(x.UserCode))
.Where(x => x.Year == TargetYear && x.Month == TargetMonth)
.ToListAsync();
var updatedRecords = new List<UserBalanceUploadRecordEntity>();
var newRecords = new List<UserBalanceUploadRecordEntity>();
foreach (var item in writeItems)
{
var cost = Math.Abs(item.Amount);
var record = existRecords.FirstOrDefault(x => x.UserCode == item.UserCode);
if (record != null)
{
record.Cost = cost;
updatedRecords.Add(record);
}
else
{
newRecords.Add(new UserBalanceUploadRecordEntity
{
UserCode = item.UserCode,
Year = TargetYear,
Month = TargetMonth,
Cost = cost
});
}
}
if (updatedRecords.Any())
{
await DbContext.UpdateAsync(updatedRecords);
}
if (newRecords.Any())
{
await DbContext.InsertAsync(newRecords);
}
writtenCount += writeItems.Count;
writtenAmount += writeItems.Sum(x => Math.Abs(x.Amount));
Logger.LogInformation("【补传】第{batch}/{total}批 成功{ok}人 失败{fail}人 累计({done}/{all})",
batchNo, batchTotal, writeItems.Count, failItems.Count, writtenCount, toUpload.Count);
}
catch (Exception ex)
{
// JIP 侧多半已成功;这里不回滚、不重试,只告警——重跑前必须先核对 JIP 侧
writeBackFailed.Add($"#{batchNo}({writeItems.Count}人)");
Logger.LogError(ex,
"【补传】第{batch}/{total}批 JIP已返回成功但**回写上传记录失败**:{codes} → 这些人在JIP侧可能已成功,重跑前必须先核对,否则会重复提交!",
batchNo, batchTotal, string.Join(',', writeItems.Select(x => x.UserCode)));
}
}
// 9) 分类汇总(便于核对"应上传/成功/各类未确认")
Logger.LogInformation(
"【补传】结束:待补传{all}人/{amount}元 → 成功回写{ok}人/{okAmount}元;MSGTY≠S {fail}人 未返回结果{no}人 响应异常工号{un}个 响应为空{empty}批 JIP请求异常{err}批 回写失败{wb}批 未处理{rest}人;耗时{elapsed}分钟",
toUpload.Count, Math.Round(totalAmount, 2), writtenCount, Math.Round(writtenAmount, 2),
failedUserCodes.Count, unreportedUserCodes.Count, unmatchedUserCodes.Count,
emptyResponseBatches.Count, errorBatches.Count, writeBackFailed.Count,
toUpload.Count - processedCount, Math.Round(stopwatch.Elapsed.TotalMinutes));
if (failedUserCodes.Any())
{
Logger.LogWarning("【补传】MSGTY≠S 工号({count}):{codes}",
failedUserCodes.Count, string.Join(',', failedUserCodes.Take(MaxLogUserCodes)));
}
if (unreportedUserCodes.Any() || unmatchedUserCodes.Any())
{
Logger.LogWarning("【补传】未确认工号 未返回结果({no}):{noCodes} | 响应异常({un}):{unCodes}",
unreportedUserCodes.Count, string.Join(',', unreportedUserCodes.Take(MaxLogUserCodes)),
unmatchedUserCodes.Count, string.Join(',', unmatchedUserCodes.Take(MaxLogUserCodes)));
}
if (emptyResponseBatches.Any() || errorBatches.Any() || writeBackFailed.Any())
{
Logger.LogWarning(
"【补传】未确认批次 响应为空[{empty}] JIP请求异常[{err}] 回写失败[{wb}] → **先核对 JIP 侧再决定是否重跑**",
string.Join(',', emptyResponseBatches), string.Join(',', errorBatches), string.Join(',', writeBackFailed));
}
Logger.LogInformation(
"【补传】请按 `20260913 每月费用上传完整性核对SQL.sql` 复核:该月记录数应达 {expect} 左右(7893+本次成功);剩余缺失应回归 0(或仅剩 {manual})",
7893 + writtenCount, string.Join(',', ManuallyUploadedUserCodes));
}
/// <summary>
/// 工号规范化:去空白 + 去前导零,用于容忍 JIP 回显工号与本地工号的格式差异
/// (依据:SI 接口样例中请求 PERNR=343 → 响应 PERNR=00000343)。
/// </summary>
private static string NormalizeUserCode(string userCode)
=> string.IsNullOrWhiteSpace(userCode) ? string.Empty : userCode.Trim().TrimStart('0');
}