WorkerCdr::setDisposition() - Code Metrics - Inspection of "Уменьшил время ожидания ответа на запрос Ping." - mikopbx/Core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( cf94b8...c1cb26 )

by Портнов

created 2020-11-23 15:01 UTC

WorkerCdr::setDisposition() B

↳ Parent: WorkerCdr

Complexity

Conditions	11
Paths	20

Size

Total Lines	22
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
eloc	15
c	3
b	0
f	0
dl	0
loc	22
rs	7.3166
cc	11
nc	20
nop	3

How to fix Complexity

<?php
/*
 * Copyright © MIKO LLC - All Rights Reserved
 * Unauthorized copying of this file, via any medium is strictly prohibited
 * Proprietary and confidential
 * Written by Alexey Portnov, 9 2020
 */

namespace MikoPBX\Core\Workers;

require_once 'Globals.php';

use MikoPBX\Common\Models\{CallDetailRecordsTmp, Extensions, Users};
use MikoPBX\Core\System\{BeanstalkClient, Processes, Util};
use Throwable;

/**
 * Class WorkerCdr
 * Обработка записей CDR. Заполение длительности звонков.
 */
class WorkerCdr extends WorkerBase
{

    public const SELECT_CDR_TUBE = 'select_cdr_tube';
    public const UPDATE_CDR_TUBE = 'update_cdr_tube';

    private BeanstalkClient $client_queue;
    private array $internal_numbers  = [];
    private array $no_answered_calls = [];


    /**
     * Entry point
     *
     * @param $argv
     *
     */
    public function start($argv): void
    {
        $filter = [
            'work_completed<>1 AND endtime<>""',
            'columns'=> 'start,answer,src_num,dst_num,dst_chan,endtime,linkedid,recordingfile,dialstatus,UNIQUEID'
        ];

        $this->client_queue = new BeanstalkClient(self::SELECT_CDR_TUBE);
        $this->client_queue->subscribe($this->makePingTubeName(self::class), [$this, 'pingCallBack']);

        $this->initSettings();

        while ($this->needRestart === false) {
            $result = CallDetailRecordsTmp::find($filter)->toArray();
            if (!empty($result)) {
                $this->updateCdr($result);
            }
            $this->client_queue->wait(5); // instead of sleep
        }
    }

    /**
     * Fills settings
     */
    private function initSettings(): void
    {
        $this->internal_numbers  = [];
        $this->no_answered_calls = [];

        $usersClass = Users::class;
        $parameters = [
            'columns'=>[
                'email'=>'email',
                'language'=>'language',
                'number'=>'Extensions.number'
            ],
            'joins'      => [
                'Extensions' => [
                    0 => Extensions::class,
                    1 => "Extensions.userid={$usersClass}.id",
                    2 => 'Extensions',
                    3 => 'INNER',
                ],
            ],
            'cache' => [
                'key'=>'Users-WorkerCdr',
                'lifetime' => 3600,
            ]
        ];

        $results   = Users::find($parameters);
        foreach ($results as $record) {
            if (empty($record->email)) {
                continue;
            }
            $this->internal_numbers[$record->number] = [
                'email'    => $record->email,
                'language' => $record->language,
            ];
        }
    }

    /**
     * Обработчик результата запроса.
     * @param $result
     */
    private function updateCdr($result): void
    {
        $this->initSettings();
        $arr_update_cdr = [];
        // Получаем идентификаторы активных каналов.
        $channels_id = $this->getActiveIdChannels();
        foreach ($result as $row) {
            if (array_key_exists($row['linkedid'], $channels_id)) {
                // Цепочка вызовов еще не завершена.
                continue;
            }

            $start      = strtotime($row['start']);
            $answer     = strtotime($row['answer']);
            $end        = strtotime($row['endtime']);
            $dialstatus = trim($row['dialstatus']);

            $duration = max(($end - $start), 0);
            $billsec  = ($end && $answer) ? ($end - $answer) : 0;

            [$disposition, $row] = $this->setDisposition($billsec, $dialstatus, $row);
            [$row, $billsec]     = $this->checkBillsecMakeRecFile($billsec, $row);

            $data = [
                'work_completed' => 1,
                'duration'       => $duration,
                'billsec'        => $billsec,
                'disposition'    => $disposition,
                'UNIQUEID'       => $row['UNIQUEID'],
                'recordingfile'  => ($disposition === 'ANSWERED') ? $row['recordingfile'] : '',
                'tmp_linked_id'  => $row['linkedid'],
            ];

            $arr_update_cdr[] = $data;
            $this->checkNoAnswerCall(array_merge($row, $data));
        }

        $this->setStatusAndPublish($arr_update_cdr);
        $this->notifyByEmail();
    }

