dotnet/molilian.core/Core/log/base.cs

using dodohold.core;
using CSRedis;
using System.Data;
using System.Diagnostics;
using YunhuiKit;
using static ICSharpCode.SharpZipLib.Zip.ExtendedUnixData;


namespace molilian.core
{
    public partial class TkLogCore
    {

        public static bool save_dailys_log = false;

        static TkLogCore()
        {
            int flag = RedisHelper.Get<int>("turn:save_dailys_log");
            if (flag == 1) save_dailys_log = true;
        }
        private static readonly SemaphoreSlim semaphore = new SemaphoreSlim(10, 10);
        private const int StatsHourlyExpireSeconds = 7 * 86400;
        private const int StatsDailyExpireSeconds = 90 * 86400;
        private const int StatsMonthlyExpireSeconds = 366 * 86400;

        public static async Task<int> BatchInsertLogDBAsync(int limit)
        {
            await semaphore.WaitAsync().ConfigureAwait(false);
            try
            {
                var tasks = EndPointCore.List()
                    .Where(node => node.is_public_api && !string.IsNullOrEmpty(node.redis_server))
                    .Where(node => CenterHub.IsCenter ? !node.is_coupon_api : node.is_coupon_api)
                    .Select(async node =>
                    {
                        try
                        {
                            var redisServer = EndPointCore.GetRedisServer(node);
                            if (string.IsNullOrEmpty(redisServer)) return 0;
                            await using var scope = await RedisClientFactory.CreateScopeAsync(redisServer);
                            return BatchInsertLogDB(limit, scope.Client);
                        }
                        catch (Exception ex)
                        {
                            // TODO: 添加日志记录
                            return 0;
                        }
                    });

                var results = await Task.WhenAll(tasks).ConfigureAwait(false);
                return results.Sum();
            }
            catch (Exception)
            {
                return 0;
            }
            finally
            {
                semaphore.Release();
            }
        }

        public static int BatchInsertLogDB(int limit, YunhuiKit.RedisClient redis)
        {
            int total = 0;
            //using var connection = DBContext.GetOpenConnection();
            //connection.Open();
            //using var transaction = connection.BeginTransaction();
            using IDbTransaction transaction = null;
            using IDbConnection connection = null;

            try
            {
                Stopwatch stopwatch = new Stopwatch(); // 创建一个计时器

                var tasks = new List<Task<int>>
                {
                    RunTaskWithLoggingAsync(() => InsertPromotionImgAsync(limit, redis), "task_insert_promotion_img_logs"),
                    RunTaskWithLoggingAsync(() => InsertParseTkLogAsync(limit, redis), "task_insert_parse_tb_logs"),
                    RunTaskWithLoggingAsync(() => InsertParseJDLogAsync(limit, redis), "task_insert_parse_jd_logs"),
                    RunTaskWithLoggingAsync(() => InsertParsePddLogAsync(limit, redis), "task_insert_parse_pdd_logs"),
                    RunTaskWithLoggingAsync(() => InsertParseDyLogAsync(limit, redis), "task_insert_parse_dy_logs"),
                    RunTaskWithLoggingAsync(() => InsertParseKsLogAsync(limit, redis), "task_insert_parse_ks_logs"),


                    RunTaskWithLoggingAsync(() => InsertToolLogAsync(limit, redis), "task_insert_parse_tool_logs"),
                    RunTaskWithLoggingAsync(() => InsertDeeplinkLogAsync(limit, redis), "task_insert_parse_deeplink_logs"),

                    RunTaskWithLoggingAsync(() => InsertCouponLogAsync(limit, redis), "task_insert_parse_coupon_logs"),
                    RunTaskWithLoggingAsync(() => InsertActivityLogAsync(limit, redis), "task_insert_activity_logs"),
                    RunTaskWithLoggingAsync(() => InsertCpsLogAsync(limit, redis), "task_insert_parse_cps_logs"),
                    RunTaskWithLoggingAsync(() => TracksCore.InsertTrackRequestLogAsync(limit, redis), "task_insert_track_request_logs"),



                    //Task.Run(() => RunTaskWithLogging(() => task_insert_tk_logs(limit, redis), "task_insert_tk_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_tb_logs(limit, redis), "task_insert_parse_tb_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_jd_logs(limit, redis), "task_insert_parse_jd_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_pdd_logs(limit, redis), "task_insert_parse_pdd_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_dy_logs(limit, redis), "task_insert_parse_dy_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_tool_logs(limit, redis), "task_insert_parse_tool_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_deeplink_logs(limit, redis), "task_insert_parse_deeplink_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_coupon_logs(limit, redis), "task_insert_parse_coupon_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_cps_logs(limit, redis), "task_insert_parse_cps_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_promotion_img_logs(limit, redis), "task_insert_promotion_img_logs")),
                    //Task.Run(() => RunTaskWithLogging(() => task_insert_parse_ks_logs(limit, redis), "task_insert_parse_ks_logs"))
                };

                // 等待所有任务完成
                Task.WhenAll(tasks).Wait();//252行

                // 计算所有任务的结果总和
                total = tasks.Select(t => t.Result).Sum();

                //transaction.Commit();
            }
            catch (Exception ex)
            {
                //transaction.Rollback();


                // 构建异常详细信息字符串
                string detailedError = $"{ex.Message}\n" +
                                       $"堆栈跟踪: {ex.StackTrace}\n";

                if (ex.InnerException != null)
                {
                    detailedError += $"内部异常: {ex.InnerException.Message}\n" +
                                     $"内部堆栈跟踪: {ex.InnerException.StackTrace}\n";
                }

                // 如果异常包含其他数据，也可以记录下来
                if (ex.Data != null && ex.Data.Count > 0)
                {
                    detailedError += "附加数据:\n";
                    foreach (var key in ex.Data.Keys)
                    {
                        detailedError += $"{key}: {ex.Data[key]}\n";
                    }
                }


                _ = new LoggerLibrary("database_error", "parse_log")
                    .Info(detailedError)
                    .SaveAsync();


                NotifyCore.Notify(new NifyMessage
                {
                    message = $"【Exception】\n{detailedError}",
                    priority = NifyMessagePriority.high,
                    tags = ["red_circle"]
                });
            }
            finally
            {
                //connection.Close();
            }
            return total;
        }



