from typing import Union
import uuid
import time
from urllib.parse import urlparse

from flask import Blueprint, request, g

from config import config
from consts.consts import ENV_PROD, Claude3, GPT_4, MODE_TYPE_BIRD, MODE_TYPE_SPIDER, ContextTypes, DataSummaryStatus, IdentifyTypes, GenSQLErrorMsg, ModuleType, AssignedAppID
from errors.errors import BadRequestException, InternalServerException
from handlers.base import abort_with, check_api_key_permission, check_org_id, fail, get_access_key, get_app_id, get_current_user_email, get_limit_offset, get_org_id_from_header, get_session_id, get_trace_id, get_user_id, get_user_token, inject_metrics, inject_metrics_for_oldapi, success, check_schema, check_permission
from models.api_key import OpenAIApiKey
from models.context import Context
from models.domain_knowledge import KnowledgeBase
from models.edatask import EDATask as EDATaskModel, EDATaskJob
from models.feedback import UserFeedback
from models.oplog import OpLog
from models.ratelimit import RateLimit
from models.user_preference import UserPreference, Preference, is_internal_api_enabled
from schema.schema import (
    CreateDomainItemSchema, CreateSessionSchema, LikeOrNotSchema, QueryExecutionSchema, SQLEditSchema, SuggestQuestionSchema,
    TaskExecutionSchema, UpdateColumnsDescriptionSchema, UpdateDomainItemSchema, UpdateEDATaskSchema,
    CheckConnectionSchema, CreateDataContextSchema, QuestionBreakdownSchema, UpdateSessionSchema,
    CreateSessionContextSchema, UpdateColumnDescriptionSchema, UpdateTableDescriptionSchema, RefineSQLSchema,
    Text2SQLSchema, CreateDataSummarySchema, Chat2DataSchema, SQLExecutionSchema, UpdateDataDescriptionSchema, UserFeedbackSchema,
    MultiRoundChat2SQLSchema, MultiRoundFeedbackSchema, CreateKnowledgeBaseSchema, UpdateKnowledgeBaseSchema,
)
from svc.dataapi import CrossRegionDataAPI
from svc.feishu import send_feishu_msg
from svc.openai import create_thread
from tasks.eda import task_gen_db_schema, task_sql_refine, task_suggest_questions, task_chat2data, task_text2sql
from tasks.utils import get_default_queue_length, enqueue
from utils.credential import get_cluster_info, is_apikey_belongs_to_org_id, get_org_ids_by_pubkey, user_has_permission
from utils.database import check_connection_pass, check_connection_pass_by_dataapi, unify_database_uri
from utils.logging import logging
from svc.data_explore import (
    update_table_description, update_column_description
)
from svc.eda_task import EDATask
from svc.measuring import measuring_api
from utils.local import set_data_context_id, set_session
from utils.sql import has_ddl_sql_permission, is_ddl_sql
from domain_info.index import remote_kg_management

edatask_bp = Blueprint("edatask_bp", __name__)
edatask_v1_bp = Blueprint("edatask_v1_bp", __name__)  # internal only, for compatible with old dataapps
edatask_v2_bp = Blueprint("edatask_v2_bp", __name__)
edatask_v3_bp = Blueprint("edatask_v3_bp", __name__)


def is_rate_limit_reatched():
    if config.DISABLE_RATE_LIMIT:
        logging.info("rate limit disabled, skip rate limit check")
        return False

    app_id = get_app_id()
    if app_id and OpenAIApiKey.get_api_key_by_app_id(app_id):
        logging.info("app_id %s use it's own api key, skip rate limit check", app_id)
        return False

    app_preference = UserPreference.load(app_id)
    if app_id and app_preference.llm_api_key:
        logging.info("app_id %s use it's own openai api key, skip rate limit check", app_id)
        return False

    org_id = get_org_id_from_header()
    assert org_id is not None

    org_preference = UserPreference.load_by_org_id(org_id)

    max_count = max(app_preference.rate_limit_per_day, org_preference.rate_limit_per_day)

    if RateLimit.by_org_id_per_day(org_id) > max_count:
        return True

    return False


def gen_job_id() -> str:
    return uuid.uuid4().hex


def get_dbname(context: Context) -> str:
    data = context.load_data()
    database_uri = data["database_uri"]
    parsed_url = urlparse(database_uri)
    db_name = parsed_url.path.removeprefix("/")

    return db_name


@edatask_v1_bp.post("/chat2data")
@edatask_v1_bp.post("/chatbot")
def handle_v1_chatbot(data_app_id: int):
    payload = request.get_json()
    logging.info("handle_v1_chatbot with payload %s", payload)
    bot_type = payload.get("bot_type", "chat2data")
    cluster_id = payload["cluster_id"]
    database = payload["database"]
    app_id = get_app_id()
    api_key = get_access_key()
    org_id = get_org_id_from_header()

    data_summary_id = try_find_data_summary(cluster_id, database, "", [DataSummaryStatus.DONE, DataSummaryStatus.INITED], org_id)
    if not data_summary_id:
        data_summary_id = init_new_data_summary(g.session, org_id, cluster_id, database, "", "", {}, app_id, api_key)

    data_context = Context.get_by_id(g.session, data_summary_id)

    session_context = gen_new_session_context(g.session, org_id, data_context, data_context.cluster_id, data_context.db_name)

    if bot_type == "chat2question":
        question_num = payload.get("question_num", 5)

        job_id = enqueue(
            task_suggest_questions, org_id, session_context.id, session_context.data_context_id,
            app_id, api_key, "", question_num=question_num, question_only=False,
        )
    else:
        evidence = ""
        sql_to_data = False
        data_to_chart = True if bot_type == "chat2chart" else False
        user_token = ""
        query_id = ""
        direct_resolve = True
        auto_run = True
        question = payload["instruction"]

        job_id = enqueue(
            task_chat2data, org_id,
            session_context.data_context_id, session_context.id, question, direct_resolve, auto_run,
            sql_to_data, data_to_chart, evidence, app_id, api_key, user_token, query_id=query_id,
            v3_format=True,
        )

    return success(result={"job_id": job_id})


@edatask_v1_bp.get("/chatbot/jobs/<job_id>")
def get_v1_chatbot_job_detail(data_app_id: int, job_id: str):
    edatask_job = EDATaskJob.get_by_job_id(g.session, job_id)
    if not edatask_job:
        return fail(404)

    job_detail = edatask_job.to_json()
    job_detail.pop("variables", None)
    job_detail.pop("output", None)

    return success(result=job_detail)


@edatask_bp.get("/jobs/<job_id>")
def get_edatask_job_detail(data_app_id: str, job_id: str):
    edatask_job = EDATaskJob.get_by_job_id(g.session, job_id)
    if not edatask_job:
        return fail(404)

    job_detail = edatask_job.to_json()
    job_detail.pop("variables", None)
    job_detail.pop("output", None)

    return success(result=job_detail)


@edatask_bp.put("/tasks/<int:edatask_id>")
def update_edatask(data_app_id: str, edatask_id: int):
    payload = check_schema(UpdateEDATaskSchema, request.get_json())

    item = EDATaskModel.get_by_id(g.session, edatask_id)
    if not item:
        return fail(404)

    item_by_url_path = EDATaskModel.get_by_url_path(g.session, payload["url_path"])
    if item_by_url_path:
        if item_by_url_path.id != item.id:
            return fail(409, "url_path duplicated")

    item.update(
        g.session, payload["name"], payload["url_path"], payload["method"], payload["description"],
        payload["timeout"], payload["prompt_template"], payload["exprs"],
    )

    return success()


@edatask_bp.delete("/tasks/<int:edatask_id>")
def delete_edatask(data_app_id: str, edatask_id: int):
    item = EDATaskModel.get_by_id(g.session, edatask_id)
    if not item:
        return fail(404)

    g.session.delete(item)

    return success()


@edatask_bp.get("/org_id")
@check_permission
def get_org_id_by_pubkey(data_app_id: str):
    access_key = get_access_key()

    org_ids = get_org_ids_by_pubkey(access_key)

    org_id = org_ids[0] if len(org_ids) else None

    result = {"org_id": org_id}
    return success(result=result)


