WorkerCallEvents::otherEvents() - Code Metrics - Inspection of "#461 Добавил настройки записи разговоров. Возможно..." - mikopbx/Core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( dcb728...9b177f )

by Портнов

created 2023-02-22 13:38 UTC

WorkerCallEvents::otherEvents() A

↳ Parent: WorkerCallEvents

Complexity

Conditions	4
Paths	8

Size

Total Lines	12
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
eloc	8
c	1
b	0
f	0
dl	0
loc	12
rs	10
cc	4
nc	8
nop	2

<?php
/*
 * MikoPBX - free phone system for small business
 * Copyright (C) 2017-2020 Alexey Portnov and Nikolay Beketov
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along with this program.
 * If not, see <https://www.gnu.org/licenses/>.
 */

namespace MikoPBX\Core\Workers;
require_once 'Globals.php';

use MikoPBX\Core\System\{BeanstalkClient, Storage, Util};
use MikoPBX\Common\Models\Extensions;
use MikoPBX\Common\Models\PbxSettings;
use MikoPBX\Common\Models\Sip;
use MikoPBX\Core\Asterisk\Configs\CelConf;
use MikoPBX\Core\Workers\Libs\WorkerCallEvents\ActionCelAnswer;
use MikoPBX\Core\Workers\Libs\WorkerCallEvents\SelectCDR;
use MikoPBX\Core\Workers\Libs\WorkerCallEvents\UpdateDataInDB;
use Phalcon\Text;

class WorkerCallEvents extends WorkerBase
{
    public    array $mixMonitorChannels = [];
    protected bool  $record_calls       = true;
    protected bool  $split_audio_thread = false;
    public    array $checkChanHangupTransfer = [];
    private   array $activeChannels = [];
    public const TIMOUT_CHANNEL_TUBE = 'CleanChannelTimout';

    private array $innerNumbers       = [];
    private array $exceptionsNumbers  = [];
    private bool  $notRecInner        = false;
    public const REC_DISABLE          = 'Conversation recording is disabled';

    /**
     * Наполняем кэш реальных каналов.
     * @param string $channel
     * @return void
     */
    public function addActiveChan(string $channel):void
    {
        if(stripos($channel, 'local') === 0){
            return;
        }
        $this->activeChannels[$channel] = true;
    }

    /**
     * Очищаем кэш реальных каналов.
     * @param string $channel
     * @return void
     */
    public function removeActiveChan(string $channel):void
    {
        unset($this->activeChannels[$channel]);
    }

    /**
     * Проверяет существует ли канал в кэш.
     * @param string $channel
     * @return void
     */
    public function existsActiveChan(string $channel):bool
    {
        return isset($this->activeChannels[$channel]);

    }

    /**
     * @param string $src
     * @param string $dst
     * @param string $error
     * @return bool
     */
    public function enableMonitor(string $src, string $dst):bool
    {
        $src = substr($src,-9);
        $dst = substr($dst,-9);
        $enable = true;
        $isInner = in_array($src, $this->innerNumbers,true) && in_array($dst, $this->innerNumbers,true);
        if(($this->notRecInner && $isInner) ||
            in_array($src, $this->exceptionsNumbers,true) || in_array($dst, $this->exceptionsNumbers,true)){
            $enable = false;
        }
        return $enable;
    }

    /**
     * Инициирует запись разговора на канале.
     *
     * @param string    $channel
     * @param ?string   $file_name
     * @param ?string   $sub_dir
     * @param ?string   $full_name
     *
     * @return string
     */
    public function MixMonitor($channel, $file_name = null, $sub_dir = null, $full_name = null, string $actionID=''): string
    {
        $resFile = $this->mixMonitorChannels[$channel]??'';
        if($resFile !== ''){
            return $resFile;
        }
        $resFile           = '';
        $file_name = str_replace('/', '_', $file_name);
        if ($this->record_calls) {
            [$f, $options] = $this->setMonitorFilenameOptions($full_name, $sub_dir, $file_name);
            $arr = $this->am->GetChannels(false);
            if(!in_array($channel, $arr, true)){
                return '';
            }
            $srcFile = "{$f}.wav";
            $resFile = "{$f}.mp3";
            $this->am->MixMonitor($channel, $srcFile, $options, '', $actionID);
            $this->mixMonitorChannels[$channel] = $resFile;
            $this->am->UserEvent('StartRecording', ['recordingfile' => $resFile, 'recchan' => $channel]);
        }
        return $resFile;
    }

