MaxKB OSS — Универсальный инструмент загрузки файлов

Загружает существующие файлы с сервера или переданный текстовый контент в MaxKB OSS. Возвращает структурированный результат и ссылку для скачивания, что удобно для построения условных ветвлений в рабочих процессах.

1. Описание функционала

Режим file_path: загружает уже существующий файл с сервера (поддерживаются все типы файлов)

Режим file_content: принимает текстовый контент, автоматически создаёт временный файл и загружает его

Возвращает структурированный dict с полями: status, message, file_id, file_name, download_url

2. Описание параметров

2.1 Параметры запуска (параметры инициализации)

Параметр	Тип компонента	Обязательный	Значение по умолчанию	Описание
`upload_url`	Текстовое поле	Да	—	Адрес сервера MaxKB, например `http://x.x.x.x:8080`
`upload_headers`	Текстовое поле	Да	—	API токен авторизации (без префикса Bearer)
`source_type`	Текстовое поле	Нет	`TEMPORARY_30_MINUTE`	Тип жизненного цикла файла (см. описание ниже)
`source_id`	Текстовое поле	Нет	Совпадает с `source_type`	ID владельца файла

Допустимые значения source_type:

Параметр source_type определяет принадлежность файла и его жизненный цикл. source_id — это ID соответствующего бизнес-объекта.

Тип	Описание
`KNOWLEDGE`	Файл базы знаний; удаляется вместе с базой знаний. `source_id` — ID базы знаний
`APPLICATION`	Файл приложения; удаляется вместе с приложением. `source_id` — ID приложения
`TOOL`	Файл инструмента; удаляется вместе с инструментом. `source_id` — ID инструмента
`DOCUMENT`	Файл документа. `source_id` — ID документа
`CHAT`	Файл диалога. `source_id` — ID диалога
`SYSTEM`	Системный файл
`TEMPORARY_30_MINUTE`	Временный файл; автоматически удаляется через 30 минут
`TEMPORARY_120_MINUTE`	Временный файл; автоматически удаляется через 120 минут
`TEMPORARY_1_DAY`	Временный файл; автоматически удаляется через 1 день

Рекомендации по типичным сценариям:

Агент генерирует файл для скачивания пользователем → TEMPORARY_1_DAY

Файл должен удаляться вместе с бизнес-объектом → используйте соответствующий тип (KNOWLEDGE / APPLICATION / TOOL и т.д.)

Только временное использование → TEMPORARY_30_MINUTE или TEMPORARY_120_MINUTE

2.2 Входные параметры

Параметр	Тип данных	Обязательный	Источник	Описание
`file_path`	string	Один из двух	Ссылочный параметр	Путь к файлу на сервере (приоритет выше). Если задан, `file_content` игнорируется
`file_content`	string	Один из двух	Ссылочный параметр	Текстовое содержимое файла (используется, если `file_path` пустой)
`filename`	string	Нет	Ссылочный параметр	Имя файла без расширения. По умолчанию: `файл`
`extension`	string	Нет	Пользовательский параметр	Расширение файла: `.html`, `.md`, `.txt`, `.pdf` и т.д.

Как получить значения параметров:

upload_url — укажите адрес вашего сервера MaxKB (включая порт)

upload_headers — сгенерируйте API токен в панели управления MaxKB и вставьте его значение (без префикса Bearer)

file_path / file_content — используйте системные переменные MaxKB, например {{file_path}}, для ссылки на выходные данные предыдущих шагов

3. Возвращаемый результат

Возвращается dict следующей структуры:

Поле	Тип	Описание
`status`	string	`success` / `failed` / `warning`
`message`	string	Описание результата
`file_id`	string	ID файла в OSS
`file_name`	string	Имя загруженного файла
`download_url`	string	Ссылка для скачивания файла

4. Код инструмента (Python)

"""
Универсальный инструмент загрузки файлов (кастомный инструмент MaxKB)

Два режима использования:
  1. Режим file_path — загружает уже существующий файл с сервера (поддерживаются все типы)
  2. Режим file_content — принимает текстовый контент, автоматически создаёт временный файл и загружает его

Возвращает структурированный dict для построения условных ветвлений в рабочих процессах.
"""
import os
import io
import re
import requests
from datetime import datetime
from mimetypes import guess_type


# ── Вспомогательные функции ──

def _plain(value) -> str:
    """Надёжное получение значения: совместимо с str / dict / объект / None"""
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, dict):
        for key in ["value", "content", "text", "result", "data"]:
            if key in value:
                return _plain(value[key])
        return str(value).strip()
    for attr in ["value", "content", "text", "result", "data"]:
        if hasattr(value, attr):
            try:
                return _plain(getattr(value, attr))
            except Exception:
                pass
    return str(value).strip()


def _cfg(name: str, value: str = "", default: str = "") -> str:
    """Резервное значение из переменных среды: переданное значение > переменная среды > значение по умолчанию"""
    value = _plain(value)
    return value or _plain(os.getenv(name)) or _plain(os.getenv(name.upper())) or default


def _result(status: str, message: str, file_name: str = "",
            download_url: str = "", file_id: str = "", extra=None) -> dict:
    """Унифицированное формирование возвращаемого dict"""
    result = {
        "status": status,
        "message": _plain(message),
        "file_id": file_id,
        "file_name": _plain(file_name),
        "download_url": download_url,
    }
    if extra is not None:
        result["extra"] = extra
    return result


