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');

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 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}");
            $provider = refill\util::getProvider($channel_name,$type);
            [$state, $balance] = $provider->balance();
            Log::record("query balance channel_name:{$channel_name}, err_msg:{$balance}");

            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')->field('*')->order('provider_id asc')->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)
        {
            $balance = $balance_getter($pid);
            if($balance !== false) {
                $mod_provider->editProvider(['balance' => $balance],['provider_id' => $pid]);
            }
        }
    }

    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()
    {
        if(defined('COMPANY_NAME') && COMPANY_NAME === 'XYZ_COMPANY')
        {
            $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 _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, 'refill_order.order_id');
        if(!empty($orders)) {
            foreach ($orders as $order) {
                $order_id = $order['order_id'];
                QueueClient::push("QueryRefillState",['order_id' => $order_id]);
            }
        }
    }
}