python-fastapi-mm-zhcs-yj-api/routers/api/gateway/__init__.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from re import S
from fastapi import APIRouter, Request, Depends, Form, Body, File, UploadFile, BackgroundTasks
from fastapi.responses import Response
from fastapi.responses import JSONResponse
from database import get_db
from sqlalchemy.orm import Session
from utils import *
from models import *
from urllib import parse
from common.DBgetdata import dbgetdata
from pprint import pprint
from datetime import datetime
# import pandas as pd
from pydantic import BaseModel
import sqlalchemy
import pymysql
import json,time,io
import time, os
import math
import uuid
import re
router = APIRouter()

def contains_special_characters(input_string, special_characters=";|&|$|#|'|\\t|@| "):
    """
    判断字符串是否包含特殊符号。

    :param input_string: 需要检查的字符串
    :param special_characters: 特殊符号的字符串，多个符号用竖线 '|' 分隔
    :return: 如果包含特殊符号返回 True，否则返回 False
    """
    # 创建正则表达式模式
    pattern = re.compile('[' + re.escape(special_characters) + ']')

    # 搜索字符串中的特殊符号
    if pattern.search(input_string):
        return True
    return False





@router.get('/v1/demo')
@router.post('/v1/demo')
async def test1(request: Request):
    data = await request.body()
    body = data.decode(encoding='utf-8')
    # print(body)
    if len(body) > 0:
        body = json.loads(body)
        print(body)
        print([body,{'msg':'good'}])
        return body
    else:
        return body


@router.post('/create_api_service')
async def create_api_service(
    request: Request,
    db: Session = Depends(get_db)
):
    # print(1)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        # print(body)
        body = json.loads(body)
    serviceid = str(uuid.uuid4()).replace("-","")
    serviname = body['servicename']
    datasource = str(uuid.uuid1())
    sqlname = body['sqlname']
    id = str(uuid.uuid4())
    dbtype = body['dbtype']
    if body['dbtype'] == 'pymysql':
        host = body['host']
        user = body['user']
        password = body['password']
        database = body['database']
        port = body['port']
        charset = body['charset']
        create_datasource = DatasourceEntity(id= datasource,name=sqlname,scope=serviceid,dbtype=dbtype,host=host,user=user,password=password,database=database,port=port,charset=charset)
        # db.add()
        # print()
        db.add(create_datasource)
    sqltext = body['sqltext']
    create_command = CommandEntity(id=id,sqltext=sqltext,datasource=datasource,scope=serviceid)
    create_api = ApiServiceEntity(id=serviceid,name=serviname)
    db.add(create_api)
    db.add(create_command)
    db.commit()
    return serviceid