@edatask_bp.post("/datacontext/connection_check")
@check_permission
def handle_connection_check(data_app_id: str):
    payload = check_schema(CheckConnectionSchema, request.get_json())
    database_uri = unify_database_uri(payload["database_uri"])

    connection_pass, table_count, message = check_connection_pass(database_uri)

    if connection_pass and table_count == 0:
        connection_pass = False
        message = "Your database is empty, please try another one."

    result = {
        "pass": connection_pass,
        "message": message,
    }
    return success(result=result)


@edatask_bp.post("/datacontext/context")
@check_permission
def create_data_context(data_app_id: str):
    payload = check_schema(CreateDataContextSchema, request.get_json())
    org_id = int(payload["org_id"])
    force = payload["force"]
    app_id = get_app_id()

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, org_id):
        return fail(403, "invalid access_key")

    database_uri = unify_database_uri(payload["database_uri"])
    logging.info("received request for create data context by %s, data_context_id: %s", str(payload), database_uri)

    if config.ENVIRONMENT == "prod" and access_key == "S0T0RH50" and "billing" in database_uri:
        return success(result={"data_context_id": 255549})

    connection_pass, table_count, msg = check_connection_pass(database_uri)
    if not connection_pass or table_count == 0:
        return fail(400, msg)

    context = Context.get_by_database_uri(g.session, database_uri)
    if context:
        ctx = context.load_data()
        if not ctx.get("db_schema"):
            logging.info("set force to True, db_schema: %s, %s", access_key, len(ctx.get("db_schema", {})))
            force = True
    else:
        logging.info("set force to True, context not found")
        force = True

    if force:
        context_name = "data-context"

        # default value
        domain_context_id, project_id, cluster_id, db_name, session_id = 0, 0, 0, "", ""

        context = Context.create(
            g.session, ContextTypes.DATA, context_name, org_id, project_id, cluster_id, db_name,
            session_id, payload["creator"], payload["data"], 0, domain_context_id, database_uri,
        )
        logging.info("create data context by %s, data_context_id: %s", str(payload), context.id)

        ctx = context.load_data()
        ctx["database_uri"] = database_uri
        ctx["app_id"] = app_id
        context.update_data(ctx)
        logging.info("save database_uri success")

    return success(result={"data_context_id": context.id})


@edatask_bp.post("/session/context")
@check_permission
def create_session_context(data_app_id: str):
    payload = check_schema(CreateSessionContextSchema, request.get_json())
    org_id = int(payload["org_id"])
    data_context_id = payload["data_context_id"]
    session_id = uuid.uuid4().hex

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, org_id):
        return fail(403, "invalid access_key")

    context_name = "session-context-of-{}".format(data_context_id)

    context = Context.create(
        g.session, ContextTypes.SESSION, context_name, org_id, 0, 0,
        "", session_id, payload["creator"], payload["data"],
        data_context_id, 0,
    )
    logging.info("create session context by %s, session_context_id: %s", str(payload), context.id)

    return success(result={"session_context_id": context.id})


@edatask_bp.post("/datacontext/<int:data_context_id>/actions/db_exploring")
@check_permission
def handle_db_exploring(data_app_id: str, data_context_id: int):
    data_context = Context.get_by_id(g.session, data_context_id)
    if not data_context:
        return fail(400, "data context not found")

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, data_context.org_id):
        return fail(403, "invalid access_key")

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    org_ids = get_org_ids_by_pubkey(access_key)
    org_id = str(org_ids[0]) if len(org_ids) else None
    job_id = enqueue(task_gen_db_schema, org_id, data_context_id)

    return success(result={"job_id": job_id})


@edatask_bp.post("/session/<int:session_context_id>/actions/question_suggestions")
@check_permission
def handle_question_suggestions(data_app_id: str, session_context_id: int):
    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    app_id = get_app_id()

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, session_context.org_id):
        return fail(403, "invalid access_key")

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    org_ids = get_org_ids_by_pubkey(access_key)
    org_id = str(org_ids[0]) if len(org_ids) else None
    job_id = enqueue(task_suggest_questions, org_id, session_context_id, session_context.data_context_id, app_id, access_key, "")

    return success(result={"job_id": job_id})


@edatask_bp.post("/session/<int:session_context_id>/actions/question_breakdown")
@check_permission
def handle_question_breakdown(data_app_id: str, session_context_id: int):
    payload = check_schema(QuestionBreakdownSchema, request.get_json())
    raw_question = payload["raw_question"]
    auto_run = payload["auto_run"]
    direct_resolve = payload["direct_resolve"]
    data_to_chart = payload["data_to_chart"]
    query_id = payload.get("query_id", "")
    app_id = get_app_id()

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, session_context.org_id):
        return fail(403, "invalid access_key")

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    org_ids = get_org_ids_by_pubkey(access_key)
    org_id = str(org_ids[0]) if len(org_ids) else None
    job_id = enqueue(
        task_chat2data, org_id, session_context.data_context_id, session_context_id,
        raw_question, direct_resolve, auto_run, True, data_to_chart, "", app_id, access_key, user_token=None,
        query_id=query_id,
    )

    return success(result={"job_id": job_id})


@edatask_bp.post("/session/<int:session_context_id>/actions/text2sql")
@check_permission
def handle_text2sql(data_app_id: str, session_context_id: int):
    payload = check_schema(Text2SQLSchema, request.get_json())
    task_id = payload["task_id"]
    question_id = payload["question_id"]

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    access_key = get_access_key()
    if not is_apikey_belongs_to_org_id(access_key, session_context.org_id):
        return fail(403, "invalid access_key")

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    org_ids = get_org_ids_by_pubkey(access_key)
    org_id = str(org_ids[0]) if len(org_ids) else None
    job_id = enqueue(task_text2sql, org_id, session_context.data_context_id, session_context_id, task_id, question_id)

    return success(result={"job_id": job_id})


@edatask_bp.put("/session/<int:session_context_id>/like_or_not")
@check_permission
def handle_like_or_not(data_app_id: str, session_context_id: int):
    payload = check_schema(LikeOrNotSchema, request.get_json())
    task_id = payload["task_id"]
    question_id = payload["question_id"]
    like_or_not = payload["like_or_not"]

    set_session(g.session)

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    inject_metrics_for_oldapi(session_context.id, session_context.data_context_id)
    eda = EDATask(session_context_id)
    eda.set_like_or_not(question_id, task_id, like_or_not)

    return success()


@edatask_bp.put("/session/<int:session_context_id>/sql")
@check_permission
def handle_update_sql(data_app_id: str, session_context_id: int):
    payload = check_schema(SQLEditSchema, request.get_json())
    task_id = payload["task_id"]
    question_id = payload["question_id"]
    sql = payload["sql"]

    set_session(g.session)

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    inject_metrics_for_oldapi(session_context.id, session_context.data_context_id)
    eda = EDATask(session_context_id)
    eda.update_task_sql(question_id, task_id, sql)

    return success()


@edatask_bp.post("/session/<int:session_context_id>/sql_execution")
@check_permission
def handle_execute_sql(data_app_id: str, session_context_id: int):
    payload = check_schema(TaskExecutionSchema, request.get_json())
    task_id = payload["task_id"]
    question_id = payload["question_id"]

    set_session(g.session)

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    set_data_context_id(session_context.data_context_id)
    inject_metrics_for_oldapi(session_context.id, session_context.data_context_id)
    eda = EDATask(session_context_id)
    result = eda.task_text2sql(question_id, task_id)

    return success(result=result)


@edatask_bp.put("/datacontext/<int:data_context_id>/table/description")
@check_permission
def handle_update_table_description(data_app_id: str, data_context_id: int):
    payload = check_schema(UpdateTableDescriptionSchema, request.get_json())
    table = payload["table"]
    description = payload["description"]

    set_data_context_id(data_context_id)
    set_session(g.session)

    data_context = Context.get_by_id(g.session, data_context_id)
    if not data_context:
        logging.info("data context by %s not found", data_context_id)
        return fail(404, "data context not found")

    update_table_description(table, description)

    return success()


@edatask_bp.put("/datacontext/<int:data_context_id>/column/description")
@check_permission
def handle_update_column_description_old(data_app_id: str, data_context_id: int):
    payload = check_schema(UpdateColumnDescriptionSchema, request.get_json())
    table = payload["table"]
    column = payload["column"]
    description = payload["description"]

    set_data_context_id(data_context_id)
    set_session(g.session)

    data_context = Context.get_by_id(g.session, data_context_id)
    if not data_context:
        logging.info("data context by %s not found", data_context_id)
        return fail(404, "data context not found")

    update_column_description(table, column, description)

    return success()