    /**
     * @param string|null $full_name
     * @param string|null $sub_dir
     * @param string|null $file_name
     * @return array
     */
    public function setMonitorFilenameOptions(?string $full_name, ?string $sub_dir, ?string $file_name): array{
        if (!file_exists((string)$full_name)) {
            $monitor_dir = Storage::getMonitorDir();
            if ($sub_dir === null) {
                $sub_dir = date('Y/m/d/H/');
            }
            $f = "{$monitor_dir}/{$sub_dir}{$file_name}";
        } else {
            $f = Util::trimExtensionForFile($full_name);
        }
        if ($this->split_audio_thread) {
            $options = "abSr({$f}_in.wav)t({$f}_out.wav)";
        } else {
            $options = 'ab';
        }
        return array($f, $options);
    }

    /**
     * Останавливает запись разговора на канале.
     * @param string $channel
     * @param string $actionID
     */
    public function StopMixMonitor($channel, string $actionID=''): void
    {
        if(isset($this->mixMonitorChannels[$channel])){
            unset($this->mixMonitorChannels[$channel]);
        }else{
            return;
        }
        if ($this->record_calls) {
            $this->am->StopMixMonitor($channel, $actionID);
        }
    }

    /**
     *
     * @param $params
     */
    public function start($params): void
    {
        $this->updateRecordingOptions();
        $this->mixMonitorChannels       = [];
        $this->checkChanHangupTransfer  = [];
        $this->am                 = Util::getAstManager('off');

        // PID сохраняем при начале работы Worker.
        $client = new BeanstalkClient(self::class);
        $client->subscribe(CelConf::BEANSTALK_TUBE,    [$this, 'callEventsWorker']);
        $client->subscribe(self::class,                [$this, 'otherEvents']);
        $client->subscribe(WorkerCdr::SELECT_CDR_TUBE, [$this, 'selectCDRWorker']);
        $client->subscribe(WorkerCdr::UPDATE_CDR_TUBE, [$this, 'updateCDRWorker']);
        $client->subscribe(self::TIMOUT_CHANNEL_TUBE,  [$this, 'cleanTimeOutChannel']);
        $client->subscribe($this->makePingTubeName(self::class), [$this, 'pingCallBack']);
        $client->setErrorHandler([$this, 'errorHandler']);

        while ($this->needRestart === false) {
            $client->wait();
        }
    }

    /**
     * @return void
     */
    private function updateRecordingOptions():void
    {
        $usersNumbers = [];
        $users = [];
        $filter = [
            'conditions' => 'userid <> "" and userid>0 ',
            'columns' => 'userid,number,type',
            'order' => 'type DESC'
        ];
        $extensionsData = Extensions::find($filter);
        /** @var Extensions $extension */
        foreach ($extensionsData as $extension){
            if($extension->type === "SIP"){
                $usersNumbers[$extension->number][] = $extension->number;
                $users[$extension->userid] = $extension->number;
            }else{
                $internalNumber = $users[$extension->userid]??'';
                if($internalNumber !==''){
                    $usersNumbers[$internalNumber][] = $extension->number;
                }
            }
        }
        unset($users, $extensionsData);
        $filter = [
            'conditions' => 'type="peer"',
            'columns'    => 'extension,enableRecording',
        ];
        $peers = Sip::find($filter);
        foreach ($peers as $peer) {
            $numbers = $usersNumbers[$peer->extension]??[];
            foreach ($numbers as $num){
                $num = substr($num,-9);
                $this->innerNumbers[] = $num;
                if($peer->enableRecording === '0'){
                    $this->exceptionsNumbers[] = $num;
                }
            }
        }
        $this->notRecInner        = PbxSettings::getValueByKey('PBXRecordCallsInner') === '0';
        $this->record_calls       = PbxSettings::getValueByKey('PBXRecordCalls') === '1';
        $this->split_audio_thread = PbxSettings::getValueByKey('PBXSplitAudioThread') === '1';
    }

