xyzshop/crontab/control/minutes.php

<?php
/**
 * 任务计划 - 分钟执行的任务
 *
 *
 *
 *
 */

defined('InShopNC') or exit('Access Invalid!');

require_once (BASE_ROOT_PATH . '/helper/message/publisher.php');
require_once(BASE_HELPER_PATH . '/refill/util.php');
require_once(BASE_HELPER_PATH . '/refill/divert_account.php');
require_once(BASE_HELPER_PATH . '/refill/RefillFactory.php');
require_once(BASE_CORE_PATH . '/framework/function/http.php');
require_once(BASE_HELPER_PATH . '/PHPExcel/PHPExcel.php');
require_once(BASE_HELPER_PATH . '/task/task_helper.php');


class minutesControl extends BaseCronControl
{
    public function indexexOp()
    {
        $this->_cron_common([1,2,3,4,5,6,7]);
    }

    public function indexOp()
    {
        Log::record(__FUNCTION__ . " start",Log::DEBUG);
        //未付款订单超期自动关闭
//        $this->_order_timeout_cancel();
        $this->_cron_common([1,2,3,4,5,6,7]);
        $this->_fetch_order_unavaliable();
        $this->_check_merchant_alarm_amount();

        Log::record(__FUNCTION__ . " end",Log::DEBUG);

//        $this->_web_index_update();
//        $this->_check_merchant_alarm_amount();
//        $this->check_refill_order_limit();
//        $this->_cron_mail_send();
//        $this->_SendOrderQuery();
    }

    //任务队列处理
    public function admin_taskOp()
    {
        Log::short_name('task');
        Log::record('start task',Log::DEBUG);
        $this->exec_task(false);
    }

    public function update_channel_balanceOp()
    {
        $balance_getter = function ($provider_id)
        {
            $provider_model = Model('refill_provider');
            $provider_info = $provider_model->getProviderInfo(['provider_id' => $provider_id, 'opened' => 1]);
            if (empty($provider_info)) {
                return false;
            }
            $channel_name = $provider_info['name'];
            if ($provider_info['type'] == 1) {
                $type = 'RefillOil';
            } elseif ($provider_info['type'] == 2 || $provider_info['type'] == 3) {
                $type = 'RefillPhone';
            } else {
                $type = 'RefillPhone';
            }

            Log::record("query balance channel_name:{$channel_name}",Log::DEBUG);
            $provider = refill\util::getProvider($channel_name,$type);

            if($provider == false) return false;
            [$state, $balance] = $provider->balance();
            Log::record("query balance channel_name:{$channel_name}, err_msg:{$balance}",Log::DEBUG);

            if(!$state) {
                return false;
            } else {
                return round(floatval($balance),4);
            }
        };

        $providers = function ()
        {
            $pids = [];

            $i = 0;
            while (true)
            {
                $start = $i * 1000;
                $items = Model()->table('refill_provider')->where(['opened' => 1])->field('*')->order('provider_id desc')->limit("{$start},1000")->select();
                if(empty($items)) {
                    return $pids;
                }
                $i++;

                foreach ($items as $item) {
                    $pid = intval($item['provider_id']);
                    $pids[] = $pid;
                }
            }
        };

        $mod_provider = Model('refill_provider');
        $pids = $providers();
        foreach ($pids as $pid)
        {
            try {
                $balance = $balance_getter($pid);
                if($balance !== false) {
                    $mod_provider->editProvider(['balance' => $balance],['provider_id' => $pid]);
                }
            }
            catch (Exception $ex)
            {
                Log::record($ex->getMessage(),Log::ERR);
            }
        }
    }