        private static async Task<int> RunTaskWithLoggingAsync(Func<Task<int>> taskFunc, string taskName)
        {
            LoggerLibrary log = new LoggerLibrary("debug", "BatchInsertLogDB"); // 创建日志对象
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();
            int taskTotal = await taskFunc();
            stopwatch.Stop();
            log.Info($"{taskName} 耗时: {stopwatch.ElapsedMilliseconds} ms,\t插入记录数: {taskTotal}");
            log.SaveAsync();
            return taskTotal;
        }

        /// <summary>
        /// 过时方法 随时删除
        /// </summary>
        /// <param name="channel"></param>
        /// <param name="accountId"></param>
        /// <param name="accountName"></param>
        /// <param name="success"></param>
        /// <param name="message"></param>
        /// <param name="reason"></param>
        private static void saveCache(string channel, int accountId, string accountName, bool success, string message, string reason)
        {
            saveAccountCache("all", success, message, reason);
            saveAccountCache($"{channel}", success, message, reason);
            saveAccountCache($"{accountName}", success, message, reason);
            if (accountId != 0)
            {
                //todo 放着跑两天，要将读取的地方改成读取accountid
                saveAccountCache($"{channel}_{accountId}", success, message, reason);
            }
        }

        private static void saveAccountCache(string accountName, bool success, string message, string reason)
        {
            SaveStatsAccountCache("total", accountName, success, message, reason);
        }

        private static async Task saveUnionCouponParseCacheAsync(TkDataDTO data)
        {
            string cacheKey = $":cache:parse:{data.ip}_{data.oaid}_{data.itemId}";
            await EndPointCore.ProcessEndPointNodesAsync(node =>
            {
                if (!node.is_coupon_api) return Task.CompletedTask;
                if (string.IsNullOrEmpty(node.redis_server)) return Task.CompletedTask;

                var redis = RedisClientManager.GetRedisClient(node.redis_server);
                redis.Set(cacheKey, 1, 2 * 86400);
                return Task.CompletedTask;
            });
        }
        private async static Task saveClientRequestTotalAsync(TkChannelEnum channel, string ip, string oaid)
        {
            await saveClientRequestTotalAsync(channel.ToString(), ip, oaid);
        }
        private static async Task saveClientRequestTotalAsync(string channel, string ip, string oaid)
        {
            await EndPointCore.ProcessEndPointNodesAsync(node =>
            {
                if (string.IsNullOrEmpty(node.redis_server)) return Task.CompletedTask;
                if (!node.is_public_api) return Task.CompletedTask;

#if DEBUG
                switch (node.name)
                {

                    case "bj":
                        node.redis_server = "101.200.152.61:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=webhook";
                        break;
                    case "gz":
                        node.redis_server = "8.138.110.158:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=webhook";
                        break;
                    case "coupon1":
                        node.redis_server = "123.56.185.166:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=coupon";
                        break;
                    default: return Task.CompletedTask;
                }

#endif


                var redis = RedisClientManager.GetRedisClient(node.redis_server);

                string cacheKey = $":cache:{channel}:ip:{DateTime.Now:yyyyMMdd}:{ip}";
                redis.IncrBy(cacheKey);
                redis.Expire(cacheKey, 86400);

                if (!string.IsNullOrEmpty(oaid))
                {
                    cacheKey = $":cache:{channel}:oaid:{DateTime.Now:yyyyMMdd}:{oaid}";
                    redis.IncrBy(cacheKey);
                    redis.Expire(cacheKey, 86400);
                }
                return Task.CompletedTask;
            });

        }
        public static bool InBlacklist(string blacklist, string oaid)
        {
            if (string.IsNullOrEmpty(blacklist)) return false;

            var arr = blacklist.Split(new[] { "\r\n" }, StringSplitOptions.None)
                   .Select(s => s.Trim()).ToArray();
            return arr.Contains(oaid);
        }


