20201218 优化数据相关操作，完成新接口

server/www/teleport/webroot/app/base/plugin.py

+# coding: utf-8
+import builtins
+import traceback
+from abc import ABCMeta, abstractmethod
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.base.logger import log
+from base.configs import tp_cfg
+
+
+class StoreDBBase(object):
+    __metaclass__ = ABCMeta
+
+    def __init__(self, host='127.0.0.1', port='8123', user='default', pwd='123456', db="default", **kwargs):
+        '''初始化构造函数'''
+        super(StoreDBBase, self).__init__()
+
+        self.host = host
+        self.port = port
+        self.user = user
+        self.pwd = pwd
+
+        connect_args = {
+            "server_host": self.host,
+            "port": self.port,
+            "user": self.user,
+            "password": self.pwd,
+            "db": db
+        }
+        connection_str = self.connection_str(connect_args)
+        self.engine = create_engine(connection_str, pool_size=100, pool_recycle=3600, pool_timeout=20)
+        # self._session = self.get_session(self.engine)
+
+    @abstractmethod
+    def connection_str(self, conf):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_session(self, engine):
+        raise NotImplementedError
+
+    @property
+    def session(self):
+        if not self._session:
+            self._session = self.get_session(self.engine)
+        return self._session
+
+    def insert_list(self, table, data_list):
+        """
+        插入列表
+        :param table:
+        :param data_list:
+        :return:
+        """
+        session = self.get_session(self.engine)
+
+        try:
+            session.execute(table.__table__.insert(), data_list)
+        except Exception as e:
+            log.e('DataBase INSERT ERROR error:{} data:{}'.format(e, data_list))
+        finally:
+            session.close()
+
+    def insert_list_for_one(self, table, data_list):
+        """
+        插入列表
+        :param table:
+        :param data_list:
+        :return:
+        """
+        session = self.get_session(self.engine)
+        try:
+            for item in data_list:
+                session.execute(table.__table__.insert(), [item])
+        except Exception as e:
+            log.e('DataBase INSERT ERROR error:{} data:{}'.format(e, data_list))
+        finally:
+            session.close()
+
+    def insert_single(self, table, data):
+        """
+        插入单条
+        :param table:
+        :param data:
+        :return:
+        """
+        session = self.get_session(self.engine)
+        try:
+            session.execute(table.__table__.insert(), [data])
+        except Exception as e:
+            log.e('DataBase INSERT ERROR error:{} data:{}'.format(e, data))
+        finally:
+            session.close()
+
+    @staticmethod
+    def generate_sql(name, fileters, fields='', limit=None):
+        assert isinstance(fileters, (dict, str))
+
+        if isinstance(fileters, str):
+            return str
+
+        where_sql = StoreDBBase.where_sql(fileters)
+
+        # 无条件默认查询1000条
+        # if not where_sql.strip() and not limit:
+        #     limit = 1000
+
+        if fields and isinstance(fields, list):
+            fields = ','.join(fields)
+        else:
+            fields = '*'
+
+        _sql = u'select {fields} from {table} {where_sql}'.format(fields=fields, table=name, where_sql=where_sql)
+        print('sql', _sql)
+        if limit and isinstance(limit, int):
+            _sql = '{sql} limit {num}'.format(sql=_sql, num=limit)
+        return _sql
+
+    @staticmethod
+    def where_sql(filters):
+        if not isinstance(filters, dict):
+            return ''
+
+        where_param = []
+        for q in filters.keys():
+            # 区间查询
+            if isinstance(filters[q], dict):
+                operate_sql = StoreDBBase.operate_sql(filters[q], q)
+                where_param.append(' ({}) '.format(operate_sql))
+            # 定值查询
+            elif isinstance(filters[q], (str,)):
+                where_param.append(u" ({field} {operate} '{value}') ".format(field=q, operate='=', value=filters[q]))
+            elif isinstance(filters[q], (int,)):
+                where_param.append(u" ({field} {operate} {value}) ".format(field=q, operate='=', value=filters[q]))
+            else:
+                continue
+        _where_sql = u' and '.join(where_param)
+        return u' {} {} '.format(u' where ' if _where_sql else u'', _where_sql)
+
+    @staticmethod
+    def operate_sql(intervals, q):
+        # todo 根据表，自动转换字段类型
+        result = []
+        for item in intervals.keys():
+            # todo python3 无long
+            if isinstance(intervals[item], (int, float)):
+                value = '{}'.format(intervals[item])
+            elif isinstance(intervals[item], str):
+                value = " '{}' ".format(intervals[item])
+            # 时间戳转换
+            elif isinstance(intervals[item], list):
+                value = "( '{}')".format("','".join(intervals[item]))
+            else:
+                value = str(intervals[item])
+                value = " '{}' ".format(value)
+
+            result.append(' {field} {operate} {value} '.format(field=q, operate=item, value=value))
+        return ' and '.join(result)
+
+    def _query(self, table_name, q):
+        sql = self.generate_sql(table_name, q)
+        log.i('query sql:{}'.format(str(sql)))
+        session = self.get_session(self.engine)
+
+    def query(self, table_name, q, fields):
+        sql = self.generate_sql(table_name, q, fields)
+        log.i('query sql:{}'.format(str(sql)))
+        session = self.get_session(self.engine)
+        cursor = session.execute(sql)
+        try:
+            fields = cursor._metadata.keys
+            return [dict(zip(fields, item)) for item in cursor.fetchall()]
+        finally:
+            cursor.close()
+            session.close()
+
+    def partial_query(self, table_name, q):
+        sql = self.generate_sql(table_name, q)
+        log.i('partial_query sql:{}'.format(str(sql)))
+        session = self.get_session(self.engine)
+        cursor = session.execute(sql)
+        try:
+            fields = cursor._metadata.keys
+            while True:
+                data = cursor.fetchmany(1000)
+                if not data:
+                    return
+                yield [dict(zip(fields, item)) for item in data]
+        finally:
+            cursor.close()
+            session.close()
+
+    def fields(self, table_name):
+        sql = self.generate_sql(table_name, {}, limit=1)
+        log.i('query sql:{}'.format(str(sql)))
+        session = self.get_session(self.engine)
+        cursor = session.execute(sql)
+        try:
+            fields = cursor._metadata.keys
+            return fields
+        finally:
+            cursor.close()
+            session.close()
+
+    def execute(self, sql):
+        session = self.get_session(self.engine)
+        try:
+            # self.logger.info('execute sql:{}'.format(str(sql)))
+            session.execute(sql)
+        except:
+            log.e("execute sql error:{}".format(traceback.format_exc()))
+        finally:
+            session.close()
+
+
+class MySql(StoreDBBase):
+    def __init__(self, host='127.0.0.1', port='8123', user='default', pwd='123456', db="micr_pay", **kwargs):
+        '''初始化构造函数'''
+        cfg = tp_cfg()
+        if 'mysql' != cfg.database.type:
+            raise Exception("Configure the mysql type!")
+        super(MySql, self).__init__(cfg.database.mysql_host, cfg.database.mysql_port, cfg.database.mysql_user,
+                                    cfg.database.mysql_password, cfg.database.mysql_db)
+
+    def connection_str(self, conf):
+        connection = 'mysql+pymysql://{user}:{password}@{server_host}:{port}/{db}'.format(**conf)
+        print(connection)
+        return connection
+
+    def get_session(self, engine):
+        Session = sessionmaker(bind=engine)
+        session = Session()
+        return session
+
+
+def get_plugin_db():
+    """
+    :rtype : TPDatabase
+    """
+    if '__plugin_db__' not in builtins.__dict__:
+        builtins.__dict__['__plugin_db__'] = MySql()
+    return builtins.__dict__['__plugin_db__']

