增加公开api

zhaoqingang

2024-12-04 cdb5f8bf45f117831959291c89e0694606ebb479

增加公开api

 app/api/auth.py

@@ -1,18 +1,23 @@
import json

from Tools.scripts.mailerdaemon import emparse_list_from
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession

from Log import logger
from app.api import Response, pwd_context, get_current_user
from app.config.config import settings
from app.config.const import RAGFLOW, BISHENG, DIFY
from app.models.app_token_model import AppToken
from app.models.base_model import get_db
from app.models.postgresql_base_model import get_pdb
from app.models.token_model import upsert_token, get_token
from app.models.token_model import upsert_token, get_token, update_token
from app.models.user import UserCreate, LoginData
from app.models.user_model import UserModel
from app.service.auth import authenticate_user, create_access_token
from app.service.bisheng import BishengService
from app.service.common.app_register import AppRegisterDao
from app.service.ragflow import RagflowService
from sqlalchemy.future import select

@@ -102,30 +107,32 @@
    user = authenticate_user(db, login_data.username, login_data.password)
    if not user:
        return Response(code=400, msg="Incorrect username or password")

    bisheng_service = BishengService(settings.sgb_base_url)
    ragflow_service = RagflowService(settings.fwr_base_url)

    # 登录到毕昇
    try:
        bisheng_token = await bisheng_service.login(login_data.username, login_data.password)
    except Exception as e:
        return Response(code=500, msg=f"Failed to login with Bisheng: {str(e)}")

    # 登录到ragflow
    try:
        ragflow_token = await ragflow_service.login(login_data.username, login_data.password)
    except Exception as e:
        return Response(code=500, msg=f"Failed to login with Ragflow: {str(e)}")
    app_register = AppRegisterDao(db).get_apps()
    token_dict = {}
    for app in app_register:
        if app["id"] == RAGFLOW:
            service = RagflowService(settings.fwr_base_url)
        elif app["id"] == BISHENG:
            service = BishengService(settings.sgb_base_url)
        elif app["id"] == DIFY:
            continue
        else:
            logger.error("未知注册应用---")
            continue
        try:
            token = await service.login(login_data.username, login_data.password)
            token_dict[app["id"]] = token
        except Exception as e:
            return Response(code=500, msg=f"Failed to login with {app['id']}: {str(e)}")

    # 创建本地token
    access_token = create_access_token(data={"sub": user.username, "user_id": user.id})

    upsert_token(db, user.id, access_token, bisheng_token, ragflow_token)
    await update_token(db, user.id, access_token, token_dict)
    result = await pdb.execute(select(AppToken).where(AppToken.id == user.id))
    db_app_token = result.scalars().first()
    if not db_app_token:
        app_token_str = json.dumps({"rag_token": ragflow_token, "bs_token":bisheng_token})
        app_token_str = json.dumps(token_dict)
        # print(app_token_str)
        app_token = AppToken(id=user.id, token=access_token.decode(), app_token=app_token_str)
        pdb.add(app_token)
@@ -133,7 +140,7 @@
        await pdb.refresh(app_token)
    else:
        db_app_token.token = access_token.decode()
        db_app_token.app_token = json.dumps({"rag_token": ragflow_token, "bs_token":bisheng_token})
        db_app_token.app_token = json.dumps(token_dict)
        await pdb.commit()
        await pdb.refresh(db_app_token)
    return Response(code=200, msg="Login successful", data={
@@ -143,3 +150,39 @@
        "nickname": "",
        # "user": user.to_login_json()
    })



