helpers.py 3.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
  1. import mimetypes
  2. import os
  3. import re
  4. import urllib.parse
  5. from collections.abc import Mapping
  6. from typing import Any
  7. from uuid import uuid4
  8. import httpx
  9. from pydantic import BaseModel
  10. from configs import dify_config
  11. class FileInfo(BaseModel):
  12. filename: str
  13. extension: str
  14. mimetype: str
  15. size: int
  16. def guess_file_info_from_response(response: httpx.Response):
  17. url = str(response.url)
  18. # Try to extract filename from URL
  19. parsed_url = urllib.parse.urlparse(url)
  20. url_path = parsed_url.path
  21. filename = os.path.basename(url_path)
  22. # If filename couldn't be extracted, use Content-Disposition header
  23. if not filename:
  24. content_disposition = response.headers.get("Content-Disposition")
  25. if content_disposition:
  26. filename_match = re.search(r'filename="?(.+)"?', content_disposition)
  27. if filename_match:
  28. filename = filename_match.group(1)
  29. # If still no filename, generate a unique one
  30. if not filename:
  31. unique_name = str(uuid4())
  32. filename = f"{unique_name}"
  33. # Guess MIME type from filename first, then URL
  34. mimetype, _ = mimetypes.guess_type(filename)
  35. if mimetype is None:
  36. mimetype, _ = mimetypes.guess_type(url)
  37. if mimetype is None:
  38. # If guessing fails, use Content-Type from response headers
  39. mimetype = response.headers.get("Content-Type", "application/octet-stream")
  40. extension = os.path.splitext(filename)[1]
  41. # Ensure filename has an extension
  42. if not extension:
  43. extension = mimetypes.guess_extension(mimetype) or ".bin"
  44. filename = f"{filename}{extension}"
  45. return FileInfo(
  46. filename=filename,
  47. extension=extension,
  48. mimetype=mimetype,
  49. size=int(response.headers.get("Content-Length", -1)),
  50. )
  51. def get_parameters_from_feature_dict(*, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]):
  52. return {
  53. "opening_statement": features_dict.get("opening_statement"),
  54. "suggested_questions": features_dict.get("suggested_questions", []),
  55. "suggested_questions_after_answer": features_dict.get("suggested_questions_after_answer", {"enabled": False}),
  56. "speech_to_text": features_dict.get("speech_to_text", {"enabled": False}),
  57. "text_to_speech": features_dict.get("text_to_speech", {"enabled": False}),
  58. "retriever_resource": features_dict.get("retriever_resource", {"enabled": False}),
  59. "annotation_reply": features_dict.get("annotation_reply", {"enabled": False}),
  60. "more_like_this": features_dict.get("more_like_this", {"enabled": False}),
  61. "user_input_form": user_input_form,
  62. "sensitive_word_avoidance": features_dict.get(
  63. "sensitive_word_avoidance", {"enabled": False, "type": "", "configs": []}
  64. ),
  65. "file_upload": features_dict.get(
  66. "file_upload",
  67. {
  68. "image": {
  69. "enabled": False,
  70. "number_limits": 3,
  71. "detail": "high",
  72. "transfer_methods": ["remote_url", "local_file"],
  73. }
  74. },
  75. ),
  76. "system_parameters": {
  77. "image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,
  78. "video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,
  79. "audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT,
  80. "file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,
  81. "workflow_file_upload_limit": dify_config.WORKFLOW_FILE_UPLOAD_LIMIT,
  82. },
  83. }