@edatask_bp.route("/generate_200", methods=["GET", "POST", "HEAD", "PUT", "DELETE"])
def generate_200(data_app_id: str):
    return success()


##################################################################################

@edatask_v2_bp.get("/dataSummaries")
@check_api_key_permission
def handle_get_data_summaries_v2(data_app_id: str):
    """
    Get Data Summaries
    ---
    get:
        description: Get data summary by given cluster_id and database, returns the data summary of the database.
        operationId: getDataSummaries
        parameters:
            - name: cluster_id
              in: query
              required: true
              schema:
                type: string
              description: The ID of the cluster to query
            - name: database
              in: query
              required: true
              schema:
                  type: string
              description: The name of the database to query
        responses:
            200:
                description: data summary detail
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ListDataSummaryV2Response"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    cluster_id = request.args.get("cluster_id", type=str)
    database = request.args.get("database", type=str)

    if not (cluster_id and database):
        return fail(400, "cluster_id and database is required")

    show_summary_detail = is_internal_api_enabled(get_current_user_email(), get_org_id_from_header())

    data_summary_id = None
    data = None
    first_updating_data_summary = None
    for data_context in Context.get_all_data_context_by_cluster_id_and_database(g.session, cluster_id, database):
        # return the first data summary with status done(if exists)
        # or return the first updating data summary
        data_summary_id = data_context.id
        data = data_context.load_data()
        if not first_updating_data_summary:  # only set first_updating_data_summary once
            first_updating_data_summary = data
        if data.get("status") == DataSummaryStatus.DONE:
            result={"data_summary_id": data_summary_id, "status": "done"}
            if show_summary_detail:
                result["data_summary"] = data
            return success(result=result)

    if data_summary_id is None:
        return fail(404, "No data summary found for the cluster and database")

    result={"data_summary_id": data_summary_id, "status": "updating"}
    if show_summary_detail:
        result["data_summary"] = first_updating_data_summary
    return success(result=result)


@edatask_v3_bp.get("/dataSummaries")
@check_api_key_permission
def handle_get_data_summaries(data_app_id: str):
    """
    Get Data Summaries
    ---
    get:
        description: Get data summary by given cluster_id and database, returns the data summary of the database.
        operationId: getDataSummaries
        parameters:
            - name: cluster_id
              in: query
              required: true
              schema:
                type: string
              description: The ID of the cluster to query
            - name: database
              in: query
              required: true
              schema:
                  type: string
              description: The name of the database to query
        responses:
            200:
                description: data summary detail
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ListDataSummaryResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    cluster_id = request.args.get("cluster_id", type=str)
    database = request.args.get("database", type=str)
    org_id = get_org_id_from_header()

    if not (cluster_id and database):
        return fail(400, "cluster_id and database is required")

    data_summaries = []
    for data_context in Context.get_all_data_context_by_cluster_id_and_database(g.session, cluster_id, database):
        if data_context.org_id != org_id:
            continue

        data_summaries.append(data_context.fmt_data_summary_v2(simple=True))

    return success(result={"data_summaries": data_summaries})


def try_find_data_summary(cluster_id, database, database_uri, status_filters, org_id):
    # try to find if there is an existing data context with the given cluster_id and database
    # if we found, we can use the existing data context
    if cluster_id and database:
        data_contexts = Context.get_all_data_context_by_cluster_id_and_database(g.session, cluster_id, database)
    else:
        data_contexts = Context.get_all_data_context_by_database_uri(g.session, database_uri, org_id)

    for status in status_filters:
        for data_context in data_contexts:
            data = data_context.load_data()
            logging.info("try_find_data_summary, data_context_id: %s, data summary status: %s, is_default: %s", data_context.id, data.get("status"), data_context.is_default)

            if data.get("status") == status and data_context.is_default:
                return data_context.id

    # fallback
    for status in status_filters:
        for data_context in data_contexts:
            data = data_context.load_data()
            logging.info("try_find_data_summary, data_context_id: %s, data summary status: %s", data_context.id, data.get("status"))

            if data.get("status") == status:
                return data_context.id

    return None


def init_new_data_summary(s, org_id, cluster_id, database, database_uri, created_by, data: dict, app_id, public_key, user_token: Union[str, None] = None):
    domain_context_id, project_id, session_id = 0, "", "",

    if cluster_id and database:
        try:
            cluster_info = get_cluster_info(cluster_id)["cluster"]
        except Exception:
            raise InternalServerException(GenSQLErrorMsg.GET_DATABASE_COLUMN_ERROR)

        region = cluster_info["region_name"]
        project_id = cluster_info["project_id"]
        cloud_provider = cluster_info.get("cloud_provider_name", "")
        data.update({
            "project_id": project_id,
            "region": region,
            "cluster_id": cluster_id,
            "database": database,
            "cloud_provider": cloud_provider,
        })

        logging.info("check connection pass by dataapi, data: %s", str(data))
        connection_pass, _, msg, code  = check_connection_pass_by_dataapi(data, app_id, public_key, user_token, trace_id=get_trace_id())
        if not connection_pass:
            logging.info("connection_pass is %s, return 400, error msg: %s, original msg and code: %s, %s", connection_pass, GenSQLErrorMsg.GET_DATABASE_COLUMN_ERROR, msg, code)
            abort_with(400, GenSQLErrorMsg.GET_DATABASE_COLUMN_ERROR)
    elif database_uri:
        data.update({
            "database_uri": database_uri,
        })

    context_name = "data-context"
    data_summary = Context.create(
        s, ContextTypes.DATA, context_name, org_id, project_id, cluster_id, database,
        session_id, created_by, data, 0, domain_context_id, database_uri,
    )
    s.commit()

    return data_summary.id


@edatask_v2_bp.post("/dataSummaries")
@edatask_v3_bp.post("/dataSummaries")
@check_api_key_permission
def handle_create_data_summaries(data_app_id: str):
    """
    Generate Data Summaries
    ---
    post:
        description: This is an async API, analysis the database and column schema, explore the structure of your database, the response contains a data_summary_id and job_id, use the /v2/jobs to query the job status, when the status changed to done, you can explore dataset by using /v2/chat2data.
        operationId: createDataSummaries
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/CreateDataSummary"
    responses:
        200:
            description: if job started
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/CreateDataSummaryResponse"
        400:
            description: if params invalid
        403:
            description: if permission denied
    """
    time_before = int(time.time())
    payload = check_schema(CreateDataSummarySchema, request.get_json())
    try:
        cluster_id = int(payload["cluster_id"] or 0)
    except ValueError:
        return fail(400, "cluster_id must be integer")
    reuse = payload["reuse"]
    default = payload["default"]
    database_uri = payload["database_uri"]
    database = payload["database"]
    creator = payload["creator"]
    description = payload["description"]
    logging.info("create data context, payload: %s", str(payload))
    if not (database_uri or (cluster_id and database)):
        return fail(400, "database_uri or (cluster_id with database) is required")

    org_id = get_org_id_from_header()
    public_key = get_access_key()
    app_id = get_app_id()
    created_by = get_current_user_email()
    project_id = ""
    data = {
        "org_id": org_id,
        "app_id": app_id,
        "public_key": public_key,
        "created_by": created_by,
    }

    if is_rate_limit_reatched():
        return fail(429, "Too many requests in your org today")

    if database_uri:
        if database_uri.startswith("spider://"):
            preference = UserPreference.load(app_id)
            preference.mode_type = MODE_TYPE_SPIDER
            Preference.upsert(g.session, app_id, preference.to_json())
        elif database_uri.startswith("bird://"):
            preference = UserPreference.load(app_id)
            preference.mode_type = MODE_TYPE_BIRD
            Preference.upsert(g.session, app_id, preference.to_json())

        database_uri = unify_database_uri(payload["database_uri"])
        connection_pass, table_count, msg = check_connection_pass(database_uri)
        logging.info("check connection pass by dataapi, connection_pass: %s, time_cost: %s, table_count: %s, msg: %s, database_uri: %s", connection_pass, str(int(time.time()) - time_before), table_count, msg, database_uri)
        if (not connection_pass) or table_count == 0:
            return fail(400, msg)

        data.update({
            "database_uri": database_uri,
        })
    else:
        # cluster binding mode
        try:
            cluster_info = get_cluster_info(cluster_id)["cluster"]
        except Exception:
            return fail(400, "Failed to connect your database")

        region = cluster_info["region_name"]
        project_id = cluster_info["project_id"]
        tenant_id = cluster_info["tenant_id"]
        cloud_provider = cluster_info.get("cloud_provider_name", "")
        if str(org_id) != str(tenant_id):
            return fail(400, "You can only connect clusters in your org")

        data.update({
            "project_id": project_id,
            "cluster_id": cluster_id,
            "region": region,
            "database": database,
            "cloud_provider": cloud_provider,
        })

        connection_pass, table_count, msg, code = check_connection_pass_by_dataapi(data, app_id, public_key, trace_id=get_trace_id())
        if not connection_pass:
            return fail(code, msg)

    measuring_api()

    if reuse:
        data_summary_id = try_find_data_summary(cluster_id, database, database_uri, [DataSummaryStatus.DONE], org_id)
        logging.info("try_find_data_summary, data_context_id: %s, time_cost: %s", str(data_summary_id), str(int(time.time()) - time_before))
        if data_summary_id:
            job_id = enqueue(task_gen_db_schema, org_id, data_summary_id, set_default=default)
            return success(result={"data_summary_id": data_summary_id, "job_id": job_id})

    # if we can't find an existing and ready data context, we need to create a new one
    # default value
    domain_context_id, session_id = 0, ""
    context_name = "data-context"
    context = Context.create(
        g.session, ContextTypes.DATA, context_name, org_id, project_id, cluster_id, database,
        session_id, creator, data, 0, domain_context_id, database_uri, description=description,
    )
    # if it is a new data summary by (cluster_id + database), set it as default automatically
    if cluster_id and database:
        all_data_summaries = Context.get_all_data_context_by_cluster_id_and_database(g.session, cluster_id, database)
        if len(all_data_summaries) == 1:
            context.update_description_and_default(g.session, description, True)

    logging.info("create data context by %s, with data %s, data_context_id: %s", str(payload), str(data), context.id)
    g.session.commit()

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    metrics = {
        "type": "create_data_summary",
    }
    job_id = enqueue(
        task_gen_db_schema, org_id, context.id, set_default=default, metrics=metrics,
    )

    OpLog.create(g.session, OpLog.EVENT_CREATE_DATA_SUMMARY, public_key, created_by, job_id, cluster_id, database, context.id)

    result = {
        "data_summary_id": context.id,
        "job_id": job_id,
    }

    return success(result=result)