        public static int getClientRequestTotalByOAID(TkChannelEnum channel, string oaid)
        {
            if (string.IsNullOrEmpty(oaid)) return 0;
            string cacheKey = $":cache:{channel}:oaid:{DateTime.Now:yyyyMMdd}:{oaid}";
            return RedisHelper.Get<int>(cacheKey);
        }
        public static int getClientRequestTotalByIp(TkChannelEnum channel, string ip)
        {
            if (string.IsNullOrEmpty(ip)) return 0;
            string cacheKey = $":cache:{channel}:ip:{DateTime.Now:yyyyMMdd}:{ip}";
            return RedisHelper.Get<int>(cacheKey);
        }
        public static int getClientRequestTotalByOAID(string channel, string oaid)
        {
            if (string.IsNullOrEmpty(oaid)) return 0;
            string cacheKey = $":cache:{channel}:oaid:{DateTime.Now:yyyyMMdd}:{oaid}";
            return RedisHelper.Get<int>(cacheKey);
        }
        public static int getClientRequestTotalByIp(string channel, string ip)
        {
            if (string.IsNullOrEmpty(ip)) return 0;
            string cacheKey = $":cache:{channel}:ip:{DateTime.Now:yyyyMMdd}:{ip}";
            return RedisHelper.Get<int>(cacheKey);
        }


        private static void saveClientRequestTotal(CpsChannelEnum channel, string ip, string oaid)
        {
            string cacheKey = $":cache:cps_{channel}:ip:{DateTime.Now:yyyyMMdd}:{ip}";
            RedisHelper.IncrBy(cacheKey);
            RedisHelper.Expire(cacheKey, 86400);

            if (!string.IsNullOrEmpty(oaid))
            {
                cacheKey = $":cache:cps_{channel}:oaid:{DateTime.Now:yyyyMMdd}:{oaid}";
                RedisHelper.IncrBy(cacheKey);
                RedisHelper.Expire(cacheKey, 86400);
            }
        }
        public static int getClientRequestTotalByOAID(CpsChannelEnum channel, string oaid)
        {
            if (string.IsNullOrEmpty(oaid)) return 0;
            string cacheKey = $":cache:cps_{channel}:oaid:{DateTime.Now:yyyyMMdd}:{oaid}";
            return RedisHelper.Get<int>(cacheKey);
        }
        public static int getClientRequestTotalByIp(CpsChannelEnum channel, string ip)
        {
            if (string.IsNullOrEmpty(ip)) return 0;
            string cacheKey = $":cache:cps_{channel}:ip:{DateTime.Now:yyyyMMdd}:{ip}";
            return RedisHelper.Get<int>(cacheKey);
        }


        private static async Task SaveParseCacheAsync(string channel, int accountId, string accountName,
            bool success, string message, string reason, string deeplink)
        {
            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",
            };

            //关于dp的缓存
            await SaveParseAccountCacheAsync($"dp_{dp_flag}:all", success, message, reason);
            await SaveParseAccountCacheAsync($"dp_{dp_flag}:{channel}", success, message, reason);
            await SaveParseAccountCacheAsync($"dp_{dp_flag}:{accountName}", success, message, reason);
            if (accountId != 0)
            {
                await SaveParseAccountCacheAsync($"dp_{dp_flag}:{channel}_{accountId}", success, message, reason);
            }

            await SaveParseAccountCacheAsync("all", success, message, reason);
            await SaveParseAccountCacheAsync($"dp_channel:{channel}", success, message, reason);
            await SaveParseAccountCacheAsync($"{channel}", success, message, reason);
            await SaveParseAccountCacheAsync($"{accountName}", success, message, reason);
            if (accountId != 0)
            {
                //todo 放着跑两天，要将读取的地方改成读取accountid
                await SaveParseAccountCacheAsync($"{channel}_{accountId}", success, message, reason);
            }
        }

