Sjzs_admin/application/admin/controller/Queue.php

<?php

// +----------------------------------------------------------------------
// | ThinkAdmin
// +----------------------------------------------------------------------
// | 版权所有 2014~2019 广州楚才信息科技有限公司 [ http://www.cuci.cc ]
// +----------------------------------------------------------------------
// | 官方网站: http://demo.thinkadmin.top
// +----------------------------------------------------------------------
// | 开源协议 ( https://mit-license.org )
// +----------------------------------------------------------------------
// | gitee 代码仓库：https://gitee.com/zoujingli/ThinkAdmin
// | github 代码仓库：https://github.com/zoujingli/ThinkAdmin
// +----------------------------------------------------------------------

namespace app\admin\controller;

use library\Controller;
use library\service\ProcessService;
use think\Console;
use think\Db;
use think\exception\HttpResponseException;
use app\service\queue\ImportTable1Queue;
use app\service\queue\ImportTable2Queue;

/**
 * 上传管理
 * Class Queue
 * @package app\admin\controller
 */
class Queue extends Controller
{
    /**
     * 绑定数据表
     * @var string
     */
    protected $table = 'system_queue';

    /**
     * 上传记录
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     * @throws \think\exception\PDOException
     */
    public function index()
    {
        if (session('user.username') === 'admin') try {
            $this->message = Console::call('xtask:state')->fetch();
            $this->command = ProcessService::instance()->think('xtask:start');
            $this->listen = preg_match('/process.*?\d+.*?running/', $this->message, $attr);

//            $listen = ProcessService::instance()->query('xtask:_work 71 -');
//            dump($listen);
        } catch (\Exception $exception) {
            $this->listen = false;
            $this->message = $exception->getMessage();
        }
        $this->title = '上传记录';
        sysoplog('上传记录', '访问上传记录页面');
        $this->iswin = ProcessService::instance()->iswin();
        $query = $this->_query($this->table)->dateBetween('create_at,start_at,end_at,date');
        $query->like('title,preload')->equal('status,type,is_undo')->order('id desc')->page();
    }

    /**
     * 数据列表处理
     * @auth true
     * @menu true
     * @param array $data
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\ModelNotFoundException
     * @throws \think\exception\DbException
     */
    protected function _index_page_filter(&$data)
    {
        foreach ($data as &$v){
            $v['listen'] = 0;
            if ($v['status']==2){
                $listen = ProcessService::instance()->query('xtask:_work '.$v['id'].' -');
                $v['listen'] = count($listen);
            }
        }
    }

    /**
     * 导入表一数据
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function import_table1()
    {
//        $query = "SHOW FULL COLUMNS FROM system_courier";
//        $table = Db::query($query); //获取表所有字段
//        $field = array_column($table,'Field'); //提取表字段，放到一维数组中

        $username = $this->app->session->get('user')['username'];

        $get = $this->request->get();
        $import_path = $_SERVER['DOCUMENT_ROOT'].'/'.$get['path'];
        $url = $get['url'];
        $ordfilename = $get['ordfilename'];
        if (getConfigValue('storage_type')=='oss'){
            $save_dir = "./upload/excle/"; // 服务资源目录
            $filename = date('Ymd').time().".xlsx"; // 自定义名称
            $res = getFile($url,$save_dir,$filename,1);
            $import_path = $_SERVER['DOCUMENT_ROOT'].'/upload/excle/'.$res;
        }

        try {
            $result = sysqueue($ordfilename, ImportTable1Queue::URI, 0, ['type' => 1,'local_url'=>$import_path,'import_name'=>$username], 0);
            if ($result['status']){
                sysoplog('上传记录', '上传表一数据成功');
                $this->success($result['info']);
            }else{
                sysoplog('上传记录', '上传表一数据失败');
                $this->error($result['info'],[],3);
            }
        } catch (\think\exception\ValidateException $e) {
            $this->error($e->getMessage());
        }
    }


    /**
     * 导入表二数据
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function import_table2()
    {
        set_time_limit(0);
        $username = $this->app->session->get('user')['username'];

        $get = $this->request->get();
        $import_path = $_SERVER['DOCUMENT_ROOT'].'/'.$get['path'];
        $url = $get['url'];
        $ordfilename = $get['ordfilename'];
        if (getConfigValue('storage_type')=='oss'){
            $save_dir = "./upload/excle/"; // 服务资源目录
            $filename = date('Ymd').time().".xlsx"; // 自定义名称
            $res = getFile($url,$save_dir,$filename,1);
            $import_path = $_SERVER['DOCUMENT_ROOT'].'/upload/excle/'.$res;
        }
        try {

            $result = sysqueue($ordfilename, ImportTable2Queue::URI, 0, ['type' => 2,'local_url'=>$import_path,'import_name'=>$username], 0);
            if ($result['status']){
                sysoplog('上传记录', '上传表二数据成功');
                $this->success($result['info']);
            }else{
                sysoplog('上传记录', '上传表二数据失败');
                $this->error($result['info'],[],3);
            }
        } catch (\think\exception\ValidateException $e) {
            $this->error($e->getMessage());
        }
    }



    /**
     * 重置任务
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function redo()
    {
        $id = $this->request->get('id');
        Db::startTrans();
        try {
            $queue = Db::name('system_queue')->where('id',$id)->find();
            if ($queue['deal_count']>0){
                Db::name('system_queue')->where('id',$id)->update(
                    [
                        'deal_count'=>0,
                    ]
                );
            }
            if ($queue['type']==1){
                Db::name('system_values')->where('import_log_id',$id)->delete();
            }else{
                Db::name('system_table2')->where('import_log_id',$id)->delete();
                Db::name('system_values2')->where('import_log_id',$id)->delete();
                Db::name('system_values3')->where('import_log_id',$id)->delete();
            }
            Db::commit();
        }catch (\think\exception\ValidateException $e) {
            Db::rollback();
            sysoplog('上传记录', '重置任务ID:'.$id.'失败');
            $this->error('重置任务失败');
        }

        $log = ['action'=>'上传记录','content'=>'重置任务'];
        $this->_save($this->table,['status' => '1'],$log);
    }
    /**
     * WIN开始监听任务
     * @auth false
     */
    public function start()
    {
        try {
            $message = nl2br(Console::call('xtask:start')->fetch());
            if (preg_match('/process.*?\d+/', $message, $attr)) {
                $this->success('任务监听主进程启动成功！');
            } else {
                $this->error($message);
            }
        } catch (HttpResponseException $exception) {
            throw $exception;
        } catch (\Exception $e) {
            $this->error($e->getMessage());
        }
    }