@edatask_v3_bp.get("/dataSummaries/<int:data_summary_id>")
@check_api_key_permission
def handle_get_data_summary(data_app_id: str, data_summary_id: int):
    """
    Get Data Summary Detail
    ---
    get:
        description: Get data summary detail by data_summary_id
        operationId: getDataSummaryDetail
        parameters:
            - name: data_summary_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the data summary to query
        responses:
            200:
                description: data summary detail
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/DataSummaryDetail"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    check_org_id(context.org_id)

    result = context.fmt_data_summary_v2()

    measuring_api()

    return success(200, "", result)


@edatask_v3_bp.put("/dataSummaries/<int:data_summary_id>")
@check_api_key_permission
def handle_update_data_summary(data_app_id: str, data_summary_id: int):
    """
    Update Data Summary
    ---
    put:
        description: Update data summary by data_summary_id
        operationId: updateDataSummary
        parameters:
            - name: data_summary_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the data summary to update
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateDataSummary"
        responses:
            200:
                description: if update success
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/DataSummaryDetail"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    payload = check_schema(UpdateDataDescriptionSchema, request.get_json())
    description = payload["description"]
    is_default = payload["default"]

    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    check_org_id(context.org_id)

    context.update_description_and_default(g.session, description, is_default)

    return success(result=context.fmt_data_summary_v2(simple=False))


@edatask_v3_bp.put("/dataSummaries/<int:data_summary_id>/tables/<table_name>")
@check_api_key_permission
def handle_update_table_description(data_app_id: str, data_summary_id: int, table_name: str):
    """
    Update Table Description
    ---
    put:
        description: Update table description by data_summary_id and table name
        operationId: updateTableDescription
        parameters:
            - name: data_summary_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the data summary to update
            - name: table_name
              in: path
              required: true
              schema:
                type: string
              description: The name of the table to update
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateTableDescription"
        responses:
            200:
                description: if update success
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/DataSummaryDetail"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    payload = check_schema(UpdateTableDescriptionSchema, request.get_json())
    description = payload["description"]

    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    check_org_id(context.org_id)

    context.update_table_description(g.session, table_name, description)

    return success(result=context.fmt_data_summary_v2(simple=False))


@edatask_v3_bp.put("/dataSummaries/<int:data_summary_id>/tables/<table_name>/columns")
@check_api_key_permission
def handle_update_column_description(data_app_id: str, data_summary_id: int, table_name: str):
    """
    Update Columns Description
    ---
    put:
        description: Update columns description by data_summary_id, table name and column name
        operationId: updateColumnsDescription
        parameters:
            - name: data_summary_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the data summary to update
            - name: table_name
              in: path
              required: true
              schema:
                type: string
              description: The name of the table to update
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateColumnsDescription"
        responses:
            200:
                description: if update success
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/DataSummaryDetail"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    payload = check_schema(UpdateColumnsDescriptionSchema, request.get_json())
    columns = payload["columns"]
    update_type = request.args.get("updateType", "override")
    if update_type not in ("override", "merge"):
        return fail(400, "invalid updateType")

    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    check_org_id(context.org_id)

    context.update_columns_description(g.session, table_name, columns, update_type)

    return success(result=context.fmt_data_summary_v2(simple=False))


def gen_new_session_context(session, org_id, data_context, cluster_id, database):
    context_name = "session-context-of-{}".format(data_context.id)
    session_context = Context.create(
        session, ContextTypes.SESSION, context_name, org_id, 0, cluster_id,
        database, "", data_context.creator, {}, data_context.id, 0,
    )
    session.add(session_context)
    session.flush()

    return session_context


@edatask_v3_bp.post("/suggestQuestions")
@check_api_key_permission
def handle_generate_suggest_questions(data_app_id: str):
    """
    Suggest Questions
    ---
    post:
        description: This is an async API, suggest questions based on the data summary, the response contains a job_id, use the /v2/jobs to query the job status
        operationId: suggestQuestions
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/SuggestQuestion"
        responses:
            200:
                description: if job started
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/SuggestQuestionResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data summary not found
    """
    payload = check_schema(SuggestQuestionSchema, request.get_json())
    cluster_id = payload["cluster_id"]
    database = payload["database"]
    database_uri = payload["database_uri"]
    if database_uri:
        database_uri = unify_database_uri(database_uri)
    app_id = get_app_id()
    public_key = get_access_key()

    if not (cluster_id and database) and not database_uri:
        return fail(400, "cluster_id with database or database_uri is required")

    org_id = get_org_id_from_header()
    status_filters = [DataSummaryStatus.DONE, DataSummaryStatus.INITED]
    data_summary_id = try_find_data_summary(cluster_id, database, database_uri, status_filters, org_id)
    if not data_summary_id:
        data_summary_id = init_new_data_summary(g.session, org_id, cluster_id, database, database_uri, "", {}, app_id, public_key)

    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    check_org_id(context.org_id)

    set_data_context_id(data_summary_id)
    set_session(g.session)

    org_id = check_org_id(context.org_id)

    if is_rate_limit_reatched():
        return fail(429, "Too many requests in your org today")

    # as we need to keep api v1, so we still need to create a session context here. when we drop the support of
    # v1(used by tiinsight), update the arguments of task_suggest_questions to drop session context id
    session_context = gen_new_session_context(g.session, org_id, context, context.cluster_id, context.db_name)

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    job_id = enqueue(
        task_suggest_questions, org_id, session_context.id, session_context.data_context_id,
        app_id, public_key, ""
    )

    result = {
        "job_id": job_id,
    }

    measuring_api()

    return success(200, "", result)


@edatask_v2_bp.get("/dataSummaries/<int:data_summary_id>/suggestedQuestions")
@check_api_key_permission
def handle_get_generated_suggest_questions(data_app_id: str, data_summary_id: int):
    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    set_data_context_id(data_summary_id)
    set_session(g.session)

    check_org_id(context.org_id)

    ctx = context.load_data()
    result = ctx.get("suggest_questions")

    measuring_api()

    return success(200, "", result)