    private function _fetch_order_unavaliable()
    {
        try {
            $mod_fetch_order = Model('fetch_order');
            $trans = new trans_wapper($mod_fetch_order,__METHOD__);
            $items = $mod_fetch_order->timeout_orders(900);
            foreach ($items as $item) {
                $fetch_id = $item['fetch_id'];
                $mod_fetch_order->unavaliable($fetch_id);
            }
            $trans->commit();

            foreach ($items as $item) {
                $order_sn = $item['order_sn'];
                $chname = $item['channel_name'];
                refill\util::push_notify($chname,['order_sn' => $order_sn,'state' => 'CANCEL']);
            }
        }
        catch (Exception $ex) {
            $trans->rollback();
            Log::record(__METHOD__ . " " . $ex->getMessage(),Log::ERR);
        }
    }

    private function _check_merchant_alarm_amount()
    {
        $mch_cache = rcache("merchant-notify", 'refill-');
        $caches = empty($mch_cache['data']) ? [] : unserialize($mch_cache['data']);

        $new_caches = [];
        $merchants = Model('merchant')->getMerchantList(['merchant.merchant_state' => 1], '', '', true, 'merchant.*,member.available_predeposit', "0,1000");
        foreach ($merchants as $merchant) {
            $mchid = $merchant['mchid'];
            $phones = empty($merchant['warning_phone']) ? [] : unserialize($merchant['warning_phone']);
            $available_pd = intval($merchant['available_predeposit']);
            $alarm_pd = intval($merchant['alarm_amount']);

            if (array_key_exists($mchid, $caches)) {
                $mch_cache = $caches[$mchid];
            } else {
                $mch_cache = ['last_time' => 0, 'send_count' => 0];
            }
            if ($available_pd < $alarm_pd || $available_pd < 10000) {
                $counts = $mch_cache['send_count'];
                if (($mch_cache['last_time'] + 300 < time()) && $counts < 2) {
                    $mch_cache = ['last_time' => time(), 'send_count' => $counts + 1];
                    foreach ($phones as $phone) {
                        if (!empty($phone)) {
                            QueueClient::push('sendSMS', ['mobile' => $phone,
                                'type' => 'balance_warning', 'datas' => [date("m月d日H时"), $merchant['available_predeposit']]]);
                        }
                    }
                }
            } else {
                $mch_cache = ['last_time' => 0, 'send_count' => 0];
            }
            $new_caches[$mchid] = $mch_cache;
        }

        wcache("merchant-notify", ['data' => serialize($new_caches)], 'refill-');
    }

    public function check_refill_order_limit()
    {
        $mch_cache = rcache("storge_limit" , 'merchant-');
        $caches = empty($mch_cache['data']) ? [] : unserialize($mch_cache['data']);

        $reader = function ($mchid,$time)
        {
            $cond['mchid'] = $mchid;
            $cond['inner_status'] = 0;
            $cond['order_state'] = ORDER_STATE_SUCCESS;
            $cond['refill_order.order_time'] = ['egt', $time];

            $items = Model('')->table('refill_order,vr_order')->join('inner')
                ->on('refill_order.order_id=vr_order.order_id')
                ->field('refill_order.mchid,refill_order.card_type,refill_order.refill_amount,count(*) as num')
                ->group('refill_order.card_type,refill_order.refill_amount')
                ->where($cond)->select();

            return $items;
        };


        $merger = function ($limits,$reals,$mchid,$card_type)
        {
            $result = [];
            foreach ($limits as $amount => $limit)
            {
                $limit = intval($limit);

                if($limit === -1) {
                    $allow = true;
                }
                elseif($limit === 0) {
                    $allow = false;
                }
                else
                {
                    $finded = false;
                    foreach ($reals as $item)
                    {
                        $tmp = intval($item['refill_amount'] + 0.005);
                        if($item['card_type'] == $card_type && $tmp == $amount) {
                            $num = $item['num'];
                            $allow = $limit > $num;
                            $finded = true;
                            break;
                        }
                    }

                    if(!$finded) {
                        $allow = true;
                    }
                }

                $result["{$mchid}-{$card_type}-{$amount}"] = $allow;
            }

            return $result;
        };

        $type_map = ['petrochina' => 1,'sinopec' => 2];
        $abilitys = [];
        foreach ($caches as $cache)
        {
            $mchid = intval($cache['mchid']);
            $start = $cache['time'];

            $petros = $cache['petrochina'];
            $sinos  = $cache['sinopec'];
            $items = $reader($mchid,$start);

            $prets = $merger($petros,$items,$mchid,$type_map['petrochina']);
            $srets = $merger($sinos,$items,$mchid,$type_map['sinopec']);
            foreach ($prets as $key => $val) {
                $abilitys[$key] = $val;
            }
            foreach ($srets as $key => $val) {
                $abilitys[$key] = $val;
            }
        }

        $old = rcache("refill_able",'merchant-');
        $old = empty($old) ? "" : $old['data'];

        ksort($abilitys);
        $new = serialize($abilitys);
        if($new != $old) {
            wcache("refill_able" , ['data' => $new] , 'merchant-');
//            $publisher = new message\publisher();
//            $publisher->modify_refill_merchant();
        }
    }

