drb/app/Service/TPlusServerService.php

<?php

namespace App\Service;

use App\Jobs\ProcessDataJob;
use App\Model\Depart;
use App\Model\Employee;
use App\Model\EmployeeDepartPermission;
use App\Model\Product;
use App\Model\RevenueCost;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;

class TPlusServerService extends Service
{
    /**
     * @var TPlusDatabaseServerService
     */
    protected $databaseService;

    /**
     * @var string|null
     */
    protected $error;

    /**
     * TPlusServerService constructor.
     */
    public function __construct()
    {
        $service = new TPlusDatabaseServerService();
        $this->databaseService = $service->db;
        $this->error = $service->error;
    }

    /**
     * 获取错误信息
     *
     * @return string|null
     */
    public function getError()
    {
        return $this->error;
    }

    private $table = "tmp_revenue_cost_data";

    /**
     * 同步人员部门
     *
     * @param array $data
     * @param array $user
     * @return array
     */
    public function synPersonDepart($data, $user)
    {
        try {
            $this->databaseService->table('AA_Department')
                ->select('id','idparent as parent_id','name as title','code','disabled as is_use')
                ->chunkById(100, function ($data) {
                    DB::transaction(function () use ($data) {
                        $dataArray = Collect($data)->map(function ($object) {
                            return (array)$object;
                        })->toArray();

                        $d_id = Depart::whereIn('id', array_column($dataArray,'id'))
                            ->pluck('id')
                            ->toArray();

                        $insert = $update = [];
                        foreach ($dataArray as $value){
                            $is_use = $value['is_use'] ? 0 : 1;
                            if(in_array($value['id'], $d_id)){
                                $update[] = [
                                    'id' => $value['id'],
                                    'parent_id' => $value['parent_id'],
                                    'title' => $value['title'],
                                    'code' => $value['code'],
                                    'is_use' => $is_use
                                ];
                            }else{
                                $insert[] = [
                                    'id' => $value['id'],
                                    'parent_id' => $value['parent_id'],
                                    'title' => $value['title'],
                                    'code' => $value['code'],
                                    'is_use' => $is_use
                                ];
                            }
                        }
                        if(! empty($insert)) Depart::insert($insert);
                        if(! empty($update)) {
                            foreach ($update as $value){
                                Depart::where('id', $value['id'])
                                    ->update($value);
                            }
                        }
                    });
                });
            $this->databaseService->table('AA_Person')
                ->select('id','code as number','name as emp_name','mobilePhoneNo as mobile','iddepartment as depart_id','disabled as state')
                ->chunkById(100, function ($data) {
                    DB::transaction(function () use ($data) {
                        $dataArray = Collect($data)->map(function ($object) {
                            return (array)$object;
                        })->toArray();

                        $employee_id = Employee::whereIn('id', array_column($dataArray,'id'))
                            ->pluck('id')
                            ->toArray();

                        $insert = $update = $depart_update = [];
                        foreach ($dataArray as $value){
                            $state = $value['state'] ? Employee::NOT_USE : Employee::USE;
                            if(in_array($value['id'], $employee_id)){
                                $update[] = [
                                    'id' => $value['id'],
                                    'number' => $value['number'],
                                    'emp_name' => $value['emp_name'],
                                    'mobile' => $value['mobile'],
                                    'state' => $state
                                ];
                            }else{
                                $insert[] = [
                                    'id' => $value['id'],
                                    'number' => $value['number'],
                                    'emp_name' => $value['emp_name'],
                                    'mobile' => $value['mobile'],
                                    'state' => $state
                                ];
                            }
                            $depart_update[] = [
                                'employee_id' => $value['id'],
                                'depart_id' => $value['depart_id']
                            ];
                        }
                        if(! empty($insert)) Employee::insert($insert);
                        if(! empty($update)) {
                            foreach ($update as $value){
                                Employee::where('id', $value['id'])
                                    ->update($value);
                            }
                        }
                        if(! empty($depart_update)){
                            EmployeeDepartPermission::whereIn('employee_id',array_column($depart_update,'employee_id'))->delete();
                            EmployeeDepartPermission::insert($depart_update);
                        }
                    });
                });
        } catch (\Throwable $e) {
            return [false, $e->getMessage()];
        }

        return [true, ''];
    }