@router.post("/register_test", response_model=Response)
async def register_test(user: UserCreate, db=Depends(get_db)):
    db_user = db.query(UserModel).filter(UserModel.username == user.username).first()
    if db_user:
        return Response(code=200, msg="Username already registered")
    app_register = AppRegisterDao(db).get_apps()
    register_dict = {}
    for app in app_register:
        if app["id"] == RAGFLOW:
            service = RagflowService(settings.fwr_base_url)
        elif app["id"] == BISHENG:
            service = BishengService(settings.sgb_base_url)
        elif app["id"] == DIFY:
            continue
        else:
            logger.error("未知注册应用---")
            continue
        try:
            register_info = await service.register(user.username, user.password)
            register_dict[app['id']] = register_info.get("id") if app['id'] == RAGFLOW else register_info.get("user_id") if app['id'] == BISHENG else ""
        except Exception as e:
            return Response(code=500, msg=f"Failed to register with {app['id']}: {str(e)}")

    # 存储用户信息
    hashed_password = pwd_context.hash(user.password)
    db_user = UserModel(username=user.username, hashed_password=hashed_password, email=user.email)
    db_user.password = db_user.encrypted_password(user.password)
    for k, v in register_dict.items():
        setattr(db_user, k.replace("app", "id"), v)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return Response(code=200, msg="User registered successfully",data={"username": db_user.username})

 app/api/chat.py

@@ -10,9 +10,11 @@
from Log import logger
from app.api import get_current_user_websocket
from app.config.config import settings
from app.config.const import IMAGE_TO_TEXT, DOCUMENT_TO_REPORT, DOCUMENT_TO_CLEANING
from app.models.agent_model import AgentModel, AgentType
from app.models.base_model import get_db
from app.models.user_model import UserModel
from app.service.common.api_token import DfTokenDao
from app.service.dialog import update_session_history
from app.service.basic import BasicService
from app.service.difyService import DifyService
@@ -310,7 +312,10 @@
        try:
            async def forward_to_dify():
                if agent.type == "imageTalk":
                    token = settings.dify_api_token
                    token = DfTokenDao(db).get_token_by_id(IMAGE_TO_TEXT)
                    if not token:
                        await websocket.send_json({"message": "Invalid token", "type": "error"})

                    while True:
                        image_list = []
                        is_image = False
@@ -334,9 +339,11 @@
                        except Exception as e:
                            logger.error(e)
                        # complete_response = ""

                        answer_str = ""
                        async for rag_response in dify_service.chat(token, current_user.id, question, upload_file_id,
                                                                    conversation_id):
                            # print(rag_response)
                            try:
                                if rag_response[:5] == "data:":
                                    # 如果是，则截取掉前5个字符，并去除首尾空白符
@@ -403,8 +410,10 @@
                                await websocket.send_json(result)
                                print(f"Error process message of ragflow: {e2}")
                elif agent.type == "reportWorkflow":
                    print(2323333232)
                    token = settings.dify_workflow_clean

                    token = DfTokenDao(db).get_token_by_id(DOCUMENT_TO_CLEANING)
                    if not token:
                        await websocket.send_json({"message": "Invalid token document_to_cleaning", "type": "error"})
                    while True:
                        receive_message = await websocket.receive_json()
                        print(f"Received from client {chat_id}: {receive_message}")
@@ -440,10 +449,13 @@
                        if workflow_type == 2:
                            inputs["file_list"] = files
                            inputs["Completion_of_main_indicators"] = title
                            token = settings.dify_workflow_report
                            token = DfTokenDao(db).get_token_by_id(DOCUMENT_TO_REPORT)
                            if not token:
                                await websocket.send_json(
                                    {"message": "Invalid token document_to_cleaning", "type": "error"})
                        complete_response = ""
                        async for rag_response in dify_service.workflow(token, current_user.id, inputs):
                            print(rag_response)
                            # print(rag_response)
                            try:
                                if rag_response[:5] == "data:":
                                    # 如果是，则截取掉前5个字符，并去除首尾空白符

 app/api/files.py

@@ -15,6 +15,7 @@
from app.models.user_model import UserModel
from app.service.basic import BasicService
from app.service.bisheng import BishengService
from app.service.common.api_token import DfTokenDao
from app.service.difyService import DifyService
from app.service.ragflow import RagflowService
from app.service.service_token import get_ragflow_token, get_bisheng_token
@@ -96,9 +97,11 @@
            # result = await service.paper_file_upload(chat_id, file.filename, file_content)

    elif agent.agent_type == AgentType.DIFY:
        token = settings.dify_api_token
        dify_service = DifyService(base_url=settings.dify_base_url)
        if agent.type == "imageTalk":
            token = DfTokenDao(db).get_token_by_id("image_and_text_conversion")
            if not token:
                raise HTTPException(status_code=500, detail="获取token失败，image_and_text_conversion！")
            file = file[0]
            # 读取上传的文件内容
            try:
