@@ -0,0 +1,401 @@
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 消费费用补传（ ）
///
/// 背景：
/// 2 月仅成功上传 7,893 人 / 应有 20,972 人，缺失 13,078 人 / 5,333,866.15 元。
/// 排查与名单见 `_运维笔记/人脸平台/问题与解决汇总/20260913 *`（已三方核对一致）。
///
/// 与 <see cref="CostUploadJob"/> 的关系与差异：
/// 相同： （
/// EMealVoucher/Birthday、合计≠0）
/// BETRG=|合计|、只按 MSGTY=='S' 回写 UserBalanceUploadRecords；
/// 不同：① 目标年月写死为 2026-02（
/// ② 先一次 GROUP BY 聚合成"工号→金额"，不再逐人查询（原任务 N+1 查询是耗时主因）；
/// ③ **逐批 try/catch**，单批失败不中断后续（原任务缺此保护，是本次事故的根因）；
/// ④ 回写单独 try/catch，
/// ⑤ 响应工号按"去前导零"二次匹配，匹配不到的人进告警清单而不是被静默跳过；
/// ⑥ 数量闸门 + 时长预算 + 演练模式（见下方常量），避免误发整月数据或超 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 确认**，
/// 因此对请求异常/超时的批次**不做任何自动重试**（
/// 超时/未返回结果的批次必须先人工核对 JIP 侧是否已落库，再决定是否重跑本任务。
/// 2. 手工已补传但未回写上传记录的工号必须排除，见 <see cref="ManuallyUploadedUserCodes"/>。
/// 3. 上报金额沿用原任务写法（<c>Math.Abs(合计)</c>，未做 ROUND） ， ，
///
/// 当前状态（
/// - 执行结果：待补传 13,078 人 → 成功写入 13,076 人（ ， ） ；
/// 未写入 2 人：`30002488`(23.20 元， ， ；
/// 另有 `30011368`(388.00 元) 系 2026-03 用 Postman 手工补传、未回写平台记录（
/// - 屏蔽方式：<see cref="Disabled"/> = true ⇒ 任务即使被触发也只写一行警告日志、**不取数、不上传、不写库**；
/// 代码保留不删除，便于日后核对/复用。恢复方式：把 Disabled 改为 false 并重新发版。
/// - 注意：
/// 它只会在兜底时间（ ） ，
///
/// 历史执行方式（
/// - 触发入口：**https://face.jinkosolar.com/hangfire/**（走 ingress； ）
/// → `Recurring Jobs` → **勾选 `SupplementCostUploadJob` 这一行** → 列表上方出现 **`Trigger now`** 按钮 → 点它
/// （注意：不勾选时行内看不到按钮，容易误以为没有该功能）；
/// - 千万不要用"每天/每小时"这类会反复命中的 cron： ，
/// 但"未确认"批次（超时/未返回）会被反复重试，那是唯一会造成重复提交的场景；
/// - 本地调试需要注册时，给特性加 RunOnDevelopment = true（ ， ）
///
/// 补传完成后：**删除本文件**并从 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：
private const string SubTypeMealCardCost = "2086" ;
/// <summary>日志中最多打印多少个工号，避免刷爆日志</summary>
private const int MaxLogUserCodes = 500 ;
/// <summary>
/// 【安全闸门】待补传人数上限：预期 13,078 人（该月有消费 20,972 − − ）
/// 超过此上限说明取数口径异常（
/// </summary>
private const int MaxAllowedUserCount = 15000 ;
/// <summary>
/// 【安全闸门】总时长预算（ ）
/// （见 <see cref="Packs.HangfirePack"/>），超时任务会被判定"丢失"并重新入队；
/// 这里主动提前停止后续批次，避免越界，同时保证已处理部分已正确回写。
/// </summary>
private const int MaxRunMinutes = 90 ;
/// <summary>
/// 【演练模式】true = 只取数、只打日志，**不调用 JIP 接口、不写任何数据**。
/// 当前值 = <c>false</c>： （ ） ，
/// 若需退回演练：把本值改回 true，
/// 注意：故意声明为 static readonly（ ） ， ；
/// 改动本值后**必须重新编译并重新发版**才生效。
/// </summary>
private static readonly bool DryRun = false ;
/// <summary>
/// 已通过其它方式（
/// 依据：`20260913 2月费用上传缺失排查与补传方案.md` 第 6.1 节 —— 30011368 已于 2026-03 手工补传 388 元。
/// </summary>
private static readonly List < string > ManuallyUploadedUserCodes = new ( ) { "30011368" } ;
/// <summary>
/// 【总开关】true = **整体屏蔽**（不取数、不调 JIP、不写库） ，
/// 2026-09-30 补传完成后按运维要求置为 true， ；
/// 与 <see cref="DryRun"/> 的区别： ，
/// 同样故意声明为 static readonly（ ） ，
/// </summary>
private static readonly bool Disabled = true ;
public SupplementCostUploadJob ( IServiceProvider serviceProvider ) : base ( serviceProvider )
{
}
protected override async Task RunAsync ( IServiceProvider serviceProvider )
{
if ( Disabled )
{
Logger . LogWarning (
"【补传】本任务已屏蔽(Disabled=true)， ) ;
return ;
}
var jipApi = serviceProvider . GetService < IJinkoJipApi > ( ) ;
var stopwatch = Stopwatch . StartNew ( ) ;
// 目标区间： （ ， ）
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批； ,
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 = 0 m ;
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：
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}元； ,
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（ ） ,
7893 + writtenCount , string . Join ( ',' , ManuallyUploadedUserCodes ) ) ;
}
/// <summary>
/// 工号规范化：去空白 + 去前导零，用于容忍 JIP 回显工号与本地工号的格式差异
/// （ ： ）
/// </summary>
private static string NormalizeUserCode ( string userCode )
= > string . IsNullOrWhiteSpace ( userCode ) ? string . Empty : userCode . Trim ( ) . TrimStart ( '0' ) ;
}
