kuaileorder/queue/crontab/sync_push_data.php

<?php

/**
 * 同步推送数据
 */
include_once(dirname(dirname(dirname(__FILE__)))."/library/publicBase.php");

class syncPushData {
    private $table;

    protected $dbflag = 'pushdata';
    private static $db_w = null;
    private static $db_r = null;
    private static $link;

    private $ismaster = false;


    public function __construct() {
        $obj = new mOrder();
$obj->writeLog('pdd', 'sync_push_data.log', 'exec');

        $info = $obj->getSyncDataModifyTime();
$obj->writeLog('pdd', 'sync_push_data.log', $info['pdp_modified']);
        if ($info['pdp_modified'] == '0000-00-00 00:00:00') {
            $sql = "select min(pdp_modified) as pdp_modified from `pdp_tb_trade`";
            $r = $this->execute($sql, false, true);

            $start = $r['pdp_modified'];
        } else {
            $start = $info['pdp_modified'];
        }

        $end = date("Y-m-d H:i:s", strtotime($start)+7200); // 每次取2小时数据

$obj->writeLog('pdd', 'sync_push_data.log', $start."|".$end);

        $list = $this->selectAll('pdp_tb_trade', array('sql'=>'`pdp_modified` >=? and `pdp_modified` <= ?', 'vals'=>array($start, $end)), 'pdp_modified desc');

        $this->adddata($list);

        $modified = strtotime($end) <= time() ? $end : date("Y-m-d H:i:s");

        $obj->updateSyncDataModifyTime($info['id'], $modified);
    }

    private function adddata($list) {
        $obj = new mOrder();

        if (empty($list)) {
            $obj->writeLog('pdd', 'sync_push_data.log', 'empty list');
            return true;
        }

        foreach ($list as $info) {
            if (empty($info['pdp_response'])) continue;

            $order = json_decode($info['pdp_response'], true);

            /*
             array(17) {
             ["order_sn"]=>
             string(22) "220811-669893293090781"
             ["trade_type"]=>
             int(0)
             ["order_status"]=>
             int(2)
             ["group_status"]=>
             int(1)
             ["confirm_status"]=>
             int(1)
             ["refund_status"]=>
             int(1)
             ["after_sales_status"]=>
             int(0)
             ["risk_control_status"]=>
             int(0)
             ["created_time"]=>
             string(19) "2022-08-11 10:58:31"
             ["pay_time"]=>
             string(19) "2022-08-11 10:58:40"
             ["confirm_time"]=>
             string(19) "2022-08-11 10:58:40"
             ["last_ship_time"]=>
             string(19) "2022-08-13 10:58:40"
             ["updated_at"]=>
             string(19) "2022-08-11 11:01:12"
             ["pay_amount"]=>
             float(10.3)
             ["item_list"]=>
             array(1) {
             [0]=>
             array(6) {
             ["goods_id"]=>
             int(329073140054)
             ["sku_id"]=>
             int(1119777051922)
             ["goods_name"]=>
             string(87) "万方查重万方数据官网本科硕士博士毕业论文查重论文检测系统2.0"
             ["goods_price"]=>
             float(2.6)
             ["goods_spec"]=>
             string(42) "万方通用版（期刊作业报告等）"
             ["goods_count"]=>
             int(4)
             }
             }
             ["buyer_memo"]=>
             string(0) ""
             ["urge_shipping_time"]=>
             string(0) ""
             }

             */

            $obj->order_sn = $order['order_sn'];
            $obj->order_status = $order['order_status'];
            $obj->refund_status = $order['refund_status'];
            $obj->after_sales_status = $order['after_sales_status'];
            $obj->pay_amount = $order['pay_amount'];
            $obj->pay_time = $order['pay_time'];
            $obj->goods_id = $order['item_list'][0]['goods_id'];
            $obj->sku_id = $order['item_list'][0]['sku_id'];
            $obj->goods_price = $order['item_list'][0]['goods_price'];
            $obj->goods_count = $order['item_list'][0]['goods_count'];
            $obj->urge_shipping_time = $order['urge_shipping_time'];
            $obj->last_ship_time = $order['last_ship_time'];
            $obj->buyer_memo = $order['buyer_memo'];
            $obj->risk_control_status = $order['risk_control_status'];

            $sobj = new mShop();
            $shop = $sobj->getShopByOwnerid($info['mall_id']);

            $res = $obj->addOrder($shop['id'], $shop['uid']);
            $obj->writeLog('pdd', 'sync_push_data.log', $order['order_sn'].'|'.$res);
        }

        return true;
    }

    /**
     * 解析数据库配置信息
     * 获得数据库的相关连接信息
     * @param sting $dbflag 数据库标识
     * @param boolean $master 是否为主库
     */
    private function parseDbCnf($dbflag,$master=false) {
        $configs = parse_ini_file(_Storage_CNF_PATH, true);
        $config = $configs[strtolower($dbflag)];

        if($master)	{
            $host_port = $config['master'];
        } else {
            $host_port = $config['slave'];
        }
        $hps = explode(',', $host_port);
        $random = rand(0, count($hps)-1);
        $hp = $hps[$random];

        list($host, $port) = explode(':', $hp);

        $cnf = array();
        $cnf['host'] = trim($host);
        $cnf['port'] = trim($port);
        $cnf['user'] = $master==false ? trim($config['user_r']) : trim($config['user']);
        $cnf['pwd'] = $master==false ? trim($config['passwd_r']) : trim($config['passwd']);
        $cnf['db'] = trim($config['db']);
        return $cnf;
    }