def _handle_chat2data(sql_to_data: bool, sqleditor_mode: bool = False, project_id: int = 0, v3_format=False, data_to_chart=False):
    payload = check_schema(Chat2DataSchema, request.get_json())
    data_summary_id = payload["data_summary_id"]
    raw_question = payload["raw_question"] or payload["question"]
    evidence = payload.get("evidence", "")
    query_id = payload.get("query_id", "")
    direct_resolve = payload["sql_generate_mode"] == "direct"
    # flag can be set by user
    flags = {"auto_run": True, "data_to_chart": data_to_chart}
    # update by user
    for key, value in payload.get("flags", {}).items():
        flags[key] = value

    if (not sqleditor_mode) and flags["auto_run"]:
        sql_to_data = True

    # api: chat2data which need to return data, must resolve directly
    if sql_to_data:
        direct_resolve = True

    cluster_id = payload["cluster_id"]
    database = payload["database"]
    api_key = get_access_key()
    created_by = get_current_user_email()
    org_id = get_org_id_from_header()
    app_id = get_app_id()
    data = {
        "org_id": org_id,
        "created_by": created_by,
        "cluster_id": cluster_id,
        "database": database,
        "public_key": api_key,
    }

    user_token = get_user_token()
    # 检查是否有data_summary_id，如果没有，就要根据cluster_id和database来创建一个新的data_summary
    if sqleditor_mode:
        resource_uri = f"orgs/{org_id}/projects/{project_id}/clusters/{cluster_id}/connections/list"
        if not user_has_permission(user_token, resource_uri):
            return fail(403, "You don't have permission to access this cluster")

        if not cluster_id:
            return fail(400, GenSQLErrorMsg.INVALID_PARAMS_ERROR)
        if not database:
            return fail(400, GenSQLErrorMsg.NO_DATABASE_GUESSED)
    else:
        if not ((cluster_id and database) or data_summary_id):
            return fail(400, "cluster_id with database or data_summary_id is required")

    # 新的OpenAPI 接口，用cluster_id + database 来访问
    if cluster_id and database:
        if sqleditor_mode:
            data.update({
                "app_id": AssignedAppID.SQLEDITOR,
            })
            data_summary_id = init_new_data_summary(g.session, org_id, cluster_id, database, "", created_by, data, app_id, api_key, user_token=user_token)
        else:
            # find the data_summary_id by cluster_id and database
            status_filters = [DataSummaryStatus.DONE, DataSummaryStatus.INITED] if sqleditor_mode else [DataSummaryStatus.DONE]
            data_summary_id = try_find_data_summary(cluster_id, database, "", status_filters, org_id)

            if not data_summary_id:
                if sqleditor_mode:
                    user_token = get_user_token()
                    data_summary_id = init_new_data_summary(g.session, org_id, cluster_id, database, "", created_by, data, app_id, api_key, user_token=user_token)
                else:
                    return fail(400, "No data summary generated for the cluster and database, please use /v2/dataSummaries to generate one")
    else:
        # 用户传的是data_summary_id，直接用
        # 还有一种情况，就是老的OpenAPI 接口，用data_summary_id 来访问，就直接走下面的检查
        pass

    context = Context.get_by_id(g.session, data_summary_id)
    if not context:
        return fail(404, "data summary not found")

    set_data_context_id(data_summary_id)
    set_session(g.session)

    org_id = check_org_id(context.org_id)

    if not sqleditor_mode:
        if is_rate_limit_reatched():
            return fail(429, "Too many requests in your org today")

    session_context = gen_new_session_context(g.session, org_id, context, cluster_id, database)

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    g.session.commit()  # commit before enqueue job

    metrics = {
        "sql_generate_mode": payload["sql_generate_mode"],
        "sqleditor_mode": sqleditor_mode,
        "type": "chat2data",
        "question": raw_question,
    }
    job_id = enqueue(
        task_chat2data, org_id,
        session_context.data_context_id, session_context.id, raw_question, direct_resolve, flags["auto_run"],
        sql_to_data, flags["data_to_chart"], evidence, app_id, api_key, user_token, query_id=query_id,
        v3_format=v3_format, metrics=metrics,
    )

    OpLog.create(g.session, OpLog.EVENT_CHAT2DATA, api_key, created_by, job_id, cluster_id, database, context.id, raw_question)

    if not sqleditor_mode:
        measuring_api()

    if v3_format:
        result = {
            "job_id": job_id,
            "session_id": session_context.id,
            "cluster_id": cluster_id,
            "database": database,
        }
    else:
        result={
            "job_id": job_id,
            "data_summary_id": data_summary_id,
            "session_context_id": session_context.id,
            "session_id": get_session_id(),
        }

    return success(result=result)


@edatask_v2_bp.post("/chat2data")
@check_api_key_permission
def handle_chat2data(data_app_id: str):
    """
    Chat2Data
    ---
    post:
        description: This is an async API, Chat2Data generate a SQL by given question described in nature language, the response contains a job_id, use the /v2/jobs to query the job status, when the status changed from running to done, get generated sql and execution result from response.
        operationId: chat2data
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/Chat2Data"
        responses:
            200:
                description: Job initiates successfully, it will generate a unique job ID that can be used for detailed tracking and specific inquiries.
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/Chat2DataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    return _handle_chat2data(True, data_to_chart=False)


@edatask_v3_bp.post("/chat2data")
@check_api_key_permission
def handle_chat2data_v3(data_app_id: str):
    """
    Chat2Data V3
    ---
    post:
        description: This is an async API, Chat2Data generate a SQL by given question described in nature language, the response contains a job_id, use the /v2/jobs to query the job status, when the status changed from running to done, get generated sql and execution result from response.
        operationId: chat2data
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/Chat2Data"
        responses:
            200:
                description: Job initiates successfully, it will generate a unique job ID that can be used for detailed tracking and specific inquiries.
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/Chat2DataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    return _handle_chat2data(True, v3_format=True, data_to_chart=True)


@edatask_v2_bp.post("/chat2sql")
@check_api_key_permission
def handle_chat2sql(data_app_id: str):
    return _handle_chat2data(False)


@edatask_v2_bp.post("/executeSQL")
@check_api_key_permission
def handle_execute_sql(data_app_id: str):
    payload = check_schema(SQLExecutionSchema, request.get_json())
    session_context_id = payload["session_context_id"]
    task_id = payload["task_id"]
    question_id = payload["question_id"]

    set_session(g.session)

    if is_rate_limit_reatched():
        return fail(429, "Too many requests in your org today")

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "data context not found")

    set_data_context_id(session_context.data_context_id)
    eda = EDATask(session_context_id)
    result = eda.task_text2sql(question_id, task_id)

    measuring_api()

    return success(result=result)