    /**
     * 未付款订单超期自动关闭
     */
    private function _order_timeout_cancel()
    {
        Log::record(__FUNCTION__,Log::DEBUG);
        //实物订单超期未支付系统自动关闭
        $_break = false;
        $model_order = Model('order');
        $logic_order = Logic('order');
        $condition = [];
        $condition['order_state'] = ORDER_STATE_NEW;
        $condition['add_time'] = ['lt',time() - ORDER_AUTO_CANCEL_DAY * 86400];

        //分批，每批处理100个订单，最多处理5W个订单
        for ($i = 0; $i < 500; $i++)
        {
            if ($_break) {
                break;
            }
            $order_list = $model_order->getOrderList($condition, '', '*', '', 100);
            if (empty($order_list)) break;
            foreach ($order_list as $order_info)
            {
                Log::record("1",Log::DEBUG);
                $result = $logic_order->changeOrderStateCancel($order_info,'system','系统','超期未支付系统自动关闭订单',true,false);
                Log::record("2",Log::DEBUG);
                if (!$result['state']) {
                    $this->log('实物订单超期未支付关闭失败SN:'.$order_info['order_sn']); $_break = true; break;
                } else {
                    Log::record("3",Log::DEBUG);
                    account_helper::onPredeposit('order_cancel',$order_info['buyer_id'],$order_info['order_sn']);
                    Log::record("4",Log::DEBUG);

                }
            }
        }

        //虚拟订单超期未支付系统自动关闭
        $_break = false;
        $model_vr_order = Model('vr_order');
        $logic_vr_order = Logic('vr_order');
        $condition = [];
        $condition['order_state'] = ORDER_STATE_NEW;
        $condition['add_time'] = ['lt',time() - VRORDER_AUTO_CANCEL_MINUTE * 60];

        //分批，每批处理100个订单，最多处理5W个订单
        for ($i = 0; $i < 500; $i++)
        {
            if ($_break) {
                break;
            }
            $order_list = $model_vr_order->getOrderList($condition, '', '*', '',100);
            if (empty($order_list)) break;
            foreach ($order_list as $order_info) {
                $result = $logic_vr_order->changeOrderStateCancel($order_info,'system','超期未支付系统自动关闭订单',false);
            }

            if (!$result['state']) {
                $this->log('虚拟订单超期未支付关闭失败SN:'.$order_info['order_sn']);
                $_break = true;
            }
        }
    }

    /**
     * 更新首页的商品价格信息
     */
    private function _web_index_update()
    {
         Model('web_config')->updateWebGoods();
    }

    /**
     * 发送邮件消息
     */
    private function _cron_mail_send()
    {
        //每次发送数量
        $_num = 50;
        $model_storemsgcron = Model('mail_cron');
        $cron_array = $model_storemsgcron->getMailCronList([], $_num);
        if (!empty($cron_array))
        {
            $email = new Email();
            $mail_array = [];
            foreach ($cron_array as $val)
            {
                $return = $email->send_sys_email($val['mail'],$val['subject'],$val['contnet']);
                if ($return) {
                    // 记录需要删除的id
                    $mail_array[] = $val['mail_id'];
                }
            }
            // 删除已发送的记录
            $model_storemsgcron->delMailCron(['mail_id' => ['in', $mail_array]]);
        }
    }

