new_sdk_wu/www/new_sdk/application/service/RetaineService.php

<?php

namespace app\service;

use app\admin\controller\RetaineGame;
use think\Cache;
use think\Db;
use think\Env;
use think\Exception;
use think\Log;

/**
 * QuickSDK游戏同步加解密算法描述
 * @copyright quicksdk 2015
 * @author quicksdk
 * @version quicksdk v 0.0.1 2014/9/2
 */
class RetaineService
{
    /**
     * 需要统计的游戏
     * @return array|false|string
     */
    public function getGameIds()
    {
        return model('common/Game')->cache('retaine:game', 600)->where('cooperation_status', 1)->column('id');
    }

    /**
     * 当天注册账号
     * @param $day
     * @return array|false|string
     */
    public function getMembers($day)
    {
        return model('common/Members')->cache('retaine:members:' . strtotime($day), 600)->whereBetween('reg_time', [strtotime($day), strtotime($day . ' 23:59:59')])->column('id');
    }
    
    /**
     * 注册玩家总数
     *
     * @param $where
     * @return array|false|string
     */
    public function getMembersTotal($game_id, $channel_id)
    {
        $where = [];
        if($game_id){
            $where['gameid'] = $game_id;
        }
        if($channel_id){
            $where['channel_id'] = $channel_id;
        }
        return model('common/Members')
            ->cache('retaine:roleInfo_total:'.$game_id.'_'.$channel_id, 500)
            ->where($where)->fetchSql(false)->count();
    }

