m

zhaoqingang

2024-12-09 52ba4076f5ad55fdf3239a33a2a376eaa0e0dea5

m

 alembic/versions/07b5185945e0_token_table.py

New file
@@ -0,0 +1,30 @@
"""token table

Revision ID: 07b5185945e0
Revises: 4b3a7c69ceac
Create Date: 2024-12-06 16:27:30.329519

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '07b5185945e0'
down_revision: Union[str, None] = '4b3a7c69ceac'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###

 alembic/versions/4b3a7c69ceac_init_table.py

New file
@@ -0,0 +1,79 @@
"""init table

Revision ID: 4b3a7c69ceac
Revises: 
Create Date: 2024-12-06 16:12:04.760402

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

# revision identifiers, used by Alembic.
revision: str = '4b3a7c69ceac'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index('ix_user_test_id', table_name='user_test')
    op.drop_index('ix_user_test_username', table_name='user_test')
    op.drop_table('user_test')
    op.alter_column('user', 'ragflow_id',
               existing_type=mysql.VARCHAR(length=32),
               nullable=True)
    op.alter_column('user', 'bisheng_id',
               existing_type=mysql.INTEGER(),
               nullable=True)
    op.alter_column('user', 'status',
               existing_type=mysql.VARCHAR(length=10),
               nullable=True,
               existing_server_default=sa.text("'1'"))
    op.alter_column('user', 'permission',
               existing_type=mysql.VARCHAR(length=16),
               nullable=True,
               existing_server_default=sa.text("'general'"))
    op.drop_column('user', 'updated_at11')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user', sa.Column('updated_at11', mysql.INTEGER(), autoincrement=False, nullable=True))
    op.alter_column('user', 'permission',
               existing_type=mysql.VARCHAR(length=16),
               nullable=False,
               existing_server_default=sa.text("'general'"))
    op.alter_column('user', 'status',
               existing_type=mysql.VARCHAR(length=10),
               nullable=False,
               existing_server_default=sa.text("'1'"))
    op.alter_column('user', 'bisheng_id',
               existing_type=mysql.INTEGER(),
               nullable=False)
    op.alter_column('user', 'ragflow_id',
               existing_type=mysql.VARCHAR(length=32),
               nullable=False)
    op.create_table('user_test',
    sa.Column('id', mysql.INTEGER(), autoincrement=True, nullable=False),
    sa.Column('username', mysql.VARCHAR(length=255), nullable=True),
    sa.Column('hashed_password', mysql.VARCHAR(length=255), nullable=True),
    sa.Column('password', mysql.VARCHAR(length=255), nullable=True),
    sa.Column('compellation', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('phone', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('email', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('description', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('ragflow_id', mysql.VARCHAR(length=32), nullable=True),
    sa.Column('bisheng_id', mysql.INTEGER(), autoincrement=False, nullable=True),
    sa.PrimaryKeyConstraint('id'),
    mysql_collate='utf8mb4_0900_ai_ci',
    mysql_default_charset='utf8mb4',
    mysql_engine='InnoDB'
    )
    op.create_index('ix_user_test_username', 'user_test', ['username'], unique=True)
    op.create_index('ix_user_test_id', 'user_test', ['id'], unique=False)
    # ### end Alembic commands ###

 alembic/versions/580e984b9882_initial_migration.py

File was deleted

 alembic/versions/abc6bb9129ed_user_app_update.py

New file
@@ -0,0 +1,44 @@
"""user app update

Revision ID: abc6bb9129ed
Revises: d4c8f204280f
Create Date: 2024-12-09 16:57:06.559644

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

# revision identifiers, used by Alembic.
revision: str = 'abc6bb9129ed'
down_revision: Union[str, None] = 'd4c8f204280f'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user_app', sa.Column('access_token', sa.String(length=1000), nullable=True))
    op.add_column('user_app', sa.Column('refresh_token', sa.String(length=1000), nullable=True))
    op.add_column('user_app', sa.Column('token_at', sa.DateTime(), nullable=True))
    op.alter_column('user_app', 'app_type',
               existing_type=mysql.INTEGER(),
               type_=sa.String(length=16),
               existing_nullable=True)
    op.drop_index('ix_user_app_username', table_name='user_app')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_index('ix_user_app_username', 'user_app', ['username'], unique=True)
    op.alter_column('user_app', 'app_type',
               existing_type=sa.String(length=16),
               type_=mysql.INTEGER(),
               existing_nullable=True)
    op.drop_column('user_app', 'token_at')
    op.drop_column('user_app', 'refresh_token')
    op.drop_column('user_app', 'access_token')
    # ### end Alembic commands ###

 alembic/versions/d4c8f204280f_user_app_add.py

New file
@@ -0,0 +1,67 @@
"""user app add

Revision ID: d4c8f204280f
Revises: 07b5185945e0
Create Date: 2024-12-09 15:43:14.470291

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

# revision identifiers, used by Alembic.
revision: str = 'd4c8f204280f'
down_revision: Union[str, None] = '07b5185945e0'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('user_app',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('username', sa.String(length=255), nullable=True),
    sa.Column('password', sa.String(length=255), nullable=True),
    sa.Column('email', sa.String(length=255), nullable=True),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.Column('app_id', sa.String(length=36), nullable=True),
    sa.Column('app_type', sa.Integer(), nullable=True),
    sa.Column('status', sa.String(length=10), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('user_id', 'app_type', name='user_app_id_ix')
    )
    op.create_index(op.f('ix_user_app_id'), 'user_app', ['id'], unique=False)
    op.create_index(op.f('ix_user_app_username'), 'user_app', ['username'], unique=True)
    op.drop_table('dialog')
    op.drop_table('flow')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('flow',
    sa.Column('id', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('name', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('status', mysql.INTEGER(), autoincrement=False, nullable=False),
    sa.PrimaryKeyConstraint('id'),
    mysql_collate='utf8mb4_0900_ai_ci',
    mysql_default_charset='utf8mb4',
    mysql_engine='InnoDB'
    )
    op.create_table('dialog',
    sa.Column('id', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('name', mysql.VARCHAR(length=255), nullable=False),
    sa.Column('status', mysql.VARCHAR(length=1), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    mysql_collate='utf8mb4_0900_ai_ci',
    mysql_default_charset='utf8mb4',
    mysql_engine='InnoDB'
    )
    op.drop_index(op.f('ix_user_app_username'), table_name='user_app')
    op.drop_index(op.f('ix_user_app_id'), table_name='user_app')
    op.drop_table('user_app')
    # ### end Alembic commands ###

 app/api/auth.py

@@ -1,4 +1,5 @@
import json
import time

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
@@ -14,9 +15,11 @@
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.auth import authenticate_user, create_access_token, is_valid_password, save_register_user, \
    update_user_token, UserAppDao
from app.service.bisheng import BishengService
from app.service.common.app_register import AppRegisterDao
from app.service.difyService import DifyService
from app.service.ragflow import RagflowService
from sqlalchemy.future import select

@@ -103,7 +106,7 @@


@router.post("/v2/login", response_model=Response)
async def login_test(login_data: LoginData, db: Session = Depends(get_db), pdb: AsyncSession = Depends(get_pdb)):
async def login_v2(login_data: LoginData, db: Session = Depends(get_db), pdb: AsyncSession = Depends(get_pdb)):
    user = authenticate_user(db, login_data.username, login_data.password)
    if not user:
        return Response(code=400, msg="Incorrect username or password")
@@ -120,7 +123,10 @@
            logger.error("未知注册应用---")
            continue
        try:
            token = await service.login(login_data.username, login_data.password)
            user_app = UserAppDao(db).get_data_by_id(user.id, app["id"])
            if user_app:
                name  = user_app.username
            token = await service.login(name, 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)}")
@@ -129,6 +135,7 @@
    access_token = create_access_token(data={"sub": user.username, "user_id": user.id})

    await update_token(db, user.id, access_token, token_dict)
    await update_user_token(db, user.id, token_dict)
    result = await pdb.execute(select(AppToken).where(AppToken.id == user.id))
    db_app_token = result.scalars().first()
    if isinstance(access_token, bytes):
@@ -156,7 +163,9 @@


@router.post("/v2/register", response_model=Response)
async def register_test(user: UserCreate, db=Depends(get_db)):
async def register_v2(user: UserCreate, db=Depends(get_db)):
    if not is_valid_password(user.password):
        return Response(code=400, msg="The password must be at least 8 and contain both numbers and letters")
    db_user = db.query(UserModel).filter(UserModel.username == user.username).first()
    if db_user:
        return Response(code=200, msg="Username already registered")
@@ -168,23 +177,29 @@
        elif app["id"] == BISHENG:
            service = BishengService(settings.sgb_base_url)
        elif app["id"] == DIFY:
            continue
            service = DifyService(settings.dify_base_url)
        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 ""
            name = app["id"] + str(int(time.time()))
            register_info = await service.register(name, user.password)
            print(register_info)
            register_dict[app['id']] = {"id":register_info.get("id"), "name": name, "email": register_info.get("email")}
        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})
    # 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)

    is_sava = await save_register_user(db, user.username, user.password, user.email, register_dict)
    if not is_sava:
        return Response(code=500, msg=f"Failed to register with app")
    return Response(code=200, msg="User registered successfully",data={"username": user.username})

 app/config/const.py

@@ -8,4 +8,10 @@
### -----------app register --------------
RAGFLOW = "ragflow_app"
BISHENG = "bisheng_app"
DIFY = "dify_app"
DIFY = "dify_app"

### ---------------app type-----------------
BASIC_ID = 3
RAGFLOW_ID = 1
BISHENG_ID = 2
DIFY_ID = 4

 app/models/__init__.py

@@ -11,6 +11,9 @@
from .resource_model import *
from .role_model import *
from .user_model import *
from .token_model import *
from .session_model import SessionModel
from .public_api_model import *


# 获取当前时区的时间

 app/models/session_model.py

@@ -1,11 +1,17 @@
import json
from datetime import datetime
from enum import IntEnum

import pytz
from sqlalchemy import Column, String, Enum as SQLAlchemyEnum, Integer, DateTime, JSON, TEXT

from app.models import AgentType, current_time
from app.models.agent_model import AgentType
# from app.models import current_time
from app.models.base_model import Base

def current_time():
    tz = pytz.timezone('Asia/Shanghai')
    return datetime.now(tz)

class SessionModel(Base):
    __tablename__ = "sessions"

 app/models/user_model.py

@@ -1,7 +1,7 @@
from datetime import datetime

from cryptography.fernet import Fernet
from sqlalchemy import Column, Integer, String, Table, ForeignKey, DateTime
from sqlalchemy import Column, Integer, String, Table, ForeignKey, DateTime, UniqueConstraint
from sqlalchemy.orm import relationship, backref

from app.config.config import settings
@@ -27,21 +27,20 @@
    username = Column(String(255), unique=True, index=True)
    hashed_password = Column(String(255))
    password = Column(String(255))
    compellation = Column(String(255), nullable=False, default="")
    phone = Column(String(255), nullable=False, default="")
    email = Column(String(255), nullable=False, default="")
    description = Column(String(255), nullable=False, default="")
    compellation = Column(String(255), default="")
    phone = Column(String(255),  default="")
    email = Column(String(255),  default="")
    description = Column(String(255), default="")
    ragflow_id = Column(String(32))
    bisheng_id = Column(Integer)
    login_name = Column(String(100))
    status = Column(String(10), nullable=False, default="1")
    status = Column(String(10),  default="1")
    creator = Column(String(36))
    sex = Column(String(1))
    permission = Column(String(16), nullable=False, default="general")
    permission = Column(String(16), default="general")
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now())
    updated_at11 = Column(Integer)



@@ -181,15 +180,33 @@



class UserModel(Base):
    __tablename__ = "user_test"
class UserAppModel(Base):
    __tablename__ = "user_app"
    __table_args__ = (UniqueConstraint('user_id', 'app_type', name='user_app_id_ix'),)
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(255), unique=True, index=True)
    hashed_password = Column(String(255))
    username = Column(String(255))
    password = Column(String(255))
    compellation = Column(String(255), nullable=False, default="")
    phone = Column(String(255), nullable=False, default="")
    email = Column(String(255), nullable=False, default="")
    description = Column(String(255), nullable=False, default="")
    ragflow_id = Column(String(32))
    bisheng_id = Column(Integer)
    email = Column(String(255),  default="")
    user_id = Column(Integer)
    app_id = Column(String(36))
    app_type = Column(String(16))
    status = Column(String(10),  default="1")
    access_token = Column(String(1000))
    refresh_token = Column(String(1000))
    token_at = Column(DateTime, default=datetime.now())
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now())

    def to_json(self):
        return {
            'id': self.id,
            'userName': self.username,
            'createTime': self.created_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else "",
            'updateTime': self.updated_at.strftime('%Y-%m-%d %H:%M:%S') if self.created_at else "",
            'password': self.password,
            'email': self.email,
            'user_id': self.user_id,
            'app_id': self.app_id,
            "app_type": self.app_type,
            'status': self.status,
        }

 app/service/auth.py

@@ -1,10 +1,15 @@
import re
from datetime import datetime, timedelta
from typing import Type

from jwt import encode, decode, exceptions
from passlib.context import CryptContext
from fastapi import HTTPException, status
from sqlalchemy.orm import Session

from Log import logger
from app.config.config import settings
from app.models.user_model import UserModel
from app.models.user_model import UserModel, UserAppModel

SECRET_KEY = settings.secret_key
ALGORITHM = "HS256"
@@ -47,3 +52,98 @@
        return payload
    except exceptions.DecodeError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")


def is_valid_password(password: str) -> bool:
    if len(password) <= 8:
        return False
    has_digit = re.search(r'[0-9]', password)
    has_letter = re.search(r'[A-Za-z]', password)

    # 如果密码包含数字和字母，则返回True，否则返回None
    return has_digit is not None and has_letter is not None


async def save_register_user(db, username, password, email, register_dict):
    user_id = ""
    try:
        hashed_password = pwd_context.hash(password)
        db_user = UserModel(username=username, hashed_password=hashed_password, email=email)
        pwd = db_user.encrypted_password(password)
        db_user.password = pwd
        db.add(db_user)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        user_id = db_user.id
        for k, v in register_dict.items():
            UserAppDao(db).update_and_insert_token(v.get("name"), pwd, v.get("email"), user_id, str(v.get("id")), k)

    except Exception as e:
        logger.error(e)
        # db.roolback()
        if user_id:
            db.query(UserModel).filter(UserModel.id == user_id).delete()
        return False
    return True


async def update_user_token(db, user_id, token_dict):

    try:
        for k, v in token_dict.items():
            UserAppDao(db).update_user_app_data({"user_id": user_id, "app_type": k},
                                                {"access_token": v, "token_at": datetime.now()})

    except Exception as e:
        logger.error(e)
        return False
    return True


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

    def get_data_by_id(self, user_id: int, app_type: int) -> Type[UserAppModel] | None:
        session = self.db.query(UserAppModel).filter_by(user_id=user_id, app_type=app_type).first()
        return session

    def update_user_app_data(self, query: int, update_data: str):

        logger.error("更新数据df update_app_data---------------------------")
        try:
            self.db.query(UserAppModel).filter_by(**query).update(update_data)
            self.db.commit()
        except Exception as e:
            logger.error(e)
            self.db.rollback()
            raise Exception("更新失败！")

    def insert_user_app_data(self, username: str, password: str, email: str, user_id: int, app_id: str, app_type: int):
        logger.error("新增数据df insert_user_app_data---------------------------")
        new_session = UserAppModel(
            username=username,
            password=password,
            email=email,
            user_id=user_id,
            app_id=app_id,
            app_type=app_type,
        )
        self.db.add(new_session)
        self.db.commit()
        self.db.refresh(new_session)
        return new_session

    def update_and_insert_token(self, username: str, password: str, email: str, user_id: int, app_id: str,
                                app_type: int):

        logger.error("更新或者添加数据 update_and_insert_token---------------------------")
        token_boj = self.get_data_by_id(user_id, app_type)
        if token_boj:
            self.update_user_app_data({"id": token_boj.id}, {"username": username,
                                                             "password": password, "email": email, "username": username,
                                                             "updated_at": datetime.now(),
                                                             })
        else:
            self.insert_user_app_data(username, password, email, user_id, app_id, app_type)

 app/service/bisheng.py

@@ -35,7 +35,10 @@
                json={"user_name": username, "password": password},
                headers={'Content-Type': 'application/json'}
            )
            return self._check_response(response)
            res = self._check_response(response)
            if isinstance(res, dict):
                res["id"] = res.get("user_id")
            return res

    async def login(self, username: str, password: str) -> str:
        public_key = await self.get_public_key_api()

 app/service/difyService.py

@@ -1,11 +1,13 @@
import json
from datetime import datetime
from urllib.parse import urlparse, parse_qs

import httpx
from typing import Union, Dict, List
from fastapi import HTTPException
from starlette import status

# from Log import logger
from app.config.config import settings
from app.utils.rsa_crypto import RagflowCrypto

@@ -38,32 +40,90 @@
        else:
            return data

    async def register(self, username: str, password: str):
        password = RagflowCrypto(settings.PUBLIC_KEY, settings.PRIVATE_KEY).encrypt(password)
    async def register(self, username: str, password: str, token=None):
        if not token:
            token = await get_df_token()
        email = f"{username}@df.com"
        invite_res = await self.invite_workspaces_member(token, [email], "admin")
        # print(invite_res)
        if invite_res.get("result") != "success" or not invite_res.get("invitation_results"):
            # logger.error(invite_res)
            return {}
        invite_token = \
        parse_qs(urlparse(invite_res.get("invitation_results")[0].get("url", "")).query).get('token', [None])[0]
        # print(invite_token)
        if not invite_token:
            return {}
        await self.login(email, password, True, invite_token)
        activate_res = await self.activate(email, username, invite_token, "", "")
        activate_res["email"] = email
        activate_res["id"] = invite_token
        return activate_res


    async def invite_workspaces_member(self, token, emails: list, role: str):
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/v1/user/register",
                headers={'Content-Type': 'application/json'},
                json={"nickname": username, "email":  f"{username}@example.com", "password": password}
                f"{self.base_url}/console/api/workspaces/current/members/invite-email",
                headers={'Content-Type': 'application/json',"Authorization": f'Bearer {token}'},
                json={"emails": emails, "language":  "zh-Hans", "role": role}
            )
            if response.status_code != 200:
                raise Exception(f"Ragflow registration failed: {response.text}")
            print(response.text)
            return self._handle_response(response)

    async def login(self, username: str, password: str) -> str:
        password = RagflowCrypto(settings.PUBLIC_KEY, settings.PRIVATE_KEY).encrypt(password)
    async def login(self, email: str, password: str, remember_me,invite_token:str="") -> str:
        # password = RagflowCrypto(settings.PUBLIC_KEY, settings.PRIVATE_KEY).encrypt(password)
        data = {"email": email, "password": password, "remember_me": remember_me, "invite_token": invite_token,
         "language": "zh-Hans"}

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/v1/user/login",
                f"{self.base_url}/console/api/login",
                headers={'Content-Type': 'application/json'},
                json={"email": f"{username}@example.com", "password": password}
                json=data
            )
            if response.status_code != 200:
                raise Exception(f"Ragflow login failed: {response.text}")
            authorization = response.headers.get('Authorization')
            if not authorization:
                raise Exception("Authorization header not found in response")
            return authorization
                raise Exception(f"df login failed: {response.text}")
            return self._handle_response(response)

    async def email_check(self, token, email: str):
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/console/api/activate/check?email={email}&token={token}",
                headers={'Content-Type': 'application/json'}
            )
            return self._handle_response(response)

    async def activate(self, email: str, name: str, token, workspace_id:str, access_token) -> str:
        # password = RagflowCrypto(settings.PUBLIC_KEY, settings.PRIVATE_KEY).encrypt(password)
        data = {"email": email, "name": name, "token": token, #  "workspace_id": workspace_id,
         "interface_language": "en-US", "timezone": "Asia/Shanghai"}
        print(data)
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/console/api/activate",
                headers={'Content-Type': 'application/json'},  # , 'Authorization': f'Bearer {access_token}'
                json=data
            )
            if response.status_code != 200:
                raise Exception(f"df login failed: {response.text}")
            return self._handle_response(response)

    async def invite_member_activate(self, token:str, email: str, name: str, password:str) -> str:
        invite_res  =  await self.invite_workspaces_member(token, [email], "admin")
        print(invite_res)
        if invite_res.get("result") != "success" or not invite_res.get("invitation_results"):
            # logger.error(invite_res)
            return {}
        invite_token = parse_qs(urlparse(invite_res.get("invitation_results")[0].get("url", "")).query).get('token', [None])[0]
        # print(invite_token)
        if not invite_token:
            return {}
        await self.login(email, password, True, invite_token)
        activate_res = await self.activate(email, name,invite_token, "", "")
        return activate_res



    async def chat(self, token: str, user_id: int,  message: str, upload_file_id: str, conversation_id: str):

