知识服务管理

api/controllers/console/__init__.py

@@ -43,7 +43,7 @@ api.add_resource(AppImportConfirmApi, "/apps/imports/<string:import_id>/confirm"
 api.add_resource(AppImportCheckDependenciesApi, "/apps/imports/<string:app_id>/check-dependencies")
 
 # Import other controllers
-from . import admin, apikey, extension, feature, ping, setup, version
+from . import admin, apikey, extension, external_application, feature, ping, setup, version
 
 # Import app controllers
 from .app import (

api/controllers/console/external_application.py

@@ -0,0 +1,108 @@
+from flask import request
+from flask_login import login_required
+from flask_restful import Resource, marshal, marshal_with, reqparse
+from werkzeug.exceptions import NotFound
+
+from controllers.console import api
+from controllers.console.wraps import account_initialization_required, setup_required
+from fields.external_application_fields import external_application_fields
+from models.external_application import ExternalApplication
+from services.external_application_service import ExternalApplicationService
+
+
+class ExternalApplicationListApi(Resource):
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        page = request.args.get("page", default=1, type=int)
+        limit = request.args.get("limit", default=20, type=int)
+        search = request.args.get("search", default=None, type=str)
+        type = request.args.get("type", default=None, type=str)
+        url = request.args.get("url", default=None, type=str)
+        method = request.args.get("method", default=None, type=str)
+
+        external_applications, total = ExternalApplicationService.get_external_applications(
+            page, limit, search, type, url, method)
+
+        data = marshal(external_applications, external_application_fields)
+        response = {"data": data, "has_more": len(external_applications) == limit, "limit": limit,
+                    "total": total, "page": page}
+        return response, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(external_application_fields)
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name", location="json", nullable=False, required=True, help="Name must be between 1 to 50 characters."
+        )
+        parser.add_argument(
+            "type", type=str, location="json", choices=ExternalApplication.EXTERNAL_APPLICATION_TYPE_LIST,
+            nullable=False, help="Invalid external_application type."
+        )
+        parser.add_argument(
+            "url", type=str, location="json", nullable=False, help="Invalid external_application url."
+        )
+        parser.add_argument(
+            "method", type=str, location="json", nullable=False, help="Invalid external_application method."
+        )
+        parser.add_argument(
+            "status", type=bool, location="json", nullable=False, help="Invalid external_application status."
+        )
+        args = parser.parse_args()
+        external_application = ExternalApplicationService.save_external_application(args)
+        return external_application, 200
+
+class ExternalApplicationApi(Resource):
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(external_application_fields)
+    def get(self, external_application_id):
+        external_application = ExternalApplicationService.get_external_application(external_application_id)
+        if not external_application:
+            raise NotFound(f"ExternalApplication with id {id} not found.")
+        return external_application, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(external_application_fields)
+    def patch(self, external_application_id):
+        external_application_id = str(external_application_id)
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name", location="json", nullable=False, required=True, help="Name must be between 1 to 50 characters."
+        )
+        parser.add_argument(
+            "type", type=str, location="json", nullable=False, help="Invalid external_application type."
+        )
+        parser.add_argument(
+            "url", type=str, location="json", nullable=False, help="Invalid external_application url."
+        )
+        parser.add_argument(
+            "method", type=str, location="json", nullable=False, help="Invalid external_application method."
+        )
+        parser.add_argument(
+            "status", type=bool, location="json", nullable=False, help="Invalid external_application status."
+        )
+        args = parser.parse_args()
+        external_application = ExternalApplicationService.update_external_application(args, external_application_id)
+        return external_application, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, external_application_id):
+        external_application_id = str(external_application_id)
+        ExternalApplicationService.delete_external_application(external_application_id)
+        return 200
+
+api.add_resource(ExternalApplicationListApi, '/external_applications')
+api.add_resource(ExternalApplicationApi, '/external_applications/<external_application_id>')

api/fields/external_application_fields.py

@@ -0,0 +1,16 @@
+from flask_restful import fields
+
+from libs.helper import TimestampField
+
+external_application_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "type": fields.String,
+    "url": fields.String,
+    "method": fields.String,
+    "status": fields.Boolean,
+    "updated_at": TimestampField,
+    "updated_by": fields.String,
+    "created_at": TimestampField,
+    "created_by": fields.String,
+}

api/models/external_application.py