    /**
     * WIN停止监听任务
     * @auth false
     */
    public function stop()
    {
        try {
            $message = nl2br(Console::call('xtask:stop')->fetch());
            if (stripos($message, 'succeeded')) {
                $this->success('停止任务监听主进程成功！');
            } elseif (stripos($message, 'finish')) {
                $this->success('没有找到需要停止的进程！');
            } else {
                $this->error($message);
            }
        } catch (HttpResponseException $exception) {
            throw $exception;
        } catch (\Exception $e) {
            $this->error($e->getMessage());
        }
    }

    /**
     * 清理3天前的记录
     * @auth false
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function clear()
    {
        $map = [['time', '<', strtotime('-3days')]];
        $result = Db::name($this->table)->where($map)->delete();
        if ($result !== false) {
            $this->success('成功清理3天前的任务记录！');
        } else {
            $this->error('清理3天前的任务记录失败！');
        }
    }

    /**
     * 删除任务
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function remove()
    {
        $log = ['action'=>'上传记录'];
        $this->_delete($this->table,$log);
    }

    /**
     * 回撤
     * @auth true
     * @menu true
     * @throws \think\Exception
     * @throws \think\exception\PDOException
     */
    public function undo()
    {
        $id = $this->request->get('id');
        Db::startTrans();
        try {
            $queue = Db::name('system_queue')->where('id',$id)->find();
            $queue_result = Db::name('system_queue')->where('id',$id)->update(
                [
                    'is_undo'=>2,
                    'undo_at'=>date('Y-m-d H:i:s'),
                    'undo_name'=>$this->app->session->get('user')['username']
                ]
            );
            if ($queue['type']==1){
                $values = Db::name('system_values')->where('import_log_id',$id)->delete();
                if ($queue_result && $values){
                    Db::commit();
                    sysoplog('上传记录', '回撤ID:'.$id.'成功');
                    $this->success('回撤成功');
                }else{
                    Db::rollback();
                    sysoplog('上传记录', '回撤ID:'.$id.'失败');
                    $this->error('撤销失败');
                }
            }else{
                $table2 = Db::name('system_table2')->where('import_log_id',$id)->delete();
                $values2 = Db::name('system_values2')->where('import_log_id',$id)->delete();
                $values3 = Db::name('system_values3')->where('import_log_id',$id)->delete();
                if ($queue_result && $table2 && $values2 && $values3){
                    Db::commit();
                    sysoplog('上传记录', '回撤ID:'.$id.'成功');
                    $this->success('回撤成功');
                }else{
                    Db::rollback();
                    sysoplog('上传记录', '回撤ID:'.$id.'失败');
                    $this->error('撤销失败');
                }
            }
        }catch (\think\exception\ValidateException $e) {
            Db::rollback();
            sysoplog('上传记录', '回撤ID:'.$id.'失败');
            $this->error('撤销失败');
        }

    }

}