bin/synchCdr.php - master - mikopbx/ModuleBeelinePbx

bin/synchCdr.php
243 строки10 KB
boffart
Инициализация модуля ModuleBeelinePbx
08 июл 2026, 17:11
08 июл 2026, 17:111cc4e1a
<?php
/*
 * MikoPBX - free phone system for small business
 * Copyright © 2017-2024 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/>.
 */

use MikoPBX\Core\System\BeanstalkClient;
use MikoPBX\Core\System\Util;
use MikoPBX\Core\Workers\WorkerCallEvents;
use Modules\ModuleBeelinePbx\Lib\BeelineApi;
use Modules\ModuleBeelinePbx\Lib\BeelineCdrHelper;
use Modules\ModuleBeelinePbx\Lib\Logger;
use Modules\ModuleBeelinePbx\Models\CallHistory;
use Modules\ModuleBeelinePbx\Models\ModuleBeelinePbx;

require_once 'Globals.php';

// Лаг от текущего времени: активные звонки ещё не завершены и не попадают в статистику.
const BEELINE_SYNC_LAG_MINUTES = 2;

// Перекрытие с предыдущим окном на каждом запуске (перехват «пограничных» звонков).
// Сумма OVERLAP + LAG задаёт «окно отлова» для звонков, которые Билайн индексирует с задержкой.
const BEELINE_SYNC_OVERLAP_MINUTES = 18;

// Максимальный размер окна запроса статистики.
const BEELINE_MAX_WINDOW_DAYS = 5;

// Предохранитель от бесконечной пагинации внутри окна.
const BEELINE_MAX_PAGES_PER_WINDOW = 5000;

$pidFile = '/var/run/beeline-sync.pid';

// Режим глубокой дозагрузки (--lookback=N минут): фиксированное окно назад, offset не трогаем.
// Нужен для длинных/отложенных звонков, попавших в индекс уже после ухода основного окна вперёд.
$deepLookbackMinutes = 0;
foreach ($argv as $arg) {
    if (preg_match('/^--lookback=(\d+)$/', $arg, $m)) {
        $deepLookbackMinutes = (int)$m[1];
    }
}
$isDeep = $deepLookbackMinutes > 0;

$logger = new Logger($isDeep ? 'SyncCdrDeep' : 'SyncCdr', 'ModuleBeelinePbx');

// Перехватываем все необработанные исключения — иначе MikoPBX может автоматически
// отключить модуль из-за ошибки в воркере крона.
set_exception_handler(static function (\Throwable $e) use ($logger) {
    $logger->writeError([
        'exception' => $e->getMessage(),
        'file' => $e->getFile(),
        'line' => $e->getLine(),
        'trace' => $e->getTraceAsString(),
    ], 'Uncaught exception in synchCdr.php');
    exit(1);
});

/**
 * Single-instance guard: основной и глубокий проходы не должны идти одновременно.
 */
function beelineProcessExists(string $pidFile): bool
{
    $pid = posix_getpid();
    if (file_exists($pidFile)) {
        $psPath = Util::which('ps');
        $busyboxPath = Util::which('busybox');
        $oldPid = trim(file_get_contents($pidFile));
        $output = trim(shell_exec("$psPath -A -o 'pid' | $busyboxPath awk '{print $1}' | $busyboxPath grep '^$oldPid\$'") ?? '');
        if (!empty($output)) {
            echo "Old PID: $oldPid is still running...".PHP_EOL;
            return true;
        }
    }
    file_put_contents($pidFile, $pid);
    return false;
}

/**
 * Сохраняет строку CDR в таблицу модуля.
 *
 * Правила слияния:
 *  - строку SIP-транка не даём затирать уже сохранённую строку сотрудника (та же UNIQUEID);
 *  - успешно скачанную запись (bee_rec_status=ok) при пересинхронизации не сбрасываем.
 *
 * @return bool была ли строка записана/обновлена
 */
function saveCdrRow(array $tmpCdr, bool $incomingIsTrunk, Logger $logger): bool
{
    $attempts = 0;
    while ($attempts < 15) {
        $attempts++;
        try {
            $dbCDR = CallHistory::findFirst([
                'linkedid=:linkedid:',
                'bind' => ['linkedid' => $tmpCdr['linkedid']],
            ]);
            if (!$dbCDR) {
                $dbCDR = new CallHistory();
            } else {
                $trunkSuffix = BeelineCdrHelper::TRUNK_USER_SUFFIX;
                // substr-endsWith для совместимости с PHP 7.4 (str_ends_with — PHP 8+).
                $existingIsTrunk = substr((string)$dbCDR->bee_user_id, -strlen($trunkSuffix)) === $trunkSuffix;
                if ($incomingIsTrunk && !$existingIsTrunk) {
                    // Реальная нога сотрудника уже сохранена — транковую ногу игнорируем.
                    return false;
                }
                if ($dbCDR->bee_rec_status === 'ok'
                    && !empty($dbCDR->recordingfile)
                    && file_exists($dbCDR->recordingfile)) {
                    unset($tmpCdr['recordingfile'], $tmpCdr['bee_rec_status']);
                }
            }
            foreach ($tmpCdr as $key => $value) {
                $dbCDR->{$key} = $value;
            }
            if ($dbCDR->save()) {
                return true;
            }
            $logger->writeError($dbCDR->getMessages(), 'Fail save CDR');
        } catch (\Throwable $e) {
            $logger->writeError(['exception' => $e->getMessage(), 'cdr' => $tmpCdr], 'Exception save CDR');
            sleep(1);
        }
    }
    return false;
}