    /**
     * 收入成本统计同步
     *
     * @param array $data
     * @param array $user
     * @return array
     */
    public function synRevenueCost($data, $user){
        if(empty($data['crt_time']) || ! is_array($data['crt_time'])) return [false, '同步时间不能为空'];
        list($start_time, $end_time) = $this->changeDateToTimeStampAboutRange($data['crt_time'],false);
        if ($start_time === null || $end_time === null || $start_time > $end_time) return [false, "同步时间：时间区间无效"];
        list($bool, $bool_msg) = $this->isOverThreeMonths($start_time, $end_time);
        if(! $bool) return [false, $bool_msg];
        $data['start_timeStamp'] = $start_time;
        $data['end_timeStamp'] = $end_time;
        $start = date('Y-m-d H:i:s.000', $start_time);
        $end = date('Y-m-d H:i:s.000', $end_time);
        $data['start_time'] = $start;
        $data['end_time'] = $end;
        $data['operation_time'] = time();
        if(empty($data['type'])) return [false, '同步类型不能为空'];
        if(in_array($data['type'],[1,2,3,4])){
            list($status,$msg) = $this->limitingSendRequest($this->table);
            if(! $status) return [false, '收入成本相关信息同步正在后台运行,请稍后'];

//            //同步
//            list($status, $msg) = $this->synRevenueCostFromTPlus($data, $user);
//            if(! $status) {
//                $this->clearTmpTable();
//                $this->dellimitingSendRequest($this->table);
//                return [false, $msg];
//            }

            //队列
            ProcessDataJob::dispatch($data, $user)->onQueue(RevenueCost::job);
        }else{
            return [false, '同步类型错误'];
        }

        return [true, '收入成本相关信息同步已进入后台任务'];
    }

    public function synRevenueCostFromTPlus($data, $user){
        //创建临时表 如果不存在
        $this->createTmpTable();
        //清理临时表 如果内容不为空
        $this->clearTmpTable();
        $type = $data['type'];

        //写入临时数据
        if($type == 1){
            list($status, $msg) = $this->xhdTPlus($data, $user);
            if(! $status) return [false, $msg];
            list($status, $msg) = $this->xsfpTPlus($data, $user);
            if(! $status) return [false, $msg];
            list($status, $msg) = $this->hkdTPlus($data, $user);
            if(! $status) return [false, $msg];
        }elseif ($type == 2){
            list($status, $msg) = $this->xhdTPlus($data, $user);
            if(! $status) return [false, $msg];
        }elseif ($type == 3){
            list($status, $msg) = $this->xsfpTPlus($data, $user);
            if(! $status) return [false, $msg];
        }elseif ($type == 4){
            list($status, $msg) = $this->hkdTPlus($data, $user);
            if(! $status) return [false, $msg];
        }

        //更新主表数据
        list($status,$msg) = $this->updateRevenueCost($data);
        if(! $status) return [false, $msg];

        //都成功后 清理临时表
        $this->clearTmpTable();
        //释放redis
        $this->delTableKey();

        return [true, '同步成功'];
    }