@@ -110,6 +113,9 @@
            except Exception as e:
                raise HTTPException(status_code=500, detail=str(e))
        elif agent.type == "reportWorkflow":
            token = DfTokenDao(db).get_token_by_id("document_to_report")
            if not token:
                raise HTTPException(status_code=500, detail="获取token失败，document_to_report！")
            result = []
            for f in file:
                try:

 app/api/public_api.py

New file
@@ -0,0 +1,62 @@
import json

from fastapi import APIRouter, Depends

from Log import logger
from app.api import Response
from app.api.auth import login
from app.config.const import IMAGE_TO_TEXT, DOCUMENT_TO_CLEANING, DOCUMENT_TO_REPORT, DIFY, BISHENG, RAGFLOW
from app.models.base_model import get_db
from app.models.public_api_model import DfToken, AppRegister
from app.service.common.api_token import DfTokenDao
from app.service.common.app_register import AppRegisterDao
from app.task.sync_resources import sync_knowledge, sync_dialog, sync_agent, sync_llm, sync_resource

public_api = APIRouter()


@public_api.post("/sync/df_token", response_model=Response)
async def sync_df_token(df: DfToken, db=Depends(get_db)):
    token_dict = {IMAGE_TO_TEXT: df.image, DOCUMENT_TO_CLEANING: df.clean, DOCUMENT_TO_REPORT: df.report}

    try:
        for api_id, token in token_dict.items():
            if not token:
                continue
            DfTokenDao(db).update_and_insert_token(api_id, token)
    except Exception as e:
        logger.error(e)
        return Response(code=500, msg="failed", data={})

    return Response(code=200, msg="successfully", data={})


@public_api.post("/sync/app_register", response_model=Response)
async def sync_app_register(apr: AppRegister, db=Depends(get_db)):
    app_dict = {RAGFLOW: apr.rg, BISHENG: apr.bs, DIFY: apr.df}

    try:
        for app_id, status in app_dict.items():

            AppRegisterDao(db).update_and_insert_app(app_id, status)
    except Exception as e:
        logger.error(e)
        return Response(code=500, msg="failed", data={})

    return Response(code=200, msg="successfully", data={})


@public_api.get("/sync/resource", response_model=Response)
async def user_group_list(resource_type:int, db=Depends(get_db)):
    if resource_type == 1:
        await sync_knowledge()
    elif resource_type == 2:
        await sync_dialog()
    elif resource_type == 3:
        await sync_agent()
    elif resource_type == 4:
        await sync_llm()
    else:
        await sync_resource()

    return Response(code=200, msg="", data={})

 app/api/sync_data.py

File was deleted

 app/api/user.py

@@ -1,6 +1,6 @@
from fastapi import APIRouter, Depends
from app.api import Response, pwd_context, get_current_user
from app.models.app_model import AppRegisterModel
from app.models.public_api_model import AppRegisterModel
from app.models.base_model import get_db
from app.models.user import PageParameter, UserStatus, UserInfo, LoginData
from app.models.user_model import UserModel

 app/config/const.py

New file
@@ -0,0 +1,11 @@

### ----------dify------api token
DOCUMENT_TO_CLEANING = "document_to_cleaning"
DOCUMENT_TO_REPORT = "document_to_report"
IMAGE_TO_TEXT = "image_and_text_conversion"


### -----------app register --------------
RAGFLOW = "ragflow_app"
BISHENG = "bisheng_app"
DIFY = "dify_app"

 app/models/app_model.py

File was deleted

 app/models/postgresql_base_model.py

@@ -7,7 +7,7 @@

DATABASE_URL = os.getenv('POSTGRESQL_DATABASE_URL') or settings.postgresql_database_url

engine = create_async_engine(DATABASE_URL, echo=True)
engine = create_async_engine(DATABASE_URL, echo=False)
PostgresqlSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)
PostgresqlBase = declarative_base()


 app/models/public_api_model.py