    /**
     * 新增玩家
     *
     * @param $day
     * @param array $mebmers 账户ID列表
     * @param int $server 是否区服数据
     * @param int $channel 是否渠道数据
     * @param int $platform 是否平台数据
     * @param string $type count=统计数据, sum=汇总数据
     *
     * @return array|bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function regNum($day, $mebmers, $server = 0, $channel = 0, $platform = 0, $type='reg_num')
    {
        // if (!$mebmers || $type == 'reg_total') {
            $sql = Db::table('cy_role_info')->whereIn('gameid', $this->getGameIds());
                // ->where('serverid', '<>', '')
                // ->where('channel_id', '<>', 0);
            if($type == 'reg_total'){
                $sql = $sql->where('create_time', '<', strtotime($day . ' 23:59:59'));
            }else{
                // ->where('userid', 'in', $mebmers)
                $sql = $sql->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
            }
            // $sql->fetchSql(true);

            if ($platform) {
                $field = ' 0 as game_id,count(DISTINCT userid) as '.$type.', ';
                if ($channel) {
                    $list = $sql->field($field . 'channel_id,0 as server_id')
                        ->group('channel_id')->select();
                } else {
                    $list = $sql->field($field . ' 0 as channel_id,0 as server_id')
                        ->select();
                }
            } else {
                $field = ' gameid as game_id,count(DISTINCT userid) as '.$type.', ';
                //当天登陆游戏
                if ($channel) {
                    if ($server) {
                        $list = $sql->field($field . 'channel_id,serverid as server_id')
                            ->group('gameid,channel_id,serverid')->select();
                    } else {
                        $list = $sql->field($field . 'channel_id,0 as server_id')
                            ->group('gameid,channel_id')->select();
                    }
                } else {
                    if ($server) {
                        $list = $sql->field($field . ' 0 as channel_id,serverid as server_id')
                            ->group('gameid,serverid')->select();
                    } else {
                        $list = $sql->field($field . ' 0 as channel_id,0 as server_id')
                            ->group('gameid')->select();
                    }
                }
            }

            // dump($list);
            return $list??[];
        // }
        // return [];
    }
    public function regNumNew($day, $mebmers, $server = 0, $channel = 0, $platform = 0, $type='reg_num')
    {
        $sql = Db::table('nw_member_game_server')
            ->whereIn('game_id', $this->getGameIds())
            ->where('serverid', '<>', '')
            ->where('channel_id', '<>', '')
            ->where('roleid', '<>', '');
        if($type == 'reg_total'){
            $sql->where('create_time', '<=', strtotime($day . ' 23:59:59'));
        }else{
            if($mebmers){
                $sql->where('member_id', 'in', $mebmers);
            }
            $sql->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
        }
        // $sql->fetchSql(true);

        if ($platform) {
            $field = ' 0 as game_id,count(DISTINCT member_id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->select();
            }
        } else {
            $field = ' game_id,count(DISTINCT member_id) as '.$type.', ';
            if ($channel) {
                if ($server) {
                    $list = $sql->field($field . 'channel_id,serverid as server_id')->group('game_id,channel_id,serverid,member_id')->select();
                } else {
                    $list = $sql->field($field . 'channel_id,0 as server_id')->group('game_id,channel_id')->select();
                }
            } else {
                if ($server) {
                        $list = $sql->field($field . ' 0 as channel_id,serverid as server_id')->group('game_id,serverid,member_id')->select();
                } else {
                    $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->group('game_id')->select();
                }
            }
        }

        // dump($list);
        return $list??[];
    }
    /**
     * 新增玩家V2
     * @param        $day
     * @param int    $channel  是否渠道数据
     * @param int    $platform 是否平台数据
     * @param string $type     count=统计数据, sum=汇总数据
     *
     * @return array|bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function regNumV2s($day, $channel = 0, $platform = 0, $type='reg_num')
    {
        $sql = Db::table('nw_subaccount')->where('game_id', 'in', $this->getGameIds());
        if($type == 'reg_total'){
            $sql = $sql->where('create_time', '<', strtotime($day . ' 23:59:59'));
        }else{
            $sql = $sql->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
        }
        // $sql->fetchSql(true);

        if ($platform) {
            $field = ' 0 as game_id,count(id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->select();
            }
        } else {
            $field = ' game_id,count(id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('game_id,channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->group('game_id')->select();
            }
        }

        // dump($list);
        return $list??[];
    }
    public function regNumV2($day, $channel = 0, $platform = 0, $type='reg_num')
    {
        $gameIds = implode(',', $this->getGameIds());
        $sql = Db::table('nw_subaccount')->alias('nw')->join('cy_game cg', "cg.id=nw.game_id", 'left')
            ->where('cg.id', 'in', $gameIds);
        if($type == 'reg_total'){
            $sql = $sql->where('nw.create_time', '<=', strtotime($day . ' 23:59:59'));
        }else{
            $sql = $sql->whereBetween('nw.create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
        }
        // $sql->fetchSql(true);

        if ($platform) {
            $field = ' 0 as game_id,count(nw.id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'nw.channel_id,0 as server_id')->group('nw.channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->select();
            }
        } else {
            $field = ' nw.game_id,count(nw.id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'nw.channel_id,0 as server_id')->group('nw.game_id,nw.channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->group('game_id')->select();
            }
        }

        // dump($list);
        return $list??[];
    }

    /**
     * 新增玩家留存
     *
     * @param $day
     * @param $server
     * @param $channel
     * @param $platform
     * @param $type
     *
     * @return array|bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function regNumStay($day, $server = 0, $channel = 0, $platform = 0, $type='reg_num')
    {
        $sql = model('common/Subaccount')
            ->where('channel_id', '<>', 0);
        if($type == 'reg_total'){
            $sql = $sql->where('create_time', '<', strtotime($day . ' 23:59:59'));
        }else{
            $sql = $sql->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
        }
        // $sql->fetchSql(true);

        if ($platform) {
            $field = ' 0 as game_id,count(id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->select();
            }
        } else {
            $field = ' game_id,count(id) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('game_id,channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id,0 as server_id')->group('game_id')->select();
            }
        }

        // dump($list);
        return $list??[];
    }

    /**
     * 新增角色
     *
     * @param $day
     * @param int $server 是否根据游戏分组：0=否、1=是
     * @param int $channel 是否根据游戏分组：0=否、1=是
     * @param int $platform 是否根据游戏分组：0=否、1=是
     * @param string $type role_num=统计数据, reg_total=汇总数据
     *
     * @return bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function getFirstRoleMember($day, $server = 0, $channel = 0, $platform = 0, $type='role_num')
    {
        $sql = Db::table('nw_member_game_server')
            ->whereIn('game_id', $this->getGameIds())
            ->where('serverid', '<>', '')->where('channel_id', '<>', '');
        if($type == 'role_num'){
            $sql = $sql->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')]);
        }else{
            $sql = $sql->where('create_time', '<=', strtotime($day . ' 23:59:59'));
        }
        // $sql->fetchSql(true);

        if ($platform) {
            $field = ' 0 as game_id,count(DISTINCT game_id,member_id,serverid,roleid) as '.$type.', ';
            if ($channel) {
                $list = $sql->field($field . 'channel_id,0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . '0 as channel_id,0 as server_id')->select();
            }
        } else {
            $field = ' game_id,count(DISTINCT game_id,member_id,serverid,roleid) as '.$type.', ';
            if ($channel) {
                if ($server) {
                    $list = $sql->field($field . 'channel_id,serverid as server_id')->group('game_id,channel_id,serverid')->select();
                } else {
                    $list = $sql->field($field . 'channel_id,0 as server_id')->group('game_id,channel_id')->select();
                }
            } else {
                if ($server) {
                    $list = $sql->field($field . '0 as channel_id,serverid as server_id')->group('game_id,serverid')->select();
                } else {
                    $list = $sql->field($field . '0 as channel_id,0 as server_id')->group('game_id')->select();
                }
            }
        }

        // dump($list);
        return $list;
    }

    /**
     * 活跃玩家
     *
     * @param $day
     * @param $param
     * @param int $server
     * @param int $channel
     * @param array $members
     * @param int $platform
     *
     * @return bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function activeMember($day, $param, $server = 0, $channel = 0, $members = [], $platform = 0)
    {
        $getGameIds = $this->getGameIds();
        $sql = Db::table('cy_role_info')
            ->whereBetween('update_time', [strtotime($day), strtotime($day . ' 23:59:59')])
            ->where('serverid', '<>', '')
            ->where('channel_id', '<>', '')
            ->where(function ($query) use ($members, $getGameIds) {
                if ($members) {
                    $query->where('userid', 'in', $members);
                }
                if ($getGameIds) {
                    $query->where('gameid', 'in', $getGameIds);
                }
            });
        // $sql->fetchSql(true);

        if ($platform) {
            $field = '0 as game_id,count(DISTINCT gameid, userid, serverid, roleid) as ' . $param . ', ';
            if ($channel) {
                $list = $sql->field($field . ' channel_id, 0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->select();
            }
        }else{
            $field = 'gameid as game_id,count(DISTINCT gameid, userid, serverid) as ' . $param . ', ';
            if ($channel) {
                if ($server) {
                    $list = $sql->field($field . ' channel_id, serverid as server_id')->group('gameid,channel_id,serverid')->select();
                } else {
                    $list = $sql->field($field . ' channel_id, 0 as server_id')->group('gameid,channel_id')->select();
                }
            } else {
                if ($server) {
                    $list = $sql->field($field . ' 0 as channel_id, serverid as server_id')->group('gameid,serverid')->select();
                } else {
                    $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->group('gameid')->select();
                }
            }
        }

        // dump($list);
        return $list;
    }
    /**
     * 活跃玩家V2
     *
     * @param $day
     * @param $param
     * @param $channel
     * @param $members
     * @param $platform
     *
     * @return bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function activeMemberV2($day, $param, $channel = 0, $members = [], $platform = 0, $wheres = [])
    {
        $sql = Db::table('cy_logininfo')->alias((['cy_logininfo' => 'cl', 'nw_subaccount' => 'ns']))
            ->join('nw_subaccount', 'cl.sub_id = ns.id', 'left')
            ->whereIn('gameid', $this->getGameIds())
            ->whereBetween('cl.login_time', [strtotime($day), strtotime($day . ' 23:59:59')])
            // ->where('channel_id', '<>', 0)
            ->where(function ($query) use ($members, $wheres) {
                if ($members) {
                    $query->where('cl.userid', 'in', $members);
                }
                if ($wheres) {
                    $query->where( $wheres);
                }
            });
        // $sql->fetchSql(true);

        if ($platform) {
            $field = '0 as game_id,count(DISTINCT cl.userid) as ' . $param . ', ';
            if ($channel) {
                $list = $sql->field($field . ' cl.channel_id, 0 as server_id')->group('cl.channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->select();
            }
        }else{
            $field = 'cl.gameid as game_id,count(DISTINCT cl.gameid, cl.userid, cl.sub_id) as ' . $param . ', ';
            if ($channel) {
                $list = $sql->field($field . ' cl.channel_id, 0 as server_id')->group('cl.gameid,cl.channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->group('cl.gameid')->select();
            }
        }

        // dump($list);
        return $list;
    }

    /**
     * 付费人数&充值金额
     *
     * @param $day
     * @param $server
     * @param $channel
     * @param $platform
     * @param $members
     * @param $param ['付费人数', '付费金额']
     *
     * @return bool|\PDOStatement|string|\think\Collection
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    public function rechargeMoney($day, $server = 0, $channel = 0, $platform = 0, $members = [], $param = ['pay_num', 'recharge_num'])
    {
        $sql = Db::table('cy_pay')
            ->whereBetween('pay_time', [strtotime($day), strtotime($day . ' 23:59:59')])
            ->where(['status' => 1])
            ->whereIn('gameid', $this->getGameIds())
            ->where(function ($query) use ($members) {
                if ($members) {
                    $query->where('userid', 'in', $members);
                }
            });
        // $sql->fetchSql(true);

        if($platform){
            $field = "0 as game_id,sum(pay_amount) as {$param[1]},count(DISTINCT userid) as {$param[0]}, ";
            if ($channel) {
                $list = $sql->field($field . 'channel_id, 0 as server_id')->group('channel_id')->select();
            } else {
                $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->select();
            }
        }else{
            $field = "gameid as game_id,sum(pay_amount) as {$param[1]},count(DISTINCT userid) as {$param[0]}, ";
            if ($channel) {
                if ($server) {
                    $list = $sql->field($field . 'channel_id,serverid as server_id')->group('gameid,channel_id,server_id')->select();
                } else {
                    $list = $sql->field($field . 'channel_id, 0 as server_id')->group('gameid,channel_id')->select();
                }

            } else {

                if ($server) {
                    $list = $sql->field($field . ' 0 as channel_id,serverid as server_id')->group('gameid,server_id')->select();
                } else {
                    $list = $sql->field($field . ' 0 as channel_id, 0 as server_id')->group('gameid')->select();
                }
            }
        }
        // dump($list);

        return $list;
    }

    // 获取当天有新增角色的会员ID
    public function getFirstRoleMemberList($day)
    {
        $list = model('common/MemberGameServer')
            ->cache('retaine:getFirstRoleMemberList:' . strtotime($day), 500)
            ->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')])->column(' DISTINCT member_id');
        return $list;

    }

    // 获取当天有子账号注册的会员ID
    public function getSubIds($day)
    {
        $list = Db::table('nw_subaccount')
            ->cache('retaine:getSubIds:' . strtotime($day), 500)
            ->whereBetween('create_time', [strtotime($day), strtotime($day . ' 23:59:59')])
            ->where('game_id', 'in', $this->getGameIds())
            ->column('DISTINCT member_id');
        return $list;

    }

    public function insertOrUpdate($list, $type_id, $day, $param)
    {
        // Db::startTrans();
        try {
            foreach ($list as $k => $v) {
                // echo "retaine:insertOrUpdate:{$param}_" . $type_id . "_{$day}_" . $v['game_id'] . '_' . $v['channel_id'].PHP_EOL;

                $retaineGame = Db::table('cy_retaine_game')
                    // 缓存查询会导致脏数据问题，先注释掉
                    // ->cache("retaine:insertOrUpdate:{$param}_" . $type_id . '_' . strtotime($day) . '_' . $v['game_id'] . '_' . $v['channel_id'], 10)
                    ->where(['type_id' => $type_id, 'day' => $day, 'game_id' => $v['game_id'], 'channel_id' => $v['channel_id']])
                    ->find();

                if ($retaineGame) {
                    $data = [];
                    if ($retaineGame['s_channel_id'] == 0) {
                        $data = $this->getChannel($v['channel_id']);
                    }
                    $data[$param] = $v[$param];
                    if ($param == 'recharge_num') {
                        $this->saveInfo($type_id, $day, $v['game_id'], $v['channel_id'], ['recharge_num' => $v['recharge_num'], 'pay_num' => $v['pay_num']]);
                    } else {
                        $this->saveInfo($type_id, $day, $v['game_id'], $v['channel_id'], $data);
                    }
                } else {
                    $data = $this->getChannel($v['channel_id']);
                    $data[$param] = $v[$param];
                    $this->insertInfo($type_id, $day, $v['game_id'], $v['channel_id'], $data);
                }

            }
            // Db::commit();
        } catch (Exception $e) {
            // Db::rollback();
            // curlDD('【风谷】数据统计异常.insertOrUpdate: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile(). $e->getLine().' - err_id:'.makeOrderid('ERR'), Env::get('dingtalk.warning_url'));
            log_message('【风谷】数据统计异常.insertOrUpdate: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile(). $e->getLine().' - err_id:'.makeOrderid('ERR'), 'error', LOG_PATH . '/complex_retaine/service');
        }
    }

    /**
     * 批量同步统计数据至留存表
     * @param array $data 统计好的数组数据
     * @param string $statDay 统计日期 e.g. '2026-01-30'
     * @param int $typeId 类型 1游戏 2推广员
     */
    /**
     * @param array $list   原始数据列表
     * @param int $type_id  类型
     * @param string $day   日期
     * @param string $param 对应数据库中的目标字段名 (如: 'reg_total' 或 'act_num')
     */
    public function syncGameStatData($list, $type_id, $day, $param)
    {
        if (empty($list) || empty($param)) return;

        // 1. 明确我们要更新的字段
        // 强制指定更新 $param 传入的字段名和 update_time
        $updateField = $param;

        // 2. 分批处理
        $chunks = array_chunk($list, 500);
        $now = time();

        foreach ($chunks as $item) {
            $values = [];
            foreach ($item as $row) {
                // 基础必有字段
                $gameId    = intval($row['game_id']);
                $channelId = intval($row['channel_id']);

                /**
                 * 重点：动态取值
                 * 无论你的数组里那个数字的键名是什么（可能是 reg_total, 也可能是 count）
                 * 我们通过 array_diff_key 排除掉固定键名，剩下的那个就是我们要的值
                 */
                $tempRow = $row;
                unset($tempRow['game_id'], $tempRow['channel_id'], $tempRow['server_id']);

                $temp = [
                    $type_id,
                    "'$day'",
                    $gameId,
                    $channelId,
                    $now,
                    (float)$row[$param] // 对应 $param 的值
                ];

                $values[] = "(" . implode(',', $temp) . ")";
            }

            // 3. 动态拼装 SQL
            // 注意：这里手动指定 $updateField
            $fieldStr = "type_id, `day`, game_id, channel_id, update_time, `$updateField` ";
            $valStr   = implode(',', $values);

            // 构建唯一索引冲突时的更新逻辑
            // 只有 update_time 和 你传入的这个 $param 字段会被更新
            $sql = "INSERT INTO cy_retaine_game ($fieldStr) VALUES $valStr 
                ON DUPLICATE KEY UPDATE 
                update_time = VALUES(update_time), 
                `$updateField` = VALUES(`$updateField`)";

            // dump($sql);
            try {
                $res = \think\Db::execute($sql);
                // dump($res);

            } catch (\Exception $e) {
                \think\Log::error("同步字段 {$param} 失败: " . $e->getMessage());
            }
        }
    }

