autocall

#!/usr/bin/env php
<?php
    
    require_once 'vendor/autoload.php';
    
    use splitbrain\phpcli\CLI;
    use splitbrain\phpcli\Colors;
    use splitbrain\phpcli\Options;
    use splitbrain\phpcli\TableFormatter;
    use Symfony\Component\Yaml\Yaml;
    use React\EventLoop\Loop;
    
    class Minimal extends CLI
    {
        
        public array $companies = [];
        public int $wait = 1;
        public int $channels = 1;
        public int $queue_size = 3;
        public DateTime $start_time;
        public bool $is_debug = false;
        public string $pool_path_out = '/storage/usbdisk1/mikopbx/astspool/outgoing/';
        public string $pool_path_done = '/storage/usbdisk1/mikopbx/astspool/outgoing_done/';
        
        /**
         * register options and arguments
         * @param Options $options
         */
        protected function setup(Options $options): void
        {
            $options->setHelp('Скрипт автообзвона для Asterisk от Asko.Host');
            $options->registerCommand('run', 'Запустить обзвон');
            $options->registerCommand('list', 'Список компаний');
            $options->registerOption('wait',
                'Пауза между звонками в рамках транка кампании, в секундах', 'w',
                'string');
            $options->registerOption('queue_size', 'Максимальное количество звонков в процессе.', 'q', 'count');
            $options->registerOption('channels', 'Кол-во одновременных каналов транка.', 'c', 'count');
            $options->registerOption('is_debug', 'Режим отладки.', 'd', '1|0');
        }
        
        /**
         * @param $name
         * @param $message
         * @throws Exception
         */
        public function plog($name, $message): void
        {
            $time = date('Y-m-d H:i:s');
            $message = "$time: $message";
            file_put_contents($this->log_path($name), "$message\n", FILE_APPEND);
        }
        
        /**
         * implement your code
         * @param Options $options
         * @throws Exception
         */
        protected function main(Options $options): void
        {
            $wait = $options->getOpt('wait');
            $this->wait = $wait ?: $this->wait;
            
            $channels = $options->getOpt('channels');
            $this->channels = $channels ?: $this->channels;
            
            $queue_size = $options->getOpt('queue_size');
            $this->queue_size = $queue_size ?: $this->queue_size;
            
            $is_debug = $options->getOpt('is_debug');
            $this->is_debug = $is_debug ? (bool)$is_debug : $this->is_debug;
            
            switch ($options->getCmd()) {
                case 'run':
                    $this->c_run();
                    break;
                case 'list':
                    $this->c_list();
                    $this->show_progress(false);
                    break;
                default:
                    $this->error('Ни одна известная команда не была вызвана, вместо этого мы показываем справку по умолчанию:');
                    echo $options->help();
                    exit;
            }
        }
        
        /**
         * @throws Exception
         */
        public function c_list(): void
        {
            $companies = glob(__DIR__ . "/data/companies/*", GLOB_ONLYDIR);
            
            foreach ($companies as $dir) {
                $name = basename($dir);
                
                if (!file_exists("$dir/number.list")) {
                    $error = "Нашел папку компании " . basename($dir) .
                             ", но в ней нет списка обзвона number.list";
                    throw new Exception($error);
                }
                
                $path_numbers = "$dir/number.list";
                
                $this->check_dublicate_numbers($name, $path_numbers);
                
                $done_file = $this->splfile("$dir/finished.list");
                $file = $this->splfile($path_numbers);
                
                while ($phone = $file->current()) {
                    if ($done_file->current()) {
                        if ($phone !== $done_file->current()) {
                            // todo: Нелья проверять последовательность так как в finished.list номера могут попадать в другой
                            // throw new Exception("Нарушена последовательность списка номеров и завершенных");
                        }
                        $done_file->next();
                    }
                    $file->next();
                }
                
                $this->companies[ $name ] = [
                    'name'     => $name,
                    'dir'      => $dir,
                    'log'      => date('Y-m-d_H:i:s') . ".log",
                    'spl'      => $file,
                    'spl_done' => $done_file,                                   // Текущая пауза в секундах
                    'count'    => $file->key(),                                 // Всего номеров в компании
                    'queue'    => [],                                           // В процессе
                    'done'     => $done_file->key(),                            // Завершено
                    'template' => [],
                ];
                
                $file->seek($done_file->key());
                
                $trunks_path = "$dir/trunks.list";
                
                if (!file_exists($trunks_path)) {
                    throw new Exception("Для компании $name не нашел список транков");
                }
                
                $trunks = file($trunks_path, FILE_SKIP_EMPTY_LINES | FILE_IGNORE_NEW_LINES);
                
                if ($trunks === false) {
                    throw new Exception("Не смог прочитать файл со списком транков $trunks_path");
                }
                
                if (!count($trunks)) {
                    throw new Exception("В списке транков должен быть хотя бы один транк");
                }
                
                // Делаем транки ключами
                $trunks = array_flip($trunks);
                // Задаем всем транкам текущее время
                $trunks = array_map(static fn() => date('Y-m-d H:i:s'), $trunks);
                $this->companies[ $name ][ 'trunks' ] = $trunks;
                
                $cafile_path = "$dir/call_file.tpl";
                
                if (!file_exists($cafile_path)) {
                    throw new Exception("Для компании $name не нашел шаблон Call файла!");
                }
                
                $cafile = file_get_contents($cafile_path);
                $this->companies[ $name ][ 'template' ][ 'raw' ] = $cafile;
                
                $ini = Yaml::parse($cafile);
                
                $this->companies[ $name ][ 'template' ][ 'direction' ] = $ini[ 'Context' ] . ' [' . $ini[ 'Extension' ] . ']';
            }
        }
        
        /**
         * @param $path
         * @return SplFileObject
         */
        public function splfile($path): SplFileObject
        {
            $file = new SplFileObject($path, 'a+');
            $file->setFlags(SplFileObject::READ_AHEAD | SplFileObject::SKIP_EMPTY | SplFileObject::DROP_NEW_LINE);
            return $file;
        }
        
        public function show_progress($is_update = true): void
        {
            ob_implicit_flush(false);
            ob_start();
            
            $tf = new TableFormatter($this->colors);
            $tf->setBorder(' | '); // nice border between columns
            
            $tf->setMaxWidth(120);
            
            // show a header
            echo $tf->format(
                ['*', '20%', '13%', '12%', '12%', '12%', '6%'],
                ['Список компаний', 'Направление', 'Всего номеров', 'В процессе', 'Завершено', 'Осталось', '']
            );
            
            // a line across the whole width
            echo str_pad('', $tf->getMaxWidth(), '-') . "\n";
            
            ksort($this->companies);
            
            foreach ($this->companies as $name => $company) {
                $remain = $company[ 'count' ] - $company[ 'done' ];
                $progress = round($company[ 'done' ] * 100 / $company[ 'count' ]);
                $direction = $company[ 'template' ][ 'direction' ];
                
                echo $tf->format(
                    ['*', '20%', '12%', '12%', '12%', '12%', '6%'],
                    [
                        $name,
                        $direction,
                        $company[ 'count' ],
                        $this->size_queue($name),
                        $company[ 'done' ],
                        $remain,
                        $progress . '%'
                    ],
                    [
                        Colors::C_CYAN,
                        Colors::C_WHITE,
                        Colors::C_WHITE,
                        Colors::C_BLUE,
                        Colors::C_GREEN,
                        Colors::C_RED,
                        Colors::C_YELLOW
                    ]
                );
                
                // todo:: Убрать контроль окончания кампании из вывода прогресса
                if ($is_update && $remain === 0) {
                    // Отмечаем что очередь вызовов закончена
                    $this->companies[ $name ][ 'is_over' ] = 1;
                    // Отключаем обработку очереди вызовов
                    Loop::cancelTimer($this->companies[ $name ][ 'timer_queue' ]);
                }
            }
            
            $out = ob_get_clean();
            
            if ($is_update) {
                system('clear');
            }
            
            if ($is_update) {
                fwrite(STDOUT, PHP_EOL);
                $this->info("Интервал в секундах между звонками транка  : {$this->wait}");
                $this->info("Максимальное кол-во одновременных вызовов  : " . $this->queue_size);
                $this->info("Кол-во одновременных каналов одного транка : " . $this->channels);
                fwrite(STDOUT, PHP_EOL);
                $elapsed = $this->start_time->diff(new DateTime)->format('%h:%i:%s');
                $this->alert("Прошло времени: $elapsed");
            }
            
            fwrite(STDOUT, PHP_EOL . $out);
        }
        
        /**
         * Свободно каналов для транка
         * @param $name
         * @param $trunk
         * @return int
         */
        public function free_channel($name, $trunk): int
        {
            $queue = $this->companies[ $name ][ 'queue' ];
            return array_reduce($queue, static function ($carry, $_trunk) use ($trunk)
            {
                return $_trunk === $trunk ? $carry - 1 : $carry;
            }, $this->channels);
        }
        
        /**
         * Извлекаем транк.
         * В зависимости от режима либо по кругу, либо случайный
         * Та же учитываем паузу и время последнего завершенного звонка
         * @param        $name
         * @return false|string
         * @throws Exception
         */
        public function get_trunk($name)
        {
            $trunk = array_key_first($this->companies[ $name ][ 'trunks' ]);
            
            // Неважно извлечем мы транк для работы или нет, перемещаем его в конец очереди
            $future_time = array_shift($this->companies[ $name ][ 'trunks' ]);
            $this->companies[ $name ][ 'trunks' ][ $trunk ] = $future_time;
            
            $free_channel = $this->free_channel($name, $trunk);
            // Когда достаточно свободных каналов транка и пауза прошла
            if ($free_channel > 0 && date_create($future_time) <= date_create()) {
                return $trunk;
            }
            
            return false;
        }
        
        /**
         * Возвращаем время в будущем на основании заданной паузы
         */
        public function future_time(): string
        {
            $sec = $this->wait;
            $time = new DateTime();
            $interval = new DateInterval("PT{$sec}S");
            $time->add($interval);
            
            return $time->format('Y-m-d H:i:s');
        }
        
        public function temp_call($name, $phone): string
        {
            return "data/temporary/{$name}{$phone}.call";
        }
        
        public function pool_call($name, $phone): string
        {
            $filename = $name . $phone . ".call";
            if ($this->is_debug) {
                return "data/pool_out/$filename";
            }
            return $this->pool_path_out . $filename;
        }
        
        public function pool_call_done($name, $phone): string
        {
            $filename = $name . $phone . ".call";
            if ($this->is_debug) {
                return "data/pool_done/$filename";
            }
            return $this->pool_path_done . $filename;
        }
    
        /**
         * @throws Exception
         */
        public function call_cli($command) {
            ob_start();
            passthru($command, $b);
            $out = ob_get_clean();
            
            if ($b !== 0) {
                throw new Exception($out);
            }
            
            return $out;
        }
        
        /**
         * @throws Exception
         */
        public function asterisk_cli_passthru($command): string
        {
            return $this->call_cli("asterisk -rx '$command'");
        }
        
        /**
         * Проверяем нет ли дубликатов телефонов в файле
         * @throws Exception
         */
        public function check_dublicate_numbers($name, $path): string
        {
            
            $out = $this->call_cli("sort $path | uniq -dc");

            if(!empty($out)) {
                $this->warning("В кампании $name есть дубликаты номеров, удалить их (y|n)?");
                $confirm = fgets(STDIN);

                if(trim($confirm) === 'y') {
                    $this->call_cli("sort -u -o $path $path");
                }
            }
            
            return $out;
        }
        
        /**
         * @throws Exception
         */
        public function log_path($name): string
        {
            $filename = $this->companies[ $name ][ 'log' ];
            $log_dir = $this->companies[ $name ][ 'dir' ] . "/logs";
            if (!file_exists($log_dir) && !mkdir($log_dir, 0755, true) && !is_dir($log_dir)) {
                throw new Exception(sprintf('Директория "%s" не может быть создана', $log_dir));
            }
            return "$log_dir/$filename";
        }
        
        /**
         * Проверяем активность транка на текущий момент
         * @param $trunk
         * @return bool
         * @throws Exception
         */
        public function is_active_trunk($trunk): bool
        {
            $out = $this->asterisk_cli_passthru('core show channels concise');
            
            $trunk = explode('-', $trunk);
            $trunk = implode('-', [$trunk[ 0 ], $trunk[ 1 ]]);
            
            return strpos($out, $trunk) !== false;
        }
        
        // Смотрим какие звонки ушли из очереди
        
        /**
         * @throws Exception
         */
        public function id_done_call($name, $phone, $trunk): bool
        {
            // Так как call файл может быть удален или перемещен не нами, очистим кеш
            clearstatcache();
            return !file_exists($this->pool_call($name, $phone)) &&         // Если удален call файл
                   file_exists($this->pool_call_done($name, $phone)) &&     // Если создан call файл
                   !$this->is_active_trunk($trunk);                         // Если нет активности транка
        }
        
        /**
         * @throws Exception
         */
        public function harvest_info_pool(): void
        {
            foreach ($this->companies as $name => $company) {
                foreach ($company[ 'queue' ] as $phone => $trunk) {
                    if ($this->id_done_call($name, $phone, $trunk)) {
                        unset($this->companies[ $name ][ 'queue' ][ $phone ]);
                        $this->companies[ $name ][ 'done' ]++;
                        
                        $future_time = $this->future_time();
                        
                        $message = "Звонок на $phone через $trunk завершен. ";
                        $this->plog($name, $message);
                        $message = "Активных звонков: " . $this->size_queue($name);
                        $this->plog($name, $message);
                        $message = "Следующий звонок через $trunk возможен после : $future_time";
                        $this->plog($name, $message);
                        
                        // После завершения звонка назначим транку новую паузу (дату в будущем)
                        $this->companies[ $name ][ 'trunks' ][ $trunk ] = $future_time;
                        
                        // Удалим
                        if (!unlink($this->pool_call_done($name, $phone))) {
                            throw new Exception("Не смогу удалить " . $this->pool_call_done($name, $phone));
                        }
                    }
                }
            }
        }
        
        /**
         * Количество звонков в процессе
         * @param $name
         * @return int
         */
        public function size_queue($name): int
        {
            return count($this->companies[ $name ][ "queue" ]);
        }
        
        /**
         * @throws Exception
         */
        public function c_run(): void
        {
            $this->start_time = new DateTime();
            
            $this->c_list();
            
            foreach ($this->companies as $name => $company) {
                $this->plog($name, "Добавил таймер обработки очереди");
                
                $this->companies[ $name ][ 'timer_queue' ] = Loop::addPeriodicTimer(1, function () use ($name)
                {
                    try {
                        if (!$phone = $this->companies[ $name ][ 'spl' ]->current()) {
                            return;
                        }
                        
                        $free_calls = $this->queue_size - $this->size_queue($name);

                        // Если есть окно, то добавляем еще звонок
                        if ($free_calls > 0) {
                            $trunk = $this->get_trunk($name);
                            
                            // Если нет свободного транка с вышедшей паузой, пропустим
                            if ($trunk === false) {
                                return;
                            }
                            
                            $this->add_queue($name, $phone, $trunk);
                        }
                    } catch (Throwable $e) {
                        $this->error($e->getMessage());
                        $this->plog($name, "Error: " . $e->getMessage());
                        Loop::stop();
                    }
                });
            }
            
            // Запускаем обработку
            Loop::addPeriodicTimer(0.8, function ()
            {
                try {
                    $this->harvest_info_pool();
                    $this->show_progress();
                    
                    $overs = array_sum(array_column($this->companies, "is_over"));
                    
                    if ($overs === count($this->companies)) {
                        fwrite(STDOUT, PHP_EOL);
                        $this->success('Обзвон законен!');
                        Loop::stop();
                    }
                } catch (Throwable $e) {
                    $this->error($e->getMessage());
                    Loop::stop();
                }
            });
            
            Loop::addSignal(SIGINT, function (int $signal)
            {
                fprintf(STDOUT, "\033[?25h"); //show cursor
                Loop::stop();
            });
            
            register_shutdown_function(static function ()
            {
                fprintf(STDOUT, "\033[?25h"); //show cursor
            });
            
            fprintf(STDOUT, "\033[?25l"); // hide cursor
            
            Loop::run();
        }
        
        /**
         * @throws Exception
         */
        public function add_queue($name, $phone, $trunk): void
        {
            $message = "Добавляем звонок [$phone] в очередь";
            $this->plog($name, $message);
            
            $cafile = str_replace("__number__", $phone, $this->companies[ $name ][ 'template' ][ 'raw' ]);
            $cafile = str_replace("__trunk__", $trunk, $cafile);
            
            $path_temp = $this->temp_call($name, $phone);
            if (file_put_contents($path_temp, $cafile) === false) {
                throw new Exception("Не смог сохранить временный файл по пути $path_temp");
            }
            
            $path_pool = $this->pool_call($name, $phone);
            if (rename($path_temp, $path_pool) === false) {
                throw new Exception("Не смог переместить временный файл по пути $path_pool");
            }
            
            $this->companies[ $name ][ 'queue' ][ $phone ] = $trunk;
            $this->companies[ $name ][ 'spl_done' ]->fwrite($phone . PHP_EOL);
            $this->companies[ $name ][ 'spl' ]->next();
        }
        
    }
    
    // execute it
    $cli = new Minimal();
    $cli->run();