    /**
     * Ping callback for keep alive check
     *
     * @param BeanstalkClient $message
     */
    public function pingCallBack(BeanstalkClient $message): void
    {
        parent::pingCallBack($message);
        $this->updateRecordingOptions();
    }

    /**
     * @param $tube
     * @param $data
     * @return void
     */
    public function otherEvents($tube, array $data=[]): void
    {
        if(empty($data)){
            $data = json_decode($tube->getBody(), true);
        }
        $funcName = "Action_".$data['action']??'';
        if ( method_exists($this, $funcName) ) {
            $this->$funcName($data);
        }
        $className = __NAMESPACE__.'\Libs\WorkerCallEvents\\'.Text::camelize($funcName, '_');
        if( method_exists($className, 'execute') ){
            $className::execute($this, $data);
        }
    }

    /**
     * Обработчик событий изменения состояния звонка.
     *
     * @param array | BeanstalkClient $tube
     */
    public function callEventsWorker($tube): void
    {
        $data  = json_decode($tube->getBody(), true);
        $event = $data['EventName']??'';
        if('ANSWER' === $event){
            ActionCelAnswer::execute($this, $data);
            return;
        }elseif('USER_DEFINED' !== $event){
            return;
        }
        try {
            $data = json_decode(
                base64_decode($data['AppData']??''),
                true,
                512,
                JSON_THROW_ON_ERROR
            );
        }catch (\Throwable $e){
            $data = [];
        }
        $this->otherEvents($tube, $data);
    }

    /**
     * Получения CDR к обработке.
     *
     * @param array | BeanstalkClient $tube
     */
    public function updateCDRWorker($tube): void
    {
        $task    = $tube->getBody();
        $data = json_decode($task, true);
        UpdateDataInDB::execute($data);
        $tube->reply(json_encode(true));
    }

    /**
     * Получения CDR к обработке.
     *
     * @param array | BeanstalkClient $tube
     */
    public function cleanTimeOutChannel($tube): void
    {
        $task        = $tube->getBody();
        $taskData    = json_decode($task, true);
        $srcChannel  = $taskData['srcChannel']??'';
        $this->am->SetVar($srcChannel, "MASTER_CHANNEL(M_DIALSTATUS)", 'ANSWER');
        $tube->reply(json_encode(true));
    }

    /**
     * @param array | BeanstalkClient $tube
     */
    public function selectCDRWorker($tube): void
    {
        $filter   = json_decode($tube->getBody(), true);
        $res_data = SelectCDR::execute($filter);
        $tube->reply($res_data);
    }

    public function errorHandler($m): void
    {
        Util::sysLogMsg(self::class . '_ERROR', $m, LOG_ERR);
    }
}


// Start worker process
WorkerCallEvents::startWorker($argv??null);