    public function insertOrUpdatev2($list, $type_id, $day, $param)
    {
        if (empty($list)) {
            return;
        }
        
        // 分批处理，每批100条记录
        $batchSize = 100;
        $batches = array_chunk($list, $batchSize);
        
        foreach ($batches as $batchIndex => $batch) {
            // Db::startTrans();
            try {
                // 1. 批量查询已存在的记录
                $gameIds = array_unique(array_column($batch, 'game_id'));
                $channelIds = array_unique(array_column($batch, 'channel_id'));
                
                $existingList = Db::table('cy_retaine_game')
                    ->where('type_id', $type_id)
                    ->where('day', $day)
                    ->where('game_id', 'in', $gameIds)
                    ->where('channel_id', 'in', $channelIds)
                    ->select();

                // 构建已存在记录的键值映射
                $existingRecords = [];
                foreach ($existingList as $record) {
                    $key = $record['game_id'] . '_' . $record['channel_id'];
                    $existingRecords[$key] = $record;
                }
                
                // 2. 分离新增和更新数据
                $insertData = [];  // 需要新增的
                $updateData = [];  // 需要更新的
                
                foreach ($batch as $v) {
                    $key = $v['game_id'] . '_' . $v['channel_id'];
                    
                    // 获取渠道信息（如果记录不存在，或者s_channel_id为0）
                    $channelData = [];
                    if (!isset($existingRecords[$key]) || (isset($existingRecords[$key]) && $existingRecords[$key]['s_channel_id'] == 0)) {
                        $channelData = $this->getChannel($v['channel_id']);
                    }
                    
                    // 统一数据结构：确保所有记录都有相同字段（避免 insertAll 字段不匹配错误）
                    $data = [
                        's_channel_id' => isset($channelData['s_channel_id']) ? intval($channelData['s_channel_id']) : 0,
                        'b_channel_id' => isset($channelData['b_channel_id']) ? intval($channelData['b_channel_id']) : 0,
                        'bz_channel_id' => isset($channelData['bz_channel_id']) ? intval($channelData['bz_channel_id']) : 0,
                        $param => $v[$param],
                        'update_time' => time()
                    ];
                    
                    // 处理充值金额和付费人数
                    if ($param == 'recharge_num') {
                        $data['pay_num'] = $v['pay_num'];
                    }
                    
                    // 判断是新增还是更新
                    if (!isset($existingRecords[$key])) {
                        // 新增记录：添加必要字段
                        $data['type_id'] = $type_id;
                        $data['day'] = $day;
                        $data['game_id'] = $v['game_id'];
                        $data['channel_id'] = $v['channel_id'];
                        $insertData[] = $data;
                    } else {
                        // 更新记录：只更新非0的渠道字段
                        $updateFields = [];
                        if ($data['s_channel_id'] > 0) {
                            $updateFields['s_channel_id'] = $data['s_channel_id'];
                        }
                        if ($data['b_channel_id'] > 0) {
                            $updateFields['b_channel_id'] = $data['b_channel_id'];
                        }
                        if ($data['bz_channel_id'] > 0) {
                            $updateFields['bz_channel_id'] = $data['bz_channel_id'];
                        }
                        $updateFields[$param] = $data[$param];
                        $updateFields['update_time'] = $data['update_time'];
                        
                        if ($param == 'recharge_num') {
                            $updateFields['pay_num'] = $data['pay_num'];
                        }
                        
                        $updateFields['_where'] = [
                            'type_id' => $type_id,
                            'day' => $day,
                            'game_id' => $v['game_id'],
                            'channel_id' => $v['channel_id']
                        ];
                        $updateData[] = $updateFields;
                    }
                }
                
                // 3. 批量插入新记录
                if (!empty($insertData)) {

                    $insertResult = Db::table('cy_retaine_game')->insertAll($insertData);
                    Log::write("## {$param}@insertOrUpdatev2.insertAll: data=".json_encode($insertData).", result=".$insertResult,'notice');
                }
                
                // 4. 批量更新已存在记录
                if (!empty($updateData)) {
                    foreach ($updateData as $data) {
                        $where = $data['_where'];
                        unset($data['_where']);
                        
                        if (!empty($data)) {
                            $updateResult = Db::table('cy_retaine_game')->where($where)->update($data);
                            Log::write("## {$param}@insertOrUpdatev2.update: where=".json_encode($where).", data=".json_encode($data).", result=".$updateResult,'notice');

                        }
                    }
                }
                
                // Db::commit();
            } catch (Exception $e) {
                // Db::rollback();
                log_message('【风谷】数据统计异常.insertOrUpdatev2: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile(). $e->getLine().' - err_id:'.makeOrderid('ERR'), 'error', LOG_PATH . '/complex_retaine/service');
            }
        }
    }

    public function insertInfo($type_id, $day, $game_id, $channel_id, $param)
    {
        $data = array_merge($param, [
            'type_id' => $type_id,
            'day' => $day,
            'game_id' => $game_id,
            'channel_id' => $channel_id,
            'update_time' => time(),
        ]);
        $result = model('common/RetaineGame')->insert($data);

        Log::write("## {$type_id}@insertInfo: data=".json_encode($data).", result=".$result,'notice');
    }

    public function saveInfo($type_id, $day, $game_id, $channel_id, $param)
    {
        $param = array_merge($param, ['update_time' => time()]);
        $result = model('common/RetaineGame')->where(['type_id' => $type_id, 'day' => $day, 'game_id' => $game_id, 'channel_id' => $channel_id])->update($param);

        Log::write("## {$type_id}@saveInfo: where=".json_encode(['type_id' => $type_id, 'day' => $day, 'game_id' => $game_id, 'channel_id' => $channel_id]).", data=".json_encode($param).", result=".$result,'notice');
    }


    public function insertOrUpdateServer($list, $type_id, $day, $param)
    {
        // Db::startTrans();
        try {
            foreach ($list as $k => $v) {
                // dump($v);

                // if (in_array($v['game_id'], [175])) {
                //     if (!is_numeric($v['server_id'])) {
                //         $gameServer = model('common/GameServer')->field('serverid')->cache('retaineService:insertOrUpdateServer:' . $v['game_id'] . "_" . $v['server_id'])->where(['game_id' => $v['game_id'], 'servername' => $v['server_id']])->where('serverid', '<>', $v['server_id'])->find();
                //         $v['server_id'] = $gameServer['serverid'];
                //     }
                // }
                if ($retaineGame = Db::table('cy_retaine_server')
                        ->cache('retaine:insertOrUpdate:' . $type_id . '_' . strtotime($day) . '_' . $v['game_id'] . '_' . $v['channel_id'] . '_' . $v['server_id'], 60)
                        ->where(['type_id' => $type_id, 'day' => $day, 'game_id' => $v['game_id'], 'channel_id' => $v['channel_id'], 'server_id' => $v['server_id']])->find()) {
                    $data = [];
                    // dump($retaineGame);
                    // 判断是否商务
                    if ($retaineGame['s_channel_id'] == 0) {
                        $data = $this->getChannel($v['channel_id']);
                    }
                    // $data[$param] = $v[$param];
                    // if ($param == 'recharge_num') {
                    //     $this->saveInfoServer($type_id, $day, $v['game_id'], $v['channel_id'], $v['server_id'], ['recharge_num' => $v['recharge_num'], 'pay_num' => $v['pay_num']]);
                    // } else {
                    //     $this->saveInfoServer($type_id, $day, $v['game_id'], $v['channel_id'], $v['server_id'], $data);
                    // }

                } else {
                    //
                    // $data = $this->getChannel($v['channel_id']);
                    // $data[$param] = $v[$param];
                    // $this->insertInfoServer($type_id, $day, $v['game_id'], $v['channel_id'], $v['server_id'], $data);
                }
            }
            // Db::commit();
        } catch (Exception $e) {

            curlDD('【风谷】数据统计异常.insertOrUpdateServer: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile().''. $e->getLine().' - err_id:'.makeOrderid('ERR'), Env::get('dingtalk.warning_url'));
            // Db::rollback();
        }
    }
    public function insertOrUpdateServerv2($list, $type_id, $day, $param)
    {
        if (empty($list)) {
            return;
        }

        // 分批处理，每批100条记录
        $batchSize = 500;
        $batches = array_chunk($list, $batchSize);
        $time = time();

        foreach ($batches as $batch) {
            Db::startTrans();
            try {
                // 1. 批量查询已存在的记录
                $gameIds = array_unique(array_column($batch, 'game_id'));
                $channelIds = array_unique(array_column($batch, 'channel_id'));
                $serverIds = array_unique(array_column($batch, 'server_id'));
                
                $existingList = Db::table('cy_retaine_server')
                    ->where('type_id', $type_id)
                    ->where('day', $day)
                    ->where('game_id', 'in', $gameIds)
                    ->where('channel_id', 'in', $channelIds)
                    ->where('server_id', 'in', $serverIds)
                    ->select();
                
                // 构建已存在记录的键值映射（包含 server_id）
                $existingRecords = [];
                foreach ($existingList as $record) {
                    $key = $record['game_id'] . '_' . $record['channel_id'] . '_' . $record['server_id'];
                    $existingRecords[$key] = $record;
                }
                
                // 2. 分离新增和更新数据
                $insertData = [];  // 需要新增的
                $updateData = [];  // 需要更新的
                
                foreach ($batch as $v) {
                    $key = $v['game_id'] . '_' . $v['channel_id'] . '_' . $v['server_id'];
                    
                    // 获取渠道信息（如果记录不存在，或者s_channel_id为0）
                    $channelData = [];
                    if (!isset($existingRecords[$key]) || (isset($existingRecords[$key]) && $existingRecords[$key]['s_channel_id'] == 0)) {
                        $channelData = $this->getChannel($v['channel_id']);
                    }
                    
                    // 统一数据结构：确保所有记录都有相同字段
                    $data = [
                        's_channel_id' => isset($channelData['s_channel_id']) ? intval($channelData['s_channel_id']) : 0,
                        'b_channel_id' => isset($channelData['b_channel_id']) ? intval($channelData['b_channel_id']) : 0,
                        'bz_channel_id' => isset($channelData['bz_channel_id']) ? intval($channelData['bz_channel_id']) : 0,
                        $param => $v[$param],
                        'update_time' => $time
                    ];
                    
                    // 处理充值金额和付费人数
                    if ($param == 'recharge_num') {
                        $data['pay_num'] = $v['pay_num'];
                    }
                    
                    // 判断是新增还是更新
                    if (!isset($existingRecords[$key])) {
                        // 新增记录：添加必要字段
                        $data['type_id'] = $type_id;
                        $data['day'] = $day;
                        $data['game_id'] = $v['game_id'];
                        $data['channel_id'] = $v['channel_id'];
                        $data['server_id'] = $v['server_id'];
                        $insertData[] = $data;
                    } else {
                        // 更新记录：只更新非0的渠道字段
                        $updateFields = [];
                        if ($data['s_channel_id'] > 0) {
                            $updateFields['s_channel_id'] = $data['s_channel_id'];
                        }
                        if ($data['b_channel_id'] > 0) {
                            $updateFields['b_channel_id'] = $data['b_channel_id'];
                        }
                        if ($data['bz_channel_id'] > 0) {
                            $updateFields['bz_channel_id'] = $data['bz_channel_id'];
                        }
                        $updateFields[$param] = $data[$param];
                        $updateFields['update_time'] = $data['update_time'];
                        
                        if ($param == 'recharge_num') {
                            $updateFields['pay_num'] = $data['pay_num'];
                        }
                        
                        $updateFields['_where'] = [
                            'type_id' => $type_id,
                            'day' => $day,
                            'game_id' => $v['game_id'],
                            'channel_id' => $v['channel_id'],
                            'server_id' => $v['server_id']
                        ];
                        $updateData[] = $updateFields;
                    }
                }
                
                // 3. 批量插入新记录
                if (!empty($insertData)) {
                    Db::table('cy_retaine_server')->insertAll($insertData);
                }
                
                // 4. 批量更新已存在记录
                if (!empty($updateData)) {
                    foreach ($updateData as $data) {
                        $where = $data['_where'];
                        unset($data['_where']);
                        
                        if (!empty($data)) {
                            Db::table('cy_retaine_server')->where($where)->update($data);
                        }
                    }
                }
                
                Db::commit();
            } catch (Exception $e) {
                Db::rollback();
                log_message('【风谷】数据统计异常.insertOrUpdateServerv2: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile().''. $e->getLine().' - err_id:'.makeOrderid('ERR'), 'error', LOG_PATH . '/complex_retaine/service');
            }
        }
    }

    public function insertInfoServer($type_id, $day, $game_id, $channel_id, $server_id, $param)
    {
        model('common/RetaineServer')->insert(array_merge($param, [
            'type_id' => $type_id,
            'day' => $day,
            'game_id' => $game_id,
            'server_id' => $server_id,
            'channel_id' => $channel_id,
            'update_time' => time(),
        ]));
    }

    public function saveInfoServer($type_id, $day, $game_id, $channel_id, $server_id, $param)
    {
        $param = array_merge($param, ['update_time' => time()]);
        model('common/RetaineServer')->where(['type_id' => $type_id, 'day' => $day, 'game_id' => $game_id, 'channel_id' => $channel_id, 'server_id' => $server_id])->update($param);
    }

    public function getChannels($channel_id)
    {
        $data = [];
        if ($channel_id > 0) {
            return $data;
        }

        $channelList = model('common/Channel')->cache('retaine:channelList', 120)->column('id,name,id_path,level', 'id');

        if (isset($channelList[$channel_id])) {
            $channel = $channelList[$channel_id];
            $id_path = explode(',', $channel['id_path']);
            foreach ($id_path as $kk => $vv) {
                if (isset($vv) && isset($channelList[$vv]) && $channelList[$vv] && $vv != $channel_id) {
                    $channel_ch = $channelList[$vv];

                    if ($channel_ch['level'] == 0) {
                        $data['s_channel_id'] = $channel_ch['id'];
                    } elseif ($channel_ch['level'] == 1) {
                        $data['b_channel_id'] = $channel_ch['id'];
                    } elseif ($channel_ch['level'] == 2) {
                        $data['bz_channel_id'] = $channel_ch['id'];
                    }
                }

            }
        }
        return $data;
    }
    public function getChannel($channel_id)
    {

        // dump($channel_id);
        $data = [];
        if ($channel_id > 0) {
            return $data;
        }

        // 进程级静态缓存，一次请求只读一次文件
        static $channelList = null;
        if ($channelList === null) {
            $channelList = Cache::store('File')->get('cache_retaine_channel_list');
            if (!$channelList) {
                $channelList = model('common/Channel')->column('id,name,id_path,level', 'id');
                Cache::store('File')->set('cache_retaine_channel_list', $channelList, 3600);
            }
        }
        // dump($channelList);


        if (!isset($channelList[$channel_id])) {
            return [];
        }

        $data = [];
        $id_path = explode(',', $channelList[$channel_id]['id_path']);
        foreach ($id_path as $vv) {
            if ($vv == $channel_id || !isset($channelList[$vv])) {
                continue;
            }
            $ch = $channelList[$vv];
            if ($ch['level'] == 0) {
                $data['s_channel_id'] = $ch['id'];
            } elseif ($ch['level'] == 1) {
                $data['b_channel_id'] = $ch['id'];
            } elseif ($ch['level'] == 2) {
                $data['bz_channel_id'] = $ch['id'];
            }
        }
        // dump($data);
        return $data;
    }

    public function insertOrUpdatePlatform($list, $type_id, $day, $param)
    {
        Db::startTrans();
        try {
            foreach ($list as $k => $v) {
                $retaine = model('common/Retaine')
                    // ->cache('retaine:insertOrUpdatePlatform:' . $type_id . '_' . strtotime($day) . '_' . $v['channel_id'], 60)
                    ->where(['type_id' => $type_id, 'day' => $day, 'channel_id' => $v['channel_id']])->find();
                if ($retaine) {
                    $data = [];
                    if ($retaine['s_channel_id'] == 0) {
                        $data = $this->getChannel($v['channel_id']);
                    }
                    $data[$param] = $v[$param];
                    if ($param == 'recharge_num') {
                        $this->saveInfoPlatform($type_id, $day, $v['channel_id'], ['recharge_num' => $v['recharge_num'], 'pay_num' => $v['pay_num']]);
                    } else {
                        $this->saveInfoPlatform($type_id, $day, $v['channel_id'], $data);
                    }

                } else {
                    $data = $this->getChannel($v['channel_id']);
                    $data[$param] = $v[$param];
                    $this->insertInfoPlatform($type_id, $day, $v['channel_id'], $data);
                }
            }
            Db::commit();
        } catch (Exception $e) {
            curlDD('【风谷】数据统计异常.insertOrUpdatePlatform: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile().''. $e->getLine().' - err_id:'.makeOrderid('ERR'), Env::get('dingtalk.warning_url'));
            Db::rollback();
        }
    }
    public function insertOrUpdatePlatformv2($list, $type_id, $day, $param)
    {
        if (empty($list)) {
            return;
        }
        
        // 分批处理，每批100条记录
        $batchSize = 100;
        $batches = array_chunk($list, $batchSize);
        
        foreach ($batches as $batch) {
            try {
                // 1. 批量查询已存在的记录
                $channelIds = array_unique(array_column($batch, 'channel_id'));

                $existingList = Db::table('cy_retaine')
                    ->where('type_id', $type_id)
                    ->where('day', $day)
                    ->where('channel_id', 'in', $channelIds)
                    ->select();

                // 构建已存在记录的键值映射
                $existingRecords = [];
                foreach ($existingList as $record) {
                    $key = $record['channel_id'];
                    $existingRecords[$key] = $record;
                }

                // 2. 分离新增和更新数据
                $insertData = [];  // 需要新增的
                $updateData = [];  // 需要更新的

                foreach ($batch as $v) {
                    $key = $v['channel_id'];

                    // 获取渠道信息（如果记录不存在，或者s_channel_id为0）
                    $channelData = [];
                    if (!isset($existingRecords[$key]) || (isset($existingRecords[$key]) && $existingRecords[$key]['s_channel_id'] == 0)) {
                        $channelData = $this->getChannel($v['channel_id']);
                    }

                    // 统一数据结构：确保所有记录都有相同字段
                    $data = [
                        's_channel_id' => isset($channelData['s_channel_id']) ? intval($channelData['s_channel_id']) : 0,
                        'b_channel_id' => isset($channelData['b_channel_id']) ? intval($channelData['b_channel_id']) : 0,
                        'bz_channel_id' => isset($channelData['bz_channel_id']) ? intval($channelData['bz_channel_id']) : 0,
                        $param => $v[$param],
                        'update_time' => time()
                    ];

                    // 处理充值金额和付费人数
                    if ($param == 'recharge_num') {
                        $data['pay_num'] = $v['pay_num'];
                    }

                    // 判断是新增还是更新
                    if (!isset($existingRecords[$key])) {
                        // 新增记录：添加必要字段
                        $data['type_id'] = $type_id;
                        $data['day'] = $day;
                        $data['channel_id'] = $v['channel_id'];
                        $insertData[] = $data;
                    } else {
                        // 更新记录：只更新非0的渠道字段
                        $updateFields = [];
                        if ($data['s_channel_id'] > 0) {
                            $updateFields['s_channel_id'] = $data['s_channel_id'];
                        }
                        if ($data['b_channel_id'] > 0) {
                            $updateFields['b_channel_id'] = $data['b_channel_id'];
                        }
                        if ($data['bz_channel_id'] > 0) {
                            $updateFields['bz_channel_id'] = $data['bz_channel_id'];
                        }
                        $updateFields[$param] = $data[$param];
                        $updateFields['update_time'] = $data['update_time'];

                        if ($param == 'recharge_num') {
                            $updateFields['pay_num'] = $data['pay_num'];
                        }

                        $updateFields['_where'] = [
                            'type_id' => $type_id,
                            'day' => $day,
                            'channel_id' => $v['channel_id']
                        ];
                        $updateData[] = $updateFields;
                    }
                }

            } catch (\Exception $e){
                throw new \Exception('数据统计异常.insertOrUpdatePlatformv2: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage() . ' - file:' . $e->getFile() . ' - line:' . $e->getLine());
            }

            if(empty($insertData) && empty($updateData)){
                break;
            }
            Db::startTrans();
            try {
                // 3. 批量插入新记录
                if (!empty($insertData)) {
                    Db::table('cy_retaine')->insertAll($insertData);
                }

                // 4. 批量更新已存在记录
                if (!empty($updateData)) {
                    foreach ($updateData as $data) {
                        $where = $data['_where'];
                        unset($data['_where']);

                        if (!empty($data)) {
                            Db::table('cy_retaine')->where($where)->update($data);
                        }
                    }
                }

                Db::commit();
            } catch (Exception $e) {
                Db::rollback();
                log_message('【风谷】数据统计异常.insertOrUpdatePlatformv2: ' . $type_id . '_' . $day . '_' . $param . '_' . $e->getMessage().' - file:'. $e->getFile().''. $e->getLine().' - err_id:'.makeOrderid('ERR'), 'error', LOG_PATH . '/complex_retaine/service');
            }
        }
    }

    public function insertInfoPlatform($type_id, $day, $channel_id, $param)
    {
        model('common/Retaine')->insert(array_merge($param, [
            'type_id' => $type_id,
            'day' => $day,
            'channel_id' => $channel_id,
            'update_time' => time(),
        ]));
    }

    public function saveInfoPlatform($type_id, $day, $channel_id, $param)
    {
        $param = array_merge($param, ['update_time' => time()]);
        model('common/Retaine')->where(['type_id' => $type_id, 'day' => $day, 'channel_id' => $channel_id])->update($param);
    }
}