cpa-api/app/queue/redis/fast/SyncAccount.php

<?php

namespace app\queue\redis\fast;

use app\extra\ocean\Account;
use app\model\cpa\CpaAccount;
use Webman\RedisQueue\Consumer;

class SyncAccount implements Consumer
{

    /**
     * 消费队列名
     * 核销队列
     * @var string
     */
    public string $queue = "sync-account";

    /**
     * 连接配置
     * @var string
     */
    public string $connection = "default";

    public function consume($data): bool
    {
        try {
            if (empty($data['account_id'])) return true;
            echo("开始同步====>" . getDateFull());
            $nextPage = 1;
            while (is_numeric($nextPage)) {
                $resp = (new Account)->config($this->getConfig())->token()->getAccountList($data['account_id'],"AD",$nextPage);
                if (empty($resp['account_list'])) {
                    echo getDateFull()."==没有数据了==\n";
                    $nextPage = null;
                    return true;
                }
                foreach ($resp['account_list'] as $item) {
                    $account = (new CpaAccount)->where("advertiser_id",$item['account_id'])->findOrEmpty();
                    if ($account->isEmpty()) {
                        $account->strict(false)->insertGetId([
                            "advertiser_id" => $item['account_id'],
                            "group_id"      => $data['account_id'],
                            "advertiser_name"   => $item['account_name']
                        ]);
                        echo getDateFull()."==新账户入库=={$item['account_id']}====\n";
                    } else {
                        $account->save([
                            "advertiser_id" => $item['account_id'],
                            "advertiser_name"   => $item['account_name']
                        ]);
                        echo getDateFull()."==更新账户=={$item['account_id']}====\n";
                    }
                }
                $nextPage = $nextPage+1;
                echo("进入下一页=={$nextPage}==".getDateFull());
            }
        } catch (\Throwable $throwable) {
            echo getDateFull()."===同步账户列表报错\n";
            echo $throwable->getFile()."\n";
            echo $throwable->getLine()."\n";
            echo $throwable->getMessage()."\n";
            return true;
        }
        return true;
    }



    /**
     * 配置信息
     * @param string $type
     * @return array
     */
    protected function getConfig(string $type = ""): array
    {
        return [
            "appid"     => "1874672050081403",
            "secret"    => "584e48eab19f6ec69118686b40998e2082eb77ee",
            "prefix"    => "ocean",
        ];
    }

}