    /**
     * Функция позволяет получить активные каналы.
     * Возвращает ассоциативный массив. Ключ - Linkedid, значение - массив каналов.
     *
     * @return array
     */
    private function getActiveIdChannels(): array
    {
        $am           = Util::getAstManager('off');
        return $am->GetChannels(true);
    }

    /**
     * Анализируем не отвеченные вызовы. Наполняем временный массив для дальнейшей обработки.
     *
     * @param $row
     */
    private function checkNoAnswerCall($row): void
    {
        if ($row['disposition'] === 'ANSWERED') {
            $this->no_answered_calls[$row['linkedid']]['NOANSWER'] = false;
            return;
        }
        if ( ! array_key_exists($row['dst_num'], $this->internal_numbers)) {
            // dst_num - не является номером сотрудника. Это исходящий.
            return;
        }
        $is_internal = false;
        if ((array_key_exists($row['src_num'], $this->internal_numbers))) {
            // Это внутренний вызов.
            $is_internal = true;
        }

        $this->no_answered_calls[$row['linkedid']][] = [
            'from_number' => $row['src_num'],
            'to_number'   => $row['dst_num'],
            'start'       => $row['start'],
            'answer'      => $row['answer'],
            'endtime'     => $row['endtime'],
            'email'       => $this->internal_numbers[$row['dst_num']]['email'],
            'language'    => $this->internal_numbers[$row['dst_num']]['language'],
            'is_internal' => $is_internal,
            'duration'    => $row['duration'],
        ];
    }


    /**
     * Постановка задачи в очередь на оповещение по email.
     */
    private function notifyByEmail(): void
    {
        foreach ($this->no_answered_calls as $call) {
            $this->client_queue->publish(json_encode($call), WorkerNotifyByEmail::class);
        }
        $this->no_answered_calls = [];
    }

    /**
     * @param array $arr_update_cdr
     */
    private function setStatusAndPublish(array $arr_update_cdr): void{
        foreach ($arr_update_cdr as $data) {
            $linkedid = $data['tmp_linked_id'];
            $data['GLOBAL_STATUS'] = $data['disposition'];
            if (isset($this->no_answered_calls[$linkedid]['NOANSWER']) && $this->no_answered_calls[$linkedid]['NOANSWER'] === false) {
                $data['GLOBAL_STATUS'] = 'ANSWERED';
                // Это отвеченный вызов (на очередь). Удаляем из списка.
                unset($this->no_answered_calls[$linkedid]);
            }
            unset($data['tmp_linked_id']);
            $this->client_queue->publish(json_encode($data), self::UPDATE_CDR_TUBE);
        }
    }

    /**
     * @param int $billsec
     * @param     $row
     * @return array
     */
    private function checkBillsecMakeRecFile(int $billsec, $row): array{
        if ($billsec <= 0) {
            $row['answer'] = '';
            $billsec = 0;

            if (!empty($row['recordingfile'])) {
                // Удаляем файлы
                $p_info = pathinfo($row['recordingfile']);
                $fileName = $p_info['dirname'] . '/' . $p_info['filename'];
                $file_list = [$fileName . '.mp3', $fileName . '.wav', $fileName . '_in.wav', $fileName . '_out.wav',];
                foreach ($file_list as $file) {
                    if (!file_exists($file) || is_dir($file)) {
                        continue;
                    }
                    Processes::mwExec("rm -rf '{$file}'");
                }
            }
        } elseif (trim($row['recordingfile']) !== '') {
            // Если каналов не существует с ID, то можно удалить временные файлы.
            $p_info = pathinfo($row['recordingfile']);
            // Запускаем процесс конвертации в mp3
            $wav2mp3Path = Util::which('wav2mp3.sh');
            $nicePath = Util::which('nice');
            Processes::mwExecBg("{$nicePath} -n 19 {$wav2mp3Path} '{$p_info['dirname']}/{$p_info['filename']}'");
            // В последствии конвертации (успешной) исходные файлы будут удалены.
        }
        return array($row, $billsec);
    }