1			<?php
2			/*
3			* MikoPBX - free phone system for small business
4			* Copyright (C) 2017-2020 Alexey Portnov and Nikolay Beketov
5			*
6			* This program is free software: you can redistribute it and/or modify
7			* it under the terms of the GNU General Public License as published by
8			* the Free Software Foundation; either version 3 of the License, or
9			* (at your option) any later version.
10			*
11			* This program is distributed in the hope that it will be useful,
12			* but WITHOUT ANY WARRANTY; without even the implied warranty of
13			* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			* GNU General Public License for more details.
15			*
16			* You should have received a copy of the GNU General Public License along with this program.
17			* If not, see <https://www.gnu.org/licenses/>.
18			*/
19
20			namespace MikoPBX\Core\Workers;
21			require_once 'Globals.php';
22
23			use MikoPBX\Core\System\{BeanstalkClient, Storage, Util};
24			use MikoPBX\Common\Models\Extensions;
25			use MikoPBX\Common\Models\PbxSettings;
26			use MikoPBX\Common\Models\Sip;
27			use MikoPBX\Core\Asterisk\Configs\CelConf;
28			use MikoPBX\Core\Workers\Libs\WorkerCallEvents\ActionCelAnswer;
29			use MikoPBX\Core\Workers\Libs\WorkerCallEvents\SelectCDR;
30			use MikoPBX\Core\Workers\Libs\WorkerCallEvents\UpdateDataInDB;
31			use Phalcon\Text;
32
33			class WorkerCallEvents extends WorkerBase
34			{
35			public array $mixMonitorChannels = [];
36			protected bool $record_calls = true;
37			protected bool $split_audio_thread = false;
38			public array $checkChanHangupTransfer = [];
39			private array $activeChannels = [];
40			public const TIMOUT_CHANNEL_TUBE = 'CleanChannelTimout';
41
42			private array $innerNumbers = [];
43			private array $exceptionsNumbers = [];
44			private bool $notRecInner = false;
45			public const REC_DISABLE = 'Conversation recording is disabled';
46
47			/**
48			* Наполняем кэш реальных каналов.
49			* @param string $channel
50			* @return void
51			*/
52			public function addActiveChan(string $channel):void
53			{
54			if(stripos($channel, 'local') === 0){
55			return;
56			}
57			$this->activeChannels[$channel] = true;
58			}
59
60			/**
61			* Очищаем кэш реальных каналов.
62			* @param string $channel
63			* @return void
64			*/
65			public function removeActiveChan(string $channel):void
66			{
67			unset($this->activeChannels[$channel]);
68			}
69
70			/**
71			* Проверяет существует ли канал в кэш.
72			* @param string $channel
73			* @return void
74			*/
75			public function existsActiveChan(string $channel):bool
76			{
77			return isset($this->activeChannels[$channel]);
			0 ignored issues – show Bug Best Practice introduced 2023-02-22 13:44 UTC by Report Bug Copy Issue Report The expression `return IssetNode` returns the type `boolean` which is incompatible with the documented return type `void`. Loading history...
78			}
79
80			/**
81			* @param string $src
82			* @param string $dst
83			* @param string $error
84			* @return bool
85			*/
86			public function enableMonitor(string $src, string $dst):bool
87			{
88			$src = substr($src,-9);
89			$dst = substr($dst,-9);
90			$enable = true;
91			$isInner = in_array($src, $this->innerNumbers,true) && in_array($dst, $this->innerNumbers,true);
92			if(($this->notRecInner && $isInner) \|\|
93			in_array($src, $this->exceptionsNumbers,true) \|\| in_array($dst, $this->exceptionsNumbers,true)){
94			$enable = false;
95			}
96			return $enable;
97			}
98
99			/**
100			* Инициирует запись разговора на канале.
101			*
102			* @param string $channel
103			* @param ?string $file_name
104			* @param ?string $sub_dir
105			* @param ?string $full_name
106			*
107			* @return string
108			*/
109			public function MixMonitor($channel, $file_name = null, $sub_dir = null, $full_name = null, string $actionID=''): string
110			{
111			$resFile = $this->mixMonitorChannels[$channel]??'';
112			if($resFile !== ''){
113			return $resFile;
114			}
115			$resFile = '';
116			$file_name = str_replace('/', '_', $file_name);
117			if ($this->record_calls) {
118			[$f, $options] = $this->setMonitorFilenameOptions($full_name, $sub_dir, $file_name);
119			$arr = $this->am->GetChannels(false);
120			if(!in_array($channel, $arr, true)){
121			return '';
122			}
123			$srcFile = "{$f}.wav";
124			$resFile = "{$f}.mp3";
125			$this->am->MixMonitor($channel, $srcFile, $options, '', $actionID);
126			$this->mixMonitorChannels[$channel] = $resFile;
127			$this->am->UserEvent('StartRecording', ['recordingfile' => $resFile, 'recchan' => $channel]);
128			}
129			return $resFile;
130			}
131
132			/**
133			* @param string\|null $full_name
134			* @param string\|null $sub_dir
135			* @param string\|null $file_name
136			* @return array
137			*/
138			public function setMonitorFilenameOptions(?string $full_name, ?string $sub_dir, ?string $file_name): array{
139			if (!file_exists((string)$full_name)) {
140			$monitor_dir = Storage::getMonitorDir();
141			if ($sub_dir === null) {
142			$sub_dir = date('Y/m/d/H/');
143			}
144			$f = "{$monitor_dir}/{$sub_dir}{$file_name}";
145			} else {
146			$f = Util::trimExtensionForFile($full_name);
147			}
148			if ($this->split_audio_thread) {
149			$options = "abSr({$f}_in.wav)t({$f}_out.wav)";
150			} else {
151			$options = 'ab';
152			}
153			return array($f, $options);
154			}
155
156			/**
157			* Останавливает запись разговора на канале.
158			* @param string $channel
159			* @param string $actionID
160			*/
161			public function StopMixMonitor($channel, string $actionID=''): void
162			{
163			if(isset($this->mixMonitorChannels[$channel])){
164			unset($this->mixMonitorChannels[$channel]);
165			}else{
166			return;
167			}
168			if ($this->record_calls) {
169			$this->am->StopMixMonitor($channel, $actionID);
170			}
171			}
172
173			/**
174			*
175			* @param $params
176			*/
177			public function start($params): void
178			{
179			$this->updateRecordingOptions();
180			$this->mixMonitorChannels = [];
181			$this->checkChanHangupTransfer = [];
182			$this->am = Util::getAstManager('off');
183
184			// PID сохраняем при начале работы Worker.
185			$client = new BeanstalkClient(self::class);
186			$client->subscribe(CelConf::BEANSTALK_TUBE, [$this, 'callEventsWorker']);
187			$client->subscribe(self::class, [$this, 'otherEvents']);
188			$client->subscribe(WorkerCdr::SELECT_CDR_TUBE, [$this, 'selectCDRWorker']);
189			$client->subscribe(WorkerCdr::UPDATE_CDR_TUBE, [$this, 'updateCDRWorker']);
190			$client->subscribe(self::TIMOUT_CHANNEL_TUBE, [$this, 'cleanTimeOutChannel']);
191			$client->subscribe($this->makePingTubeName(self::class), [$this, 'pingCallBack']);
192			$client->setErrorHandler([$this, 'errorHandler']);
193
194			while ($this->needRestart === false) {
195			$client->wait();
196			}
197			}
198
199			/**
200			* @return void
201			*/
202			private function updateRecordingOptions():void
203			{
204			$usersNumbers = [];
205			$users = [];
206			$filter = [
207			'conditions' => 'userid <> "" and userid>0 ',
208			'columns' => 'userid,number,type',
209			'order' => 'type DESC'
210			];
211			$extensionsData = Extensions::find($filter);
212			/** @var Extensions $extension */
213			foreach ($extensionsData as $extension){
214			if($extension->type === "SIP"){
215			$usersNumbers[$extension->number][] = $extension->number;
216			$users[$extension->userid] = $extension->number;
217			}else{
218			$internalNumber = $users[$extension->userid]??'';
219			if($internalNumber !==''){
220			$usersNumbers[$internalNumber][] = $extension->number;
221			}
222			}
223			}
224			unset($users, $extensionsData);
225			$filter = [
226			'conditions' => 'type="peer"',
227			'columns' => 'extension,enableRecording',
228			];
229			$peers = Sip::find($filter);
230			foreach ($peers as $peer) {
231			$numbers = $usersNumbers[$peer->extension]??[];
232			foreach ($numbers as $num){
233			$num = substr($num,-9);
234			$this->innerNumbers[] = $num;
235			if($peer->enableRecording === '0'){
236			$this->exceptionsNumbers[] = $num;
237			}
238			}
239			}
240			$this->notRecInner = PbxSettings::getValueByKey('PBXRecordCallsInner') === '0';
241			$this->record_calls = PbxSettings::getValueByKey('PBXRecordCalls') === '1';
242			$this->split_audio_thread = PbxSettings::getValueByKey('PBXSplitAudioThread') === '1';
243			}
244
245			/**
246			* Ping callback for keep alive check
247			*
248			* @param BeanstalkClient $message
249			*/
250			public function pingCallBack(BeanstalkClient $message): void
251			{
252			parent::pingCallBack($message);
253			$this->updateRecordingOptions();
254			}
255
256			/**
257			* @param $tube
258			* @param $data
259			* @return void
260			*/
261			public function otherEvents($tube, array $data=[]): void
262			{
263			if(empty($data)){
264			$data = json_decode($tube->getBody(), true);
265			}
266			$funcName = "Action_".$data['action']??'';
267			if ( method_exists($this, $funcName) ) {
268			$this->$funcName($data);
269			}
270			$className = __NAMESPACE__.'\Libs\WorkerCallEvents\\'.Text::camelize($funcName, '_');
271			if( method_exists($className, 'execute') ){
272			$className::execute($this, $data);
273			}
274			}
275
276			/**
277			* Обработчик событий изменения состояния звонка.
278			*
279			* @param array \| BeanstalkClient $tube
280			*/
281			public function callEventsWorker($tube): void
282			{
283			$data = json_decode($tube->getBody(), true);
284			$event = $data['EventName']??'';
285			if('ANSWER' === $event){
286			ActionCelAnswer::execute($this, $data);
287			return;
288			}elseif('USER_DEFINED' !== $event){
289			return;
290			}
291			try {
292			$data = json_decode(
293			base64_decode($data['AppData']??''),
294			true,
295			512,
296			JSON_THROW_ON_ERROR
297			);
298			}catch (\Throwable $e){
299			$data = [];
300			}
301			$this->otherEvents($tube, $data);
302			}
303
304			/**
305			* Получения CDR к обработке.
306			*
307			* @param array \| BeanstalkClient $tube
308			*/
309			public function updateCDRWorker($tube): void
310			{
311			$task = $tube->getBody();
312			$data = json_decode($task, true);
313			UpdateDataInDB::execute($data);
314			$tube->reply(json_encode(true));
315			}
316
317			/**
318			* Получения CDR к обработке.
319			*
320			* @param array \| BeanstalkClient $tube
321			*/
322			public function cleanTimeOutChannel($tube): void
323			{
324			$task = $tube->getBody();
325			$taskData = json_decode($task, true);
326			$srcChannel = $taskData['srcChannel']??'';
327			$this->am->SetVar($srcChannel, "MASTER_CHANNEL(M_DIALSTATUS)", 'ANSWER');
328			$tube->reply(json_encode(true));
329			}
330
331			/**
332			* @param array \| BeanstalkClient $tube
333			*/
334			public function selectCDRWorker($tube): void
335			{
336			$filter = json_decode($tube->getBody(), true);
337			$res_data = SelectCDR::execute($filter);
338			$tube->reply($res_data);
339			}
340
341			public function errorHandler($m): void
342			{
343			Util::sysLogMsg(self::class . '_ERROR', $m, LOG_ERR);
344			}
345			}
346
347
348			// Start worker process
349			WorkerCallEvents::startWorker($argv??null);

mikopbx / Core

Push — develop ( dcb728...9b177f )

WorkerCallEvents::otherEvents() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like