    public function second_queueOp()
    {
        while (true)
        {
            do{
                try {
                    $ret = $this->_cron_common([8]);
                    perfor_clear();
                } catch (Exception $ex) {
                    Log::record($ex->getMessage(),Log::ERR);
                }
            }
            while($ret === true);
            sleep(1);
        }
    }

    public function account_editOp()
    {
        $stamp = 0;
        while (true)
        {
            try {
                $divertor = new refill\divert_account();
                $next = time();
                $divertor->update_account($stamp);
                $stamp = $next;
            } catch (Exception $ex) {
                Log::record($ex->getMessage(),Log::ERR);
            }

            sleep(1);
        }
    }

    private function exec_task($merchant)
    {
        $task_manager = new task\manager();
        $task_id = 0;
        while (true)
        {
            try
            {
                $task = $task_manager->pop_task($task_id,$merchant);
                if (!empty($task)) {
                    $task_manager->handle($task);
                    $task_id = $task['task_id'];
                }
            }
            catch (Exception $ex) {
                Log::record($ex->getMessage(),Log::ERR);
            }

            sleep(1);
        }
    }
    //任务队列处理
    public function merchant_taskOp()
    {
        Log::short_name('merchant_task');
        Log::record('start task',Log::DEBUG);
        $this->exec_task(true);
    }

    //更新统计业务数据
    public function stat_utilOp()
    {
        Log::short_name('stat_util');
        while (true)
        {
            try {
                $this->_update_earlist_send();
                $this->_calc_arrears();

            } catch (Exception $ex) {
                Log::record($ex->getMessage(), Log::ERR);
            }

            sleep(1);
        }
    }

    //更新统计业务数据
    public function statOp()
    {
        Log::short_name('stat');
        while (true)
        {
            try {
                $this->_update_earlist_send();
                $this->_calc_arrears();

            } catch (Exception $ex) {
                Log::record($ex->getMessage(), Log::ERR);
            }

            sleep(1);
        }
    }

    private function _update_earlist_send()
    {
        //查找最早的充值中的订单时间
//        $update_earliest_ordertime = function () {
//            $mod = Model('refill_detail');
//            $time = $mod->getEarliestSendTime();
//            wcache('stat-info', ['earliest_order_time' => $time], 'refill-');
//        };
//        $update_earliest_ordertime();

        //按各个机构，更新最早的充值中的订单时间
        $update_earliest_ordertime_bymerchant = function () {
            $mod = Model('refill_detail');
            $mtimes = $mod->getEarliestSendTimeByMerchant();
            $val = json_encode($mtimes);

            wkcachex('stat-earliest-ordertime', $val, 'refill-');
        };
        $update_earliest_ordertime_bymerchant();
    }