    private function getInstance() {
        $dbpara = $this->ismaster==false ? 'db_r' : 'db_w';

        if(!is_null(self::$$dbpara)) {
            self::$link = self::$$dbpara;

            // 判断mysql是否gone away
            $status = self::$link->getAttribute(PDO::ATTR_SERVER_INFO);
            if($status == 'MySQL server has gone away') {
                self::$link = self::$$dbpara = $this->toDb();
            }
        } else {
            self::$link = self::$$dbpara = $this->toDb();
        }
        return self::$link;
    }

    private function initDb($table, $ismaster=false) {
        $this->table = $table;
        $this->ismaster = $ismaster;

        return self::getInstance();
    }

    private function toDb() {
        $type = strtolower($this->dbflag);
        $cnf = $this->parseDbCnf($this->dbflag, $this->ismaster);

        if (count($cnf)<=0 && $cnf['host'] == '') {
            return false;
        }
        try {
            $conn[PDO::ATTR_TIMEOUT] = 3;
            if($GLOBALS['pconnect_db']===true) $conn[PDO::ATTR_PERSISTENT] = true;
            $conn[PDO::MYSQL_ATTR_INIT_COMMAND] = "SET NAMES UTF8;";
            $db = new PDO('mysql:host='.$cnf['host'].';port='.$cnf['port'].';dbname='.$cnf['db'],$cnf['user'],$cnf['pwd'], $conn);
        } catch(PDOException $e) {
            //error_log('['.date('Y-m-d H:i:s').'][first-fail]'.implode('|', $cnf).':'.$e->getMessage()."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');

            try {
                $conn[PDO::ATTR_TIMEOUT] = 3;
                if($GLOBALS['pconnect_db']===true) $conn[PDO::ATTR_PERSISTENT] = true;
                $conn[PDO::MYSQL_ATTR_INIT_COMMAND] = "SET NAMES UTF8;";
                $db = new PDO('mysql:host='.$cnf['host'].';port='.$cnf['port'].';dbname='.$cnf['db'],$cnf['user'],$cnf['pwd'], $conn);
            } catch(PDOException $e) {
                //error_log('['.date('Y-m-d H:i:s').'][second-fail]'.implode('|', $cnf).':'.$e->getMessage()."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
                return false;
            }

            //error_log('['.date('Y-m-d H:i:s').'][second-succ]'.implode('|', $cnf)."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
            return $db;
        }

        //error_log('['.date('Y-m-d H:i:s').'][first-succ]'.implode('|', $cnf)."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
        return $db;
    }

    /**
     * 数据库Select多数据
     * 封装标准方法
     * @param string $table 表名
     * @param string $where where条件，比如：array('sql'=>'name1=? and name2=?', 'vals'=>array('name1', 'name2'))
     * @param string $orderby 排序，比如："ctime desc, name asc"
     * @param array $limit 读取数量 array(起始数, 数量)
     */
    public function selectAll($table, $where=array(), $orderby='', $limit=array()) {
        $this->initDb($table);

        try {
            $wheresql = '';
            if(!empty($where)) {
                $wheresql = "WHERE {$where['sql']}";
                $vals = $where['vals'];
            }

            $orderby = "ORDER BY {$orderby}";
            if($orderby === null) $orderby = '';

            $limitstr = empty($limit) ? '' : "LIMIT {$limit[0]}, {$limit[1]}";
            $sql = "SELECT * FROM `{$this->table}` {$wheresql} {$orderby} {$limitstr}";
            $st = self::$link->prepare($sql);
            $res = $st->execute($vals);
            if(!$res) return false;

            return $st->fetchAll(PDO::FETCH_ASSOC);

        } catch(PDOException $e) {
            return false;
        }
    }

    /**
     * 执行SQL查询语句
     *
     * 当SQL语句为查询语句时返回执行后的全部数据
     * @access public
     * @param string $sql SQL语句
     * @param boolean $all_rows 是否显示全部数据开关，当为true时，显示全部数据，为false时，显示一行数据，默认为true
     * @param boolean $is_select 是否为查询语句
     * @return array | void
     */
    public function execute($sql, $all_rows=true, $is_select=false) {
        $sql = trim($sql);
        if(!$sql) return false;

        $this->ismaster = $is_select==false ? true : false;
        self::getInstance();
        try {
            $st = self::$link->prepare($sql);
            $res = $st->execute();
            if(!$res) {
                error_log('['.date('Y-m-d H:i:s').']'.$sql.':'.json_encode($st->errorInfo())."\n", 3, LOG_PATH_BASE.'/mysql/execute_'.date('Y-m-d').'.log');
                return false;
            }

            if(!$is_select) return $res;
            if($all_rows==true) {
                return $st->fetchAll(PDO::FETCH_ASSOC);
            } else {
                return $st->fetch(PDO::FETCH_ASSOC);
            }
        } catch(PDOException $e) {
            return false;
        }
    }

}

new syncPushData();