    /**
     * @param int    $billsec
     * @param string $dialstatus
     * @param        $row
     * @return array
     */
    private function setDisposition(int $billsec, string $dialstatus, $row): array{
        $disposition = 'NOANSWER';
        if ($billsec > 0) {
            $disposition = 'ANSWERED';
        } elseif ('' !== $dialstatus) {
            $disposition = ($dialstatus === 'ANSWERED') ? $disposition : $dialstatus;
        }

        if ($disposition !== 'ANSWERED') {
            if (file_exists($row['recordingfile']) && !is_dir($row['recordingfile'])) {
                Processes::mwExec("rm -rf {$row['recordingfile']}");
            }
        } elseif (!empty($row['recordingfile']) &&
            !file_exists(Util::trimExtensionForFile($row['recordingfile']) . '.wav') &&
            !file_exists($row['recordingfile'])) {
            /** @var CallDetailRecordsTmp $rec_data */
            $rec_data = CallDetailRecordsTmp::findFirst("linkedid='{$row['linkedid']}' AND dst_chan='{$row['dst_chan']}'");
            if ($rec_data !== null) {
                $row['recordingfile'] = $rec_data->recordingfile;
            }
        }
        return array($disposition, $row);
    }

}

// Start worker process
$workerClassname = WorkerCdr::class;
if (isset($argv) && count($argv) > 1 && $argv[1] === 'start') {
    cli_set_process_title($workerClassname);
    try {
        $worker = new $workerClassname();
        $worker->start($argv);
    } catch (Throwable $e) {
        global $errorLogger;
        $errorLogger->captureException($e);
        Util::sysLogMsg("{$workerClassname}_EXCEPTION", $e->getMessage());
    }
}