    private function _calc_arrears()
    {
        $send_amount_reader = function () {
            $val = rkcachex('stat-merchant-sendamount', 'refill-');
            $val = json_decode($val, true);
            return $val;
        };

        $del_stop = function ($amounts)
        {
            $ins = Cache::getInstance('cacheredis');
            $items = $ins->hkeys('merchant-debts-detail','refill-');
            if(empty($items)) {
                return;
            }

            $keys = [];
            foreach ($items as $key) {
                $keys[] = intval($key);
            }

            foreach ($keys as $mchid)
            {
                if(array_key_exists($mchid,$amounts)) {
                    continue;
                } else {
                    $ins->hdel('merchant-debts-detail','refill-',$mchid);
                }
            }
        };

        $debts_detail_fun = function ($amounts)
        {
            $debts_detail = [];

            $mod = Model();
            $i = 0;
            while (true)
            {
                $start = $i * 1000;

                $items = $mod->table('merchant,member')->join('inner')->on('member.member_id=merchant.admin_id')
                    ->field('mchid,credit_bonus,co_id,available_predeposit')->where(['merchant_state' => 1, 'co_id' => ['gt', 0]])->order('mchid asc')->limit("{$start},1000")->select();
                if(empty($items)) {
                    break;
                }
                $i++;

                foreach ($items as $item)
                {
                    $mchid = intval($item['mchid']);
                    $co_id = intval($item['co_id']);
                    $credit_bonus = round(floatval($item['credit_bonus']), 2);
                    $available_predeposit = round(floatval($item['available_predeposit']), 2);

                    if (array_key_exists($mchid, $amounts)) {
                        $send_amounts = $amounts[$mchid]['send_amounts'];
                        $lack_amounts = $amounts[$mchid]['lack_amounts'];
                    } else {
                        $send_amounts = 0;
                        $lack_amounts = 0;
                    }
                    Log::record("co_id={$co_id} mchid={$mchid} send_amounts={$send_amounts} lack_amounts={$lack_amounts}",Log::DEBUG);

                    $available = $available_predeposit - $credit_bonus + $send_amounts;// - $lack_amounts;
                    $debt = -$available;
                    $debts_detail[$co_id][$mchid] = $debt;

                    $val = ['debt' => $debt, 'send_amounts' => $send_amounts, 'lack_amounts' => $lack_amounts];
                    wcache('merchant-debts-detail', [$mchid => serialize($val)], 'refill-');
                }
            }


            return $debts_detail;
        };

        $company_debt_fun = function ()
        {
            $co_debts = [];

            $mod = Model('refill_company');
            $i = 0;
            while (true)
            {
                $start = $i * 1000;
                $items = $mod->field('co_id,max_debt')->where(['opened' => 1, 'co_type' => 'merchant'])->order('co_id asc')->limit("{$start},1000")->select();
                if(empty($items)) {
                    break;
                }
                $i++;

                foreach ($items as $item)
                {
                    $co_id = intval($item['co_id']);
                    $max_debt = round(floatval($item['max_debt']), 2);
                    $co_debts[$co_id] = $max_debt;
                }
            }

            return $co_debts;
        };

        $calc_debts = function ($co_id,$detail)
        {
            if(!array_key_exists($co_id,$detail)) {
                return [[], 0];
            }

            $mchids=[];
            $sum_debts = 0;

            $co_details = $detail[$co_id];
            foreach ($co_details as $mchid => $debts) {
                $mchids[] = $mchid;
                $sum_debts += $debts;
            }

            return [$mchids,$sum_debts];
        };

        $judge_stop = function ($co_debts, $debts_detail) use ($calc_debts)
        {
            foreach ($co_debts as $co_id => $max_debts)
            {
                [$mchids, $sum_debts] = $calc_debts($co_id,$debts_detail);
                wcache('company-debts', [$co_id => $sum_debts], 'refill-');

                if(empty($mchids)) continue;

                if($max_debts < $sum_debts) {
                    $fStop = 1;
                } else {
                    $fStop = 0;
                }

                foreach ($mchids as $mchid) {
                    wcache('merchant-debt-judge', [$mchid => $fStop], 'refill-');
                }
            }
        };

        $need_calc = function ($clac_time)
        {
            static $latest_time = 0;
            if($clac_time > $latest_time) {
                $latest_time = $clac_time;
                return true;
            } else {
                return false;
            }
        };

        $result = $send_amount_reader();
        $del_stop($result);

        if (!empty($result) && $need_calc(intval($result['time']))) {
            $mch_amounts = $result['send_amounts'];
            $debts_detail = $debts_detail_fun($mch_amounts);
            $co_debts = $company_debt_fun();
            $judge_stop($co_debts, $debts_detail);
        }

    }

