dify-1.8.1/api/controllers/console/app/workflow_statistic.py

from datetime import datetime
from decimal import Decimal

import pytz
import sqlalchemy as sa
from flask import jsonify
from flask_login import current_user
from flask_restx import Resource, reqparse

from controllers.console import api
from controllers.console.app.wraps import get_app_model
from controllers.console.wraps import account_initialization_required, setup_required
from extensions.ext_database import db
from libs.helper import DatetimeString
from libs.login import login_required
from models.enums import WorkflowRunTriggeredFrom
from models.model import AppMode


class WorkflowDailyRunsStatistic(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def get(self, app_model):
        account = current_user

        parser = reqparse.RequestParser()
        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        args = parser.parse_args()

        timezone = pytz.timezone(account.timezone)
        utc_timezone = pytz.utc

        # Check if querying "today" data
        is_today = False
        if args["start"] and args["end"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")

            now = datetime.now(timezone)
            today_date = now.date()
            if start_datetime.date() == today_date and end_datetime.date() == today_date:
                is_today = True

        # Choose time truncation based on whether it's today
        # Use TO_CHAR with HH24 to ensure 24-hour format (not 12-hour)
        if is_today:
            time_trunc = "TO_CHAR(DATE_TRUNC('hour', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD HH24:MI:SS')"
        else:
            time_trunc = "TO_CHAR(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD')"

        sql_query = f"""SELECT
    {time_trunc} AS date,
    COUNT(id) AS runs
FROM
    workflow_runs
WHERE
    app_id = :app_id
    AND triggered_from = :triggered_from"""
        arg_dict = {
            "tz": account.timezone,
            "app_id": app_model.id,
            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
        }

        if args["start"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            start_datetime = start_datetime.replace(second=0)

            start_datetime_timezone = timezone.localize(start_datetime)
            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

        if args["end"]:
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
            end_datetime = end_datetime.replace(second=0)

            end_datetime_timezone = timezone.localize(end_datetime)
            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

        sql_query += " GROUP BY date ORDER BY date"

        response_data = []

        with db.engine.begin() as conn:
            rs = conn.execute(sa.text(sql_query), arg_dict)
            for i in rs:
                # date is already formatted string from TO_CHAR
                response_data.append({"date": i.date, "runs": i.runs})

        return jsonify({"data": response_data})


class WorkflowDailyTerminalsStatistic(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def get(self, app_model):
        account = current_user

        parser = reqparse.RequestParser()
        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        args = parser.parse_args()

        timezone = pytz.timezone(account.timezone)
        utc_timezone = pytz.utc

        # Check if querying "today" data
        is_today = False
        if args["start"] and args["end"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")

            now = datetime.now(timezone)
            today_date = now.date()
            if start_datetime.date() == today_date and end_datetime.date() == today_date:
                is_today = True

        # Choose time truncation based on whether it's today
        # Use TO_CHAR with HH24 to ensure 24-hour format (not 12-hour)
        if is_today:
            time_trunc = "TO_CHAR(DATE_TRUNC('hour', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD HH24:MI:SS')"
        else:
            time_trunc = "TO_CHAR(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD')"

        sql_query = f"""SELECT
    {time_trunc} AS date,
    COUNT(DISTINCT workflow_runs.created_by) AS terminal_count
FROM
    workflow_runs
WHERE
    app_id = :app_id
    AND triggered_from = :triggered_from"""
        arg_dict = {
            "tz": account.timezone,
            "app_id": app_model.id,
            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
        }

        if args["start"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            start_datetime = start_datetime.replace(second=0)

            start_datetime_timezone = timezone.localize(start_datetime)
            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

        if args["end"]:
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
            end_datetime = end_datetime.replace(second=0)

            end_datetime_timezone = timezone.localize(end_datetime)
            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

        sql_query += " GROUP BY date ORDER BY date"

        response_data = []

        with db.engine.begin() as conn:
            rs = conn.execute(sa.text(sql_query), arg_dict)
            for i in rs:
                # date is already formatted string from TO_CHAR
                response_data.append({"date": i.date, "terminal_count": i.terminal_count})

        return jsonify({"data": response_data})


class WorkflowDailyTokenCostStatistic(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model
    def get(self, app_model):
        account = current_user

        parser = reqparse.RequestParser()
        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        args = parser.parse_args()

        timezone = pytz.timezone(account.timezone)
        utc_timezone = pytz.utc

        # Check if querying "today" data
        is_today = False
        if args["start"] and args["end"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")

            now = datetime.now(timezone)
            today_date = now.date()
            if start_datetime.date() == today_date and end_datetime.date() == today_date:
                is_today = True

        # Choose time truncation based on whether it's today
        # Use TO_CHAR with HH24 to ensure 24-hour format (not 12-hour)
        if is_today:
            time_trunc = "TO_CHAR(DATE_TRUNC('hour', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD HH24:MI:SS')"
        else:
            time_trunc = "TO_CHAR(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD')"

        sql_query = f"""SELECT
    {time_trunc} AS date,
    SUM(workflow_runs.total_tokens) AS token_count
FROM
    workflow_runs
WHERE
    app_id = :app_id
    AND triggered_from = :triggered_from"""
        arg_dict = {
            "tz": account.timezone,
            "app_id": app_model.id,
            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
        }

        if args["start"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            start_datetime = start_datetime.replace(second=0)

            start_datetime_timezone = timezone.localize(start_datetime)
            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at >= :start"
            arg_dict["start"] = start_datetime_utc

        if args["end"]:
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
            end_datetime = end_datetime.replace(second=0)

            end_datetime_timezone = timezone.localize(end_datetime)
            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)

            sql_query += " AND created_at < :end"
            arg_dict["end"] = end_datetime_utc

        sql_query += " GROUP BY date ORDER BY date"

        response_data = []

        with db.engine.begin() as conn:
            rs = conn.execute(sa.text(sql_query), arg_dict)
            for i in rs:
                # date is already formatted string from TO_CHAR
                response_data.append(
                    {
                        "date": i.date,
                        "token_count": i.token_count,
                    }
                )

        return jsonify({"data": response_data})


class WorkflowAverageAppInteractionStatistic(Resource):
    @setup_required
    @login_required
    @account_initialization_required
    @get_app_model(mode=[AppMode.WORKFLOW])
    def get(self, app_model):
        account = current_user

        parser = reqparse.RequestParser()
        parser.add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
        args = parser.parse_args()

        timezone = pytz.timezone(account.timezone)
        utc_timezone = pytz.utc

        # Check if querying "today" data
        is_today = False
        if args["start"] and args["end"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")

            now = datetime.now(timezone)
            today_date = now.date()
            if start_datetime.date() == today_date and end_datetime.date() == today_date:
                is_today = True

        # Choose time truncation based on whether it's today
        # Use TO_CHAR with HH24 to ensure 24-hour format (not 12-hour)
        if is_today:
            time_trunc = "TO_CHAR(DATE_TRUNC('hour', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD HH24:MI:SS')"
        else:
            time_trunc = "TO_CHAR(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz), 'YYYY-MM-DD')"

        sql_query = f"""SELECT
    AVG(sub.interactions) AS interactions,
    sub.date
FROM
    (
        SELECT
            {time_trunc} AS date,
            c.created_by,
            COUNT(c.id) AS interactions
        FROM
            workflow_runs c
        WHERE
            c.app_id = :app_id
            AND c.triggered_from = :triggered_from
            {{{{start}}}}
            {{{{end}}}}
        GROUP BY
            date, c.created_by
    ) sub
GROUP BY
    sub.date"""
        arg_dict = {
            "tz": account.timezone,
            "app_id": app_model.id,
            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
        }

        if args["start"]:
            start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
            start_datetime = start_datetime.replace(second=0)

            start_datetime_timezone = timezone.localize(start_datetime)
            start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)

            sql_query = sql_query.replace("{{start}}", " AND c.created_at >= :start")
            arg_dict["start"] = start_datetime_utc
        else:
            sql_query = sql_query.replace("{{start}}", "")

        if args["end"]:
            end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
            end_datetime = end_datetime.replace(second=0)

            end_datetime_timezone = timezone.localize(end_datetime)
            end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)

            sql_query = sql_query.replace("{{end}}", " AND c.created_at < :end")
            arg_dict["end"] = end_datetime_utc
        else:
            sql_query = sql_query.replace("{{end}}", "")

        response_data = []

        with db.engine.begin() as conn:
            rs = conn.execute(sa.text(sql_query), arg_dict)
            for i in rs:
                # date is already formatted string from TO_CHAR
                response_data.append(
                    {"date": i.date, "interactions": float(i.interactions.quantize(Decimal("0.01")))}
                )

        return jsonify({"data": response_data})


api.add_resource(WorkflowDailyRunsStatistic, "/apps/<uuid:app_id>/workflow/statistics/daily-conversations")
api.add_resource(WorkflowDailyTerminalsStatistic, "/apps/<uuid:app_id>/workflow/statistics/daily-terminals")
api.add_resource(WorkflowDailyTokenCostStatistic, "/apps/<uuid:app_id>/workflow/statistics/token-costs")
api.add_resource(
    WorkflowAverageAppInteractionStatistic, "/apps/<uuid:app_id>/workflow/statistics/average-app-interactions"
)