journal/application/master/service/DbPushManager.php

<?php

namespace app\master\service;

use think\Db;
use think\Queue;
use app\master\service\push\PushResult;
use app\master\service\push\PorticoPusher;
use app\master\service\push\ScopusPusher;
use app\master\service\push\EbscoPusher;
use app\master\service\push\CnkiPusher;
use app\master\service\push\EmailDataPusher;

/**
 * 数据库收录推送编排器
 * 职责：根据分期解析应推送的渠道 -> 逐个执行(或入队) -> 写推送日志。
 */
class DbPushManager
{
    /** 推送器标识 => 实现类 */
    protected static $registry = [
        'portico'    => PorticoPusher::class,
        'scopus'     => ScopusPusher::class,
        'ebsco'      => EbscoPusher::class,
        'cnki'       => CnkiPusher::class,
        'email_data' => EmailDataPusher::class,
    ];

    /** @var DbPushLogger */
    protected $logger;

    public function __construct()
    {
        $this->logger = new DbPushLogger();
    }

    /**
     * 实例化推送器
     * @return \app\master\service\push\PusherInterface|null
     */
    public static function makePusher($code)
    {
        if (!isset(self::$registry[$code])) {
            return null;
        }
        $class = self::$registry[$code];
        return new $class();
    }

    /**
     * 解析某期刊应推送的渠道列表（配置驱动，数据表优先，回退默认配置）
     * @return array 每项 [channel_key, name, pusher, options]
     */
    public function resolveChannels($journalId)
    {
        try {
            $rows = Db::name('db_channel')->where('state', 0)->where('is_enable', 1)->order('sort desc')->select();
        } catch (\Throwable $e) {
            // 渠道表尚未创建时回退到默认配置
            $rows = [];
        }

        $channels = [];
        if (!empty($rows)) {
            foreach ($rows as $r) {
                $journalIds = trim($r['journal_ids']) === '' ? [] : array_map('intval', explode(',', $r['journal_ids']));
                $options = empty($r['config']) ? [] : (json_decode($r['config'], true) ?: []);
                $channels[] = [
                    'channel_key' => $r['channel_key'],
                    'name'        => $r['name'],
                    'pusher'      => $r['pusher'] !== '' ? $r['pusher'] : $r['channel_key'],
                    'journal_ids' => $journalIds,
                    'options'     => $options,
                ];
            }
        } else {
            foreach ((array) config('dbpush.default_channels') as $c) {
                if (empty($c['is_enable'])) {
                    continue;
                }
                $channels[] = [
                    'channel_key' => $c['channel_key'],
                    'name'        => $c['name'],
                    'pusher'      => !empty($c['pusher']) ? $c['pusher'] : $c['channel_key'],
                    'journal_ids' => isset($c['journal_ids']) ? $c['journal_ids'] : [],
                    'options'     => isset($c['config']) ? $c['config'] : [],
                ];
            }
        }

        // 按期刊过滤：journal_ids 为空表示全部
        $result = [];
        foreach ($channels as $c) {
            if (empty($c['journal_ids']) || in_array((int) $journalId, $c['journal_ids'])) {
                $result[] = $c;
            }
        }
        return $result;
    }

    /**
     * 推送整期到所有匹配渠道
     * @param int         $stageId
     * @param string      $operator 触发人
     * @param bool|null   $async    是否异步，null 用配置默认
     * @return array 各渠道执行概况
     */
    public function run($stageId, $operator = '', $async = null)
    {
        if ($async === null) {
            $async = (bool) config('dbpush.async');
        }
        $stage_info = Db::name('journal_stage')->where('journal_stage_id', $stageId)->find();
        if (!$stage_info) {
            return [];
        }
        $journalId = $stage_info['journal_id'];
        $channels = $this->resolveChannels($journalId);

        $summary = [];
        foreach ($channels as $c) {
            $logId = $this->logger->create($journalId, $stageId, $c['channel_key'], $c['name'], $operator);
            if ($async) {
                Queue::push('app\api\job\dbpush@handle', [
                    'log_id'   => $logId,
                    'stage_id' => $stageId,
                    'pusher'   => $c['pusher'],
                    'options'  => $c['options'],
                ], 'dbpush');
                $summary[] = ['channel' => $c['channel_key'], 'log_id' => $logId, 'queued' => true];
            } else {
                $ok = $this->executeAndLog($logId, $stageId, $c['pusher'], $c['options']);
                $summary[] = ['channel' => $c['channel_key'], 'log_id' => $logId, 'ok' => $ok];
            }
        }
        return $summary;
    }

    /**
     * 执行单个推送器并写入日志结果（供同步执行 / 队列任务调用）
     * @return bool 是否成功
     */
    public function executeAndLog($logId, $stageId, $pusherCode, array $options = [])
    {
        try {
            $pusher = self::makePusher($pusherCode);
            if (!$pusher) {
                $this->logger->finish($logId, false, '未知的推送器: ' . $pusherCode);
                return false;
            }
            $result = $pusher->pushStage($stageId, $options);
            if (!$result instanceof PushResult) {
                $result = PushResult::success();
            }
            $this->logger->finish($logId, $result->ok, $result->message);
            return $result->ok;
        } catch (\Throwable $e) {
            $this->logger->finish($logId, false, $e->getMessage());
            return false;
        }
    }

    /**
     * 重推某条日志
     * @return array [ok=>bool, queued=>bool]
     */
    public function retry($logId, $async = null)
    {
        if ($async === null) {
            $async = (bool) config('dbpush.async');
        }
        $log = $this->logger->find($logId);
        if (!$log) {
            return ['ok' => false, 'queued' => false, 'msg' => '日志不存在'];
        }
        // 依据日志里的渠道标识重新解析推送器与配置
        $channels = $this->resolveChannels($log['journal_id']);
        $pusher = $log['channel'];
        $options = [];
        foreach ($channels as $c) {
            if ($c['channel_key'] === $log['channel']) {
                $pusher = $c['pusher'];
                $options = $c['options'];
                break;
            }
        }
        // 重置为待推送
        Db::name('db_push_log')->where('log_id', $logId)->update(['status' => DbPushLogger::STATUS_PENDING, 'utime' => time()]);

        if ($async) {
            Queue::push('app\api\job\dbpush@handle', [
                'log_id'   => $logId,
                'stage_id' => $log['journal_stage_id'],
                'pusher'   => $pusher,
                'options'  => $options,
            ], 'dbpush');
            return ['ok' => true, 'queued' => true];
        }
        $ok = $this->executeAndLog($logId, $log['journal_stage_id'], $pusher, $options);
        return ['ok' => $ok, 'queued' => false];
    }
}