    /**
     * 执行通用任务
     */
    private function _cron_common($types)
    {
        Log::record(__FUNCTION__,Log::DEBUG);
        //查找待执行任务
        $model_cron = Model('cron');
        $cron = $model_cron->getCronList(['exetime'=> ['elt',time()],'type' => ['in',$types]]);
        if (!is_array($cron)) return false;

        $count = count($cron);
        if ($count <= 0) return false;
        Log::record("match cron count={$count}",Log::DEBUG);

        $cron_array = [];
        $cronid = [];

        $exeid = 1;
        foreach ($cron as $v)
        {
            $type = intval($v['type']);
            if($type == 8) {
                $cron_array[$v['type']][$exeid] = $v;
                $exeid++;
            }
            else {
                $cron_array[$v['type']][$v['exeid']] = $v;
            }
        }

        foreach ($cron_array as $k=>$v)
        {
            // 如果方法不存是，直接删除id
            if (!method_exists($this,'_cron_'.$k)) {
                $tmp = current($v);
                $cronid[] = $tmp['id'];
                continue;
            }

            $method = '_cron_'.$k;
            Log::record("crontab minutest:{$method}",Log::DEBUG);

            $result = call_user_func_array([$this, '_cron_' . $k], [$v]);
            if (is_array($result)) {
                $cronid = array_merge($cronid, $result);
            } else {
                $method = '_cron_' . $k;
                Log::record("crontab minutest err:{$method}", Log::ERR);
            }
        }

        //删除执行完成的cron信息
        if (!empty($cronid) && is_array($cronid)) {
            $model_cron->delCron(['id' => ['in', $cronid]]);
        }

        return true;
    }

    //'任务类型 1商品上架 2根据商品id更新商品促销价格 3优惠套装过期 4推荐展位过期 5团购开始更新商品促销价格 6团购过期 7限时折扣过期',

    //1商品上架
    private function _cron_1($cron = array())
    {
        $condition = ['goods_commonid' => ['in',array_keys($cron)]];
        $update = Model('goods')->editProducesOnline($condition);
        if ($update)
        {
            //返回执行成功的cronid
            $cronid = [];
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        }
        else {
            return false;
        }
        return $cronid;
    }

    //2根据商品id更新商品促销价格
    private function _cron_2($cron = array())
    {
        $condition = ['goods_id' => ['in',array_keys($cron)]];
        $update = Model('goods')->editGoodsPromotionPrice($condition);
        if ($update){
            //返回执行成功的cronid
            $cronid = [];
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        }else{
            return false;
        }
        return $cronid;
    }

    //3优惠套装过期
    private function _cron_3($cron = [])
    {
        $condition = ['store_id' => ['in', array_keys($cron)]];
        $update = Model('p_bundling')->editBundlingQuotaClose($condition);
        if ($update) {
            //返回执行成功的cronid
            $cronid = [];
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        } else {
            return false;
        }
        return $cronid;
    }

    //4推荐展位过期
    private function _cron_4($cron = [])
    {
        $condition = array('store_id' => array('in', array_keys($cron)));
        $update = Model('p_booth')->editBoothClose($condition);
        if ($update) {
            //返回执行成功的cronid
            $cronid = array();
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        } else {
            return false;
        }
        return $cronid;
    }

    //5团购开始更新商品促销价格
    private function _cron_5($cron = array())
    {
        $condition = [];
        $condition['goods_commonid'] = ['in', array_keys($cron)];
        $condition['start_time'] = ['lt', time()];
        $condition['end_time'] = ['gt', time()];
        $groupbuy = Model('groupbuy')->getGroupbuyList($condition);
        foreach ($groupbuy as $val) {
            Model('goods')->editGoods(['goods_promotion_price' => $val['groupbuy_price'], 'goods_promotion_type' => 1],
                ['goods_commonid' => $val['goods_commonid']]);
        }
        //返回执行成功的cronid
        $cronid = array();
        foreach ($cron as $v) {
            $cronid[] = $v['id'];
        }
        return $cronid;
    }

    /**
     * 抢购过期
     *
     * @param array $cron
     */
    private function _cron_6($cron = array())
    {
        $condition = ['goods_commonid' => ['in', array_keys($cron)]];
        //抢购活动过期
        $update = Model('groupbuy')->editExpireGroupbuy($condition);
        if ($update){
            //返回执行成功的cronid
            $cronid = [];
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        }else{
            return false;
        }
        return $cronid;
    }

