sync data

zhaoqingang

2024-12-16 88360b4ac6f051f62a91e93d602fd393935071ab

 app/task/fetch_agent.py

@@ -1,19 +1,25 @@
from pickle import PROTO
from typing import Dict, List, Tuple

from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy import create_engine, Column, String, Integer, Text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker

from app.config.config import settings
from app.config.const import RAGFLOW, BISHENG, DIFY
from app.models import DialogModel
from app.models.agent_model import AgentModel
from app.models.base_model import SessionLocal, Base
from app.service.v2.app_register import AppRegisterDao

# 创建数据库引擎和会话工厂
engine_bisheng = create_engine(settings.sgb_db_url)
engine_ragflow = create_engine(settings.fwr_db_url)
engine_dify = create_engine(settings.dify_database_url)

SessionBisheng = sessionmaker(autocommit=False, autoflush=False, bind=engine_bisheng)
SessionRagflow = sessionmaker(autocommit=False, autoflush=False, bind=engine_ragflow)
SessionDify = sessionmaker(autocommit=False, autoflush=False, bind=engine_dify)


class Flow(Base):
@@ -21,6 +27,8 @@
    id = Column(String(255), primary_key=True)
    name = Column(String(255), nullable=False)
    status = Column(Integer, nullable=False)
    description = Column(String(255), nullable=False)
    user_id = Column(Integer, nullable=False)


class Dialog(Base):
@@ -28,6 +36,17 @@
    id = Column(String(255), primary_key=True)
    name = Column(String(255), nullable=False)
    status = Column(String(1), nullable=False)
    description = Column(String(255), nullable=False)
    tenant_id = Column(String(36), nullable=False)


class DfApps(Base):
    __tablename__ = 'apps'
    id = Column(String(36), primary_key=True)
    name = Column(String(255), nullable=False)
    status = Column(String(16), nullable=False)
    description = Column(Text, nullable=False)
    tenant_id = Column(String(36), nullable=False)


# 解析名字
@@ -149,5 +168,150 @@
        print(f"Failed to sync agents: {str(e)}")


def update_ids_in_local(data: List[Tuple]):
    db = SessionLocal()
    try:
        for row in data:
            name = row[1]
            new_id = row[0]
            existing_agent = db.query(AgentModel).filter_by(name=name).first()
            if existing_agent:
                existing_agent.id = new_id
                db.add(existing_agent)
        db.commit()
    except IntegrityError:
        db.rollback()
        raise
    finally:
        db.close()

def get_data_from_bisheng_v2(names: List[str]) -> List[Dict]:
    db = SessionBisheng()
    try:
        if names:
            query = db.query(Flow.id, Flow.name, Flow.description, Flow.status, Flow.user_id) \
                .filter(Flow.name.in_(names), Flow.status==2)
        else:
            query = db.query(Flow.id, Flow.name, Flow.description, Flow.status, Flow.user_id).filter(Flow.status==2)

        results = query.all()
        # print(f"Executing query: {query}")
        # 格式化id为UUID
        formatted_results = [{"id":format_uuid(row[0]), "name": row[1], "description": row[2], "status": str(row[3]-1), "user_id": str(row[4])} for row in results]
        return formatted_results
    finally:
        db.close()

def get_data_from_ragflow_v2(names: List[str]) -> List[Dict]:
    db = SessionRagflow()
    try:
        if names:
            query = db.query(Dialog.id, Dialog.name, Dialog.description, Dialog.status, Dialog.tenant_id) \
                .filter( Dialog.name.in_(names))
        else:
            query = db.query(Dialog.id, Dialog.name, Dialog.description, Dialog.status, Dialog.tenant_id)

        results = query.all()
        formatted_results = [
            {"id": format_uuid(row[0]), "name": row[1], "description": row[2], "status": str(row[3]),
             "user_id": str(row[4])} for row in results]
        return formatted_results
    finally:
        db.close()


def get_data_from_dify_v2(names: List[str]) -> List[Dict]:
    db = SessionDify()
    try:
        if names:
            query = db.query(DfApps.id, DfApps.name, DfApps.description, DfApps.status, DfApps.tenant_id) \
                .filter( DfApps.name.in_(names))
        else:
            query = db.query(DfApps.id, DfApps.name, DfApps.description, DfApps.status, DfApps.tenant_id)

        results = query.all()
        formatted_results = [
            {"id": str(row[0]), "name": row[1], "description": row[2], "status": "1",
             "user_id": str(row[4])} for row in results]
        return formatted_results
    finally:
        db.close()



def update_ids_in_local_v2(data: List[Dict], dialog_type:str):
    db = SessionLocal()
    agent_id_list = []
    print("----------------------------------------")
    print(data)
    print("*********************************************")
    try:
        for row in data:
            agent_id_list.append(row["id"])
            existing_agent = db.query(DialogModel).filter_by(id=row["id"]).first()
            if existing_agent:
                existing_agent.name = row["name"]
                existing_agent.description = row["description"]
            else:
                existing = DialogModel(id=row["id"], name=row["name"], description=row["description"], tenant_id=row["user_id"], dialog_type=dialog_type)
                db.add(existing)
        db.commit()
        for dialog in db.query(DialogModel).filter_by(dialog_type=dialog_type).all():
            if dialog.id not in agent_id_list:
                db.query(DialogModel).filter_by(id=dialog.id).delete()
                db.commit()
    except IntegrityError:
        db.rollback()
        raise
    finally:
        db.close()



def get_data_from_ragflow_knowledge():
    ...

def sync_agents_v2():
    db = SessionLocal()

    try:
        app_register = AppRegisterDao(db).get_apps()
        for app in app_register:
            if app["id"] == RAGFLOW:
                ragflow_data = get_data_from_ragflow_v2([])
                update_ids_in_local_v2(ragflow_data, "1")
            elif app["id"] == BISHENG:
                bisheng_data = get_data_from_bisheng_v2([])
                update_ids_in_local_v2(bisheng_data, "2")
            elif app["id"] == DIFY:
                dify_data = get_data_from_dify_v2([])
                update_ids_in_local_v2(dify_data, "4")
        print("Agents synchronized successfully")
    except Exception as e:
        print(f"Failed to sync agents: {str(e)}")



def sync_knowledge():
    db = SessionLocal()

    try:
        app_register = AppRegisterDao(db).get_apps()
        for app in app_register:
            if app["id"] == RAGFLOW:
                ragflow_data = get_data_from_ragflow_knowledge([])
                update_ids_in_local_v2(ragflow_data, "1")
            # elif app["id"] == BISHENG:
            #     bisheng_data = get_data_from_bisheng_v2([])
            #     update_ids_in_local_v2(bisheng_data, "2")
            # elif app["id"] == DIFY:
            #     dify_data = get_data_from_dify_v2([])
            #     update_ids_in_local_v2(dify_data, "4")
        print("Agents synchronized successfully")
    except Exception as e:
        print(f"Failed to sync agents: {str(e)}")


if __name__ == "__main__":
    a = get_data_from_dify_v2([])
    print(a)