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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from fastapi import APIRouter, Request, Depends,Response,HTTPException,status,Query
from fastapi.responses import StreamingResponse
from fastapi.responses import JSONResponse
from cachetools.keys import hashkey
from sqlalchemy.orm import Session
from typing import List, Optional
from cachetools import LRUCache
from pydantic import BaseModel
from database import get_db
from urllib import parse
from models import *
import requests
import hashlib
import random
import string
import json
import time
import io


router = APIRouter()

def GetSign(signTime,nonce,passtoken):
    data = signTime+passtoken+nonce+signTime
    return hashlib.sha256(data.encode('utf-8')).hexdigest()

def GetNonce(length):
    characters = string.ascii_letters + string.digits
    # 随机选择字符集的长度个字符
    random_string = ''.join(random.choice(characters) for _ in range(length))
    return random_string

def GetTime():
    return int(time.time()*1000)


# 设置缓存大小，例如 100 个缓存项
cache_size = 10000
cache = LRUCache(maxsize=cache_size)

# 缓存过期时间，例如 10 分钟
cache.ttl = 86400  # ttl(Time to Live) 以秒为单位



@router.post('/proxyHandler/{city_code:path}/{service_code:path}')
@router.get('/proxyHandler/{city_code:path}/{service_code:path}')
async def mine(request: Request,db: Session = Depends(get_db)):

    target = str(request.url) # 获取接口地址

    cache_key = hashkey(target)

    cached_response = cache.get(cache_key)

    if cached_response:
        if 'image/png' in cached_response.headers.get('Content-Type', ''):
            return StreamingResponse(content=io.BytesIO(cached_response.content), media_type='image/png')
        return cached_response
    # 获取接口id 判断接口是否存在
    service_code = request.path_params.get('service_code')
    service_info = db.query(OneShareApiEntity).filter(OneShareApiEntity.servercode==service_code).first()
    if service_info is None:
        return JSONResponse(status_code=410, content={
           'code': 410,
           'msg': f'server_code{service_code}服务不存在'
       })
    city_code = request.path_params.get('city_code')
    if city_code == "mm":
        url = 'https://19.155.242.125/GatewayMsg/http/api/proxy/invoke'
    elif city_code == "gd":
        url = 'https://19.15.75.180:8581/GatewayMsg/http/api/proxy/invoke'
    else:
        return JSONResponse(status_code=410, content={
           'code': 410,
           'msg': f'city_code{city_code}不存在'
       })

    # 获取请求方式
    method = request.method

    # 获取请求体
    body = await request.body()
    body = body.decode(encoding='utf-8')
    if len(body) > 0:
        body = json.loads(body)

    # 获取默认params 1
    params_default = service_info.params_default
    if len(params_default)>0:
        print(params_default)
        params_default = json.loads(params_default)


    # 获取params
    query_list = parse.parse_qsl(str(request.query_params))
    params = {}
    for (key, val) in query_list:
        params[key]=val
    params_default.update(params)
    params = params_default

    # 生成请求头主体
    signTime = str(GetTime()//1000)
    nonce = GetNonce(5)
    sign = GetSign(signTime,nonce,service_info.passtoken)

    # 初始请求头
    headers = {
        # 'Content-Type': 'application/json',
        'x-tif-signature': sign,
        'x-tif-timestamp': signTime,
        'x-tif-nonce': nonce,
        'x-tif-paasid': service_info.passid,
        'x-tif-serviceId': service_code
    }

    # 加入默认请求头
    headers_default = service_info.headers_default
    if len(headers_default)>0:
        headers_default = json.loads(headers_default)
    headers.update(headers_default)

    # 判断接口类型
    # 1 普通接口 请求头请求体用默认，前端传输的请求体嵌入到请求体query中
    # 2 地图接口
    # 3 自定义接口
    if service_info.servertype == 1:
        query_timestamp = str(GetTime())
        data = {
        "system_id": service_info.passid,
        "vender_id": 'xx',
        "department_id": 'xx',
        "query_timestamp": query_timestamp,
        "UID": GetNonce(5),
        "query": body,
        "audit_info": {
            "operator_id": 'xx',
            "operator_name": 'xx',
            "query_object_id": 'xx',
            "query_object_id_type": 'xx',
            "item_id": 'xx',
            "item_code": 'xx',
            "item_sequence": 'xx',
            "terminal_info": 'xx',
            "query_timestamp": query_timestamp
        }
        }
        body=data
    else:
        body_default = service_info.body_default
        if len(body_default)>0:
            body_default = json.loads(body_default)
        body_default.update(body)


    # 根据请求方式请求获取数据
    if method == "GET":
        response = requests.get(url=url,params=params,headers=headers,json=body,verify=False)
    elif method == "POST":
        response = requests.post(url=url,params=params,headers=headers,json=body,verify=False)
    else:
        return JSONResponse(status_code=410, content={
           'code': 410,
           'msg': f'请求方式{method}不支持'
       })

    # 根据响应头数据类型返回对应数据类型
    content_type = response.headers.get('Content-Type', '')

    if 'application/json' in content_type:
        return JSONResponse(content=response.json(), media_type='application/json',status_code=response.status_code)
    elif 'application/xml' in content_type or 'text/xml' in content_type:
        return Response(content=response.text, media_type='application/xml',status_code=response.status_code)
    elif 'text/html' in content_type:

        return Response(content=response.text, media_type='application/html',status_code=response.status_code)
    elif 'image/png' in content_type:
        cache[cache_key] = response
        return StreamingResponse(content=io.BytesIO(response.content), media_type='image/png')
    # 可以继续添加更多的条件分支来处理其他类型
    else:
        return Response(content=response.text, media_type=content_type)


class OneShareApiCreateForm(BaseModel):
    passid: str
    passtoken: str
    servercode: str
    servertype: int
    params_default: str = ""
    body_default: str = ""
    headers_default: str = ""
    servername: str
@router.post("/create")
async def create_one_share_api(
    api_data: OneShareApiCreateForm,
    db: Session = Depends(get_db)
):
    try:
        # 创建一个新的 OneShareApiEntity 实例
        new_api = OneShareApiEntity(
            passid=api_data.passid,
            passtoken=api_data.passtoken,
            servercode=api_data.servercode,
            servertype=api_data.servertype,
            params_default=api_data.params_default,
            body_default=api_data.body_default,
            headers_default=api_data.headers_default,
            servername=api_data.servername
        )

        # 添加到数据库会话并提交
        db.add(new_api)
        db.commit()
        db.refresh(new_api)  # 刷新实例以包含新的 ID 等信息

        # 构建并返回响应
        return {
            "code": 200,
            "msg": "操作成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

class OneShareApiUpdateForm(BaseModel):
    passid: str = None
    passtoken: str = None
    servercode: str = None
    servertype: int = None
    params_default: str = None
    body_default: str = None
    headers_default: str = None
    servername: str = None

@router.put("/update/{id}")  # 或者使用 @app.patch 如果你只想更新部分字段
async def update_one_share_api(
    id: int,
    api_data: OneShareApiUpdateForm,
    db: Session = Depends(get_db)
):
    try:
        # 从数据库中获取现有的 OneShareApiEntity 实例
        api = db.query(OneShareApiEntity).filter(OneShareApiEntity.id == id).first()
        if not api:
            raise HTTPException(status_code=404, detail="接口不存在")

        # 更新字段
        if api_data.passid:
            api.passid = api_data.passid
        if api_data.passtoken:
            api.passtoken = api_data.passtoken
        if api_data.servercode:
            api.servercode = api_data.servercode
        if api_data.servertype:
            api.servertype = api_data.servertype
        if api_data.params_default:
            api.params_default = api_data.params_default
        if api_data.body_default:
            api.body_default = api_data.body_default
        if api_data.headers_default:
            api.headers_default = api_data.headers_default
        if api_data.servername:
            api.servername = api_data.servername

        # 更新时间
        api.update_time = datetime.now()

        # 提交更改
        db.commit()

        # 构建并返回响应
        return {
            "code": 200,
            "msg": "操作成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

@router.delete("/delete/{id}")  # 使用 ID 来标识要删除的接口
async def delete_one_share_api(
    id: int,
    db: Session = Depends(get_db)
):
    # try:
        # 从数据库中获取要删除的 OneShareApiEntity 实例
    api = db.query(OneShareApiEntity).filter(OneShareApiEntity.id == id).first()
    if api is None:
        raise HTTPException(status_code=404, detail="接口不存在")

    # 删除实例
    db.delete(api)
    db.commit()

    # 构建并返回响应
    return {
        "code": 200,
        "msg": "操作成功",
        "data": None
    }
    # except Exception as e:
    #     # 处理异常
    #     print(e)
    #     raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

class OneShareApiOut(BaseModel):
    id: int
    passid: str
    passtoken: str
    servercode: str
    servertype: int
    params_default: str
    body_default: str
    headers_default: str
    servername: str
    create_time: str
    update_time: str

# 定义查询接口的路由
@router.get("/list")
async def get_one_share_apis(
    page: int = Query(1, gt=0),
    page_size: int = Query(10, gt=0),
    servername: Optional[str] = Query(None),
    servertype:int = Query(None),
    servercode: Optional[str] = Query(None),
    db: Session = Depends(get_db)
):
    try:
        # 构建查询
        query = db.query(OneShareApiEntity)

        # 应用查询参数
        if servername:
            query = query.filter(OneShareApiEntity.servername.contains(servername))
        if servercode:
            query = query.filter(OneShareApiEntity.servercode.contains(servercode))
        if servertype:
            query = query.filter(OneShareApiEntity.servertype==servertype)

        # 获取总记录数
        total_count = query.count()

        # 执行分页查询
        items = query.offset((page - 1) * page_size).limit(page_size).all()

        # 将查询结果转换为 Pydantic 模型列表
        apis_out = [
            OneShareApiOut(
                id=item.id,
                passid=item.passid,
                passtoken=item.passtoken,
                servercode=item.servercode,
                servertype=item.servertype,
                params_default=item.params_default,
                body_default=item.body_default,
                headers_default=item.headers_default,
                servername=item.servername,
                create_time=item.create_time.strftime('%Y-%m-%d %H:%M:%S'),
                update_time=item.update_time.strftime('%Y-%m-%d %H:%M:%S')
            ) for item in items
        ]

        # 构建并返回响应
        return {
            "code": 200,
            "msg": "查询成功",
            "data": {
                "total": total_count,
                "list": apis_out
            }
        }
    except Exception as e:
        # 处理异常
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))