@edatask_v2_bp.get("/jobs/<job_id>")
@check_api_key_permission
def handle_query_job_detail(data_app_id: str, job_id: str):
    """
    Query EDATask Job Detail By JobID
    ---
    get:
        description: Query the job detail of async job, the status indicates the job status, which can be init, running, done, failed.
        operationId: queryJobDetail
        parameters:
            - name: job_id
              in: path
              required: true
              schema:
                type: string
              description: The ID of the job to query.
        responses:
            200:
                description: if job started
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/JobDetail"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    edatask_job = EDATaskJob.get_by_job_id(g.session, job_id)
    if not edatask_job:
        return fail(404)

    check_org_id(edatask_job.org_id)

    job_detail = edatask_job.to_json()
    job_detail.pop("variables", None)
    job_detail.pop("output", None)

    measuring_api()

    return success(result=job_detail)


@edatask_v2_bp.post("/executeQuery")
@check_api_key_permission
def handle_execute_query(data_app_id: str):
    """
    Execute SQL Query
    ---
    post:
        description: Execute SQL query by given cluster_id and database, returns the execution result of the query.
        operationId: executeQuery
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/ExecuteQuery"
        responses:
            200:
                description: SQL Execution result
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ExecuteQueryResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    user_email = get_current_user_email()
    if not is_internal_api_enabled(user_email, get_org_id_from_header()):
        return fail(403, "Only PingCAP employees can use this API(your email: {})".format(user_email))

    payload = check_schema(QueryExecutionSchema, request.get_json())
    try:
        cluster_id = int(payload["cluster_id"])
    except ValueError:
        return fail(400, "cluster_id must be integer")

    database = payload["database"]
    sql = payload["sql"]
    limit = payload["limit"]
    org_id = get_org_id_from_header()

    try:
        cluster_info = get_cluster_info(cluster_id)["cluster"]
        region = cluster_info["region_name"]
        project_id = cluster_info["project_id"]
        tenant_id = cluster_info["tenant_id"]
        cloud_provider = cluster_info.get("cloud_provider_name", "")
        if str(org_id) != str(tenant_id):
            return fail(400, "You can only connect clusters in your org")
    except Exception:
        return fail(400, "Failed to connect your database")

    api_key = get_access_key()
    app_id = get_app_id()
    data = {
        "org_id": org_id,
        "app_id": app_id,
        "public_key": api_key,
        "project_id": project_id,
        "cluster_id": cluster_id,
        "region": region,
        "database": database,
        "cloud_provider": cloud_provider,
    }

    connection_pass, _, msg, code = check_connection_pass_by_dataapi(data, app_id, api_key, trace_id=get_trace_id())
    if not connection_pass:
        return fail(code, msg)

    set_session(g.session)

    if is_rate_limit_reatched():
        return fail(429, "Too many requests in your org today")

    if is_ddl_sql(sql) and not has_ddl_sql_permission(api_key, app_id):
        return fail(403, "You don't have permission to execute this SQL, please check the role of your api key")

    try:
        dataapi_client = CrossRegionDataAPI(api_key, app_id)
        resp = dataapi_client.exec_sql_in_o11y_with_data_context(data, sql, limit)
        resp_json = resp.json()
        result = resp_json["data"]
        return success(result=result)
    except BadRequestException as e:
        return fail(400, str(e))


@edatask_v2_bp.post("/feedback")
@check_api_key_permission
def handle_user_feedback(data_app_id: str):
    """
    User Feedback
    ---
    post:
        description: User feedback for Chat2Data
        operationId: userFeedback
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UserFeedback"
        responses:
            200:
                description: User feedback response
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/UserFeedbackResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    user_email = get_current_user_email()
    if not is_internal_api_enabled(user_email, get_org_id_from_header()):
        return fail(403, "Only PingCAP employees can use this API(your email: {})".format(user_email))

    payload = check_schema(UserFeedbackSchema, request.get_json())
    session_context_id = payload["session_context_id"]
    question_id = payload["question_id"]
    feedback = payload["feedback"]
    comment = payload["comment"]
    created_by = get_current_user_email()

    raw_question = ""
    sql = ""
    session_context = Context.get_session_context_by_id(g.session, session_context_id)
    if session_context:
        data = session_context.load_data()
        task_tree = data.get("task_tree", {})
        section = task_tree.get("0", {})
        sql = section.get("sql", "")
        raw_question = data.get("raw_question", "")

    feedback = UserFeedback.create(
        g.session, created_by, session_context_id, question_id, feedback, comment, raw_question, sql,
    )

    if config.ENVIRONMENT == ENV_PROD:
        content = f"New feedback: {comment}, from: {created_by}, id: {feedback.id}, raw_question: {raw_question}, sql: {sql}"
        send_feishu_msg(content)

    return success()


@edatask_v3_bp.get("/sessions")
@check_api_key_permission
def handle_get_session_contexts(data_app_id: str):
    """
    Get Session Contexts
    ---
    get:
        description: Get session contexts by app_id
        operationId: getSessionContexts
        parameters:
            - name: app_id
              in: query
              required: true
              schema:
                type: string
              description: The ID of the app to query
            - name: page
              in: query
              required: false
              schema:
                type: integer
              description: The page of the result
            - name: per_page
              in: query
              required: false
              schema:
                type: integer
              description: The number of items per page
        responses:
            200:
                description: session contexts
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/SessionContexts"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    app_id = get_app_id()
    limit, offset = get_limit_offset()

    sessions = []
    items, count = Context.get_all_session_contexts_by_identifier_and_type(g.session, IdentifyTypes.APP_ID, app_id, limit, offset)
    for i in items:
        sessions.append(i.fmt_session_context_v2(simple=True))

    return success(result={"sessions": sessions, "count": count})


def _handle_create_session_context(sqleditor_mode=False):
    time_before = int(time.time())
    payload = check_schema(CreateSessionSchema, request.get_json())
    cluster_id = payload["cluster_id"]
    database = payload["database"]
    database_uri = payload["database_uri"]
    if database_uri:
        database_uri = unify_database_uri(database_uri)
    name = payload["name"]
    org_id = get_org_id_from_header()
    public_key = get_access_key()
    app_id = get_app_id()
    created_by = get_current_user_email()
    if not (database_uri or (cluster_id and database)):
        return fail(400, "cluster_id and database are required")

    data = {
        "org_id": org_id,
        "app_id": app_id,
        "public_key": public_key,
        "created_by": created_by,
    }
    status_filters = [DataSummaryStatus.DONE]

    old_session_context_id = payload["session_context_id"]

    qa_list = []
    old_session_context = Context.get_session_context_by_id(g.session, old_session_context_id)
    if old_session_context_id and not old_session_context:
        return fail(404, "session context not found")

    data_summary_id = try_find_data_summary(cluster_id, database, database_uri, status_filters, org_id)
    logging.info(
        "try_find_data_summary, time_cost: %s, cluster_id: %s, database: %s, database_uri: %s, data_summary_id: %s",
        str(int(time.time()) - time_before), cluster_id, database, database_uri, data_summary_id,
    )
    if not data_summary_id:
        return fail(400, GenSQLErrorMsg.DATA_CONTEXT_NOT_COMPLETE_ERROR)

    # create session context
    context_name = "{}-session".format(database) if name == "" else name
    session_id = uuid.uuid4().hex

    logging.info("create session context, context_name: %s, session_id: %s, data_summary_id: %s", context_name, session_id, data_summary_id)
    thread_id = create_thread(app_id, sqleditor_mode=sqleditor_mode, qa_list=qa_list)
    logging.info("create openai thread success, context_name: %s, session_id: %s, thread_id: %s, time_cost: %s", context_name, session_id, data_summary_id, str(int(time.time()) - time_before))
    identifier = str(get_user_id()) if sqleditor_mode else str(app_id)
    identifier_type = IdentifyTypes.USER_ID if sqleditor_mode else IdentifyTypes.APP_ID
    data = old_session_context.load_data() if old_session_context else {}

    session_context = Context.create(
        g.session, ContextTypes.SESSION_CHAT, context_name, org_id, 0, cluster_id,
        database, session_id, created_by, data, data_summary_id, 0, "", identifier=identifier, identifier_type=identifier_type,
        thread_id=thread_id,
    )

    return success(result=session_context.fmt_session_context_v2(simple=False))


