hotline/src/Hotline.Application/Jobs/XingTangCallsSyncJob.cs

using Hotline.Application.CallCenter;
using Hotline.Share.Tools;
using Quartz;
using SqlSugar;
using Hotline.CallCenter.Calls;
using Hotline.Repository.SqlSugar;
using Hotline.Users;
using MapsterMapper;
using Microsoft.Extensions.Logging;
using XF.Domain.Repository;
using XingTang.Sdk;
using DotNetCore.CAP;
using Hotline.Share.Dtos.Order;
using Hotline.Share.Dtos.TrCallCenter;
using Hotline.Share.Enums.CallCenter;
using System.Dynamic;
using Hotline.Share.Dtos.CallCenter;
using Hotline.Caching.Interfaces;
using Mapster;
using Hotline.Share.Mq;

namespace Hotline.Application.Jobs
{
    /// <summary>
    /// 查询兴唐通话记录
    /// </summary>
    public class XingTangCallsSyncJob : IJob, IDisposable
    {
        private readonly IRepository<CallNative> _callRepository;
        private readonly IRepository<User> _userRepository;
        private readonly ICallApplication _callApplication;
        private readonly ICapPublisher _capPublisher;
        private readonly IMapper _mapper;
        private readonly ILogger<XingTangCallsSyncJob> _logger;
        private readonly ISqlSugarClient _db;
        private readonly ISystemSettingCacheManager _systemSettingCacheManager;

        public XingTangCallsSyncJob(
            ISugarUnitOfWork<XingTangDbContext> uow,
            IRepository<CallNative> callRepository,
            IRepository<User> userRepository,
            ICallApplication callApplication,
            ICapPublisher capPublisher,
            IMapper mapper,
            ILogger<XingTangCallsSyncJob> logger,
            ISystemSettingCacheManager systemSettingCacheManager)
        {
            _callRepository = callRepository;
            _userRepository = userRepository;
            _callApplication = callApplication;
            _capPublisher = capPublisher;
            _mapper = mapper;
            _logger = logger;
            _db = uow.Db;
            _systemSettingCacheManager = systemSettingCacheManager;
        }

