WorkerCdr::notifyByEmail() - Code Metrics - Inspection of "Обновил описание в composer.json" - mikopbx/Core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#16)

by Nikolay

created 2020-10-30 13:03 UTC

WorkerCdr::notifyByEmail() A

↳ Parent: WorkerCdr

Complexity

Conditions	2
Paths	2

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	2
nc	2
nop	0

<?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, Users};
use MikoPBX\Core\System\{BeanstalkClient, Util};

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

    public const SELECT_CDR_TUBE = 'select_cdr_tube';
    public const UPDATE_CDR_TUBE = 'update_cdr_tube';
    protected int $maxProc=1;


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


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


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

        $this->initSettings();

        while (true) {
            $result = $this->client_queue->request(json_encode($filter), 10);

            if ($result !== false) {
                $this->updateCdr();
            }
            $this->client_queue->wait(5); // instead of sleep
        }
    }

    private function initSettings()
    {
        $this->internal_numbers  = [];
        $this->no_answered_calls = [];
        $users                   = Users::find();
        foreach ($users as $user) {
            if (empty($user->email)) {
                continue;
            }

            foreach ($user->Extensions as $exten) {
                $this->internal_numbers[$exten->number] = [
                    'email'    => $user->email,
                    'language' => $user->language,
                ];
            }
        }
    }

    /**
     * Обработчик результата запроса.
     *
     */
    private function updateCdr(): void
    {
        $this->initSettings();
        $result_data = $this->client_queue->getBody();
        // Получаем результат.
        $result = json_decode($result_data, true);
        if (file_exists($result)) {
            $file_data = json_decode(file_get_contents($result), true);
            if(!is_dir($result)){
                Util::mwExec("rm -rf {$result}");
            }
            $result = $file_data;
        }
        if ( ! is_array($result) && ! is_object($result)) {
            return;
        }
        if (count($result) < 1) {
            return;
        }
        $arr_update_cdr = [];
        // Получаем идентификаторы активных каналов.
        $channels_id = $this->getActiveIdChannels();
        foreach ($result as $row) {
            if (array_key_exists($row['linkedid'], $channels_id)) {
                // Цепочка вызовов еще не завершена.
                continue;
            }
            if (trim($row['recordingfile']) !== '') {
                // Если каналов не существует с ID, то можно удалить временные файлы.
                $p_info = pathinfo($row['recordingfile']);
                $fname  = $p_info['dirname'] . '/' . $p_info['filename'] . '.wav';
                if (file_exists($fname) && !is_dir($fname)) {
                    Util::mwExec("rm -rf {$fname}");
                }
            }
            $start      = strtotime($row['start']);
            $answer     = strtotime($row['answer']);
            $end        = strtotime($row['endtime']);
            $dialstatus = trim($row['dialstatus']);

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

            $disposition = 'NOANSWER';
            if ($billsec > 0) {
                $disposition = 'ANSWERED';
            } elseif ('' !== $dialstatus) {
                $disposition = ($dialstatus === 'ANSWERED') ? $disposition : $dialstatus;
            }

            if ($billsec <= 0) {
                $row['answer'] = '';
                $billsec       = 0;

                if ( ! empty($row['recordingfile'])) {
                    $p_info    = pathinfo($row['recordingfile']);
                    $file_list = [
                        $p_info['dirname'] . '/' . $p_info['filename'] . '.mp3',
                        $p_info['dirname'] . '/' . $p_info['filename'] . '.wav',
                        $p_info['dirname'] . '/' . $p_info['filename'] . '_in.wav',
                        $p_info['dirname'] . '/' . $p_info['filename'] . '_out.wav',
                    ];
                    foreach ($file_list as $file) {
                        if ( ! file_exists($file)) {
                            continue;
                        }
                        if(!is_dir($file)){
                            Util::mwExec("rm -rf {$file}");
                        }
                    }
                }
            }

            if ($disposition !== 'ANSWERED') {
                if (file_exists($row['recordingfile']) && !is_dir($row['recordingfile'])) {
                    Util::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;
                }
            }

            $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));
        }

        foreach ($arr_update_cdr as $data) {
            $linkedid              = $data['tmp_linked_id'];
            $data['GLOBAL_STATUS'] = $data['disposition'];
            if (isset($this->no_answered_calls[$linkedid]) &&
                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), null, self::UPDATE_CDR_TUBE);
        }

        $this->notifyByEmail();
    }

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

        return $active_chans;
    }

    /**
     * Анализируем не отвеченные вызовы. Наполняем временный массив для дальнейшей обработки.
     *
     * @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), null, WorkerNotifyByEmail::class);
        }
        $this->no_answered_calls = [];
    }

}

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

mikopbx / Core

Pull Request — master (#16)

WorkerCdr::notifyByEmail() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like