if (beelineProcessExists($pidFile)) {
    echo "Process exists...".PHP_EOL;
    exit(12);
}

$settings = ModuleBeelinePbx::findFirst();
if (!$settings || empty($settings->authApiKey)) {
    echo "empty settings".PHP_EOL;
    exit(1);
}

$gap = (int)$settings->gap;
$employeeMode = (string)$settings->employeeNumberField ?: BeelineCdrHelper::EMPLOYEE_FIELD_EXTENSION;
$excluded = BeelineCdrHelper::parseExcludedNumbers((string)$settings->excludedNumbers);
if (!empty($excluded)) {
    $logger->writeInfo('Excluded numbers list size: ' . count($excluded));
}
$api = new BeelineApi($settings->authApiKey, $logger);

// Определяем левую границу интервала.
if ($isDeep) {
    $windowStart = (new DateTimeImmutable('now'))->modify('-' . $deepLookbackMinutes . ' minutes');
    $logger->writeInfo('Deep lookback sync ' . $deepLookbackMinutes . 'm (offset untouched)...');
} elseif (empty($settings->offset)) {
    $windowStart = (new DateTimeImmutable('now'))->modify('-30 day');
    $logger->writeInfo('Offset is empty, start sync -30 day...');
} else {
    $windowStart = (new DateTimeImmutable($settings->offset))->modify('-' . BEELINE_SYNC_OVERLAP_MINUTES . ' minutes');
    $logger->writeInfo('Continue sync from ' . $windowStart->format('c') . ' (overlap ' . BEELINE_SYNC_OVERLAP_MINUTES . 'm)...');
}
$now = (new DateTimeImmutable('now'))->modify('-' . BEELINE_SYNC_LAG_MINUTES . ' minutes');

$clientBeanstalk = new BeanstalkClient(WorkerCallEvents::class);
$cdrData = ['action' => 'insert_cdr', 'rows' => []];
$haveError = false;
$totalRows = 0;
$skippedExcluded = 0;

while ($windowStart < $now) {
    $windowEnd = $windowStart->modify('+' . BEELINE_MAX_WINDOW_DAYS . ' days');
    if ($windowEnd > $now) {
        $windowEnd = $now;
    }
    $dateFrom = BeelineApi::formatDate($windowStart);
    $dateTo = BeelineApi::formatDate($windowEnd);
    $logger->writeInfo("Sync window {$dateFrom} - {$dateTo}...");

    $page = 0;
    do {
        $rows = $api->getStatisticsPage($dateFrom, $dateTo, $page, BeelineApi::STATISTICS_PAGE_SIZE);
        if ($rows === null) {
            $haveError = true;
            $logger->writeError(['window' => [$dateFrom, $dateTo], 'page' => $page], 'Fail get statistics page');
            break 2; // на ошибке окно не считаем синхронизированным
        }
        foreach ($rows as $row) {
            if (!is_array($row)) {
                continue;
            }
            // Чёрный список: пропускаем звонки с любым исключённым номером (добавочным/внешним).
            if (BeelineCdrHelper::isRowExcluded($row, $excluded)) {
                $skippedExcluded++;
                continue;
            }
            $tmpCdr = BeelineCdrHelper::buildCdrRow($row, $gap, $employeeMode);
            if ($tmpCdr === null) {
                continue;
            }
            $incomingIsTrunk = BeelineCdrHelper::isTrunkAbonent($row);
            $saved = saveCdrRow($tmpCdr, $incomingIsTrunk, $logger);
            if (!$saved) {
                continue;
            }
            $totalRows++;
            $cdrData['rows'][] = $tmpCdr;
            if (count($cdrData['rows']) > 10) {
                $clientBeanstalk->publish(json_encode($cdrData), WorkerCallEvents::class);
                $cdrData['rows'] = [];
                usleep(100000);
            }
        }
        $page++;
        usleep(150000);
    } while (count($rows) >= BeelineApi::STATISTICS_PAGE_SIZE && $page < BEELINE_MAX_PAGES_PER_WINDOW);

    // Окно синхронизировано. Двигаем offset только в обычном режиме.
    if (!$isDeep) {
        $settings->offset = $windowEnd->format('Y-m-d\TH:i:s');
        $settings->save();
        $logger->writeInfo('Update offset ' . $settings->offset . '...');
    }

    if ($windowEnd >= $now) {
        break;
    }
    $windowStart = $windowEnd;
}

if (!empty($cdrData['rows'])) {
    $clientBeanstalk->publish(json_encode($cdrData), WorkerCallEvents::class);
}

$logger->writeInfo("Finished: rows={$totalRows}, skipped_excluded={$skippedExcluded}, error=" . ($haveError ? 'yes' : 'no'));
exit($haveError ? 11 : 0);