bin/downloadRecords.php - master - mikopbx/ModuleBeelinePbx

bin/downloadRecords.php
239 строк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\Storage;
use MikoPBX\Core\System\Util;
use MikoPBX\Core\Workers\WorkerCallEvents;
use Modules\ModuleBeelinePbx\Lib\AudioRecodeHelper;
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';

// Окно, за которое каждый запуск проверяет наличие записей (Билайн хранит ~3 месяца;
// берём небольшое скользящее окно — уже скачанные файлы пропускаются по существованию).
const BEELINE_RECORDS_LOOKBACK_DAYS = 3;
// Допуск при сопоставлении записи с CDR: record.date (время разговора) может отличаться
// от startDate статистики на несколько секунд (наблюдалось ~10с).
const BEELINE_RECORDS_MATCH_WINDOW_SEC = 120;
const BEELINE_RECORDS_MAX_PAGES = 2000;

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

$logger = new Logger('DownloadRecords', 'ModuleBeelinePbx');
AudioRecodeHelper::setLogger($logger);

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 downloadRecords.php');
    AudioRecodeHelper::setLogger(null);
    exit(1);
});

function beelineRecordsProcessExists(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)) {
            return true;
        }
    }
    file_put_contents($pidFile, $pid);
    return false;
}

/**
 * Находит строку CDR модуля, соответствующую записи разговора, по добавочному номеру,
 * внешнему номеру и окну времени (record.date может отличаться от start на несколько секунд).
 *
 * @return CallHistory|null
 */
function findCdrForRecord(string $ext, string $externalPhone, DateTime $recDate): ?CallHistory
{
    if ($ext === '' || $externalPhone === '') {
        return null;
    }
    $lo = (clone $recDate)->modify('-' . BEELINE_RECORDS_MATCH_WINDOW_SEC . ' seconds')->format('Y-m-d H:i:s');
    $hi = (clone $recDate)->modify('+' . BEELINE_RECORDS_MATCH_WINDOW_SEC . ' seconds')->format('Y-m-d H:i:s');
    // Окно по start (индексировано) — вернёт немного строк, среди них выбираем ту,
    // где стороны {src,dst} == {ext, externalPhone} и время ближе всего к record.date.
    $rows = CallHistory::find([
        "from_account = '" . BeelineCdrHelper::FROM_ACCOUNT . "' AND start >= :lo: AND start <= :hi:",
        'bind' => ['lo' => $lo, 'hi' => $hi],
    ]);
    $best = null;
    $bestDiff = PHP_INT_MAX;
    $target = $recDate->getTimestamp();
    foreach ($rows as $row) {
        $pair = [(string)$row->src_num, (string)$row->dst_num];
        if (!in_array($ext, $pair, true) || !in_array($externalPhone, $pair, true)) {
            continue;
        }
        $diff = abs(strtotime((string)$row->start) - $target);
        if ($diff < $bestDiff) {
            $bestDiff = $diff;
            $best = $row;
        }
    }
    return $best;
}

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

$settings = ModuleBeelinePbx::findFirst();
if (!$settings || empty($settings->authApiKey)) {
    exit(1);
}
if ((int)$settings->downloadRecordings !== 1) {
    exit(0);
}

$gap = (int)$settings->gap;
$employeeMode = (string)$settings->employeeNumberField ?: BeelineCdrHelper::EMPLOYEE_FIELD_EXTENSION;
$api = new BeelineApi($settings->authApiKey, $logger);
$clientBeanstalk = new BeanstalkClient(WorkerCallEvents::class);

$dateFrom = BeelineApi::formatDate((new DateTimeImmutable('now'))->modify('-' . BEELINE_RECORDS_LOOKBACK_DAYS . ' days'));
$dateTo = BeelineApi::formatDate(new DateTimeImmutable('now'));

$logger->writeInfo("Start records sync window {$dateFrom} - {$dateTo}...");

$cursor = '';
$page = 0;
$ok = 0;
$already = 0;
$noMatch = 0;
$failed = 0;

