Underground-Digital/Workflow-Engine

1import mimetypes2import os3import re4import urllib.parse5from collections.abc import Mapping6from typing import Any7from uuid import uuid48 9import httpx10from pydantic import BaseModel11 12from configs import dify_config13 14 15class FileInfo(BaseModel):16    filename: str17    extension: str18    mimetype: str19    size: int20 21 22def guess_file_info_from_response(response: httpx.Response):23    url = str(response.url)24    # Try to extract filename from URL25    parsed_url = urllib.parse.urlparse(url)26    url_path = parsed_url.path27    filename = os.path.basename(url_path)28 29    # If filename couldn't be extracted, use Content-Disposition header30    if not filename:31        content_disposition = response.headers.get("Content-Disposition")32        if content_disposition:33            filename_match = re.search(r'filename="?(.+)"?', content_disposition)34            if filename_match:35                filename = filename_match.group(1)36 37    # If still no filename, generate a unique one38    if not filename:39        unique_name = str(uuid4())40        filename = f"{unique_name}"41 42    # Guess MIME type from filename first, then URL43    mimetype, _ = mimetypes.guess_type(filename)44    if mimetype is None:45        mimetype, _ = mimetypes.guess_type(url)46    if mimetype is None:47        # If guessing fails, use Content-Type from response headers48        mimetype = response.headers.get("Content-Type", "application/octet-stream")49 50    extension = os.path.splitext(filename)[1]51 52    # Ensure filename has an extension53    if not extension:54        extension = mimetypes.guess_extension(mimetype) or ".bin"55        filename = f"{filename}{extension}"56 57    return FileInfo(58        filename=filename,59        extension=extension,60        mimetype=mimetype,61        size=int(response.headers.get("Content-Length", -1)),62    )63 64 65def get_parameters_from_feature_dict(*, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]):66    return {67        "opening_statement": features_dict.get("opening_statement"),68        "suggested_questions": features_dict.get("suggested_questions", []),69        "suggested_questions_after_answer": features_dict.get("suggested_questions_after_answer", {"enabled": False}),70        "speech_to_text": features_dict.get("speech_to_text", {"enabled": False}),71        "text_to_speech": features_dict.get("text_to_speech", {"enabled": False}),72        "retriever_resource": features_dict.get("retriever_resource", {"enabled": False}),73        "annotation_reply": features_dict.get("annotation_reply", {"enabled": False}),74        "more_like_this": features_dict.get("more_like_this", {"enabled": False}),75        "user_input_form": user_input_form,76        "sensitive_word_avoidance": features_dict.get(77            "sensitive_word_avoidance", {"enabled": False, "type": "", "configs": []}78        ),79        "file_upload": features_dict.get(80            "file_upload",81            {82                "image": {83                    "enabled": False,84                    "number_limits": 3,85                    "detail": "high",86                    "transfer_methods": ["remote_url", "local_file"],87                }88            },89        ),90        "system_parameters": {91            "image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,92            "video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,93            "audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT,94            "file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,95            "workflow_file_upload_limit": dify_config.WORKFLOW_FILE_UPLOAD_LIMIT,96        },97    }98