feat(数据修复): 修正淘宝覆盖规则命中的转链成功记录

新增 /Test/RepairTbOverrideSuccess（默认 dryRun），按日期和节点将覆盖规则命中记录（tb、accountId=0、itemId 为空）改为失败：先修正节点 Redis 的 parse_total 与 parse_metric 计数，再更新日表、dp 日表、测试表并删除 success 汇总表记录。
抽出 dp 标记判定、统计 key 过期时间和 parse_metric success key 构造供修复复用。

molilian.api/Controllers/public/TestController.cs

@@ -1222,6 +1222,322 @@ ORDER BY l.accountId, l.end_point",
             });
         }
 
+        /// <summary>
+        /// 将淘宝覆盖规则命中产生的转链记录从 success=1 修正为 success=0，并同步修正各节点 Redis 统计。
+        /// 覆盖规则在选择账号前命中，记录特征：channel=tb、linkType=unknown、accountId=0、itemId 为空。
+        /// 按「日期 + 节点」处理：先修正该节点 Redis，成功后再更新对应 MySQL 记录；已修正的记录不会被重复计入。
+        /// </summary>
+        [HttpGet]
+        public async Task<ActionResult> RepairTbOverrideSuccess(
+            DateTime startTime = default,
+            DateTime endTime = default,
+            bool dryRun = true,
+            int commandTimeoutSeconds = 600)
+        {
+            if (startTime == default) startTime = new DateTime(2026, 10, 20, 18, 52, 42);
+            if (endTime == default) endTime = DateTime.Now;
+            commandTimeoutSeconds = Math.Clamp(commandTimeoutSeconds, 30, 3600);
+            if (endTime < startTime)
+            {
+                return new APIResult(new { success = false, message = "endTime 不能早于 startTime" });
+            }
+
+            const string condition = @"channel = @channel
+  AND linkType = @linkType
+  AND success = 1
+  AND accountId = 0
+  AND COALESCE(itemId,'') = ''
+  AND create_time BETWEEN @from AND @to";
+
+            using var conn = CenterHub.GetOpenConnection();
+            if (conn.State != ConnectionState.Open) conn.Open();
+
+            // 与 TkLogCore.GetTotalAsync 读取统计时的节点范围保持一致
+            var endpoints = EndPointCore.List()
+                .Where(node => node.is_public_api && !string.IsNullOrEmpty(EndPointCore.GetRedisServer(node)))
+                .GroupBy(node => node.name, StringComparer.Ordinal)
+                .ToDictionary(group => group.Key, group => group.First(), StringComparer.Ordinal);
+
+            var errors = new List<object>();
+            var dayReports = new List<object>();
+            foreach (var date in EachDay(startTime.Date, endTime.Date))
+            {
+                DateTime from = date < startTime ? startTime : date;
+                DateTime dayEnd = date.AddDays(1).AddTicks(-1);
+                DateTime to = dayEnd > endTime ? endTime : dayEnd;
+                string tableName = $"tk_parse_logs_{date:yyyyMMdd}";
+
+                try
+                {
+                    if (!TableExists(conn, tableName, commandTimeoutSeconds))
+                    {
+                        dayReports.Add(new { date = date.ToString("yyyy-MM-dd"), tableName, skipped = true, reason = "table not exists" });
+                        continue;
+                    }
+
+                    var args = new { channel = (int)TkChannelEnum.tb, linkType = (int)LinkTypeEnum.unknown, from, to };
+                    var groups = SqlMapper.Query<TbOverrideRepairGroupRow>(
+                        conn,
+                        $@"
+SELECT
+    COALESCE(end_point,'') end_point,
+    DATE_FORMAT(create_time,'%Y%m%d%H') hour_key,
+    COALESCE(deeplink_url,'') deeplink_url,
+    COALESCE(riskStrategy,'') riskStrategy,
+    launchScene,
+    COUNT(*) row_count
+FROM {tableName}
+WHERE {condition}
+GROUP BY COALESCE(end_point,''), DATE_FORMAT(create_time,'%Y%m%d%H'), COALESCE(deeplink_url,''), COALESCE(riskStrategy,''), launchScene",
+                        args,
+                        commandTimeout: commandTimeoutSeconds).ToList();
+
+                    var summary = SqlMapper.Query<TbOverrideRepairSummaryRow>(
+                        conn,
+                        $@"
+SELECT CAST(subCode AS CHAR) subCode, COALESCE(message,'') message, COALESCE(reason,'') reason, COUNT(*) row_count
+FROM {tableName}
+WHERE {condition}
+GROUP BY CAST(subCode AS CHAR), COALESCE(message,''), COALESCE(reason,'')
+ORDER BY row_count DESC",
+                        args,
+                        commandTimeout: commandTimeoutSeconds).ToList();
+
+                    var topDeeplinks = groups
+                        .GroupBy(item => item.deeplink_url)
+                        .Select(group => new { deeplink_url = group.Key, row_count = group.Sum(item => item.row_count) })
+                        .OrderByDescending(item => item.row_count)
+                        .Take(20)
+                        .ToList();
+
+                    var endpointReports = new List<object>();
+                    foreach (var endpointGroup in groups.GroupBy(item => item.end_point))
+                    {
+                        long rowCount = endpointGroup.Sum(item => item.row_count);
+                        if (!endpoints.TryGetValue(endpointGroup.Key, out var endpoint))
+                        {
+                            errors.Add(new { scope = "endpoint_not_found", date = date.ToString("yyyy-MM-dd"), end_point = endpointGroup.Key, rowCount });
+                            continue;
+                        }
+
+                        var deltas = BuildTbOverrideRepairDeltas(endpointGroup);
+                        var keyReports = new List<object>();
+                        int appliedKeys = 0;
+                        Dictionary<string, int> mysql;
+                        try
+                        {
+                            await using var scope = await RedisClientFactory.CreateScopeAsync(EndPointCore.GetRedisServer(endpoint));
+                            var redis = scope.Client;
+
+                            foreach (var (key, (delta, expireSeconds)) in deltas)
+                            {
+                                long before = await redis.GetAsync<long>(key);
+                                // 只减到 0 为止，避免计数出现负数
+                                long applied = delta < 0 ? -Math.Min(-delta, Math.Max(before, 0)) : delta;
+
+                                if (!dryRun && applied != 0)
+                                {
+                                    await redis.IncrByAsync(key, applied);
+                                    if (before == 0 && applied > 0 && expireSeconds > 0)
+                                    {
+                                        await redis.ExpireAsync(key, expireSeconds);
+                                    }
+                                    appliedKeys++;
+                                }
+
+                                // 小时级 key 数量较多，只在报告中列出日、月级 key
+                                if (!IsHourlyStatsKey(key))
+                                {
+                                    keyReports.Add(new { key, before, delta, applied });
+                                }
+                            }
+                        }
+                        catch (Exception ex)
+                        {
+                            errors.Add(new
+                            {
+                                scope = "redis_endpoint",
+                                date = date.ToString("yyyy-MM-dd"),
+                                end_point = endpoint.name,
+                                appliedKeys,
+                                totalKeys = deltas.Count,
+                                error = FormatRepairError(ex)
+                            });
+                            continue;
+                        }
+
+                        try
+                        {
+                            mysql = RepairTbOverrideMysqlRows(conn, date, condition, args, endpoint.name, dryRun, commandTimeoutSeconds);
+                        }
+                        catch (Exception ex)
+                        {
+                            errors.Add(new { scope = "mysql_endpoint", date = date.ToString("yyyy-MM-dd"), end_point = endpoint.name, redisApplied = !dryRun, error = FormatRepairError(ex) });
+                            continue;
+                        }
+
+                        endpointReports.Add(new
+                        {
+                            end_point = endpoint.name,
+                            rowCount,
+                            totalKeys = deltas.Count,
+                            appliedKeys,
+                            mysql,
+                            keys = keyReports
+                        });
+                    }
+
+                    dayReports.Add(new
+                    {
+                        date = date.ToString("yyyy-MM-dd"),
+                        from,
+                        to,
+                        rowCount = groups.Sum(item => item.row_count),
+                        summary,
+                        topDeeplinks,
+                        endpoints = endpointReports
+                    });
+                }
+                catch (Exception ex)
+                {
+                    errors.Add(new { scope = "day", date = date.ToString("yyyy-MM-dd"), tableName, error = FormatRepairError(ex) });
+                }
+            }
+
+            return new APIResult(new
+            {
+                success = errors.Count == 0,
+                dryRun,
+                startTime,
+                endTime,
+                days = dayReports,
+                errors
+            });
+        }
+
+        /// <summary>
+        /// 计算单个节点需要修正的 Redis 计数增量，对应 TkLogCore.ParseLogAsync 中 success 相关的写入：
+        /// parse_total 的 total/success/fail（dp_{flag} 可能由 success 变为 fail）以及 parse_metric 的 success。
+        /// </summary>
+        private static Dictionary<string, (long delta, int expireSeconds)> BuildTbOverrideRepairDeltas(IEnumerable<TbOverrideRepairGroupRow> rows)
+        {
+            const string channel = "tb";
+            var deltas = new Dictionary<string, (long delta, int expireSeconds)>(StringComparer.Ordinal);
+
+            void Add(string key, long delta, int expireSeconds)
+            {
+                deltas.TryGetValue(key, out var current);
+                deltas[key] = (current.delta + delta, expireSeconds);
+            }
+
+            foreach (var row in rows)
+            {
+                var metricTime = DateTime.ParseExact(row.hour_key, "yyyyMMddHH", null);
+                string[] timeKeys = [metricTime.ToString("yyyyMM"), metricTime.ToString("yyyyMMdd"), row.hour_key];
+
+                string oldFlag = TkLogCore.GetParseDeeplinkFlag(row.deeplink_url, true);
+                string newFlag = TkLogCore.GetParseDeeplinkFlag(row.deeplink_url, false);
+                string[] oldBuckets = [$"dp_{oldFlag}:all", $"dp_{oldFlag}:{channel}", "all", $"dp_channel:{channel}", channel];
+                string[] newBuckets = [$"dp_{newFlag}:all", $"dp_{newFlag}:{channel}", "all", $"dp_channel:{channel}", channel];
+
+                foreach (string timeKey in timeKeys)
+                {
+                    int expireSeconds = TkLogCore.GetStatsExpireSeconds(timeKey);
+                    foreach (string bucket in oldBuckets)
+                    {
+                        Add($":parse_total:{bucket}:{timeKey}", -row.row_count, expireSeconds);
+                        Add($":parse_total:{bucket}:success:{timeKey}", -row.row_count, expireSeconds);
+                    }
+                    foreach (string bucket in newBuckets)
+                    {
+                        Add($":parse_total:{bucket}:{timeKey}", row.row_count, expireSeconds);
+                        Add($":parse_total:{bucket}:fail:{timeKey}", row.row_count, expireSeconds);
+                    }
+                }
+
+                foreach (string key in TracksCore.BuildParseMetricSuccessKeys(metricTime, channel, row.riskStrategy, row.launchScene))
+                {
+                    Add(key, -row.row_count, 0);
+                }
+            }
+
+            return deltas
+                .Where(item => item.Value.delta != 0)
+                .ToDictionary(item => item.Key, item => item.Value, StringComparer.Ordinal);
+        }
+
+        private static bool IsHourlyStatsKey(string key)
+        {
+            int index = key.LastIndexOf(':');
+            string last = index >= 0 ? key[(index + 1)..] : key;
+            if (last.Length == 10 && last.All(char.IsDigit)) return true;
+            // parse_metric key 的时间段位于中间：:parse_metric:hour:success:{yyyyMMddHH}:...
+            return key.Contains(":parse_metric:hour:", StringComparison.Ordinal);
+        }
+
+        /// <summary>
+        /// 修正（dryRun 时仅统计）单个节点在指定日期的覆盖规则记录：
+        /// 日表、dp 日表、测试表改为 success=0；success 汇总表删除对应记录。
+        /// </summary>
+        private static Dictionary<string, int> RepairTbOverrideMysqlRows(
+            IDbConnection conn,
+            DateTime date,
+            string condition,
+            object args,
+            string endPoint,
+            bool dryRun,
+            int commandTimeoutSeconds)
+        {
+            var parameters = new DynamicParameters(args);
+            parameters.Add("end_point", endPoint);
+            string where = $"{condition} AND COALESCE(end_point,'') = @end_point";
+            string dpWhere = $"{where} AND parse_type = 'dp'";
+
+            var targets = new (string table, string where, bool delete)[]
+            {
+                ($"tk_parse_logs_{date:yyyyMMdd}", where, false),
+                ($"tk_dp_logs_{date:yyyyMMdd}", dpWhere, false),
+                ("tk_parse_logs_test", where, false),
+                ("tk_success_parse_logs", where, true),
+                ("tk_dp_logs_success", dpWhere, true),
+            };
+
+            var result = new Dictionary<string, int>(StringComparer.Ordinal);
+            foreach (var (table, tableWhere, delete) in targets)
+            {
+                if (!TableExists(conn, table, commandTimeoutSeconds)) continue;
+
+                string sql = dryRun
+                    ? $"SELECT COUNT(*) FROM {table} WHERE {tableWhere}"
+                    : delete
+                        ? $"DELETE FROM {table} WHERE {tableWhere}"
+                        : $"UPDATE {table} SET success = 0 WHERE {tableWhere}";
+
+                result[table] = dryRun
+                    ? SqlMapper.ExecuteScalar<int>(conn, sql, parameters, commandTimeout: commandTimeoutSeconds)
+                    : SqlMapper.Execute(conn, sql, parameters, commandTimeout: commandTimeoutSeconds);
+            }
+            return result;
+        }
+
+        private sealed class TbOverrideRepairGroupRow
+        {
+            public string end_point { get; set; } = string.Empty;
+            public string hour_key { get; set; } = string.Empty;
+            public string deeplink_url { get; set; } = string.Empty;
+            public string riskStrategy { get; set; } = string.Empty;
+            public int launchScene { get; set; }
+            public long row_count { get; set; }
+        }
+
+        private sealed class TbOverrideRepairSummaryRow
+        {
+            public string subCode { get; set; } = string.Empty;
+            public string message { get; set; } = string.Empty;
+            public string reason { get; set; } = string.Empty;
+            public long row_count { get; set; }
+        }
+
         private static HashSet<string> GetKnownDeeplinkReportChannelNames()
         {
             var channelNames = Enum.GetNames(typeof(TkChannelEnum))

molilian.core/Core/TracksParseMetricCore.cs

@@ -41,6 +41,19 @@ namespace molilian.core
             return IncrementParseResultMetricAsync(dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene, accountId, success);
         }
 