@edatask_v3_bp.post("/sessions")
@check_api_key_permission
def handle_create_session_context(data_app_id: str):
    """
    Create Session Context
    ---
    post:
        description: Create session context
        operationId: createSessionContext
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/CreateSession"
        responses:
            200:
                description: session context created
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/SessionContextDetails"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    return _handle_create_session_context(False)


def _handle_get_session_context(session_context_id: int, sqleditor_mode: bool = False):
    context = Context.get_by_id(g.session, session_context_id)
    if not context:
        return fail(404, "session context not found")

    check_org_id(context.org_id)

    result = context.fmt_session_context_v2(simple=False)

    return success(200, "", result)


@edatask_v3_bp.get("/sessions/<int:session_id>")
@check_api_key_permission
def handle_get_session_context(data_app_id: str, session_id: int):
    """
    Get Session Context
    ---
    get:
        description: Get session context by session_id
        operationId: getSessionContext
        parameters:
            - name: session_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the session context to query
        responses:
            200:
                description: session context
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/SessionContextDetails"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if session context not found
    """
    return _handle_get_session_context(session_id, False)


def _handle_reset_session_context(session_context_id: int, sqleditor_mode: bool = False):
    context = Context.get_by_id(g.session, session_context_id)
    if not context:
        return fail(404, "session context not found")

    check_org_id(context.org_id)

    app_id = get_app_id()
    thread_id = create_thread(app_id, sqleditor_mode)

    context.reset_session_context(g.session, thread_id)

    return success(result=context.fmt_session_context_v2(simple=False))


@edatask_v3_bp.put("/sessions/<int:session_id>/reset")
@check_api_key_permission
def handle_reset_session_context(data_app_id: str, session_id: int):
    """
    Reset Session Context
    ---
    put:
        description: Reset session context by session_id
        operationId: resetSessionContext
        parameters:
            - name: session_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the session context to reset
        responses:
            200:
                description: if reset success
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if session context not found
    """
    return _handle_reset_session_context(session_id, False)


def _handle_update_session_context(session_context_id: int, sqleditor_mode: bool = False):
    payload = check_schema(UpdateSessionSchema, request.get_json())
    name = payload["name"]

    context = Context.get_by_id(g.session, session_context_id)
    if not context:
        return fail(404, "session context not found")

    check_org_id(context.org_id)

    if name:
        context.name = name
        g.session.commit()

    return success(result=context.fmt_session_context_v2(simple=False))


@edatask_v3_bp.put("/sessions/<int:session_id>")
@check_api_key_permission
def handle_update_session_context(data_app_id: str, session_id: int):
    """
    Update Session Context
    ---
    put:
        description: Update session context by session_id
        operationId: updateSessionContext
        parameters:
            - name: session_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the session context to update
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateSession"
        responses:
            200:
                description: if update success
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if session context not found
    """
    return _handle_update_session_context(session_id, False)


def _handle_sql_refine_fix(sqleditor_mode: bool = False):
    payload = check_schema(RefineSQLSchema, request.get_json())
    cluster_id = payload["cluster_id"]
    database = payload["database"]
    sql = payload["sql"]
    raw_question = payload["raw_question"] or payload["question"]
    feedback = payload["feedback"]
    sql_error = payload["sql_error"]
    org_id = get_org_id_from_header()
    database_uri = ""
    created_by = get_current_user_email()
    app_id = get_app_id()
    public_key = get_access_key()
    data = {
        "org_id": org_id,
        "created_by": created_by,
        "cluster_id": cluster_id,
        "database": database,
    }

    if is_rate_limit_reatched():
        return fail(429, "Too many requests in your org today")

    status_filters = [DataSummaryStatus.DONE, DataSummaryStatus.INITED]
    data_summary_id = try_find_data_summary(cluster_id, database, None, status_filters, org_id)
    logging.info("try_find_data_summary, cluster_id: %s, database: %s, data_summary_id: %s", cluster_id, database, data_summary_id)
    user_token = get_user_token() if sqleditor_mode else None
    if not data_summary_id:
        data_summary_id = init_new_data_summary(g.session, org_id, cluster_id, database, database_uri, created_by, data, app_id, public_key, user_token=user_token)

        logging.info("data_summary not exists, create new one: %s, data: %s", data_summary_id, str(data))

    # create session context
    context_name = "{}-session".format(database)
    session_id = get_session_id()

    logging.info("session_id = %s, create session context, context_name = %s, data_summary_id = %s", session_id, context_name, data_summary_id)
    session_context = Context.create(
        g.session, ContextTypes.SESSION_FIX_REFINE, context_name, org_id, 0, 0,
        database, session_id, created_by, None, data_summary_id, 0, "", identifier=str(app_id), identifier_type=IdentifyTypes.APP_ID,
    )
    g.session.commit()

    inject_metrics(session_context.id, data_summary_id)
    set_data_context_id(data_summary_id)
    set_session(g.session)

    metrics = {
        "sqleditor_mode": sqleditor_mode,
        "type": "refine_sql",
    }
    job_id = enqueue(task_sql_refine, org_id, session_context.data_context_id, session_context.id, sql, raw_question, feedback, sql_error, user_token=user_token, metrics=metrics)
    result = {
        "job_id": job_id,
        "session_id": session_id,
    }

    return success(result=result)


@edatask_v3_bp.post("/refineSql")
@check_api_key_permission
def handle_sql_refine(data_app_id: str):
    """
    SQL Refine
    ---
    post:
        description: Refine SQL by given cluster_id and database, returns the refined SQL and execution result of the query.
        operationId: refineSql
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/RefineSQL"
        responses:
            200:
                description: SQL Refine result
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/RefineSQLResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    return _handle_sql_refine_fix()


@edatask_v3_bp.post("/knowledgeBases")
@check_api_key_permission
def handle_create_knowledge_base(data_app_id: str):
    """
    Create Knowledge Base
    ---
    post:
        description: Create knowledge base
        operationId: createKnowledgeBase
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/CreateKnowledgeBase"
        responses:
            200:
                description: knowledge base created
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/CreateKnowledgeBaseResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    payload = check_schema(CreateKnowledgeBaseSchema, request.get_json())
    org_id = get_org_id_from_header()
    cluster_id = payload["cluster_id"]
    db_name = payload["database"]
    description = payload["description"]

    knowledge_base = KnowledgeBase.create(g.session, org_id, cluster_id, db_name, description)
    all_knowledge_bases = KnowledgeBase.get_all_by_cluster_id_and_db_name(g.session, cluster_id, db_name)
    if len(all_knowledge_bases) == 1:
        knowledge_base.set_default(g.session)

    return success(result=knowledge_base.to_json())


@edatask_v3_bp.put("/knowledgeBases/<int:knowledge_base_id>")
@check_api_key_permission
def handle_update_knowledge_base(data_app_id: str, knowledge_base_id: int):
    """
    Update Knowledge Base
    ---
    put:
        description: Update knowledge base by knowledge_base_id
        operationId: updateKnowledgeBase
        parameters:
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base to update
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateKnowledgeBase"
        responses:
            200:
                description: if update success
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if knowledge base not found
    """
    payload = check_schema(UpdateKnowledgeBaseSchema, request.get_json())
    description = payload["description"]
    is_default = payload["default"]

    knowledge_base = KnowledgeBase.get_by_id(g.session, knowledge_base_id)
    if not knowledge_base:
        return fail(404, "knowledge base not found")

    check_org_id(knowledge_base.org_id)

    knowledge_base.update_description_and_default(g.session, description, is_default)

    return success(result=knowledge_base.to_json())


@edatask_v3_bp.get("/knowledgeBases")
@check_api_key_permission
def handle_list_knowledge_bases(data_app_id: str):
    """
    List Knowledge Bases
    ---
    get:
        description: List knowledge bases
        operationId: listKnowledgeBases
        responses:
            200:
                description: knowledge bases
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ListKnowledgeBasesResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    org_id = get_org_id_from_header()
    cluster_id = request.args.get("cluster_id")
    db_name = request.args.get("database")

    if not (cluster_id and db_name):
        return fail(400, "cluster_id and database are required")

    knowledge_bases = []
    for i in KnowledgeBase.get_all_by_cluster_id_and_db_name(g.session, cluster_id, db_name):
        if i.org_id != org_id:
            continue

        knowledge_bases.append(i.to_json(simple=True))

    return success(result={"knowledge_base": knowledge_bases})


@edatask_v3_bp.get("/knowledgeBases/<int:knowledge_base_id>")
@check_api_key_permission
def handle_get_knowledge_base(data_app_id: str, knowledge_base_id: int):
    """
    Get Knowledge Base Detail
    ---
    get:
        description: Get knowledge base detail by knowledge_base_id
        operationId: getKnowledgeBase
        parameters:
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base to get
        responses:
            200:
                description: knowledge base detail
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/KnowledgeBase"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if knowledge base not found
    """
    knowledge_base = KnowledgeBase.get_by_id(g.session, knowledge_base_id)
    if not knowledge_base:
        return fail(404, "knowledge base not found")

    check_org_id(knowledge_base.org_id)

    return success(result=knowledge_base.to_json(simple=False))


