duanju_crontab_script/app/Jobs/WechatPlatform/GZHSendKFMessage.php

<?php

namespace App\Jobs\WechatPlatform;

use App\Service\Util\Support\Http\HttpRequestService;
use App\Service\Util\Support\Trace\TraceContext;
use App\Service\WechatPlatform\GZHSendKFMessageService;
use App\Service\WechatPlatform\WechatPlatform;
use EasyWeChat\OfficialAccount\Application;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\DB;


class GZHSendKFMessage implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    /**
     * @var
     * <pre>
     * [
     *  'gzhId' => $gzhId,   // wechat_authorization_infos.id
     *  'messageId' => $item->id,  // wechat_kf_messages.id
     *  'traceInfo' => $traceContext->getTraceInfo()  // traceInfo
     * ]
     * </pre>
     */
    private $info;


    /**
     * @var TraceContext
     */
    private $traceContext;
    /**
     * Create a new job instance.
     */
    public function __construct($info)
    {
        $this->info = $info;
    }

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        $this->traceContext = TraceContext::newFromParent($this->info['traceInfo']);
        myLog('KFMessageSend')->info('公众号开始发送客服消息', [
            'info' => $this->info,
            'traceInfo' => $this->traceContext->getTraceInfo(),
        ]);

        $gzh = $this->getGZH();
        if(!$gzh) return;

        $message = $this->getMessage();
        if(!$message) return;


        $messageContent = collect(\json_decode($message->message_content, true));
        $messageStr = $messageContent->pluck('text')->join("\n");

        $officialAccount = $this->getOfficialAccount($gzh);
        if(false === $officialAccount) return;

        if($this->info['isTest'] ?? false) {
            $openid = $this->info['openid'] ?? '';
            if(!$openid) {
                myLog('KFMessageSend')->error('测试回传没有openid', [
                    'info' => $this->info
                ]);
            }
            GZHSendKFMessageService::sendText($officialAccount, $openid, $messageStr, $this->traceContext);
        } else {
            $next_openid = '';
            $loop = 1;
            while (true) {
                if($loop++ > 10000) {
                    break;
                }
                if(1 == $message->u_type) {
                    $info = $this->getUserOpenids($officialAccount, $next_openid);
                    foreach (($info['data']['openid'] ?? []) as $opid){
//                        GZHSendKFMessageService::sendText($officialAccount, $opid, $messageStr, $this->traceContext);
                        dump($opid);
                    }
                    $next_openid = $info['next_openid'] ?? '';
                    if(!$next_openid) {
                        break;
                    }
                } elseif (2 == $message->u_type) {
                    $info = $this->getUsersFromUG($gzh->id, $message->ug_id,  $next_openid);
                    foreach (($info['data']['openid'] ?? []) as $opid) {
                        //                        GZHSendKFMessageService::sendText($officialAccount, $opid, $messageStr, $this->traceContext);
                        dump($opid);
                    }
                    $next_openid = $info['data']['next_uid'] ?? '';
                    if(!$next_openid) {
                        break;
                    }
                }
            }
        }
    }

    private function getMessage() {
        $message = DB::table('wechat_kf_messages')
            ->where('id', $this->info['messageId'])
            ->first();
        if(!$message) {
            myLog('KFMessageSend')->error('消息不存在', [
                'info' => $this->info,
                'traceInfo' => $this->traceContext->getTraceInfo(),
            ]);
            return false;
        }
        if(1 != $message->message_type) {
            myLog('KFMessageSend')->error('不支持的消息类型', [
                'info' => $this->info,
                'traceInfo' => $this->traceContext->getTraceInfo(),
            ]);
            return false;
        }
        return $message;
    }

    /**
     * 拉取公众号粉丝
     * @param $officialAccount Application
     */
    private function getUserOpenids($officialAccount, $next_openid) {
        $result = $officialAccount->user->list($next_openid);

        if(0 != ($result['errcode'] ?? 0)) {
            return false;
        }
        return $result;
    }

    /**
     * 通过用户分群获取
     * @param $gzhId
     * @param $ugId
     * @param $nextUid
     * @return false|mixed
     */
    private function getUsersFromUG($gzhId, $ugId,  $nextUid) {
        $url = config('wechat.ug.url.listUser');
        $signKey = config('wechat.ug.signKey');
        $now = time();
        $res = HttpRequestService::simpleGet($url, [
            'timestamp' => $now,
            'sign' => md5($signKey.$now),
            'gzhId' => $gzhId,
            'ugId' => $ugId,
            'nextUid' => $nextUid,
            'limit' => 1000
        ]);
        if($res && (10000 == ($res['code'] ?? 10000))) {
            myLog('KFMessageSend')->debug('通过用户分群获取用户列表', [
                'gzhId' => $gzhId,
                'ugId' => $ugId,
                'res' => $res
            ]);
            return $res;
        }
        return false;
    }

    /**
     * 获取公众号调用对象
     * @param $gzh
     * @return \EasyWeChat\OfficialAccount\Application
     * @throws \EasyWeChat\Kernel\Exceptions\BadResponseException
     * @throws \EasyWeChat\Kernel\Exceptions\HttpException
     * @throws \EasyWeChat\Kernel\Exceptions\InvalidArgumentException
     * @throws \Psr\SimpleCache\InvalidArgumentException
     * @throws \Symfony\Contracts\HttpClient\Exception\ClientExceptionInterface
     * @throws \Symfony\Contracts\HttpClient\Exception\DecodingExceptionInterface
     * @throws \Symfony\Contracts\HttpClient\Exception\RedirectionExceptionInterface
     * @throws \Symfony\Contracts\HttpClient\Exception\ServerExceptionInterface
     * @throws \Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface
     */
    private function getOfficialAccount($gzh) {
        try{
            return WechatPlatform::buildApplication($gzh)
                ->officialAccount($gzh->authorizer_appid, $gzh->authorizer_refresh_token);
        } catch (\Throwable $exception) {
            myLog('KFMessageSend')->error('获取公众号调用对象失败', [
                'exceptionMessage' => $exception->getMessage(),
                'traceInfo' => $this->traceContext->getTraceInfo()
            ]);
            return false;
        }
    }

    private function getGZH() {
        $gzh = DB::table('wechat_authorization_infos as a')
            ->join('wechat_open_platform_infos as b', 'a.component_appid', 'b.app_id')
            ->where([
                ['a.id', '=', $this->info['gzhId']],
                ['a.is_enabled', '=', 1],
                ['b.is_enabled', '=', 1]
            ])->select('a.id', 'a.authorizer_appid', 'a.authorizer_refresh_token',
                'b.app_id', 'b.secret', 'b.token', 'b.aes_key')
            ->first();
        if(!$gzh) {
            myLog('KFMessageSend')->error('公众号不可用', [
                'traceInfo' => $this->traceContext->getTraceInfo(),
            ]);
        }
        return $gzh;
    }
}