@@ -186,8 +246,28 @@
if __name__ == "__main__":
    async def a():
        a = DifyService("http://192.168.20.116")
        b = await a.get_session_history("app-YmOAMDsPpDDlqryMHnc9TzTO", "f94c6328-8ff0-4713-af3f-e823d547682d",
                                 "63")
        # b = await a.invite_workspaces_member("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiMjE0Y2I2ODctZTZlMC00ZTE2LWExNzUtYzcyNDNlMGRhMWEwIiwiZXhwIjoxNzMzNzI2NDA5LCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.jLe1ODbcqCe79CDt6fFwnuuQL4I2FB9YTs9ynk4FeoQ", ["test05@163.com"],
        #                          "admin")

        # b = await a.email_check(
        #     "ebd36739-0272-4b3f-95ab-0c6ac1639831",
        #     "test05@163.com")
        # b = await a.login(
        #     "test05@163.com",
        #     "zhaoqg123456",
        #     True, "ebd36739-0272-4b3f-95ab-0c6ac1639831")



        # b = await a.activate(
        #     "test05@163.com",
        #     "test05", "ebd36739-0272-4b3f-95ab-0c6ac1639831", "", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiNzg5OWUzOGQtNzczOS00NGNmLTgyODItZmFlMGZhNDJlZDYwIiwiZXhwIjoxNzMzNzI5NjQyLCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.YMvypPnrvvUIfqzcESj820nP46IsFdTpF_YPz8_Exso")

        b = await a.invite_member_activate(
            "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiMDE2NTcxNjAtZTllYi00NzVhLWIzMzYtZjlmZWJlY2I5YjczIiwiZXhwIjoxNzMzNzM0ODE0LCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.khaXX3ndDe_pccEHcyTUcO2sgBCEfXCR74ZniP_b54Y",
                "zhao1@df.com",
                 "zhao1Q",
        "ZHAOQG123456")
        print(b)

    import asyncio

 app/task/fetch_agent.py

@@ -119,7 +119,8 @@
            ('basic_excel_talk', 6, '智能数据', 'BASIC', 'excelTalk'),
            ('basic_question_talk', 7, '出题组卷', 'BASIC', 'questionTalk'),
            ('9d75142a-66eb-4e23-b7d4-03efe4584915', 8, '小数绘图', 'DIFY', 'imageTalk'),
            ('basic_paper_talk', 8, '文档出卷', 'BASIC', 'paperTalk')
            ('basic_paper_talk', 8, '文档出卷', 'BASIC', 'paperTalk'),
            ('basic_report_clean', 10, '文档报告', 'DIFY', 'reportWorkflow')
        ]

        for agent in initial_agents:

 main.py

@@ -37,7 +37,7 @@
    # 在应用关闭时执行清理操作（如果需要）
    pass

init_db()
# init_db()
app = FastAPI(
    title="basic_rag_gateway",
    version="0.1",