1			<?php
2			/*
3			* Copyright © MIKO LLC - All Rights Reserved
4			* Unauthorized copying of this file, via any medium is strictly prohibited
5			* Proprietary and confidential
6			* Written by Alexey Portnov, 9 2020
7			*/
8
9			namespace MikoPBX\Core\Workers;
10
11			require_once 'Globals.php';
12
13			use MikoPBX\Common\Models\{CallDetailRecordsTmp, Extensions, Users};
14			use MikoPBX\Core\System\{BeanstalkClient, Processes, Util};
15			use Throwable;
16
17			/**
18			* Class WorkerCdr
19			* Обработка записей CDR. Заполение длительности звонков.
20			*/
21			class WorkerCdr extends WorkerBase
22			{
23
24			public const SELECT_CDR_TUBE = 'select_cdr_tube';
25			public const UPDATE_CDR_TUBE = 'update_cdr_tube';
26
27			private BeanstalkClient $client_queue;
28			private array $internal_numbers = [];
29			private array $no_answered_calls = [];
30
31
32			/**
33			* Entry point
34			*
35			* @param $argv
36			*
37			*/
38			public function start($argv): void
39			{
40			$filter = [
41			'work_completed<>1 AND endtime<>""',
42			'columns'=> 'start,answer,src_num,dst_num,dst_chan,endtime,linkedid,recordingfile,dialstatus,UNIQUEID'
43			];
44
45			$this->client_queue = new BeanstalkClient(self::SELECT_CDR_TUBE);
46			$this->client_queue->subscribe($this->makePingTubeName(self::class), [$this, 'pingCallBack']);
47
48			$this->initSettings();
49
50			while ($this->needRestart === false) {
51			$result = CallDetailRecordsTmp::find($filter)->toArray();
52			if (!empty($result)) {
53			$this->updateCdr($result);
54			}
55			$this->client_queue->wait(5); // instead of sleep
56			}
57			}
58
59			/**
60			* Fills settings
61			*/
62			private function initSettings(): void
63			{
64			$this->internal_numbers = [];
65			$this->no_answered_calls = [];
66
67			$usersClass = Users::class;
68			$parameters = [
69			'columns'=>[
70			'email'=>'email',
71			'language'=>'language',
72			'number'=>'Extensions.number'
73			],
74			'joins' => [
75			'Extensions' => [
76			0 => Extensions::class,
77			1 => "Extensions.userid={$usersClass}.id",
78			2 => 'Extensions',
79			3 => 'INNER',
80			],
81			],
82			'cache' => [
83			'key'=>'Users-WorkerCdr',
84			'lifetime' => 3600,
85			]
86			];
87
88			$results = Users::find($parameters);
89			foreach ($results as $record) {
90			if (empty($record->email)) {
91			continue;
92			}
93			$this->internal_numbers[$record->number] = [
94			'email' => $record->email,
95			'language' => $record->language,
96			];
97			}
98			}
99
100			/**
101			* Обработчик результата запроса.
102			* @param $result
103			*/
104			private function updateCdr($result): void
105			{
106			$this->initSettings();
107			$arr_update_cdr = [];
108			// Получаем идентификаторы активных каналов.
109			$channels_id = $this->getActiveIdChannels();
110			foreach ($result as $row) {
111			if (array_key_exists($row['linkedid'], $channels_id)) {
112			// Цепочка вызовов еще не завершена.
113			continue;
114			}
115
116			$start = strtotime($row['start']);
117			$answer = strtotime($row['answer']);
118			$end = strtotime($row['endtime']);
119			$dialstatus = trim($row['dialstatus']);
120
121			$duration = max(($end - $start), 0);
122			$billsec = ($end && $answer) ? ($end - $answer) : 0;
123
124			[$disposition, $row] = $this->setDisposition($billsec, $dialstatus, $row);
125			[$row, $billsec] = $this->checkBillsecMakeRecFile($billsec, $row);
126
127			$data = [
128			'work_completed' => 1,
129			'duration' => $duration,
130			'billsec' => $billsec,
131			'disposition' => $disposition,
132			'UNIQUEID' => $row['UNIQUEID'],
133			'recordingfile' => ($disposition === 'ANSWERED') ? $row['recordingfile'] : '',
134			'tmp_linked_id' => $row['linkedid'],
135			];
136
137			$arr_update_cdr[] = $data;
138			$this->checkNoAnswerCall(array_merge($row, $data));
139			}
140
141			$this->setStatusAndPublish($arr_update_cdr);
142			$this->notifyByEmail();
143			}
144
145			/**
146			* Функция позволяет получить активные каналы.
147			* Возвращает ассоциативный массив. Ключ - Linkedid, значение - массив каналов.
148			*
149			* @return array
150			*/
151			private function getActiveIdChannels(): array
152			{
153			$am = Util::getAstManager('off');
154			return $am->GetChannels(true);
155			}
156
157			/**
158			* Анализируем не отвеченные вызовы. Наполняем временный массив для дальнейшей обработки.
159			*
160			* @param $row
161			*/
162			private function checkNoAnswerCall($row): void
163			{
164			if ($row['disposition'] === 'ANSWERED') {
165			$this->no_answered_calls[$row['linkedid']]['NOANSWER'] = false;
166			return;
167			}
168			if ( ! array_key_exists($row['dst_num'], $this->internal_numbers)) {
169			// dst_num - не является номером сотрудника. Это исходящий.
170			return;
171			}
172			$is_internal = false;
173			if ((array_key_exists($row['src_num'], $this->internal_numbers))) {
174			// Это внутренний вызов.
175			$is_internal = true;
176			}
177
178			$this->no_answered_calls[$row['linkedid']][] = [
179			'from_number' => $row['src_num'],
180			'to_number' => $row['dst_num'],
181			'start' => $row['start'],
182			'answer' => $row['answer'],
183			'endtime' => $row['endtime'],
184			'email' => $this->internal_numbers[$row['dst_num']]['email'],
185			'language' => $this->internal_numbers[$row['dst_num']]['language'],
186			'is_internal' => $is_internal,
187			'duration' => $row['duration'],
188			];
189			}
190
191
192			/**
193			* Постановка задачи в очередь на оповещение по email.
194			*/
195			private function notifyByEmail(): void
196			{
197			foreach ($this->no_answered_calls as $call) {
198			$this->client_queue->publish(json_encode($call), WorkerNotifyByEmail::class);
199			}
200			$this->no_answered_calls = [];
201			}
202
203			/**
204			* @param array $arr_update_cdr
205			*/
206			private function setStatusAndPublish(array $arr_update_cdr): void{
207			foreach ($arr_update_cdr as $data) {
208			$linkedid = $data['tmp_linked_id'];
209			$data['GLOBAL_STATUS'] = $data['disposition'];
210			if (isset($this->no_answered_calls[$linkedid]['NOANSWER']) && $this->no_answered_calls[$linkedid]['NOANSWER'] === false) {
211			$data['GLOBAL_STATUS'] = 'ANSWERED';
212			// Это отвеченный вызов (на очередь). Удаляем из списка.
213			unset($this->no_answered_calls[$linkedid]);
214			}
215			unset($data['tmp_linked_id']);
216			$this->client_queue->publish(json_encode($data), self::UPDATE_CDR_TUBE);
217			}
218			}
219
220			/**
221			* @param int $billsec
222			* @param $row
223			* @return array
224			*/
225			private function checkBillsecMakeRecFile(int $billsec, $row): array{
226			if ($billsec <= 0) {
227			$row['answer'] = '';
228			$billsec = 0;
229
230			if (!empty($row['recordingfile'])) {
231			// Удаляем файлы
232			$p_info = pathinfo($row['recordingfile']);
233			$fileName = $p_info['dirname'] . '/' . $p_info['filename'];
234			$file_list = [$fileName . '.mp3', $fileName . '.wav', $fileName . '_in.wav', $fileName . '_out.wav',];
235			foreach ($file_list as $file) {
236			if (!file_exists($file) \|\| is_dir($file)) {
237			continue;
238			}
239			Processes::mwExec("rm -rf '{$file}'");
240			}
241			}
242			} elseif (trim($row['recordingfile']) !== '') {
243			// Если каналов не существует с ID, то можно удалить временные файлы.
244			$p_info = pathinfo($row['recordingfile']);
245			// Запускаем процесс конвертации в mp3
246			$wav2mp3Path = Util::which('wav2mp3.sh');
247			$nicePath = Util::which('nice');
248			Processes::mwExecBg("{$nicePath} -n 19 {$wav2mp3Path} '{$p_info['dirname']}/{$p_info['filename']}'");
249			// В последствии конвертации (успешной) исходные файлы будут удалены.
250			}
251			return array($row, $billsec);
252			}
253
254			/**
255			* @param int $billsec
256			* @param string $dialstatus
257			* @param $row
258			* @return array
259			*/
260			private function setDisposition(int $billsec, string $dialstatus, $row): array{
261			$disposition = 'NOANSWER';
262			if ($billsec > 0) {
263			$disposition = 'ANSWERED';
264			} elseif ('' !== $dialstatus) {
265			$disposition = ($dialstatus === 'ANSWERED') ? $disposition : $dialstatus;
266			}
267
268			if ($disposition !== 'ANSWERED') {
269			if (file_exists($row['recordingfile']) && !is_dir($row['recordingfile'])) {
270			Processes::mwExec("rm -rf {$row['recordingfile']}");
271			}
272			} elseif (!empty($row['recordingfile']) &&
273			!file_exists(Util::trimExtensionForFile($row['recordingfile']) . '.wav') &&
274			!file_exists($row['recordingfile'])) {
275			/** @var CallDetailRecordsTmp $rec_data */
276			$rec_data = CallDetailRecordsTmp::findFirst("linkedid='{$row['linkedid']}' AND dst_chan='{$row['dst_chan']}'");
277			if ($rec_data !== null) {
278			$row['recordingfile'] = $rec_data->recordingfile;
279			}
280			}
281			return array($disposition, $row);
282			}
283
284			}
285
286			// Start worker process
287			$workerClassname = WorkerCdr::class;
288			if (isset($argv) && count($argv) > 1 && $argv[1] === 'start') {
289			cli_set_process_title($workerClassname);
290			try {
291			$worker = new $workerClassname();
292			$worker->start($argv);
293			} catch (Throwable $e) {
294			global $errorLogger;
295			$errorLogger->captureException($e);
296			Util::sysLogMsg("{$workerClassname}_EXCEPTION", $e->getMessage());
297			}
298			}

mikopbx / Core

Push — develop ( cf94b8...c1cb26 )

WorkerCdr::setDisposition() B

Complexity

Size

Duplication

Importance

How to fix Complexity

Long Method

Duplication Side-by-Side

Filter issues like