    private function xhdTPlus($data, $user){
        try {
            $table = $this->table;
            $limit = 500;
            $lastId = 0;

            do {
                $rows = $this->databaseService->table('SA_SaleDelivery_b as sd_b')
                    ->join('SA_SaleDelivery as sd', 'sd_b.idSaleDeliveryDTO', '=', 'sd.ID')
                    ->leftJoin('AA_Partner as pn', 'sd.idsettlecustomer', '=', 'pn.ID')
                    ->leftJoin('AA_Person as ps', 'sd.idclerk', '=', 'ps.ID')
                    ->leftJoin('AA_Inventory as it', 'sd_b.idinventory', '=', 'it.ID')
                    ->leftJoin('AA_Unit as ui', 'sd_b.idbaseunit', '=', 'ui.ID')
                    ->where('sd.voucherdate', '>=', $data['start_time'])
                    ->where('sd.voucherdate', '<=', $data['end_time'])
                    ->where('sd_b.ID', '>', $lastId) // 用真实字段
                    ->orderBy('sd_b.ID')
                    ->limit($limit)
                    ->selectRaw("
                        COALESCE(sd.ID, 0) as order_id,
                        COALESCE(sd.code, '') as order_number,
                        sd.voucherdate as order_time,
                        COALESCE(ps.name, '') as employee_id_1_title,
                        COALESCE(sd.idclerk, 0) as employee_id_1,
                        COALESCE(pn.priuserdefnvc15, '') as employee_id_2_title,
                        COALESCE(pn.code, '') as customer_code,
                        COALESCE(pn.name, '') as customer_title,
                        COALESCE(sd.pubuserdefnvc11, '') as channel_finance,
                        COALESCE(sd.pubuserdefnvc12, '') as channel_details,
                        COALESCE(it.code, '') as product_code,
                        COALESCE(it.name, '') as product_title,
                        COALESCE(it.specification, '') as product_size,
                        COALESCE(ui.name, '') as unit,
                        COALESCE(sd_b.quantity, 0) as quantity,
                        COALESCE(sd_b.taxPrice, 0) as price_3,
                        COALESCE(sd_b.taxAmount, 0) as price_3_total,
                        COALESCE(sd_b.ID, 0) as id_detail
                    ")
                    ->get();

                if ($rows->isEmpty()) break;

                $dataArray = Collect($rows)->map(function ($object) {
                    return (array)$object;
                })->toArray();

                //存货档案
                $product = Product::where('del_time', 0)
                    ->whereIn('code', array_unique(array_column($dataArray, 'product_code')))
                    ->select('code', 'write_off_price', 'freight_price', 'business_cost')
                    ->get()->toArray();
                $product_map = array_column($product, null, 'code');

                //组织数据
                foreach ($dataArray as $key => $value) {
                    $p_tmp = $product_map[$value['product_code']] ?? [];
                    $dataArray[$key]['order_type'] = RevenueCost::ORDER_ONE;
                    $dataArray[$key]['order_time'] = strtotime($value['order_time']);
                    $write_off_price = $p_tmp['write_off_price'] ?? 0;
                    $dataArray[$key]['price_1'] = $write_off_price;
                    $dataArray[$key]['price_1_total'] = bcmul($write_off_price, $value['quantity'], 2);
                    $freight_price = $p_tmp['freight_price'] ?? 0;
                    $dataArray[$key]['price_2'] = $freight_price;
                    $dataArray[$key]['price_2_total'] = bcmul($freight_price, $value['quantity'], 2);
                    $business_cost = $p_tmp['business_cost'] ?? 0;
                    $dataArray[$key]['price_4'] = $business_cost;
                    $price_4_total = bcmul($business_cost, $value['quantity'], 2);
                    $dataArray[$key]['price_4_total'] = bcmul($business_cost, $value['quantity'], 2);
                    $profit = bcsub($value['price_3_total'], $price_4_total, 2);
                    $dataArray[$key]['profit'] = $profit;
                    $dataArray[$key]['profit_rate'] = $value['price_3_total'] > 0 ? bcdiv($profit, $value['price_3_total'], 2) : 0;
                }

                DB::table($table)->insert($dataArray);

                // 更新 lastId 继续下一批
                $lastId = end($dataArray)['id_detail'] ?? $lastId;

            } while (count($rows) === $limit);
        }catch  (\Throwable $exception){
            return [false, "销货单同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
        }

        return [true, ''];
    }

    private function xsfpTPlus($data, $user){
        try {
            $table = $this->table;
            $limit = 500;
            $lastId = 0;

            do {
                $rows = $this->databaseService->table('SA_SaleInvoice_b as si_b')
                    ->join('SA_SaleInvoice as si', 'si_b.idSaleInvoiceDTO', '=', 'si.ID')
                    ->leftJoin('AA_Partner as pn', 'si.idsettlecustomer', '=', 'pn.ID')
                    ->leftJoin('AA_Person as ps', 'si.idclerk', '=', 'ps.ID')
                    ->leftJoin('AA_Inventory as it', 'si_b.idinventory', '=', 'it.ID')
                    ->leftJoin('AA_Unit as ui', 'si_b.idbaseunit', '=', 'ui.ID')
                    ->where('si.voucherdate','>=',$data['start_time'])
                    ->where('si.voucherdate','<=',$data['end_time'])
                    ->where('si_b.ID', '>', $lastId) // 用真实字段
                    ->orderBy('si_b.ID')
                    ->limit($limit)
                    ->selectRaw("
                        COALESCE(si.ID, 0) as order_id,
                        COALESCE(si.code, '') as order_number,
                        si.voucherdate as order_time,
                        COALESCE(pn.code, '') as customer_code,
                        COALESCE(pn.name, '') as customer_title,
                        COALESCE(si.idclerk, 0) as employee_id_1,
                        COALESCE(ps.name, '') as employee_id_1_title,
                        COALESCE(pn.priuserdefnvc15, '') as employee_id_2_title,
                        COALESCE(it.code, '') as product_code,
                        COALESCE(it.name, '') as product_title,
                        COALESCE(it.specification, '') as product_size,
                        COALESCE(ui.name, '') as unit,
                        COALESCE(si_b.quantity, 0) as quantity,
                        COALESCE(si_b.taxPrice, 0) as price_1,
                        COALESCE(si_b.taxAmount, 0) as price_1_total,
                        COALESCE(si_b.ID, 0) as id_detail,
                        COALESCE(si_b.sourceVoucherDetailId, 0) as id_detail_upstream,
                        COALESCE(si_b.sourceVoucherCode, '') as order_number_upstream
                    ")
                    ->get();

                if ($rows->isEmpty()) break;

                $dataArray = Collect($rows)->map(function ($object) {
                    return (array)$object;
                })->toArray();

                //存货档案
                $product = Product::where('del_time',0)
                    ->whereIn('code', array_unique(array_column($dataArray,'product_code')))
                    ->select('code','business_cost')
                    ->get()->toArray();
                $product_map = array_column($product,null,'code');

                //组织数据
                foreach ($dataArray as $key => $value){
                    $p_tmp = $product_map[$value['product_code']] ?? [];
                    $dataArray[$key]['order_type'] = RevenueCost::ORDER_TWO;
                    $dataArray[$key]['order_time'] = strtotime($value['order_time']);
                    $business_cost = $p_tmp['business_cost'] ?? 0;
                    $dataArray[$key]['price_4'] = $business_cost;
                    $price_4_total = bcmul($business_cost, $value['quantity'],2);
                    $dataArray[$key]['price_4_total'] = bcmul($business_cost, $value['quantity'],2);
                    $profit = bcsub($value['price_1_total'], $price_4_total,2);
                    $dataArray[$key]['profit'] = $profit;
                    $dataArray[$key]['profit_rate'] = $value['price_1_total'] > 0 ? bcdiv($profit, $value['price_1_total'],2) : 0;
                }

                DB::table($table)->insert($dataArray);

                // 更新 lastId 继续下一批
                $lastId = end($dataArray)['id_detail'] ?? $lastId;

            } while (count($rows) === $limit);
        }catch  (\Throwable $exception){
            return [false, "销售发票同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
        }

        return [true, ''];
    }

    private function hkdTPlus($data, $user){
        try{
            $table = $this->table;
            $limit = 500;
            $lastId = 0;

            do {
                $rows = $this->databaseService->table('ARAP_ReceivePayment_b as rp_b')
                    ->join('ARAP_ReceivePayment as rp', 'rp_b.idArapReceivePaymentDTO', '=', 'rp.ID')
                    ->leftJoin('SA_SaleInvoice_b as si_b', 'rp_b.voucherDetailID', '=', 'si_b.ID')
                    ->leftJoin('SA_SaleInvoice as si', 'si_b.idSaleInvoiceDTO', '=', 'si.ID')
                    ->leftJoin('AA_Partner as pn', 'si.idsettlecustomer', '=', 'pn.ID')
                    ->leftJoin('AA_Person as ps', 'rp.idperson', '=', 'ps.ID')
                    ->leftJoin('AA_Inventory as it', 'si_b.idinventory', '=', 'it.ID')
                    ->leftJoin('AA_Unit as ui', 'si_b.idbaseunit', '=', 'ui.ID')
                    ->where('rp.voucherdate','>=',$data['start_time'])
                    ->where('rp.voucherdate','<=',$data['end_time'])
                    ->where('rp.isReceiveFlag','=', 1)
                    ->where('rp_b.ID', '>', $lastId) // 用真实字段
                    ->orderBy('rp_b.ID')
                    ->limit($limit)
                    ->selectRaw("
                        COALESCE(rp.ID, 0) as order_id,
                        COALESCE(rp.code, '') as order_number,
                        rp.voucherdate as order_time,
                        COALESCE(pn.code, '') as customer_code,
                        COALESCE(pn.name, '') as customer_title,
                        COALESCE(it.code, '') as product_code,
                        COALESCE(it.name, '') as product_title,
                        COALESCE(rp.idperson, 0) as employee_id_1,
                        COALESCE(ps.name, '') as employee_id_1_title,
                        COALESCE(pn.priuserdefnvc15, '') as employee_id_2_title,
                        COALESCE(it.specification, '') as product_size,
                        COALESCE(ui.name, '') as unit,
                        COALESCE(si_b.quantity, 0) as quantity,
                        COALESCE(si_b.taxPrice, 0) as price_1,
                        COALESCE(si_b.taxAmount, 0) as price_1_total,
                        COALESCE(rp_b.amount, 0) as payment_amount,
                        COALESCE(rp_b.ID, 0) as id_detail,
                        COALESCE(rp_b.voucherDetailID, 0) as id_detail_upstream,
                        COALESCE(rp_b.voucherCode, '') as order_number_upstream
                    ")
                    ->get();

                if ($rows->isEmpty()) break;

                $dataArray = Collect($rows)->map(function ($object) {
                    return (array)$object;
                })->toArray();

                //存货档案
                $product = Product::where('del_time',0)
                    ->whereIn('code', array_unique(array_column($dataArray,'product_code')))
                    ->select('code','business_cost')
                    ->get()->toArray();
                $product_map = array_column($product,null,'code');

                //组织数据
                foreach ($dataArray as $key => $value){
                    $p_tmp = $product_map[$value['product_code']] ?? [];
                    $dataArray[$key]['order_type'] = RevenueCost::ORDER_THREE;
                    $dataArray[$key]['order_time'] = strtotime($value['order_time']);
                    $business_cost = $p_tmp['business_cost'] ?? 0;
                    $dataArray[$key]['price_4'] = $business_cost;
                    $price_4_total = bcmul($business_cost, $value['quantity'],2);
                    $dataArray[$key]['price_4_total'] = $price_4_total;
                    $profit = bcsub($value['price_1_total'], $price_4_total,2);
                    $dataArray[$key]['profit'] = $profit;
                    $dataArray[$key]['profit_rate'] = $value['price_1_total'] > 0 ? bcdiv($profit, $value['price_1_total'],2) : 0;
                }

                DB::table($table)->insert($dataArray);

                // 更新 lastId 继续下一批
                $lastId = end($dataArray)['id_detail'] ?? $lastId;

            } while (count($rows) === $limit);

        }catch (\Throwable $exception){
            return [false, "回款单同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
        }

        return [true, ''];
    }

    private function updateRevenueCost($data){
        try {
            $start_timeStamp = $data['start_timeStamp'];
            $end_timeStamp = $data['end_timeStamp'];
            $tmpTable = $this->table;
            $time = time();

            DB::transaction(function () use ($start_timeStamp, $end_timeStamp, $tmpTable,$time) {
                // 1. 先软删除旧数据（你已有）
                RevenueCost::where('del_time', 0)
                    ->where('order_time', '>=', $start_timeStamp)
                    ->where('order_time', '<=', $end_timeStamp)
                    ->update(['del_time' => $time]);

                // 2. 分批从临时表插入新数据
                $batchSize = 500;
                $lastId = 0;

                do {
                    $chunk = DB::table($tmpTable)
                        ->where('id', '>', $lastId)
                        ->orderBy('id')
                        ->limit($batchSize)
                        ->get();

                    if ($chunk->isEmpty()) {
                        break;
                    }

                    $data = $chunk->map(function ($item) use($time){
                        return [
                            'order_id' => $item->order_id,
                            'order_number' => $item->order_number,
                            'order_time' => $item->order_time,
                            'employee_id_1_title' => $item->employee_id_1_title,
                            'employee_id_1' => $item->employee_id_1,
                            'employee_id_2_title' => $item->employee_id_2_title ?? "",
                            'customer_code' => $item->customer_code,
                            'customer_title' => $item->customer_title,
                            'channel_finance' => $item->channel_finance,
                            'channel_details' => $item->channel_details,
                            'product_code' => $item->product_code,
                            'product_title' => $item->product_title,
                            'product_size' => $item->product_size,
                            'unit' => $item->unit,
                            'quantity' => $item->quantity,
                            'price_1' => $item->price_1,
                            'price_1_total' => $item->price_1_total,
                            'price_2' => $item->price_2,
                            'price_2_total' => $item->price_2_total,
                            'price_3' => $item->price_3,
                            'price_3_total' => $item->price_3_total,
                            'price_4' => $item->price_4,
                            'price_4_total' => $item->price_4_total,
                            'profit' => $item->profit,
                            'profit_rate' => $item->profit_rate,
                            'id_detail' => $item->id_detail,
                            'order_type' => $item->order_type,
                            'payment_amount' => $item->payment_amount ?? 0,
                            'id_detail_upstream' => $item->id_detail_upstream?? 0,
                            'order_number_upstream' => $item->order_number_upstream ?? "",
                            'crt_time' => $time,
                        ];
                    })->toArray();

                    // 每批单独插入（可选：加小事务）
                    RevenueCost::insert($data);

                    // 更新 lastId
                    $lastId = $chunk->last()->id;

                } while ($chunk->count() == $batchSize);
            });
        }catch  (\Throwable $exception){
            return [false, "主表同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
        }

        return [true, ''];
    }

    private function createTmpTable(){
        $table = $this->table;
        if (! Schema::hasTable($table)) {
            // 可以通过 migration 创建，或程序启动时检查
            Schema::create($table, function (Blueprint $table) {
                $table->bigIncrements('id'); // BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY
                $table->integer('order_type')->default(0);
                $table->bigInteger('order_id')->default(0);
                $table->string('order_number', 50)->default('');
                $table->integer('order_time')->nullable();
                $table->string('employee_id_1_title', 100)->default('');
                $table->bigInteger('employee_id_1')->default(0);
                $table->string('employee_id_2_title', 100)->default('');
                $table->string('customer_code', 50)->default('');
                $table->string('customer_title', 100)->default('');
                $table->string('channel_finance', 50)->nullable();
                $table->string('channel_details', 50)->nullable();
                $table->string('product_code', 50)->default('');
                $table->string('product_title', 100)->default('');
                $table->string('product_size', 100)->nullable();
                $table->string('unit', 20)->nullable();
                $table->decimal('quantity', 12, 2)->default(0);
                $table->decimal('price_1', 12, 2)->default(0); // 销项成本
                $table->decimal('price_1_total', 12, 2)->default(0);
                $table->decimal('price_2', 12, 2)->default(0); // 运费
                $table->decimal('price_2_total', 12, 2)->default(0);
                $table->decimal('price_3', 12, 2)->default(0); // 含税单价
                $table->decimal('price_3_total', 12, 2)->default(0); // 含税金额
                $table->decimal('price_4', 12, 2)->default(0); // 业务成本
                $table->decimal('price_4_total', 12, 2)->default(0);
                $table->decimal('profit', 12, 2)->default(0);
                $table->decimal('payment_amount', 12, 2)->default(0);
                $table->decimal('profit_rate', 10, 3)->default(0);
                $table->bigInteger('id_detail')->default(0);
                $table->bigInteger('id_detail_upstream')->default(0);
                $table->string('order_number_upstream', 100)->nullable();
            });
        }
    }

    public function clearTmpTable(){
        if (Schema::hasTable($this->table)) DB::table($this->table)->truncate();
    }

    public function delTableKey(){
        $this->dellimitingSendRequest($this->table);
    }
}