        private static async Task SaveParseAccountCacheAsync(string accountName, bool success,
            string message, string reason)
        {
            await SaveStatsAccountCacheAsync("parse_total", accountName, success, message, reason);
        }



        private static void saveParseCache(string channel, int accountId, string accountName,
            bool success, string message, string reason, string deeplink)
        {
            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",
            };

            //关于dp的缓存
            saveParseAccountCache($"dp_{dp_flag}:all", success, message, reason);
            saveParseAccountCache($"dp_{dp_flag}:{channel}", success, message, reason);
            saveParseAccountCache($"dp_{dp_flag}:{accountName}", success, message, reason);
            if (accountId != 0)
            {
                saveParseAccountCache($"dp_{dp_flag}:{channel}_{accountId}", success, message, reason);
            }

            saveParseAccountCache("all", success, message, reason);
            saveParseAccountCache($"{channel}", success, message, reason);
            saveParseAccountCache($"{accountName}", success, message, reason);
            if (accountId != 0)
            {
                //todo 放着跑两天，要将读取的地方改成读取accountid
                saveParseAccountCache($"{channel}_{accountId}", success, message, reason);
            }
        }
        private static void saveParseAccountCache(string accountName, bool success,
            string message, string reason)
        {
            SaveStatsAccountCache("parse_total", accountName, success, message, reason);
        }

        private static void SaveStatsAccountCache(string prefix, string accountName, bool success,
            string message, string reason)
        {
            var now = DateTime.Now;
            string month = now.ToString("yyyyMM");
            string day = now.ToString("yyyyMMdd");
            string hour = now.ToString("yyyyMMddHH");

            SaveStatsCount(prefix, accountName, month);
            SaveStatsCount(prefix, accountName, day);
            SaveStatsCount(prefix, accountName, hour);

            string result = success ? "success" : "fail";
            SaveStatsCount(prefix, accountName, month, result);
            SaveStatsCount(prefix, accountName, day, result);
            SaveStatsCount(prefix, accountName, hour, result);

            SaveStatsDimension(prefix, accountName, "message", message, month);
            SaveStatsDimension(prefix, accountName, "message", message, day);
            SaveStatsDimension(prefix, accountName, "message", message, hour);

            SaveStatsDimension(prefix, accountName, "reason", reason, month);
            SaveStatsDimension(prefix, accountName, "reason", reason, day);
            SaveStatsDimension(prefix, accountName, "reason", reason, hour);
        }

        private static async Task SaveStatsAccountCacheAsync(string prefix, string accountName, bool success,
            string message, string reason)
        {
            var now = DateTime.Now;
            string month = now.ToString("yyyyMM");
            string day = now.ToString("yyyyMMdd");
            string hour = now.ToString("yyyyMMddHH");

            await SaveStatsCountAsync(prefix, accountName, month);
            await SaveStatsCountAsync(prefix, accountName, day);
            await SaveStatsCountAsync(prefix, accountName, hour);

            string result = success ? "success" : "fail";
            await SaveStatsCountAsync(prefix, accountName, month, result);
            await SaveStatsCountAsync(prefix, accountName, day, result);
            await SaveStatsCountAsync(prefix, accountName, hour, result);

            await SaveStatsDimensionAsync(prefix, accountName, "message", message, month);
            await SaveStatsDimensionAsync(prefix, accountName, "message", message, day);
            await SaveStatsDimensionAsync(prefix, accountName, "message", message, hour);

            await SaveStatsDimensionAsync(prefix, accountName, "reason", reason, month);
            await SaveStatsDimensionAsync(prefix, accountName, "reason", reason, day);
            await SaveStatsDimensionAsync(prefix, accountName, "reason", reason, hour);
        }

        private static void SaveStatsCount(string prefix, string accountName, string timeKey, string? dimension = null)
        {
            string key = string.IsNullOrEmpty(dimension)
                ? $":{prefix}:{accountName}:{timeKey}"
                : $":{prefix}:{accountName}:{dimension}:{timeKey}";
            RedisHelper.IncrBy(key);
            RedisHelper.Expire(key, GetStatsExpireSeconds(timeKey));
        }

        private static async Task SaveStatsCountAsync(string prefix, string accountName, string timeKey, string? dimension = null)
        {
            string key = string.IsNullOrEmpty(dimension)
                ? $":{prefix}:{accountName}:{timeKey}"
                : $":{prefix}:{accountName}:{dimension}:{timeKey}";
            await RedisKit.IncrByAsync(key);
            await RedisKit.ExpireAsync(key, GetStatsExpireSeconds(timeKey));
        }

        private static void SaveStatsDimension(string prefix, string accountName, string name, string value, string timeKey)
        {
            if (string.IsNullOrEmpty(value)) return;

            int expireSeconds = GetStatsExpireSeconds(timeKey);
            string setKey = $":{prefix}:{accountName}:{name}:{timeKey}";
            RedisHelper.SAdd(setKey, value);
            RedisHelper.Expire(setKey, expireSeconds);

            string valueKey = $":{prefix}:{accountName}:{value}:{timeKey}";
            RedisHelper.IncrBy(valueKey);
            RedisHelper.Expire(valueKey, expireSeconds);

            string namedValueKey = $":{prefix}:{accountName}:{name}:{value}:{timeKey}";
            RedisHelper.IncrBy(namedValueKey);
            RedisHelper.Expire(namedValueKey, expireSeconds);
        }

        private static async Task SaveStatsDimensionAsync(string prefix, string accountName, string name, string value, string timeKey)
        {
            if (string.IsNullOrEmpty(value)) return;

            int expireSeconds = GetStatsExpireSeconds(timeKey);
            string setKey = $":{prefix}:{accountName}:{name}:{timeKey}";
            await RedisKit.SAddAsync(setKey, value);
            await RedisKit.ExpireAsync(setKey, expireSeconds);

            string valueKey = $":{prefix}:{accountName}:{value}:{timeKey}";
            await RedisKit.IncrByAsync(valueKey);
            await RedisKit.ExpireAsync(valueKey, expireSeconds);

            string namedValueKey = $":{prefix}:{accountName}:{name}:{value}:{timeKey}";
            await RedisKit.IncrByAsync(namedValueKey);
            await RedisKit.ExpireAsync(namedValueKey, expireSeconds);
        }

        private static int GetStatsExpireSeconds(string timeKey)
        {
            return timeKey.Length switch
            {
                10 => StatsHourlyExpireSeconds,
                8 => StatsDailyExpireSeconds,
                6 => StatsMonthlyExpireSeconds,
                _ => StatsDailyExpireSeconds,
            };
        }

        public static async Task<int> GetTotalAsync(string keyname, bool all_node = true)
        {
            try
            {
                var tasks = EndPointCore.List()
                    .Where(node => node.is_public_api && !string.IsNullOrEmpty(node.redis_server))
                    .Where(node => all_node || (CenterHub.IsCenter ? !node.is_coupon_api : node.is_coupon_api))
                    .Select(async node =>
                    {
                        try
                        {

#if DEBUG
                            switch (node.name)
                            {

                                case "bj":
                                    node.redis_server = "101.200.152.61:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=webhook";
                                    break;
                                case "gz":
                                    node.redis_server = "8.138.110.158:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=webhook";
                                    break;
                                case "coupon1":
                                    node.redis_server = "123.56.185.166:6379,password=pKBiS4ka2IpXayIdcx00,defaultDatabase=0,idleTimeout=20000,preheat=3,tryit=2,ssl=false,prefix=coupon";
                                    break;
                            }

#endif


                            var redisServer = EndPointCore.GetRedisServer(node);
                            if (string.IsNullOrEmpty(redisServer))
                                return 0;

                            await using var scope = await RedisClientFactory.CreateScopeAsync(redisServer);
                            return await scope.Client.GetAsync<int>(keyname);
                        }
                        catch (Exception)
                        {
                            return 0;
                        }
                    });

                var results = await Task.WhenAll(tasks);
                return results.Sum();
            }
            catch (Exception)
            {
                return 0;
            }
        }

        public static async Task<string[]> GetTotalKeysAsync(string keyname, bool all_node = true)
        {

            var tasks = EndPointCore.List()
            .Where(node => node.is_public_api && !string.IsNullOrEmpty(node.redis_server))
            .Where(node => all_node || (CenterHub.IsCenter ? !node.is_coupon_api : node.is_coupon_api))
            .Select(async node =>
            {
                try
                {
                    try
                    {
                        var redisServer = EndPointCore.GetRedisServer(node);
                        if (string.IsNullOrEmpty(redisServer)) return [];

                        using var scope = await RedisClientFactory.CreateScopeAsync(redisServer);
                        return await scope.Client.SMembersAsync<string>(keyname);
                    }
                    catch (Exception ex) { }
                    return [];
                }
                catch (Exception ex)
                {
                    return [];
                }
            });

            var results = await Task.WhenAll(tasks);
            return results.SelectMany(x => x).Distinct().ToArray();

        }


    }
}