mirror of
https://github.com/langgenius/dify-plugin-sdks.git
synced 2026-07-22 18:35:29 -04:00
QuantumGhost
9e084d7d09
* chore(python): update .gitignore for Python SDK Adjust .gitignore for Python SDK based on the templated generated by gitignore.io, with minor custom modifications. * chore: add .editorconfig for consistent formatting Add a .editorconfig file to enforce consistent file encoding, indentation, and end-of-line characters across the project. * chore(python): upgrade ruff version to v0.11.2 * feat(python): Integrate flake8-bandit ruleset for security linting Enable the flake8-bandit (S) ruleset in ruff to enforce security-related best practices and mitigate potential vulnerabilities in the codebase. Suppress existing issues in the current code using `noqa` comments, primarily for instances of `requests` calls without a specified `timeout` parameter.
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
from enum import Enum
|
|
from typing import Optional
|
|
|
|
import requests
|
|
from pydantic import BaseModel, model_validator
|
|
|
|
from dify_plugin.core.entities.invocation import InvokeType
|
|
from dify_plugin.core.runtime import BackwardsInvocation
|
|
|
|
|
|
class UploadFileResponse(BaseModel):
|
|
class Type(str, Enum):
|
|
DOCUMENT = "document"
|
|
IMAGE = "image"
|
|
VIDEO = "video"
|
|
AUDIO = "audio"
|
|
|
|
@classmethod
|
|
def from_mime_type(cls, mime_type: str):
|
|
if mime_type.startswith("image/"):
|
|
return cls.IMAGE
|
|
if mime_type.startswith("video/"):
|
|
return cls.VIDEO
|
|
if mime_type.startswith("audio/"):
|
|
return cls.AUDIO
|
|
|
|
return cls.DOCUMENT
|
|
|
|
id: str
|
|
name: str
|
|
size: int
|
|
extension: str
|
|
mime_type: str
|
|
type: Optional[Type] = None
|
|
preview_url: Optional[str] = None
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def validate_type(cls, d):
|
|
if "type" not in d:
|
|
d["type"] = cls.Type.from_mime_type(d.get("mime_type", ""))
|
|
return d
|
|
|
|
def to_app_parameter(self) -> dict:
|
|
return {
|
|
"upload_file_id": self.id,
|
|
"transfer_method": "local_file",
|
|
"type": self.Type.from_mime_type(self.mime_type).value,
|
|
}
|
|
|
|
|
|
class File(BackwardsInvocation[dict]):
|
|
def upload(self, filename: str, content: bytes, mimetype: str) -> UploadFileResponse:
|
|
"""
|
|
Upload a file
|
|
|
|
:param filename: file name
|
|
:param content: file content
|
|
:param mimetype: file mime type
|
|
|
|
:return: file id
|
|
"""
|
|
for response in self._backwards_invoke(
|
|
InvokeType.UploadFile,
|
|
dict,
|
|
{
|
|
"filename": filename,
|
|
"mimetype": mimetype,
|
|
},
|
|
):
|
|
url = response.get("url")
|
|
if not url:
|
|
raise Exception("upload file failed, could not get signed url")
|
|
|
|
response = requests.post(url, files={"file": (filename, content, mimetype)}) # noqa: S113
|
|
if response.status_code != 201:
|
|
raise Exception(f"upload file failed, status code: {response.status_code}, response: {response.text}")
|
|
|
|
return UploadFileResponse(**response.json())
|
|
|
|
raise Exception("upload file failed, empty response from server")
|