@@ -0,0 +1,24 @@
+from sqlalchemy import func
+
+from .engine import db
+from .types import StringUUID
+
+
+class ExternalApplication(db.Model):
+    __tablename__ = "external_applications"
+    __table_args__ = (
+        db.PrimaryKeyConstraint("id", name="external_application_pkey"),
+    )
+
+    EXTERNAL_APPLICATION_TYPE_LIST = ["QUESTION_ANSWER", "SEARCH", "RECOMMEND"]
+
+    id = db.Column(StringUUID, server_default=db.text("uuid_generate_v4()"))
+    name = db.Column(db.String(255), nullable=False)
+    type = db.Column(db.String(255), nullable=False)
+    url = db.Column(db.String(255), nullable=False)
+    method = db.Column(db.String(255), nullable=False)
+    status = db.Column(db.Boolean, nullable=False, server_default=db.text("true"))
+    created_by = db.Column(StringUUID, nullable=False)
+    created_at = db.Column(db.DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_by = db.Column(StringUUID, nullable=True)
+    updated_at = db.Column(db.DateTime, nullable=False, server_default=func.current_timestamp())

api/services/errors/external_application.py

@@ -0,0 +1,5 @@
+from services.errors.base import BaseServiceError
+
+
+class ExternalApplicationNameDuplicateError(BaseServiceError):
+    pass

api/services/external_application_service.py

@@ -0,0 +1,87 @@
+import uuid
+from datetime import datetime
+from typing import Optional
+
+from flask_login import current_user
+from werkzeug.exceptions import NotFound
+
+from models import db
+from models.external_application import ExternalApplication
+from services.errors.external_application import ExternalApplicationNameDuplicateError
+
+
+class ExternalApplicationService:
+    @staticmethod
+    def get_external_applications(page, per_page, search=None, type=None, url=None, method=None):
+        query = ExternalApplication.query.order_by(ExternalApplication.created_at.desc())
+
+        if search:
+            query = ExternalApplication.query.filter(ExternalApplication.name.ilike(f"%{search}%"))
+
+        if type:
+            query = ExternalApplication.query.filter(ExternalApplication.type == type)
+
+        if url:
+            query = ExternalApplication.query.filter(ExternalApplication.url == url)
+
+        if method:
+            query = ExternalApplication.query.filter(ExternalApplication.method == method)
+
+        external_applications = query.paginate(page=page, per_page=per_page, max_per_page=100, error_out=False)
+        return external_applications.items, external_applications.total
+
+    @staticmethod
+    def get_external_application(external_application_id: str) -> Optional[ExternalApplication]:
+        external_application = (ExternalApplication.query
+                                .filter(ExternalApplication.id == external_application_id).first())
+        return external_application
+
+    @staticmethod
+    def get_external_application_by_name(name: str) -> Optional[ExternalApplication]:
+        external_application = ExternalApplication.query.filter_by(name=name).first()
+        return external_application
+
+    @staticmethod
+    def save_external_application(args: dict) -> ExternalApplication:
+        name = args["name"]
+        external_application = ExternalApplicationService.get_external_application_by_name(name)
+        if external_application:
+            raise ExternalApplicationNameDuplicateError(f"ExternalApplication with name {name} already exists.")
+        external_application = ExternalApplication(
+            id=str(uuid.uuid4()),
+            name=name,
+            type=args["type"],
+            url=args["url"],
+            method=args["method"],
+            status=args["status"],
+            created_by=current_user.id,
+            updated_by=current_user.id,
+        )
+        db.session.add(external_application)
+        db.session.commit()
+        return external_application
+
+    @staticmethod
+    def update_external_application(args: dict, external_application_id: str) -> ExternalApplication:
+        external_application = ExternalApplicationService.get_external_application(external_application_id)
+        if not external_application:
+            raise NotFound("ExternalApplication not found")
+        external_application.name = args["name"]
+        external_application.type = args["type"]
+        external_application.url = args["url"]
+        external_application.method = args["method"]
+        external_application.status = args["status"]
+        external_application.updated_by = current_user.id
+        external_application.updated_at = datetime.now()
+        db.session.add(external_application)
+        db.session.commit()
+        return external_application
+
+    @staticmethod
+    def delete_external_application(external_application_id):
+        external_application = ExternalApplicationService.get_external_application(external_application_id)
+        if not external_application:
+            raise NotFound("ExternalApplication not found")
+        db.session.delete(external_application)
+        db.session.commit()
+        return external_application