server/www/teleport/webroot/app/controller/__init__.py

@@ -287,9 +287,13 @@ controllers = [
     (r'/plugin/host_list', plugin.GetHostListHandler),
     # 服务器详情接口
     (r'/plugin/host_info', plugin.GetHostInfoHandler),
+    # 空闲主机接口
+    (r'/plugin/free_host', plugin.FreeHostHandler),
     # 服务器远程信息
     (r'/plugin/session_info', plugin.GetSessionInfoHandler),
     # 绑定支付宝账户
+    (r'/plugin/bind_list', plugin.BindAccountListHandler),
+    # 绑定支付宝账户
     (r'/plugin/bind_pay_account', plugin.BindPayAccountHandler),
     # 账户状态
     (r'/plugin/account_status', plugin.AccountStatusHandler),
@@ -305,6 +309,10 @@ controllers = [
     (r'/plugin/password_update', plugin.PasswordUpdateHandler),
     # 下载地址获取接口
     (r'/plugin/bat_download_url', plugin.BatDownloadHandler),
+    # token 获取
+    (r'/plugin/token', plugin.TokenHandler),
+    # 状态推送接口
+    (r'/plugin/push_status', plugin.PushStatusHandler),
     # websocket for real-time information
     (r'/plugin/exe_download_url', plugin.ExeDownloadHandler),
     # ws-client call 'http://ip:7190/ws/action/'

server/www/teleport/webroot/app/controller/plugin.py

@@ -21,11 +21,12 @@ from app.base.controller import TPBaseHandler
 from app.base.logger import *
 from app.const import *
 from app.model import plugin
-from app.model.plugin import free_host, host_id_lock
+from app.model.plugin import host_id_lock
 from app.base.configs import tp_cfg
 from pyDes import des, CBC, PAD_PKCS5
 import math
 
+from base.plugin import get_plugin_db
 from controller import reqparse
 
 
@@ -112,6 +113,13 @@ def execute_add_proxy(ip):
     os.system('nohup /usr/bin/python3 {}/clash_refresh.py --ip {} &'.format(path, ip))
 
 
+def execute_linux_docker_script(ip, usename, password, site, webname, sub_domain):
+    path = os.path.dirname(os.path.abspath(__file__))
+    cmd = "nohup /usr/bin/python3 {}/script.py --ip {} -u {} -p {}>> /auto.out &".format(path, ip, usename, password)
+    os.system(cmd)
+    return cmd
+
+
 class TPBasePluginHandler(TPBaseHandler):
     """
     所有返回JSON数据的控制器均从本类继承，返回的数据格式一律包含三个字段：code/msg/data
@@ -158,7 +166,10 @@ class TPBasePluginHandler(TPBaseHandler):
         try:
             return json_decode(self.request.body)
         except:
-            raise tornado.web.HTTPError(500)
+            try:
+                return {k: self.request.arguments[k][0].decode() for k in self.request.arguments.keys()}
+            except:
+                raise tornado.web.HTTPError(500)
 
     def check_ip(self, props):
         ip = props.get("ip", "")
@@ -213,6 +224,14 @@ class TPBasePluginHandler(TPBaseHandler):
     def req_parse(self, required: tuple = ()):
         pass
 
+    @staticmethod
+    def plugin_query(table, q, fields):
+        return get_plugin_db().query(table, q, fields)
+
+    @staticmethod
+    def plugin_execute(sql):
+        return get_plugin_db().execute(sql)
+
 
 class UpdateHostHandler(TPBasePluginHandler):
     def generate_assets_num(self, ip, os_type):
@@ -418,25 +437,51 @@ class GetHostInfoHandler(TPBasePluginHandler):
             args = self.req_parse()
             host_id, mch_no = args.host_id, args.mch_no
 
-            err, host_info = plugin.get_account_info(host_id=host_id, mch_no=mch_no)
-            if err != TPE_OK:
+            q = {}
+            if host_id:
+                q.update({"id": host_id})
+            elif mch_no:
+                q.update({"mch_no": mch_no})
+            else:
+                return self.finish_json("1002", msg="传入参数异常")
+
+            data = self.plugin_query("tp_remote_account_host_bind", q,
+                                     ['id', 'mch_no', 'comp_id', 'host_id', 'host_assigned', 'account_source',
+                                      'account', 'login_status', 'mch_name', 'create_time'])
+            if not data:
                 if mch_no and not host_id:
                     # 商家平台 通过商户请求
-                    self.finish_json(0, data=[])
+                    return self.finish_json(0, data=[])
                 else:
-                    self.finish_json("1004", msg="设备信息获取异常")
-                return
-
-            host_info['password'] = '*******'
-            host_info['desc'] = host_info.pop("remark", "")
-            host_info['host_id'] = host_info.pop("id", 0)
-
-            self.finish_json(0, data=[host_info])
+                    return self.finish_json("1004", msg="设备信息获取异常")
+
+            # mch_no 是否存在多个绑定
+            host_id = host_id or data[0]['host_id']
+            bind_data = self.plugin_query("tp_remote_host", {"id": host_id},
+                                          ['id', 'assets_num', 'os_type', 'ip', 'status', 'username', 'password',
+                                           'name', 'remark', 'bind_status', 'bind_remark'])
+            result = bind_data[0]
+            result['bind'] = data
+            result.pop("password", None)
+            result['desc'] = result.pop("remark", "")
+            self.finish_json(0, data=[result])
         except:
             info = traceback.format_exc()
             log.e("设备详情，异常信息：{}".format(info))
 
 
+class FreeHostHandler(TPBasePluginHandler):
+    async def get(self):
+        sql = """select a.id as id
+  from tp_remote_host a LEFT join tp_remote_account_host_bind b on a.id = b.host_id
+  	where b.id is null and os_type={}""".format(2)
+        data = self.plugin_execute(sql)
+        if data:
+            return self.finish_json(0, data=[data[0]])
+        else:
+            return self.finish_json(1001, msg="无空闲主机")
+
+
 class GetSessionInfoHandler(TPBasePluginHandler):
     def req_parse(self, required: tuple = ()):
         self.parse = reqparse.RequestParser()
@@ -518,6 +563,31 @@ class GetSessionInfoHandler(TPBasePluginHandler):
         return
 
 
+class BindAccountListHandler(TPBasePluginHandler):
+    def req_parse(self, required: tuple = ()):
+        self.parse = reqparse.RequestParser()
+        self.parse.add_argument("host_id", type=int, help='', required=True, )
+        self.parse.add_argument("account", type=str, help='', )
+        self.parse.add_argument("account_source", type=int, help='', )
+        args = self.parse.parse_args(req=self.get_payload())
+        # todo
+        # if not self.check_ip(props):
+        #     self.finish_json(1001, "IP不符合规范")
+        #     return
+        return args
+
+    async def post(self):
+        args = self.req_parse()
+        host_id, account, account_source, = args.host_id, args.account, args.account_source
+        q = {"host_id": host_id}
+        if account:
+            q.update({"account": {"like": "%{}%".format(account)}})
+        if account_source:
+            q.update({"account_source": account_source})
+        data = self.plugin_query("tp_remote_account_host_bind", q, [])
+        self.finish_json(0, data=data)
+
+
 class BindPayAccountHandler(TPBasePluginHandler):
 
     async def add_members(self, biz_id, host_id):
@@ -555,16 +625,12 @@ class BindPayAccountHandler(TPBasePluginHandler):
     def req_parse(self, required: tuple = ()):
         self.parse = reqparse.RequestParser()
         self.parse.add_argument("comp_id", type=int, required=True, help='', )
-        self.parse.add_argument("biz_id", type=int, help='', required=True, default=1)
-        self.parse.add_argument("host_assigned", type=int, help='', )
-        self.parse.add_argument("host_id", type=int, help='', )
-
-        self.parse.add_argument("ip", type=str, help='', )
-        self.parse.add_argument("username", type=str, help='', )
-        self.parse.add_argument("host_password", type=str, help='', )
-        self.parse.add_argument("account", type=str, help='', )
-        self.parse.add_argument("password", type=str, help='', )
+        self.parse.add_argument("host_id", type=int, required=True, help='', )
         self.parse.add_argument("mch_no", type=str, help='', )
+        self.parse.add_argument("biz_id", type=int, help='', )
+        self.parse.add_argument("account", type=str, required=True, help='', )
+        self.parse.add_argument("password", type=str, required=True, help='', )
+        # 绑定账户类型
 
         args = self.parse.parse_args(req=self.get_payload())
         # todo
@@ -575,60 +641,16 @@ class BindPayAccountHandler(TPBasePluginHandler):
 
     async def post(self):
         args = self.req_parse()
-        mch_no, ip, host_id, account, password, username, host_password, host_assigned, comp_id, biz_id = args.mch_no, args.ip, args.host_id, args.account, args.password, args.username, args.host_password, args.host_assigned, args.comp_id, args.biz_id
-        host_password = host_password.replace("*", "")
-
-        if host_id and not account:
-            self.finish_json(1001, "缺少必填项")
-            return
-
-        if not host_id and not host_assigned and (not ip or not username):
-            self.finish_json(1001, "手动分配，缺少必填项")
-            return
-
-        if not host_id:
-            if int(host_assigned) == 1:
-                # 查询已分配过的主机
-                host_id = self.query("remote_account_host_bind", "host_id", {"mch_no": mch_no})
-                # 自动分配功能
-                if not host_id:
-                    err, info = free_host()
-                    if err == TPE_OK:
-                        host_id = info
-                    else:
-                        self.finish_json(1010, "未发现空闲主机")
-                        return
-            else:
-                is_insert = False
-                if host_password:
-                    data = {"ip": ip, "os_type": 1, "username": username, "password": host_password}
-                    url = "http://127.0.0.1:7190/plugin/update_host"
-                    resp = await self.request_api(url, json=data)
-                    host_id = resp.get("data", 0)
-                    is_insert = True if host_id else False
-                host_id = host_id or self.query("host", 'id', {"ip": ip})
-                if not is_insert:
-                    data = {"ip": ip, "os_type": 1, "username": username, "password": host_password, "host_id": host_id}
-                    url = "http://127.0.0.1:7190/plugin/update_host"
-                    resp = await self.request_api(url, json=data, method="put")
-                    # host_id = resp.get("data", 0)
-
-        # todo 未監聽主機不允許綁定支付寶
-        if not host_id:
-            self.finish_json(1010, "无法找到对应主机信息")
-            return
+        mch_no, host_id, account, password, comp_id, biz_id = args.mch_no, args.host_id, args.account, args.password, args.comp_id, args.biz_id
 
         status = self.query("remote_host", "status", {"id": host_id})
 
         id = self.query("remote_account_host_bind", "id", {"host_id": host_id, "mch_no": mch_no})
 
         des_password = des_encrypt(password).decode()
-        args = {"mch_no": mch_no, "comp_id": comp_id, "host_id": host_id, "host_assigned": host_assigned,
-                "account_source": 1, "account": account, "login_status": 0, "mch_name": "",
-                "biz_id": biz_id, "ip": ip, "host_password": host_password, "username": username}
+        args = {"mch_no": mch_no, "comp_id": comp_id, "host_id": host_id, "host_assigned": 0,
+                "account_source": 1, "account": account, "login_status": 0, "mch_name": "", "password": des_password}
 
-        log.i("id:{}".format(id))
-        # xiugai
         if id:
             old_account = self.query('remote_account_host_bind', "account", {"id": id})
             if account and old_account and account != old_account:
@@ -674,6 +696,8 @@ class BindPayAccountHandler(TPBasePluginHandler):
             return
 
         await self.add_members(biz_id, host_id)
+        # todo 添加容器初始化功能
+        # execute_linux_docker_script()
 
 
 class AccountStatusHandler(TPBasePluginHandler):
@@ -954,3 +978,64 @@ class ExeDownloadHandler(TPBasePluginHandler):
         except:
             info = traceback.format_exc()
             log.e("设备详情，异常信息：{}".format(info))
+
+
+class TokenHandler(TPBasePluginHandler):
+    def req_parse(self, required: tuple = ()):
+        self.parse = reqparse.RequestParser()
+        self.parse.add_argument("ip", type=str, help='请求ip', required=True, trim=True)
+        args = self.parse.parse_args(req=self.get_payload())
+        return args
+
+    async def post(self):
+        try:
+            # todo 限制访问ip
+            args = self.req_parse()
+            agent = self.request.headers.get("User-agent")
+            log.i("token md5:{} \n".format(agent + args.ip))
+            token = md5(agent + args.ip)
+            self.finish_json(0, "成功", data=[{"token": token}])
+        except:
+            info = traceback.format_exc()
+            log.e("设备详情，异常信息：{}".format(info))
+
+
+class PushStatusHandler(TPBasePluginHandler):
+    def req_parse(self, required: tuple = ()):
+        self.parse = reqparse.RequestParser()
+        self.parse.add_argument("module", type=str, help='推送模块', required=True, trim=True)
+        self.parse.add_argument("unique_id", type=int, help='请求模块', required=True, )
+        self.parse.add_argument("status", type=int, help='请求模块', default=0)
+        args = self.parse.parse_args(req=self.get_payload())
+        return args
+
+    async def post(self):
+        try:
+            args = self.req_parse()
+            module, unique_id, status = args.module, args.unique_id, args.status
+            if module == "linux_docker":
+                # todo 存储容器状态
+                self.finish_json(0, "成功", data=[])
+        except:
+            info = traceback.format_exc()
+            log.e("设备详情，异常信息：{}".format(info))
+
+# class PushStatusHandler(TPBasePluginHandler):
+#     def req_parse(self, required: tuple = ()):
+#         self.parse = reqparse.RequestParser()
+#         self.parse.add_argument("module", type=str, help='推送模块', required=True, trim=True)
+#         self.parse.add_argument("unique_id", type=int, help='请求模块', required=True, )
+#         self.parse.add_argument("status", type=int, help='请求模块', default=0)
+#         args = self.parse.parse_args(req=self.get_payload())
+#         return args
+#
+#     async def post(self):
+#         try:
+#             args = self.req_parse()
+#             module, unique_id, status = args.module, args.unique_id, args.status
+#             if module == "linux_docker":
+#                 # todo 存储容器状态
+#                 self.finish_json(0, "成功", data=[])
+#         except:
+#             info = traceback.format_exc()
+#             log.e("设备详情，异常信息：{}".format(info))

server/www/teleport/webroot/app/model/plugin.py

 # coding: utf-8
+import builtins
 import datetime
+import traceback
+from abc import ABCMeta, abstractmethod
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
 
 from app.base.logger import log
 from app.base.db import get_db, SQL
 from app.const import TPE_EXISTS, TPE_DATABASE, TPE_OK, TP_GROUP_HOST, TPE_NOT_EXISTS, TPE_FAILED
+from base.configs import tp_cfg
 
 host_id_lock = {}
 
@@ -85,52 +92,6 @@ def get_shop_bind_info(host_id):
     return bind
 
 
-def get_account_info(host_id, mch_no):
-    s = SQL(get_db())
-    s.select_from('remote_account_host_bind',
-                  ['id', 'mch_no', 'comp_id', 'host_id', 'host_assigned', 'account_source', 'account',
-                   'login_status', 'mch_name', 'create_time'], alt_name='a')
-
-    # 判断
-    if mch_no:
-        s.where('a.mch_no="{}"'.format(mch_no))
-    elif host_id:
-        s.where('a.host_id={}'.format(host_id))
-
-    err = s.query()
-    if err != TPE_OK:
-        return err, None
-
-    if s.recorder:
-        host_id = host_id or s.recorder[0].host_id
-
-    # 未发现设备
-    if not host_id:
-        return TPE_DATABASE, None
-
-    sh = SQL(get_db())
-    sh.select_from('remote_host',
-                   ['id', 'assets_num', 'os_type', 'ip', 'status', 'username', 'password', 'name', 'remark', ],
-                   alt_name='h')
-    sh.where('h.id={}'.format(host_id))
-    err = sh.query()
-    if err != TPE_OK:
-        return err, None
-
-    # 使用id查询可能查询多个
-    if len(sh.recorder) != 1:
-        return TPE_DATABASE, None
-
-    result = sh.recorder[0]
-
-    bind = s.recorder
-    for i in bind:
-        i['password'] = '******'
-    result['bind'] = s.recorder
-
-    return TPE_OK, result
-
-
 def free_host():
     db = get_db()
     sql = """select a.id as id