New file
@@ -0,0 +1,61 @@
from datetime import datetime
from enum import IntEnum
from typing import Optional

from pydantic import BaseModel
from sqlalchemy import Column, String, Enum as SQLAlchemyEnum, Integer, BigInteger, DateTime, Text, Float, Boolean
from app.models.base_model import Base

class AppType(IntEnum):
    BASIC = 0
    RAGFLOW = 1
    BISHENG = 2



class AppRegisterModel(Base):
    __tablename__ = "app_register"
    id = Column(String(36), primary_key=True)
    name = Column(String(255))
    status = Column(Integer, nullable=False, default=1)
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now())


    # to_dict 方法
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'status': self.status
        }


class DfTokenModel(Base):
    __tablename__ = "df_api_token"
    id = Column(String(36), primary_key=True)
    token = Column(String(36))
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now())


    # to_dict 方法
    def to_dict(self):
        return {
            'id': self.id,
            'token': self.token,
            'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S'),
            # 'updated_at': self.updated_at.strftime('%Y-%m-%d %H:%M:%S')
        }


class DfToken(BaseModel):
    image: Optional[str] = ""
    clean: Optional[str] = ""
    report: Optional[str] = ""


class AppRegister(BaseModel):
    rg: Optional[int] = 0
    bs: Optional[int] = 0
    df: Optional[int] = 0

 app/models/token_model.py

@@ -4,6 +4,8 @@
from sqlalchemy import Column, Integer, DateTime, Text
from sqlalchemy.orm import Session

from Log import logger
from app.config.const import RAGFLOW
from app.models.base_model import Base


@@ -52,5 +54,39 @@
        db.rollback()  # 回滚事务


async def update_token(db: Session, user_id: int, access_token: str, token: dict):
    # 参数验证
    if not isinstance(user_id, int) or user_id <= 0:
        return
    db_token = None
    print(token)
    try:
        # 查询现有记录
        db_token = db.query(TokenModel).filter_by(user_id=user_id).first()
        if db_token:
            # 记录存在，进行更新
            db_token.token = access_token
            for k, v in token.items():
                setattr(db_token, k.replace("app", "token"), v)
        else:
            # 记录不存在，进行插入
            db_token = TokenModel(
                user_id=user_id,
                token=access_token,
            )
            for k, v in token.items():
                setattr(db_token, k.replace("app", "token"), v)
            db.add(db_token)

        # 提交事务
        db.commit()
        db.refresh(db_token)

    except Exception as e:
        logger.error(e)
        # 异常处理
        db.rollback()  # 回滚事务


def get_token(db: Session, user_id: int) -> Type[TokenModel] | None:
    return db.query(TokenModel).filter_by(user_id=user_id).first()

 app/models/user.py

@@ -6,6 +6,7 @@
class UserCreate(BaseModel):
    username: str
    password: str
    email: Optional[str] = ""


# 定义请求体模型

 app/service/common/api_token.py

New file
@@ -0,0 +1,48 @@
from Log import logger
from app.models import current_time
from app.models.public_api_model import DfTokenModel
from sqlalchemy.orm import Session
from typing import Type


class DfTokenDao:
    def __init__(self, db: Session):
        self.db = db

    def get_token_by_id(self, api_id: str) -> Type[DfTokenModel] | None:
        session = self.db.query(DfTokenModel).filter_by(id=api_id).first()
        if session:
            return session.token
        return None

    def update_token(self, api_id: str, token: str):

        logger.error("更新数据df api token---------------------------")
        try:
            self.db.query(DfTokenModel).filter(DfTokenModel.id==api_id).update({"token":token, "updated_at": current_time()})
            self.db.commit()
        except Exception as e:
            logger.error(e)
            self.db.rollback()
            raise Exception("更新失败！")

    def insert_token(self, api_id: str, token: str):
        logger.error("新增数据df api token---------------------------")
        new_session = DfTokenModel(
            id=api_id,
            token=token
        )
        self.db.add(new_session)
        self.db.commit()
        self.db.refresh(new_session)
        return new_session


    def update_and_insert_token(self, api_id: str, token: str):

        logger.error("更新或者添加数据df api token---------------------------")
        token_boj = self.get_token_by_id(api_id)
        if token_boj:
            self.update_token(api_id, token)
        else:
            self.insert_token(api_id, token)

 app/service/common/app_register.py