    /**
     * 限时折扣过期
     *
     * @param array $cron
     */
    private function _cron_7($cron = array())
    {
        $condition = array('xianshi_id' => array('in', array_keys($cron)));
        //限时折扣过期
        $update = Model('p_xianshi')->editExpireXianshi($condition);
        if ($update){
            //返回执行成功的cronid
            $cronid = array();
            foreach ($cron as $v) {
                $cronid[] = $v['id'];
            }
        }
        else{
            return false;
        }
        return $cronid;
    }

    private function _cron_8($cron = array())
    {
        $cronid = [];
        foreach ($cron as $v)
        {
            $cronid[] = intval($v['id']);
            $params = $v['params'];
            if(!empty($params))
            {
                $params = unserialize($params);
                if(is_array($params))
                {
                    foreach ($params as $key => $value) {
                        Log::record("delay queue: {$key}",Log::DEBUG);
                        QueueClient::push($key,$value);
                    }
                }
            }
        }

        return $cronid;
    }

    //查询超过5分钟的充值中订单
    public function _SendOrderQuery()
    {
        $model_refill_order = Model('refill_order');
        $condition['order_state'] = ORDER_STATE_SEND;
        $condition['refill_order.order_time'] = ['lt', (time() - 300)];
        $orders = $model_refill_order->getMerchantOrderList($condition, 1000, 0,'refill_order.order_id');
        if(!empty($orders)) {
            foreach ($orders as $order) {
                $order_id = $order['order_id'];
                QueueClient::push("QueryRefillState",['order_id' => $order_id]);
            }
        }
    }

    //

    public function _refill_task(): bool
    {
        Log::record(__FUNCTION__,Log::DEBUG);
        $model_refill_task = Model('refill_task');

        $task = $model_refill_task->UnDispose('order_export');
        if (empty($task)) return false;

        $task_id = $task['task_id'];
        $task_condition = unserialize($task['task_condition']);
        if (empty($task_condition)) {
            Log::record("refill task condition unserialize err, task_id：{$task_id}");
            return false;
        }

        $type = $task['type'];
        $cid = $task['cid'];
        if ($type === 'merchant') {
            $condition['refill_order.mchid'] = $cid;
        } elseif ($type === 'provider') {
            $condition['vr_order.store_id'] = $cid;
        } else {
            Log::record("refill task type err, task_id：{$task_id}");
            return false;
        }
        $order_time_type = $task_condition['order_time_type'];
        $start_unixtime = $task_condition['start_unixtime'];
        $end_unixtime = $task_condition['end_unixtime'];
        $order_state = $task_condition['order_state'];
        if ($start_unixtime > 0 && $end_unixtime > $start_unixtime) {
            $condition["refill_order.{$order_time_type}"] = [['egt', $start_unixtime], ['lt', $end_unixtime], 'and'];
            $condition['vr_order.add_time'] = ['egt', $start_unixtime-1];
        } elseif ($start_unixtime > 0) {
            $condition["refill_order.{$order_time_type}"] = ['egt', $start_unixtime];
            $condition['vr_order.add_time'] = ['egt', $start_unixtime-1];
        }
        if (in_array($order_state, ['0', '10', '20', '30', '40'], true)) {
            $condition['order_state'] = $order_state;
        }

        $model_refill_task->Dispose($task_id);
        [$state, $task_result] = $this->order_export($condition, $type, $task_id);
        if ($state) {
            $model_refill_task->DisposeFinish($task_id, $task_result);
        } else {
            $model_refill_task->DisposeErr($task_id, $task_result);
        }

        return true;
    }