@edatask_v3_bp.get("/knowledgeBases/<int:knowledge_base_id>/data")
@check_api_key_permission
def handle_list_knowledge_base_data(data_app_id: str, knowledge_base_id: int):
    """
    List KnowledgeBase Data
    ---
    get:
        description: List knowledge base data by knowledge_base_id
        operationId: listKnowledgeBaseData
        parameters:
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base to get data
        responses:
            200:
                description: knowledge base data
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/ListKnowledgeBaseDataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    app_id = get_app_id()

    domain_items = remote_kg_management.query_knowledge(g.session, app_id, knowledge_base_id)

    return success(result={"data": domain_items})


@edatask_v3_bp.post("/knowledgeBases/<int:knowledge_base_id>/data")
@check_api_key_permission
def handle_create_knowledge_base_data(data_app_id: str, knowledge_base_id: int):
    """
    Create KnowledgeBase Data
    ---
    post:
        description: Create knowledge base data
        operationId: createKnowledgeBaseData
        parameters:
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base to create data
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/CreateKnowledgeBaseData"
        responses:
            200:
                description: knowledge base data created
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/CreateKnowledgeBaseDataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    payload = check_schema(CreateDomainItemSchema, request.get_json())
    app_id = get_app_id()
    _type = payload["type"]
    raw_data = payload["raw_data"]
    module = ModuleType.ALL
    meta_data = payload.get("meta_data", {})
    created_by = get_current_user_email()
    try:
        domain_item = remote_kg_management.insert_knowledge(g.session, _type, app_id, module, raw_data, meta_data, created_by, knowledge_base_id)
        return success(result=domain_item.to_json())
    except Exception as e:
        logging.info("create domain knowledge failed, error = %s", str(e))
        return fail(400, str(e))


@edatask_v3_bp.put("/knowledgeBases/<int:knowledge_base_id>/data/<int:knowledge_data_id>")
@check_api_key_permission
def handle_update_knowledge_base_data(data_app_id: str, knowledge_base_id: int, knowledge_data_id: int):
    """
    Update KnowledgeBase Data
    ---
    put:
        description: Update knowledge base data
        operationId: updateKnowledgeBaseData
        parameters:
            - name: knowledge_data_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base data to update.
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base to get data
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/UpdateKnowledgeBaseData"
        responses:
            200:
                description: knowledge base data updated
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/UpdateKnowledgeBaseDataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    payload = check_schema(UpdateDomainItemSchema, request.get_json())
    raw_data = payload["raw_data"]
    meta_data = payload["meta_data"]
    try:
        result = remote_kg_management.update_knowledge(g.session, knowledge_data_id, knowledge_base_id, raw_data, meta_data)
        return success(result=result)
    except Exception as e:
        logging.info("update domain knowledge failed, error = %s", str(e))
        return fail(400, str(e))


@edatask_v3_bp.delete("/knowledgeBases/<int:knowledge_base_id>/data/<int:knowledge_data_id>")
@check_api_key_permission
def handle_delete_knowledge_base_data(data_app_id: str, knowledge_base_id: int, knowledge_data_id: int):
    """
    Delete KnowledgeBase Data
    ---
    delete:
        description: Delete knowledge base data
        operationId: deleteKnowledgeBaseData
        parameters:
            - name: knowledge_data_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base data to delete.
            - name: knowledge_base_id
              in: path
              required: true
              schema:
                type: integer
              description: The ID of the knowledge base
        responses:
            200:
                description: knowledge base data deleted
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/DeleteKnowledgeBaseDataResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
    """
    remote_kg_management.delete_knowledge(g.session, knowledge_data_id, knowledge_base_id)

    return success()


def _handle_multi_round_chat2sql(session_id: int, sqleditor_mode=False, data_to_chart=False):
    payload = check_schema(MultiRoundChat2SQLSchema, request.get_json())
    question = payload["question"]
    org_id = get_org_id_from_header()
    flags = {"auto_run": True, "data_to_chart": data_to_chart}
    sql_to_data = True

    if flags["auto_run"]:
        sql_to_data = True

    evidence = payload["evidence"]
    direct_resolve = payload["sql_generate_mode"] == "direct"
    
    app_id = get_app_id()
    api_key = get_access_key()
    user_token = get_user_token()
    query_id = payload["feedback_answer_id"]
    task_id = payload["feedback_task_id"]
    created_by = get_current_user_email()

    session_context = Context.get_by_id(g.session, session_id)
    logging.info("get session context by id %s", session_id)
    if not session_context:
        return fail(404, "session context not found")

    check_org_id(session_context.org_id)

    if session_context._type != ContextTypes.SESSION_CHAT:
        return fail(400, "You can only use session context generated by /sessions api")

    data_context = Context.get_by_id(g.session, session_context.data_context_id)

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    metrics = {
        "sqleditor_mode": sqleditor_mode,
        "type": "chat2data",
        "question": question,
    }
    job_id = enqueue(
        task_chat2data, org_id,
        session_context.data_context_id, session_context.id, question, direct_resolve, flags["auto_run"], sql_to_data,
        flags["data_to_chart"], evidence, app_id, api_key, user_token, query_id=query_id, task_id=task_id,
        v3_format=True, metrics=metrics,
    )

    OpLog.create(g.session, OpLog.EVENT_CHAT2DATA, api_key, created_by, job_id, data_context.cluster_id, data_context.db_name, session_context.id, question)

    if not sqleditor_mode:
        measuring_api()

    result={
        "job_id": job_id,
        "session_id": session_context.id,
    }
    return success(result=result)


@edatask_v3_bp.post("/sessions/<int:session_id>/chat2data")
@check_api_key_permission
def handle_multi_round_chat2sql(data_app_id: str, session_id: int):
    """
    Multi Round Chat2SQL
    ---
    post:
        description: Multi Round Chat2SQL generate a SQL by given question described in nature language, the response contains a job_id, use the /v2/jobs to query the job status, when the status changed from running to done, get generated sql and execution result from response.
        operationId: multiRoundChat2SQL
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/MultiRoundChat2SQL"
        parameters:
            - name: session_id
              in: path
              required: true
              schema:
                  type: integer
              description: The ID of the session context to query
        responses:
            200:
                description: Job initiates successfully, it will generate a unique job ID that can be used for detailed tracking and specific inquiries.
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/MultiRoundChat2SQLResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    return _handle_multi_round_chat2sql(session_id, False, data_to_chart=True)


def _handle_multi_round_feedback(sqleditor_mode=False):
    payload = check_schema(MultiRoundFeedbackSchema, request.get_json())
    session_context_id = payload["session_id"]
    feedback_answer_id = payload["feedback_answer_id"]
    feedback_task_id = payload["feedback_task_id"]
    question = payload["question"]
    org_id = get_org_id_from_header()
    flags = {"sql_generate_mode": "direct", "auto_run": True, "data_to_chart": False}
    sql_to_data = False
    evidence = ""
    app_id = get_app_id()
    api_key = get_access_key()
    user_token = get_user_token()
    created_by = get_current_user_email()

    session_context = Context.get_by_id(g.session, session_context_id)
    if not session_context:
        return fail(404, "session context not found")

    check_org_id(session_context.org_id)

    if session_context._type != ContextTypes.SESSION_CHAT:
        return fail(400, "You can only use session context generated by /sessions api")

    data_context = Context.get_by_id(g.session, session_context.data_context_id)

    if get_default_queue_length() > config.QUEUE_MAX_LENGTH:
        return fail(429, "Too many eda tasks right now, hold on and retry later please")

    metrics = {
        "sqleditor_mode": sqleditor_mode,
        "type": "chat2data",
    }
    job_id = enqueue(
        task_chat2data, org_id,
        session_context.data_context_id, session_context.id, question, flags["sql_generate_mode"] == "direct", flags["auto_run"], sql_to_data, flags["data_to_chart"], evidence, app_id, api_key, user_token, query_id=feedback_answer_id, task_id=feedback_task_id,
        metrics=metrics,
    )

    OpLog.create(g.session, OpLog.EVENT_CHAT2DATA, api_key, created_by, job_id, data_context.cluster_id, data_context.db_name, session_context.id, feedback_answer_id)

    if not sqleditor_mode:
        measuring_api()

    result={
        "job_id": job_id,
        "session_context_id": session_context.id,
    }
    return success(result=result)


@edatask_v3_bp.post("/multiRoundFeedback")
@check_api_key_permission
def handle_multi_round_feedback(data_app_id: str):
    """
    Multi Round Feedback
    ---
    post:
        description: Multi Round Feedback for Chat2Data
        operationId: multiRoundFeedback
        requestBody:
            required: true
            content:
                application/json:
                    schema:
                        $ref: "#/components/schemas/MultiRoundFeedback"
        responses:
            200:
                description: Multi Round Feedback response
                content:
                    application/json:
                        schema:
                            $ref: "#/components/schemas/MultiRoundFeedbackResponse"
            400:
                description: if params invalid
            403:
                description: if permission denied
            404:
                description: if data context not found
    """
    return _handle_multi_round_feedback(False)