def _extract_file_id(resp_json) -> str:
    """Извлекает file_id из ответа; совместимо с несколькими форматами"""
    def pick(text) -> str:
        text = _plain(text)
        if not text:
            return ""
        match = re.search(r"/oss/file/([^/]+)/?", text)
        if match:
            return match.group(1)
        if "/" not in text and "." not in text:
            return text
        return ""

    if not isinstance(resp_json, dict):
        return ""

    for key in ["file_id", "id", "oss_id", "uuid"]:
        if resp_json.get(key):
            return _plain(resp_json[key])

    for key in ["url", "path", "download_url", "file_url"]:
        fid = pick(resp_json.get(key))
        if fid:
            return fid

    data = resp_json.get("data")
    if isinstance(data, str):
        return pick(data)
    if isinstance(data, dict):
        for key in ["file_id", "id", "oss_id", "uuid"]:
            if data.get(key):
                return _plain(data[key])
        for key in ["url", "path", "download_url", "file_url"]:
            fid = pick(data.get(key))
            if fid:
                return fid

    return ""


def main(
    file_path: str = "",
    file_content: str = "",
    filename: str = "",
    extension: str = "",
    upload_url: str = "",
    upload_headers: str = "",
    source_type: str = "",
    source_id: str = "",
) -> dict:
    """
    Универсальный инструмент загрузки файлов

    :param file_path:      Путь к файлу на сервере (приоритет). Если задан, file_content игнорируется
    :param file_content:   Текстовое содержимое файла (используется, если file_path пустой)
    :param filename:       Имя файла без расширения. По умолчанию "файл"
    :param extension:      Расширение файла: .html, .md, .txt, .pdf
    :param upload_url:     Адрес сервера MaxKB, например http://x.x.x.x:8080
    :param upload_headers: Токен авторизации Authorization: Bearer xxx
    :param source_type:    Жизненный цикл. По умолчанию TEMPORARY_30_MINUTE
    :param source_id:      ID владельца. Если пустой, используется значение source_type
    :return: dict { status, message, file_id, file_name, download_url }
    """
    # ── Нормализация параметров ──
    file_path = _plain(file_path)
    file_content = _plain(file_content)
    filename = _plain(filename)
    extension = _plain(extension)
    if extension and not extension.startswith('.'):
        extension = '.' + extension
    upload_url = _cfg("upload_url", upload_url)
    upload_headers = _cfg("upload_headers", upload_headers)
    source_type = _cfg("source_type", source_type, "TEMPORARY_30_MINUTE")
    source_id = _cfg("source_id", source_id, source_type)

    # ── Валидация ──
    if not upload_url:
        return _result("failed", "Не указан upload_url — настройте адрес сервера MaxKB")

    if not upload_headers:
        return _result("failed", "Не указан upload_headers — настройте API токен")

    if not file_path and not file_content:
        return _result("failed", "file_path и file_content оба пустые — укажите хотя бы один из них")

    # ── Получение имени файла и данных ──
    if file_path:
        basename = os.path.basename(file_path) if '/' in file_path or '\\' in file_path else file_path
        if not filename:
            filename, auto_ext = os.path.splitext(basename)
        else:
            auto_ext = ""
        ext = extension or auto_ext
        basename = f"{filename}{ext}"
        try:
            with open(file_path, 'rb') as f:
                data_bytes = f.read()
        except Exception as e:
            return _result("failed", f"Ошибка чтения файла: {e}",
                           extra={"file_path": file_path})
    else:
        if not filename:
            filename = "файл"
        ext = extension or ".txt"
        basename = f"{filename}{ext}"
        data_bytes = file_content.encode('utf-8')

    # ── Загрузка в OSS ──
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    upload_name = f"{filename}_{timestamp}{ext}"
    api_url = upload_url.rstrip('/') + "/chat/api/oss/file"

    headers = {}
    if upload_headers:
        headers["Authorization"] = f"Bearer {upload_headers}"

    form_data = {"source_type": source_type, "source_id": source_id}
    buf = io.BytesIO(data_bytes)
    mime_type, _ = guess_type(upload_name)
    files = {"file": (upload_name, buf, mime_type) if mime_type else (upload_name, buf)}

    try:
        resp = requests.post(api_url, headers=headers, data=form_data,
                             files=files, timeout=120)
    except Exception as e:
        return _result("failed", f"Ошибка при выполнении запроса на загрузку: {e}", filename,
                       extra={"upload_url": api_url})

    if resp.status_code < 200 or resp.status_code >= 300:
        return _result("failed", "Загрузка не удалась", filename,
                       extra={"status_code": resp.status_code,
                              "response": resp.text[:500]})

    try:
        resp_json = resp.json()
    except Exception:
        return _result("failed", "Сервер вернул не JSON", filename,
                       extra={"response": resp.text[:500]})

    file_id = _extract_file_id(resp_json)
    if not file_id:
        return _result("warning", "Загрузка прошла успешно, но file_id не удалось извлечь", filename,
                       extra={"response": resp_json})

    download_url = f"{upload_url.rstrip('/')}/chat/oss/file/{file_id}/"

    return _result("success", "Загрузка успешна", basename, download_url, file_id)

5. Примеры использования

Пример 1: Загрузка файла с сервера

{
  "file_path": "/tmp/report.pdf",
  "extension": ".pdf",
  "upload_url": "http://your-maxkb-server:8080",
  "upload_headers": "your-api-token"
}

Пример 2: Загрузка текстового контента

{
  "file_content": "# Заголовок\n\nЭто содержимое",
  "filename": "отчёт",
  "extension": ".md",
  "upload_url": "http://your-maxkb-server:8080",
  "upload_headers": "your-api-token"
}