@router.get('/v1/{service_code:path}')
@router.post('/v1/{service_code:path}')
async def v1(request: Request, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    service_code = request.path_params.get('service_code')
    query_list = parse.parse_qsl(str(request.query_params))
    print(query_list)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        print('[body]', body)
        body = json.loads(body)
    # print("1111",body)

    service_info = db.query(ApiServiceEntity).filter(ApiServiceEntity.id == service_code).first()
    if service_info is None:
        return {
            'errcode': 1,
            'errmsg': '服务不存在'
    }
    # print(service_info)
    print('service_name', service_info.name)

    # print('database_name', database_info.name, database_info.dsn)

    command_info = db.query(CommandEntity).filter(CommandEntity.scope == service_code).first()
    if command_info is None:
        return {
            'errcode': 1,
            'errmsg': '查询命令没配置'
    }


    database_info = db.query(DatasourceEntity).filter(DatasourceEntity.id == command_info.datasource).first()
    if database_info is None:
        return {
            'errcode': 1,
            'errmsg': '数据库没配置'
    }



    sql = command_info.sqltext
    print('sql ==== << ', sql)
    meta_data = {}
    # 从query获取参数
    for (key, val) in query_list:
        if key != 'access_token' and key !='page' and key != 'limit' and key !='create_time' and key != 'record_name'and key != 'record_cid':
            sql = sql.replace("{" + key + "}", val.replace('全部',''))
        elif key =='create_time':
            if val != '全部':
                sql = sql.replace("{" + key + "}", ('DATE_FORMAT(create_time,\'%Y-%m-%d\') =\''+val+'\' and')) #DATE_FORMAT(create_time,'%Y-%m-%d') ='{create_time}'
            else:
                sql = sql.replace("{" + key + "}", '')
        elif key =='record_name' or  key =='record_cid' :
            if val != '全部':
                sql = sql.replace("{" + key + "}", (key+'=\''+val+'\' and')) 
            else:
                sql = sql.replace("{" + key + "}", '')
        # elif key =='record_cid':
        #     if val != '全部':
        #         sql = sql.replace("{" + key + "}", ('record_name =\''+val+'\' and')) 
        #     else:
        #         print(111)
        #         sql = sql.replace("{" + key + "}", '')
        elif key =='page' :
            meta_data['page'] = int(val)
        elif key == 'limit':
            meta_data['limit'] = int(val)

    # 从json body获取参数
    if len(body) > 0:
        print(body)
        for (key, val) in body.items():
            if key == 'page1' and 'limit1' in body:
                if val==-1:
                    pass
                if val == '':
                    val = '0'
                page1 = int(val)
                limit1=int(body['limit1'])
                print(str((page1-1)*limit1))
                sql = sql.replace('{page1}',str((page1-1)*limit1))
            # elif key == 'limit1' and 'page1' in body:
            #     page1 = int(body['page1'])
            #     limit1 = int(val)
            #     sql.replace('{page1}', str((page1-1)*limit1))
            elif key !='page' and key != 'limit' and key !='create_time':
                if isinstance(val, int):
                    val = str(val)
                sql = sql.replace("{" + key + "}", val)
            elif key =='create_time':
                if val != '全部':
                    sql = sql.replace("{" + key + "}", ('DATE_FORMAT(create_time,\'%Y-%m-%d\') =\''+val+'\' and')) #DATE_FORMAT(create_time,'%Y-%m-%d') ='{create_time}'
                else:
                    sql = sql.replace("{" + key + "}", '')
            elif key =='page' :
                if isinstance(val, str):
                    meta_data['page'] = int(val)
                elif isinstance(val, int):
                    meta_data['page'] = val
            elif key == 'limit':
                if isinstance(val, str):
                    meta_data['limit'] = int(val)
                elif isinstance(val, int):
                    meta_data['limit'] = val
            elif key == 'taskNo' and val=='':
                print("他们又输入空的taskNo啦！")
                return [{"text":"兄弟，taskNo不能为空"}]
    print('sql ==== >> ', sql)

    data = []
    if database_info.dbtype == 'psycopg2':
        '''
        print(1111)
        conn = psycopg2.connect(database_info.dsn)
        cur = conn.cursor()
        cur.execute(sql)
        rows = cur.fetchall()

        # 字段名列表
        colnames = [desc[0] for desc in cur.description]

        item = {}
        for row in rows:
            for col in range(len(colnames)):
                field_name = colnames[col]
                item[field_name] = row[col]
            data.append(item) 

        conn.close()
        '''
            

    elif database_info.dbtype == 'pymysql':
        # print(database_info)
        conn = pymysql.connect(host=database_info.host,
                               user=database_info.user,
                               password=database_info.password,
                               database=database_info.database,
                               port=database_info.port,
                               charset=database_info.charset)
        cur = conn.cursor()
        cur.execute(sql)
        rows = cur.fetchall()

        # 字段名列表
        colnames = [desc[0] for desc in cur.description]
        # print(colnames)
        # print(rows)
        pages = 1 #总页数
        current = 1 #第几页
        total = len(rows)
        size = len(rows)
        # print(size)
        if 'page' in meta_data and 'limit' in meta_data:
            current = meta_data['page']
            size = meta_data['limit']

            if (current == 0 or size == 0) and total != 0:
                 current = 1
                 size = 5
            pages = total//size
            if total%size !=0:
                pages+=1
    
        start_index = (current-1)*size
        end_index = current*size
        if pages <= current :
            # current = pages
            if total == size :
                end_index = (current-1)*size+total
            elif total%size == 0:
                end_index = current*size
            else:
                end_index = (current-1)*size+total%size
            start_index = (current-1)*size
        if total ==0:
            start_index = end_index =0
        # print(start_index,end_index)
        for row in range(start_index,end_index):
            item = {}
            for col in range(len(colnames)):
                field_name = colnames[col]
                item[field_name] = rows[row][col]
            data.append(item) 
            # print(item)

        # for row in rows:
        #     item = {}
        #     for col in range(len(colnames)):
        #         field_name = colnames[col]
        #         item[field_name] = row[col]
        #     data.append(item) 
            # print(item)
        # print(data)
        conn.close()

    return {
        'code': 0,
        'errcode': 0,
        'errmsg': '查询成功',
        'data': {"list":data,
        'pages':pages, # 总页数
        'currentPage':current, #当前页数
        # 'current':current,
        # 'total' : total,
        'total' : total, # 总数据量
        # 'size':size,
        'pageSize':size #页码
         }
    }
    # import time
    # print(time)
    # if 'ispages' in body:
    #     try:
    #         if body['ispages']=='1':
    #             return {
    #                 'page': [{"total": pages,
    #                          "page": current}],
    #                 "data":data
    #             }
    #             # data['pages'] =
    #     except:
    #         pass
    #
    # #background_tasks.add_task(post_service_method, service_code, body, db)
    #
    # return  data





class CreateApiServiceFrom(BaseModel):
        datasource:str
        name:str
        sqltext:str
        serviceid:str=None

@router.post('/v2/create_api_service')
async def create_api_service_v2(
    form_data: CreateApiServiceFrom,
    request: Request,
    db: Session = Depends(get_db)
):
    # print(1)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        # print(body)
        body = json.loads(body)


    for param in ['datasource','name','sqltext']:
        if param not in body or body[param]=='':
            return f'{param}为必填参数，不能不存在'

    # 数据库信息校验
    datasource = form_data.datasource
    database_info = db.query(DatasourceEntity).filter(DatasourceEntity.id == datasource).first()
    if database_info is None:
        return JSONResponse(status_code=410, content={
            'errcode': 410,
            'errmsg': f'数据库-{datasource}-不存在'
        })

    # 接口信息生成及获取
    if form_data.serviceid is None:
        serviceid = str(uuid.uuid4()).replace("-","")
    else:
        serviceid = form_data.serviceid
    serviname = form_data.name
    id = str(uuid.uuid4())
    sqltext = form_data.sqltext

    # 接口执行校验
    if database_info.dbtype == 'pymysql':
        try:
            conn = pymysql.connect(host=database_info.host,
                                   user=database_info.user,
                                   password=database_info.password,
                                   database=database_info.database,
                                   port=database_info.port,
                                   charset=database_info.charset)
            cur = conn.cursor()

            # 测试返回
            sql = sqltext+' limit 0;'
            print(sql)
            cur.execute(sql)
            colnames = [desc[0] for desc in cur.description]
            cur.close()
            conn.close()

        except Exception as e:
            # 捕获其他所有类型的异常
            return f"未知错误: {e}"
    else:
        return '暂不支持除pymysql以外的驱动'

    create_command = CommandEntity(id=id,sqltext=sqltext,datasource=datasource,scope=serviceid)
    create_api = ApiServiceEntity(id=serviceid,name=serviname,status=1)
    db.add(create_api)
    db.add(create_command)
    db.commit()
    return [serviceid,colnames]



@router.post('/v2/update_api_service')
async def update_api_service_v2(
    request: Request,
    db: Session = Depends(get_db)
):
    # print(1)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        # print(body)
        body = json.loads(body)

    for param in ['id','datasource','scope','name','sqltext','status']:
        if param not in body or body[param]=='':
            return f'{param}为必填参数，不能不存在'

    # 数据库信息校验
    datasource = body['datasource']
    database_info = db.query(DatasourceEntity).filter(DatasourceEntity.id == datasource).first()
    if database_info is None:
        return JSONResponse(status_code=410, content={
            'errcode': 410,
            'errmsg': f'数据库-{datasource}-不存在'
        })



    serviceid = body['scope']
    # 判断请求接口是否存在
    service_info = db.query(ApiServiceEntity).filter(ApiServiceEntity.id == serviceid).first()
    if service_info is None:
       return JSONResponse(status_code=410, content={
           'code': 410,
           'msg': f'servicecode{serviceid}服务不存在'
       })
    command_info = db.query(CommandEntity).filter(CommandEntity.scope == serviceid).first()
    if command_info is None:
        return JSONResponse(status_code=410, content={
            'code': 410,
            'msg': f'servicecode{serviceid}服务不存在'
        })

    # 接口信息生成及获取
    id = body['id']
    serviname = body['name']
    sqltext = body['sqltext']
    status = body['status']

    # 接口执行校验
    if database_info.dbtype == 'pymysql':
        try:
            conn = pymysql.connect(host=database_info.host,
                                   user=database_info.user,
                                   password=database_info.password,
                                   database=database_info.database,
                                   port=database_info.port,
                                   charset=database_info.charset)
            cur = conn.cursor()

            # 测试返回
            sql = sqltext+' limit 0;'
            print(sql)
            cur.execute(sql)
            colnames = [desc[0] for desc in cur.description]
            cur.close()
            conn.close()

        except Exception as e:
            # 捕获其他所有类型的异常
            return f"未知错误: {e}"
    else:
        return '暂不支持除pymysql以外的驱动'

    # create_command = CommandEntity(id=id,sqltext=sqltext,datasource=datasource,scope=serviceid)
    # create_api = ApiServiceEntity(id=serviceid,name=serviname,status=1)
    # db.add(create_api)
    # db.add(create_command)
    service_info.id=serviceid
    service_info.name=serviname
    service_info.status=status

    command_info.id=id
    command_info.sqltext=sqltext
    command_info.datasource=datasource
    command_info.scope=serviceid


    db.commit()
    return [serviceid,colnames]




@router.post('/v2/create_datasource')
async def create_create_datasource_v2(
    request: Request,
    db: Session = Depends(get_db)
):
    # print(1)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        # print(body)
        body = json.loads(body)

    # 检查入参
    for param in ['datasourcename','dbtype','host','user','password','database','port','charset']:
        if param not in body or body[param]=='':
            return f'{param}为必填参数，不能不存在'
    datasource = str(uuid.uuid1())
    datasourcename = body['datasourcename']
    dbtype = body['dbtype']
    if body['dbtype'] == 'pymysql':
        host = body['host']
        user = body['user']
        password = body['password']
        database = body['database']
        port = body['port']
        charset = body['charset']

        # 数据库连通性校验
        try:
            conn = pymysql.connect(host=host,
                                   user=user,
                                   password=password,
                                   database=database,
                                   port=port,
                                   charset=charset)
            cur = conn.cursor()

            # 测试返回
            sql = ' select now()'
            print(sql)
            cur.execute(sql)
            testresult = cur.fetchall()
            cur.close()
            conn.close()

        except Exception as e:
            # 捕获其他所有类型的异常
            return f"未知错误: {e}"
        create_datasource = DatasourceEntity(id=datasource, name=datasourcename, scope='', dbtype=dbtype, host=host,
                                             user=user, password=password, database=database, port=port,
                                             charset=charset)
        db.add(create_datasource)
    else:
        return f'暂不支持{dbtype}'
    db.commit()
    return [datasource,testresult]


@router.post('/v2/update_datasource')
async def create_update_datasource_v2(
    request: Request,
    db: Session = Depends(get_db)
):
    # print(1)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        # print(body)
        body = json.loads(body)

    # 检查入参
    for param in ['id','datasourcename','dbtype','host','user','password','database','port','charset']:
        if param not in body or body[param]=='':
            return f'{param}为必填参数，不能不存在'

    # 根据数据库id检测是否存在
    datasource = body['id']
    database_info = db.query(DatasourceEntity).filter(DatasourceEntity.id == datasource).first()
    if database_info is None:
        return JSONResponse(status_code=410, content={
            'errcode': 410,
            'errmsg': f'数据库{datasource}不存在'
        })
    datasourcename = body['datasourcename']
    dbtype = body['dbtype']
    if body['dbtype'] == 'pymysql':
        host = body['host']
        user = body['user']
        password = body['password']
        database = body['database']
        port = body['port']
        charset = body['charset']

        # 数据库连通性校验
        try:
            conn = pymysql.connect(host=host,
                                   user=user,
                                   password=password,
                                   database=database,
                                   port=port,
                                   charset=charset)
            cur = conn.cursor()

            # 测试返回
            sql = ' select now()'
            print(sql)
            cur.execute(sql)
            testresult = cur.fetchall()
            cur.close()
            conn.close()

        except Exception as e:
            # 捕获其他所有类型的异常
            return f"未知错误: {e}"
        # create_datasource = DatasourceEntity(id=datasource, name=datasourcename, scope='', dbtype=dbtype, host=host,
        #                                      user=user, password=password, database=database, port=port,
        #                                      charset=charset)
        # db.add(create_datasource)
        database_info.name = datasourcename
        database_info.dbtype = dbtype
        database_info.host = host
        database_info.user = user
        database_info.password = password
        database_info.database = database
        database_info.port = port
        database_info.charset = charset

    else:
        return f'暂不支持{dbtype}'
    db.commit()
    return [datasource,testresult]




@router.get('/v2/{serviceid}')
@router.post('/v2/{serviceid}')
# @router.post('/v2/{servicecode}')
async def v2(serviceid:str,request: Request, db: Session = Depends(get_db)):
    # 获取请求头 servicecode
    # if service_code is None:
    # service_code = request.headers.get('servicecode')
    # print(serviceid)
    if serviceid is None:
       return JSONResponse(status_code=410, content= {
           'code': 410,
           'msg': "请求头servicecode未传参"
       })
    # 判断请求接口是否存在
    service_info = db.query(ApiServiceEntity).filter(ApiServiceEntity.id == serviceid).first()
    if service_info is None:
       return JSONResponse(status_code=410, content={
           'code': 410,
           'msg': 'servicecode服务不存在'
       })
    # print(service_code)


    # 判断请求接口对应数据库是否存在
    command_info = db.query(CommandEntity).filter(CommandEntity.scope == serviceid).first()
    if command_info is None:
        return JSONResponse(status_code=410, content={
            'code': 410,
            'msg': '查询命令没配置'
        })

    # 判断请求接口对应数据库是否存在
    database_info = db.query(DatasourceEntity).filter(DatasourceEntity.id == command_info.datasource).first()
    if database_info is None:
        return JSONResponse(status_code=410, content={
            'errcode': 410,
            'errmsg': '数据库没配置'
        })

    # 获取接口对应的sql
    sql = command_info.sqltext
    print('sql ==== << ', sql)

    # 从params和body获取参数
    # query_list = parse.parse_qsl(str(request.query_params))
    # print(query_list)
    data = await request.body()
    body = data.decode(encoding='utf-8')
    if len(body) > 0:
        body = json.loads(body)

        # 分页器 页数和页码的设置
        size = 10
        current = 1
        if "size" in body:
            if isinstance(body['size'], str):
                size = int(body['size'])
            elif isinstance(body['size'], int):
                size = body['size']
            if size >100:
                size = 100

        if "current" in body:
            if isinstance(body['current'], str):
                current = int(body['current'])
            elif isinstance(body['current'], int):
                current = body['current']
            if current<=0:
                current=1
        # 接口sql的参数替换
        if 'query' in body:
            for (key, val) in body['query'].items():
                if isinstance(val, int):
                    val = str(val)
                if contains_special_characters(val):
                    return JSONResponse(status_code=411, content={
                                'code': 411,
                                'msg': f'参数{key}含特殊符号：;、&、$、#、\'、\\t、@、空格等'
                            })
                sql = sql.replace("{" + key + "}", val)


        print('sql ==== >> ', sql)


        data = []

        # 数据库类型为mysql情况下
        if database_info.dbtype == 'pymysql':
            # 数据库连接

            conn = pymysql.connect(host=database_info.host,
                                   user=database_info.user,
                                   password=database_info.password,
                                   database=database_info.database,
                                   port=database_info.port,
                                   charset=database_info.charset)
            cur = conn.cursor()

            # 查总数据量，分页数据处理
            totalsql = f'select count(*) from ({sql})t'
            print(totalsql)
            cur.execute(totalsql)
            total = cur.fetchone()[0]

            pages,pagesmod = divmod(total, size)
            print(total,pages,pagesmod)
            if pagesmod!=0:
                pages+=1
            print(pages,pagesmod)
            if total <size :
                size = total


            # 正式查询
            sql = sql+f' limit {size*(current-1)}, {size};'
            print(sql,size)
            cur.execute(sql)
            rows = cur.fetchall()
            colnames = [desc[0] for desc in cur.description]
            for row in range(size):
                item = {}
                for col in range(len(colnames)):
                    field_name = colnames[col]
                    item[field_name] = rows[row][col]
                data.append(item)

            # 数据库关闭
            cur.close()
            conn.close()



        else:
            return JSONResponse(status_code=410, content={
            'code': 410,
            'msg': '接口对应数据库暂不支持'
        })

        return {
            'code': 0,
            'msg': 'success',
            'data': {"list": data,
                     'pages': pages,  # 总页数
                     'currentPage': current,  # 当前页数
                     # 'current':current,
                     # 'total' : total,
                     'total': total,  # 总数据量
                     # 'size':size,
                     'pageSize': size  # 页码
                     }
        }

    else:
        return JSONResponse(status_code=410, content={
            'code': 410,
            'msg': 'body不能为空'
        })