        public async Task Execute(IJobExecutionContext context)
        {
            var xingtangCalls = await _db.Queryable<XingtangCall>()
                .Where(d => !string.IsNullOrEmpty(d.CallGuid) &&
                            !string.IsNullOrEmpty(d.Caller) &&
                            (d.IsSync == null || !d.IsSync) &&
                            (d.Tries == null || d.Tries <= 50))
                .OrderBy(d => d.Id)
                .Take(10)
                .ToListAsync(context.CancellationToken);

            if (!xingtangCalls.Any()) return;
            var occupyCalls = new List<XingtangCall>();
            foreach (var call in xingtangCalls)
            {
                call.IsSync = true;
                call.Tries += 1;

                var rows = await _db.Updateable(call)
                    .ExecuteCommandWithOptLockAsync();
                if (rows > 0)
                    occupyCalls.Add(call);
            }

            if (!occupyCalls.Any()) return;
            try
            {
                if (occupyCalls.Any(m => m.CustomerId.IsNullOrEmpty() == true))
                {
                    var calls = _mapper.Map<List<CallNative>>(occupyCalls);
                    //填充user信息
                    var staffNos = calls.Select(d => d.StaffNo).ToList();
                    var users = await _userRepository.Queryable()
                        .Where(d => staffNos.Contains(d.StaffNo) && d.StaffNo != "0")
                        .ToListAsync(context.CancellationToken);

                    foreach (var call in calls)
                    {
                        call.Id = await GetCallIdAsync(call.CallNo, context.CancellationToken);
                        var user = users.FirstOrDefault(d => d.StaffNo == call.StaffNo && d.StaffNo != "0");
                        if (user is not null)
                        {
                            call.UserId = user.Id;
                            call.UserName = user.Name;
                        }

                        if (call.RingDuration == 0)
                        {
                            if (call.BeginRingTime != null)
                            {
                                if (call.EndRingTime != null)
                                {
                                    call.RingDuration = call.EndRingTime.Value.GetDifferenceSeconds(call.BeginRingTime.Value);
                                }
                                else
                                {
                                    call.RingDuration = call.EndTime.GetDifferenceSeconds(call.BeginRingTime.Value);
                                }
                            }
                        }

                        if (call.GroupId == "0" && call.CallState != ECallState.IVRNoAccept)
                        {
                            call.CallState = ECallState.Invalid;
                        }
                    }

                    await _callRepository.AddRangeAsync(calls, context.CancellationToken);

                    // 开关
                    var unPushTime = _systemSettingCacheManager.CallSyncUnPushDateTime;
                    calls = calls.Where(m => m.BeginIvrTime >= unPushTime).ToList();
                    if (calls.Any())
                    {
                        //推省上
                        //await _capPublisher.PublishAsync(EventNames.HotlineCallAdd, calls.Adapt<List<CallNativeDto>>());
                        await PublishMessageAsync(calls);
                    }
                    _logger.LogInformation($"旧方法同步通话记录成功，数量：{calls.Count}");
                }
                else
                {
                    var calls = new List<CallNative>();

                    foreach (var occupyCall in occupyCalls)
                    {
                        var call = occupyCall.Adapt<CallNative>();
                        call.Id = await GetCallIdAsync(call.CallNo, context.CancellationToken);
                        var userSplit = occupyCall.CustomerId.Split(':');
                        call.UserId = userSplit[0];
                        call.UserName = userSplit[1];

                        if (call.RingDuration == 0)
                        {
                            if (call.BeginRingTime != null)
                            {
                                if (call.EndRingTime != null)
                                {
                                    call.RingDuration = call.EndRingTime.Value.GetDifferenceSeconds(call.BeginRingTime.Value);
                                }
                                else
                                {
                                    call.RingDuration = call.EndTime.GetDifferenceSeconds(call.BeginRingTime.Value);
                                }
                            }
                        }

                        if (call.GroupId == "0" && call.CallState != ECallState.IVRNoAccept)
                        {
                            call.CallState = ECallState.Invalid;
                        }
                        calls.Add(call);
                    }

                    await _callRepository.AddRangeAsync(calls, context.CancellationToken);

                    // 开关
                    var unPushTime = _systemSettingCacheManager.CallSyncUnPushDateTime;
                    calls = calls.Where(m => m.BeginIvrTime >= unPushTime).ToList();
                    if (calls.Any())
                    {
                        //推省上
                        await PublishMessageAsync(calls);
                        // await _capPublisher.PublishAsync(EventNames.HotlineCallAdd, calls.Adapt<List<CallNativeDto>>());
                    }
                    _logger.LogInformation($"新方法同步通话记录成功，数量：{calls.Count}");
                }
            }
            catch (Exception e)
            {
                _logger.LogError($"获取通话记录异常：{e.Message} \n {e.StackTrace}");
                foreach (var occupyCall in occupyCalls)
                {
                    occupyCall.IsSync = false;
                }

                await _db.Updateable(occupyCalls)
                    .UpdateColumns(d => new { d.IsSync })
                    .ExecuteCommandAsync(context.CancellationToken);
            }
        }

        private async Task PublishMessageAsync(List<CallNative> calls)
        {
            if (_systemSettingCacheManager.IsNoPushCallNativeOutNoFile)
            {
                var removeCalls = calls.Where(m => m.Direction == ECallDirection.Out && m.AudioFile.IsNullOrEmpty()).ToList();
                calls = calls.Except(removeCalls).ToList();
            }
            //推省上
            await _capPublisher.PublishAsync(EventNames.HotlineCallAdd, calls.Adapt<List<CallNativeDto>>());
        }

        private async Task<string> GetCallIdAsync(string callNo, CancellationToken cancellation)
        {
            if (string.IsNullOrEmpty(callNo)) return string.Empty;
            var relation = await _callApplication.GetRelationAsync(callNo, cancellation);
            if (relation is null)
            {
                relation = new CallidRelation
                {
                    Id = callNo,
                    CallId = Ulid.NewUlid().ToString(),
                    IsUsed = true
                };
                await _callApplication.AddRelationAsync(relation, cancellation);
                return relation.CallId;
            }

            if (relation.IsUsed)
                return Ulid.NewUlid().ToString();

            relation.IsUsed = true;
            var rows = await _callApplication.UpdateRelationOptLockAsync(relation, cancellation);
            if (rows > 0)
                return relation.CallId;

            //重新取relation 重新判断isUsed
            return await GetCallIdAsync(callNo, cancellation);
        }

        /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
        public void Dispose()
        {
            //_logger.LogInformation($"{nameof(XingTangCallsSyncJob)} disposed");
        }
    }
}