@@ -1,12 +1,53 @@
from app.models.app_model import AppRegisterModel
from app.models.public_api_model import AppRegisterModel
from Log import logger
from app.models import current_time
from sqlalchemy.orm import Session
from typing import Type


class AppRegister:
    app = AppRegisterModel
    def __init__(self, db):
class AppRegisterDao:
    def __init__(self, db: Session):
        self.db = db

    def get_app_by_id(self, api_id: str) -> Type[AppRegisterModel] | None:
        session = self.db.query(AppRegisterModel).filter_by(id=api_id).first()

    def get_app(self):
        return session

        return [i.app_type for i in self.db.query(self.app).filter_by(status=1).all()]
    def update_app(self, app_id: str, status: int):

        logger.error("更新数据: app register---------------------------")
        try:
            self.db.query(AppRegisterModel).filter(AppRegisterModel.id==app_id).update({"status":status, "updated_at": current_time()})
            self.db.commit()
        except Exception as e:
            logger.error(e)
            self.db.rollback()
            raise Exception("更新失败！")

    def insert_app(self, app_id: str, status: int):
        logger.error("新增数据: app register---------------------------")
        new_session = AppRegisterModel(
            id=app_id,
            status=status
        )
        self.db.add(new_session)
        self.db.commit()
        self.db.refresh(new_session)
        return new_session


    def update_and_insert_app(self, app_id: str, status: int):

        logger.error("更新或者添加数据: app register---------------------------")
        token_boj = self.get_app_by_id(app_id)
        if token_boj:
            self.update_app(app_id, status)
        else:
            self.insert_app(app_id, status)


    def get_apps(self):
        app_list = self.db.query(AppRegisterModel).filter_by(status=1).all()

        return [i.to_dict() for i in app_list]

 app/service/user.py

@@ -6,7 +6,7 @@
from app.models.user_model import UserModel
from Log import logger
from app.service.bisheng import BishengService
from app.service.common.app_register import AppRegister
from app.service.common.app_register import AppRegisterDao
from app.service.ragflow import RagflowService
from app.service.service_token import get_ragflow_token, get_bisheng_token, get_ragflow_new_token

@@ -104,7 +104,7 @@
    try:
        user = db.query(UserModel).filter(UserModel.id == user_id).first()
        pwd = user.decrypted_password()
        for app_type in AppRegister(db).get_app():
        for app_type in AppRegisterDao(db).get_app():
            if app_type == AgentType.RAGFLOW:
                token = await get_ragflow_new_token(db, user_id, app_type)
                ragflow_service = RagflowService(settings.fwr_base_url)

 app/task/sync_resources.py

@@ -5,7 +5,7 @@
from Log import logger
from app.config.config import settings
from app.models import AgentType
from app.models.app_model import AppRegisterModel
from app.models.public_api_model import AppRegisterModel
from app.models.base_model import get_db
from app.service.bisheng import BishengService
from app.service.ragflow import RagflowService

 main.py

@@ -13,9 +13,10 @@
from app.api.files import router as files_router
from app.api.knowledge import knowledge_router
from app.api.organization import dept_router
from app.api.public_api import public_api
from app.api.report import router as report_router
from app.api.resource import menu_router
from app.api.sync_data import sync_router
# from app.api.sync_data import sync_router
from app.api.user import user_router
from app.api.group import group_router
from app.api.role import role_router
@@ -71,7 +72,8 @@
app.include_router(knowledge_router, prefix='/api/knowledge', tags=["knowledge"])
app.include_router(dialog_router, prefix='/api/dialog', tags=["dialog"])
app.include_router(canvas_router, prefix='/api/canvas', tags=["canvas"])
app.include_router(sync_router, prefix='/api/sync', tags=["sync"])
# app.include_router(sync_router, prefix='/api/sync', tags=["sync"])
app.include_router(public_api, prefix='/v1/api', tags=["public_api"])
app.mount("/static", StaticFiles(directory="app/images"), name="static")

if __name__ == "__main__":