do {
    $records = $api->getRecordsPage($dateFrom, $dateTo, $cursor);
    if ($records === null) {
        $logger->writeError(['cursor' => $cursor], 'Fail get records page');
        break;
    }
    if (empty($records)) {
        break;
    }
    foreach ($records as $rec) {
        if (!is_array($rec) || empty($rec['id'])) {
            continue;
        }
        $recordId = (string)$rec['id'];
        $cursor = $recordId; // курсор пагинации: следующая страница — после этого id

        $ext = BeelineCdrHelper::employeeNumber($rec['abonent'] ?? [], $employeeMode);
        $externalPhone = BeelineApi::normalizePhone((string)($rec['phone'] ?? ''));
        $recMs = (int)($rec['date'] ?? 0);
        try {
            $recDate = (new DateTime())->setTimestamp((int)floor($recMs / 1000));
        } catch (\Throwable $e) {
            $recDate = new DateTime();
        }
        if ($gap !== 0) {
            $recDate->modify($gap . ' hour');
        }

        $dbCDR = findCdrForRecord($ext, $externalPhone, $recDate);
        if ($dbCDR === null) {
            // CDR ещё не синхронизирован (или не найден) — попробуем на следующем запуске.
            $noMatch++;
            continue;
        }

        $filename = Storage::getMonitorDir() . $recDate->format('/Y/m/d/H/') . $recordId . '.mp3';

        // Уже скачано — убеждаемся, что CDR привязан, и идём дальше.
        if (file_exists($filename) && filesize($filename) > 0) {
            if ($dbCDR->recordingfile !== $filename || $dbCDR->bee_rec_status !== 'ok') {
                $dbCDR->recordingfile = $filename;
                $dbCDR->bee_rec_status = 'ok';
                $dbCDR->save();
                publishRecordingUpdate($clientBeanstalk, $dbCDR, $filename);
            }
            $already++;
            continue;
        }

        Util::mwMkdir(dirname($filename));
        $result = $api->downloadRecordById($recordId);
        if ($result['code'] !== 200 || strlen($result['body']) === 0) {
            $logger->writeError(['recordId' => $recordId, 'code' => $result['code']], 'Record download failed');
            $failed++;
            continue;
        }

        // Пишем во временный файл на той же FS, перекодируем, затем атомарно переносим.
        $tmp = $filename . '.part.' . getmypid() . '.' . uniqid('', true);
        $written = file_put_contents($tmp, $result['body']);
        if ($written !== strlen($result['body'])) {
            @unlink($tmp);
            $logger->writeError(['recordId' => $recordId], 'Short write of record file');
            $failed++;
            continue;
        }
        // Конвертация в валидный CBR 32 kbps (как в ModuleMegafonPbx). Если транскодера нет —
        // recodeMp3 вернёт false и оставит оригинал; файл всё равно сохраняем.
        AudioRecodeHelper::recodeMp3($tmp);
        if (!@rename($tmp, $filename)) {
            @unlink($tmp);
            $logger->writeError(['recordId' => $recordId], 'Failed to move record file into place');
            $failed++;
            continue;
        }

        $dbCDR->recordingfile = $filename;
        $dbCDR->bee_rec_status = 'ok';
        $dbCDR->save();
        publishRecordingUpdate($clientBeanstalk, $dbCDR, $filename);
        $ok++;
        usleep(150000);
    }
    $page++;
} while (count($records) >= 100 && $page < BEELINE_RECORDS_MAX_PAGES);

AudioRecodeHelper::setLogger(null);
$logger->writeInfo("Finished records: downloaded={$ok}, already={$already}, no_match={$noMatch}, failed={$failed}");
exit(0);

/**
 * Публикует обновление CDR с привязанным файлом записи в тьюб ядра, чтобы запись появилась
 * и в основной истории вызовов MikoPBX (ядро дедуплицирует по UNIQUEID).
 */
function publishRecordingUpdate(BeanstalkClient $client, CallHistory $dbCDR, string $filename): void
{
    $row = $dbCDR->toArray();
    $row['recordingfile'] = $filename;
    unset($row['id'], $row['bee_ext_tracking_id'], $row['bee_user_id'], $row['bee_rec_status']);
    $client->publish(json_encode(['action' => 'insert_cdr', 'rows' => [$row]]), WorkerCallEvents::class);
}