+        /// <summary>
+        /// RecordParseResultAsync 写入的 success 计数 key（daily、hour），供历史数据修复使用
+        /// </summary>
+        public static string[] BuildParseMetricSuccessKeys(DateTime metricTime, string channel, string riskStrategy, int launchScene)
+        {
+            var dimension = ResolveParseMetricDimension(channel, riskStrategy, launchScene);
+            return
+            [
+                BuildParseMetricCountKey(ParseMetricSuccess, "daily", metricTime.ToString("yyyyMMdd"), dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene),
+                BuildParseMetricCountKey(ParseMetricSuccess, "hour", metricTime.ToString("yyyyMMddHH"), dimension.Platform, dimension.RiskStrategy, dimension.LaunchScene),
+            ];
+        }
+
         public static async Task<TrackParseMetricReportResult> GetParseMetricReportAsync(
             DateTime start,
             DateTime end,

molilian.core/Core/log/base.cs

@@ -360,16 +360,7 @@ namespace molilian.core
                 await SaveParseAccountCacheAsync($"{channel}_{accountId}", success, message, reason);
             }
 
-            string dp_flag = deeplink switch
-            {
-                "" => "none",
-                "tbopen://m.taobao.com/tbopen/index.html" or
-                "pinduoduo://com.xunmeng.pinduoduo/" or
-                "snssdk1128://feed?refer=web" or
-                "bdnetdisk://n/action.EXTERNAL_ACTIVITY" or
-                "openapp.jdmobile://virtual?params=" or "openapp.jdmobile://" => "home",
-                _ => success ? "success" : "fail",
-            };
+            string dp_flag = GetParseDeeplinkFlag(deeplink, success);
 
             //关于dp的缓存
             await SaveParseAccountCacheAsync($"dp_{dp_flag}:all", success, message, reason);
@@ -619,7 +610,24 @@ namespace molilian.core
             await RedisKit.ExpireAsync(namedValueKey, expireSeconds);
         }
 
-        private static int GetStatsExpireSeconds(string timeKey)
+        /// <summary>
+        /// parse_total 中 dp_{flag} 维度的取值：none / home / success / fail
+        /// </summary>
+        public static string GetParseDeeplinkFlag(string deeplink, bool success)
+        {
+            return deeplink switch
+            {
+                "" => "none",
+                "tbopen://m.taobao.com/tbopen/index.html" or
+                "pinduoduo://com.xunmeng.pinduoduo/" or
+                "snssdk1128://feed?refer=web" or
+                "bdnetdisk://n/action.EXTERNAL_ACTIVITY" or
+                "openapp.jdmobile://virtual?params=" or "openapp.jdmobile://" => "home",
+                _ => success ? "success" : "fail",
+            };
+        }
+
+        public static int GetStatsExpireSeconds(string timeKey)
         {
             return timeKey.Length switch
             {