    /**
     * @throws PHPExcel_Exception
     */
    private function order_export($condition, $type, $task_id)
    {
        $orders = Model('refill_order')->getAllOrders($condition);
        if(empty($orders)) {
            return [false, '统计数据为空'];
        }
        $orders = $this->orderFormat($orders);
        $objPHPExcel = new PHPExcel();

        if ($type === 'merchant') {
            $column_keys = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K'];
            $column_values = ['商户号', '客户订单号', '平台单号', '面额', '充值卡号', '充值卡类型', '下单日期', '完成日期', '官方流水号', '订单状态', '扣款金额'];
            $data_keys = ['mchid', 'mch_order', 'order_sn', 'refill_amount', 'card_no', 'card_type_text', 'order_time_text', 'notify_time_text', 'official_sn', 'order_state_text', 'mch_amount'];
        } elseif ($type === 'provider') {
            $column_keys = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J'];
            $column_values = ['上游订单号', '平台单号', '面额', '充值卡号', '充值卡类型', '下单日期', '完成日期', '官方流水号', '订单状态', '扣款金额'];
            $data_keys = ['ch_trade_no', 'order_sn', 'refill_amount', 'card_no', 'card_type_text', 'order_time_text', 'notify_time_text', 'official_sn', 'order_state_text', 'channel_amount'];
        } else {
            return [false, '主体类型错误'];
        }

        $objPHPExcel->setActiveSheetIndex(0);
        $objPHPExcel->getDefaultStyle()->getFont()->setName('Arial')->setSize(10);
        foreach ($column_keys as $key => $column_key) {
            $objPHPExcel->getActiveSheet()->getColumnDimension($column_key)->setWidth(25);
            $cell_value = $column_key . 1;
            $objPHPExcel->getActiveSheet()->setCellValue($cell_value, $column_values[$key]);
        }

        foreach ($orders as $k => $order) {
            foreach ($column_keys as $key => $column_key) {
                $field = $column_key.($k+2);
                $objPHPExcel->getActiveSheet()->setCellValueExplicit($field, $order[$data_keys[$key]],
                    PHPExcel_Cell_DataType::TYPE_STRING);
            }
        }
        try {
            $path = BASE_ROOT_PATH . "/data/upload/task/";
            if(!is_dir($path)){
                mkdir($path, 0755);
            }
            $filename = date('YmdHis',time())."-任务导出-任务ID:{$task_id}.xlsx";

            $file_path = $path . $filename;
            $objWriter = PHPExcel_IOFactory::createWriter($objPHPExcel, 'Excel2007');

            $objWriter->save($file_path);
            return [true, $filename];
        } catch (Exception $e) {
            return [false, $e->getMessage()];
        }
    }

    private function orderFormat($orders): array
    {
        $card_type_texts = [mtopcard\PetroChinaCard => '中石油', mtopcard\SinopecCard => '中石化', mtopcard\ChinaMobileCard => '中国移动', mtopcard\ChinaUnicomCard => '中国联通', mtopcard\ChinaTelecomCard => '中国电信'];
        $orderState = function ($order_info) {
            $is_retrying = $order_info['is_retrying'];
            switch ($order_info['order_state']) {
                case ORDER_STATE_CANCEL:
                    if ($is_retrying) {
                        $order_state = '重试中';
                    } else {
                        $order_state = '已取消';
                    }
                    break;
                case ORDER_STATE_NEW:
                    $order_state = '待付款';
                    break;
                case ORDER_STATE_PAY:
                    $order_state = '待发货';
                    break;
                case ORDER_STATE_SEND:
                    $order_state = '待收货';
                    break;
                case ORDER_STATE_SUCCESS:
                    $order_state = '交易完成';
                    break;
                default :
                    $order_state = '未知状态';
            }
            return $order_state;
        };
        foreach($orders as $key => $order)
        {
            $orders[$key]['card_type_text'] = $card_type_texts[$order['card_type']];
            $orders[$key]['order_time_text'] = $order['order_time'] ? date('Y-m-d H:i:s', $order['order_time']) : '';
            $orders[$key]['notify_time_text'] = $order['notify_time'] ? date('Y-m-d H:i:s', $order['notify_time']) : '';
            $orders[$key]['order_state_text'] = $orderState($order);
        }
        return $orders;
    }
}