1278 lines
51 KiB
Python
1278 lines
51 KiB
Python
"""
|
||
title: Document Generator 3.9
|
||
author: Ngo Tan Tai
|
||
author_url: https://newnol.io.vn
|
||
description: Generates Word, PowerPoint, and Excel files. Finds table data in complex JSON, multi-sheet Excel, Markdown tables in Word, images/notes in PPTX.
|
||
required_open_webui_version: 0.4.0
|
||
funding_url: https://github.com/newnol
|
||
requirements: python-docx, openpyxl, python-pptx, aiohttp, pillow
|
||
version: 3.9
|
||
licence: MIT
|
||
"""
|
||
|
||
import os
|
||
import io
|
||
import re
|
||
import csv
|
||
import json
|
||
import socket
|
||
import base64
|
||
import asyncio
|
||
import logging
|
||
import ipaddress
|
||
import tempfile
|
||
from urllib.parse import urlparse
|
||
|
||
import aiohttp
|
||
from pydantic import BaseModel, Field
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Опциональные зависимости: при отсутствии пакета отдаём понятную ошибку, а не NameError
|
||
try:
|
||
from docx import Document
|
||
from docx.shared import Pt, Inches as DocxInches
|
||
from docx.oxml import OxmlElement
|
||
from docx.oxml.ns import qn as docx_qn
|
||
from docx.enum.text import WD_ALIGN_PARAGRAPH
|
||
|
||
_HAS_DOCX = True
|
||
except ImportError:
|
||
Document = None
|
||
_HAS_DOCX = False
|
||
|
||
try:
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
_HAS_OPENPYXL = True
|
||
except ImportError:
|
||
Workbook = None
|
||
_HAS_OPENPYXL = False
|
||
|
||
try:
|
||
from pptx import Presentation
|
||
from pptx.util import Inches
|
||
from pptx.oxml.ns import qn as pptx_qn
|
||
|
||
_HAS_PPTX = True
|
||
except ImportError:
|
||
Presentation = None
|
||
Inches = None
|
||
_HAS_PPTX = False
|
||
|
||
try:
|
||
from PIL import Image
|
||
|
||
_HAS_PIL = True
|
||
except ImportError:
|
||
Image = None
|
||
_HAS_PIL = False
|
||
|
||
|
||
MIME_DOCX = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
||
MIME_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
MIME_PPTX = "application/vnd.openxmlformats-officedocument.presentationml.presentation"
|
||
|
||
# Разрешённый формат внутренних ссылок на файлы Open WebUI (без query и без "..")
|
||
_OWUI_FILE_RE = re.compile(r"/api/v1/files/[\w-]+/content")
|
||
|
||
|
||
# ======================================================================
|
||
# Вся внутренняя логика живёт здесь, а НЕ в Tools: Open WebUI строит
|
||
# спецификацию инструментов по методам класса Tools, и хелперы с "_"
|
||
# (_upload_file, _safe_unlink, _fetch_image ...) не должны быть доступны модели.
|
||
# ======================================================================
|
||
class _Impl:
|
||
def __init__(self, valves):
|
||
self.valves = valves
|
||
|
||
# ------------------------------------------------------------------
|
||
# Утилиты
|
||
# ------------------------------------------------------------------
|
||
_ILLEGAL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")
|
||
|
||
@staticmethod
|
||
def _safe_filename(filename: str, ext: str) -> str:
|
||
"""Только basename, Unicode-буквы разрешены, расширение регистронезависимо."""
|
||
name = os.path.basename(str(filename or "").strip().replace("\\", "/"))
|
||
name = re.sub(r"[^\w.\- ]", "_", name).replace(" ", "_")
|
||
name = re.sub(r"_+", "_", name).strip("._ ")
|
||
if name.lower().endswith(ext):
|
||
name = name[: -len(ext)]
|
||
name = name.strip("._")[:100] or "file"
|
||
return name + ext
|
||
|
||
@staticmethod
|
||
def _safe_unlink(path: str):
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
@staticmethod
|
||
async def _emit(emitter, text: str, done: bool = False):
|
||
if not emitter:
|
||
return
|
||
try:
|
||
await emitter(
|
||
{"type": "status", "data": {"description": text, "done": done}}
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
def _get_api_key(self, __user__) -> str:
|
||
try:
|
||
valves = (__user__ or {}).get("valves")
|
||
if not valves:
|
||
return ""
|
||
if hasattr(valves, "openwebui_api_key"):
|
||
return valves.openwebui_api_key or ""
|
||
if isinstance(valves, dict):
|
||
return valves.get("openwebui_api_key", "") or ""
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _strip_fences(s: str) -> str:
|
||
"""Убирает ```json ... ``` вокруг данных, которые любят добавлять LLM."""
|
||
s = str(s).strip()
|
||
if s.startswith("```"):
|
||
m = re.match(r"^```[\w-]*[ \t]*\n?(.*?)\n?```\s*$", s, re.DOTALL)
|
||
if m:
|
||
return m.group(1).strip()
|
||
elif s[:1] not in ("[", "{"):
|
||
m = re.search(r"```[\w-]*[ \t]*\n(.*?)```", s, re.DOTALL)
|
||
if m and m.group(1).strip()[:1] in ("[", "{"):
|
||
return m.group(1).strip()
|
||
return s
|
||
|
||
# ------------------------------------------------------------------
|
||
# Загрузка изображений (с защитой от SSRF)
|
||
# ------------------------------------------------------------------
|
||
async def _resolve_public_ip(self, host: str, port: int) -> str:
|
||
"""Резолвит хост и возвращает ПЕРВЫЙ публичный IP; приватные адреса отклоняет.
|
||
|
||
IP фиксируем один раз и дальше коннектимся именно к нему, чтобы избежать
|
||
DNS-rebinding (TOCTOU между проверкой и реальным запросом).
|
||
"""
|
||
loop = asyncio.get_running_loop()
|
||
infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
|
||
for info in infos:
|
||
ip = ipaddress.ip_address(info[4][0].split("%")[0])
|
||
if ip.is_global:
|
||
return str(ip)
|
||
raise ValueError("Адрес не публичный — запрос заблокирован")
|
||
|
||
async def _fetch_image(self, src: str, api_key: str = "") -> bytes:
|
||
"""
|
||
Возвращает байты изображения. Поддерживает:
|
||
- base64 data URI
|
||
- относительный путь /api/v1/files/<id>/content (файлы самого Open WebUI, с токеном)
|
||
- публичный http(s) URL (приватные адреса блокируются)
|
||
"""
|
||
src = str(src).strip()
|
||
max_bytes = max(1, self.valves.max_image_mb) * 1024 * 1024
|
||
|
||
if src.startswith("data:"):
|
||
try:
|
||
data = base64.b64decode(src.split(",", 1)[1])
|
||
except Exception:
|
||
raise ValueError("Некорректные base64-данные изображения")
|
||
if len(data) > max_bytes:
|
||
raise ValueError("Изображение слишком большое")
|
||
return data
|
||
|
||
headers = {}
|
||
pub = self.valves.public_domain.rstrip("/")
|
||
if pub and src.startswith(pub + "/"):
|
||
src = src[len(pub) :]
|
||
|
||
server_hostname = None
|
||
if src.startswith("/") and not src.startswith("//"):
|
||
# Строгая проверка: никаких "..", query и посторонних путей —
|
||
# иначе запрос с токеном пользователя может уйти на другой endpoint
|
||
if not _OWUI_FILE_RE.fullmatch(src):
|
||
raise ValueError("Разрешён только /api/v1/files/<id>/content")
|
||
url = self.valves.internal_api_url.rstrip("/") + src
|
||
headers["Authorization"] = f"Bearer {api_key}"
|
||
else:
|
||
p = urlparse(src)
|
||
if p.scheme not in ("http", "https") or not p.hostname:
|
||
raise ValueError("Недопустимый URL изображения")
|
||
port = p.port or (443 if p.scheme == "https" else 80)
|
||
pinned_ip = await self._resolve_public_ip(p.hostname, port)
|
||
host_part = (
|
||
f"[{pinned_ip}]:{port}" if ":" in pinned_ip else f"{pinned_ip}:{port}"
|
||
)
|
||
# Сохраняем query-строку (подписанные ссылки: ?token=..., ?X-Amz-...)
|
||
path = p.path or "/"
|
||
if p.query:
|
||
path += "?" + p.query
|
||
url = f"{p.scheme}://{host_part}{path}"
|
||
# Host с портом, если он нестандартный; userinfo отбрасываем
|
||
host_header = p.hostname
|
||
if ":" in host_header: # IPv6-литерал
|
||
host_header = f"[{host_header}]"
|
||
if p.port:
|
||
host_header += f":{p.port}"
|
||
headers["Host"] = host_header
|
||
if p.scheme == "https":
|
||
server_hostname = p.hostname # SNI, чтобы сертификат валидировался по домену
|
||
|
||
timeout = aiohttp.ClientTimeout(total=30)
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
async with session.get(
|
||
url,
|
||
headers=headers,
|
||
allow_redirects=False,
|
||
server_hostname=server_hostname,
|
||
) as resp:
|
||
if resp.status != 200:
|
||
raise ValueError(f"HTTP {resp.status}")
|
||
if resp.content_length and resp.content_length > max_bytes:
|
||
raise ValueError("Изображение слишком большое")
|
||
buf = bytearray()
|
||
async for chunk in resp.content.iter_chunked(65536):
|
||
buf += chunk
|
||
if len(buf) > max_bytes:
|
||
raise ValueError("Изображение слишком большое")
|
||
return bytes(buf)
|
||
|
||
# ------------------------------------------------------------------
|
||
# EXCEL: разбор данных
|
||
# ------------------------------------------------------------------
|
||
_DATA_KEYS = ("data", "rows", "cells")
|
||
_WRAP_KEYS = ("sheets", "tables", "data", "result", "content", "rows")
|
||
_NAME_KEYS = ("name", "title", "sheet", "sheet_name")
|
||
|
||
def _find_tables(self, data, name="Sheet1", depth=0):
|
||
"""
|
||
Рекурсивно ищет таблицы. Возвращает [(имя_листа, [[строка], ...]), ...].
|
||
Понимает: [[..]], [{..}], {"sheets":[{"name","data"}]}, {"S1":[[..]], "S2":[[..]]},
|
||
{"data":[[..]]}, плоский {"a":1,"b":2}.
|
||
"""
|
||
if depth > 6:
|
||
return []
|
||
|
||
if isinstance(data, list):
|
||
if not data:
|
||
return []
|
||
|
||
# Список «листов»: [{"name": "..", "data": [[..]]}, ...]
|
||
def is_sheet(x):
|
||
return isinstance(x, dict) and any(
|
||
isinstance(x.get(k), list) for k in self._DATA_KEYS
|
||
)
|
||
|
||
if all(is_sheet(x) for x in data):
|
||
out = []
|
||
for i, x in enumerate(data, 1):
|
||
nm = next(
|
||
(str(x[k]) for k in self._NAME_KEYS if x.get(k)), f"Sheet{i}"
|
||
)
|
||
dk = next(k for k in self._DATA_KEYS if isinstance(x.get(k), list))
|
||
out.extend(self._find_tables(x[dk], nm, depth + 1))
|
||
return out
|
||
|
||
first = data[0]
|
||
if isinstance(first, (list, tuple)):
|
||
rows = [r if isinstance(r, (list, tuple)) else [r] for r in data]
|
||
return [(name, rows)]
|
||
if isinstance(first, dict):
|
||
headers = []
|
||
for item in data:
|
||
if isinstance(item, dict):
|
||
for k in item:
|
||
if k not in headers:
|
||
headers.append(k)
|
||
rows = [headers]
|
||
for item in data:
|
||
if isinstance(item, dict):
|
||
rows.append([item.get(k, "") for k in headers])
|
||
else:
|
||
rows.append([item])
|
||
return [(name, rows)]
|
||
# список скаляров -> одна строка
|
||
return [(name, [list(data)])]
|
||
|
||
if isinstance(data, dict):
|
||
for key in self._WRAP_KEYS:
|
||
if key in data:
|
||
res = self._find_tables(data[key], name, depth + 1)
|
||
if res:
|
||
return res
|
||
|
||
values = list(data.values())
|
||
# {"Лист1": [[..]], "Лист2": [[..]]} -> несколько листов
|
||
if values and all(isinstance(v, list) and v for v in values):
|
||
out = []
|
||
for k, v in data.items():
|
||
out.extend(self._find_tables(v, str(k), depth + 1))
|
||
if out:
|
||
return out
|
||
|
||
# плоский объект -> заголовки + одна строка
|
||
if values and all(not isinstance(v, (list, dict)) for v in values):
|
||
return [(name, [list(data.keys()), values])]
|
||
|
||
for v in values:
|
||
if isinstance(v, (list, dict)):
|
||
res = self._find_tables(v, name, depth + 1)
|
||
if res:
|
||
return res
|
||
return []
|
||
|
||
@staticmethod
|
||
def _parse_plain_table(text: str) -> list:
|
||
"""Фолбэк, если пришёл не JSON: Markdown-таблица или CSV/TSV."""
|
||
lines = [l for l in str(text).strip().splitlines() if l.strip()]
|
||
if not lines:
|
||
return []
|
||
if all(l.strip().startswith("|") for l in lines):
|
||
rows = []
|
||
for l in lines:
|
||
cells = [c.strip() for c in l.strip().strip("|").split("|")]
|
||
if all(re.fullmatch(r":?-+:?", c) for c in cells if c) and any(cells):
|
||
continue
|
||
rows.append(cells)
|
||
return rows
|
||
delim = max(",;\t", key=lambda d: lines[0].count(d))
|
||
if lines[0].count(delim) == 0:
|
||
return []
|
||
return [row for row in csv.reader(lines, delimiter=delim)]
|
||
|
||
@staticmethod
|
||
def _coerce_number(s: str):
|
||
t = s.strip()
|
||
m = re.fullmatch(r"-?(0|[1-9]\d*)([.,]\d+)?", t)
|
||
if m:
|
||
return float(t.replace(",", ".")) if m.group(2) else int(t)
|
||
if re.fullmatch(r"-?[1-9]\d{0,2}([ \u00a0]\d{3})+([.,]\d+)?", t):
|
||
t = re.sub(r"[ \u00a0]", "", t).replace(",", ".")
|
||
return float(t) if "." in t else int(t)
|
||
return s
|
||
|
||
def _clean_cell(self, v):
|
||
if v is None or isinstance(v, bool):
|
||
return v
|
||
if isinstance(v, int):
|
||
return v
|
||
if isinstance(v, float):
|
||
return v if v == v and v not in (float("inf"), float("-inf")) else str(v)
|
||
if isinstance(v, (dict, list)):
|
||
v = json.dumps(v, ensure_ascii=False)
|
||
s = self._ILLEGAL.sub("", str(v))
|
||
if self.valves.convert_numeric_strings:
|
||
n = self._coerce_number(s)
|
||
if not isinstance(n, str):
|
||
return n
|
||
return s[:32767]
|
||
|
||
@staticmethod
|
||
def _sheet_title(name, used: set) -> str:
|
||
n = re.sub(r"[\[\]:*?/\\]", "_", str(name)).strip("' ")[:31] or "Sheet"
|
||
base, i = n, 2
|
||
while n.lower() in used:
|
||
suffix = f"_{i}"
|
||
n = base[: 31 - len(suffix)] + suffix
|
||
i += 1
|
||
used.add(n.lower())
|
||
return n
|
||
|
||
def _write_sheet(self, ws, rows) -> bool:
|
||
"""Пишет лист. Возвращает True, если строки были обрезаны по max_rows."""
|
||
limit = max(1, self.valves.max_rows)
|
||
truncated = len(rows) > limit
|
||
rows = rows[:limit]
|
||
widths = {}
|
||
for r_idx, row in enumerate(rows, 1):
|
||
if not isinstance(row, (list, tuple)):
|
||
row = [row]
|
||
for c_idx, val in enumerate(row, 1):
|
||
v = self._clean_cell(val)
|
||
cell = ws.cell(row=r_idx, column=c_idx, value=v)
|
||
if (
|
||
isinstance(v, str)
|
||
and v.startswith("=")
|
||
and not self.valves.allow_formulas
|
||
):
|
||
cell.data_type = "s"
|
||
if v is not None:
|
||
longest = max((len(x) for x in str(v).split("\n")), default=0)
|
||
widths[c_idx] = max(widths.get(c_idx, 0), longest)
|
||
for c_idx, w in widths.items():
|
||
ws.column_dimensions[get_column_letter(c_idx)].width = min(
|
||
max(w + 2, 8), 60
|
||
)
|
||
|
||
# Шапка: если первая строка целиком из непустых строк
|
||
if rows:
|
||
first = rows[0] if isinstance(rows[0], (list, tuple)) else [rows[0]]
|
||
if first and all(isinstance(x, str) and x.strip() for x in first):
|
||
for cell in ws[1]:
|
||
cell.font = Font(bold=True)
|
||
if len(rows) > 1:
|
||
ws.freeze_panes = "A2"
|
||
ws.auto_filter.ref = ws.dimensions
|
||
return truncated
|
||
|
||
def _build_xlsx(self, tables, path) -> list:
|
||
"""Собирает xlsx. Возвращает имена листов, обрезанных по max_rows."""
|
||
wb = Workbook()
|
||
wb.remove(wb.active)
|
||
used = set()
|
||
truncated = []
|
||
for name, rows in tables:
|
||
title = self._sheet_title(name, used)
|
||
ws = wb.create_sheet(title)
|
||
if self._write_sheet(ws, rows):
|
||
truncated.append(title)
|
||
wb.save(path)
|
||
return truncated
|
||
|
||
# ------------------------------------------------------------------
|
||
# WORD: Markdown -> docx
|
||
# ------------------------------------------------------------------
|
||
_INLINE_RE = re.compile(
|
||
r"(\*\*.+?\*\*|`[^`]+`|\*[^*\s][^*]*?\*|\[[^\]]+\]\([^)]+\))"
|
||
)
|
||
|
||
def _add_inline(self, p, text: str):
|
||
for part in self._INLINE_RE.split(text):
|
||
if not part:
|
||
continue
|
||
if part.startswith("**") and part.endswith("**") and len(part) > 4:
|
||
p.add_run(part[2:-2]).bold = True
|
||
elif part.startswith("`") and part.endswith("`") and len(part) > 2:
|
||
r = p.add_run(part[1:-1])
|
||
r.font.name = "Consolas"
|
||
elif part.startswith("*") and part.endswith("*") and len(part) > 2:
|
||
p.add_run(part[1:-1]).italic = True
|
||
elif part.startswith("[") and "](" in part:
|
||
# Было: r"\[(.+?)\]\((.+)?)\)$" — лишняя ')' -> re.error на любой ссылке
|
||
m = re.match(r"\[(.+?)\]\((.+?)\)$", part)
|
||
p.add_run(f"{m.group(1)} ({m.group(2)})" if m else part)
|
||
else:
|
||
p.add_run(part)
|
||
|
||
def _add_table(self, doc, block):
|
||
rows = []
|
||
for ln in block:
|
||
cells = [c.strip() for c in ln.strip().strip("|").split("|")]
|
||
if any(cells) and all(re.fullmatch(r":?-+:?", c) for c in cells if c):
|
||
continue # строка-разделитель |---|---|
|
||
rows.append(cells)
|
||
if not rows:
|
||
return
|
||
n = max(len(r) for r in rows)
|
||
table = doc.add_table(rows=len(rows), cols=n)
|
||
table.style = "Table Grid"
|
||
for ri, r in enumerate(rows):
|
||
for ci in range(n):
|
||
cell = table.cell(ri, ci)
|
||
p = cell.paragraphs[0]
|
||
self._add_inline(p, r[ci] if ci < len(r) else "")
|
||
if ri == 0:
|
||
for run in p.runs:
|
||
run.bold = True
|
||
doc.add_paragraph()
|
||
|
||
def _add_markdown_content_to_docx(self, doc, content: str):
|
||
lines = content.replace("\r\n", "\n").split("\n")
|
||
i = 0
|
||
while i < len(lines):
|
||
raw = lines[i].expandtabs(4).rstrip()
|
||
line = raw.strip()
|
||
|
||
if not line:
|
||
i += 1
|
||
continue
|
||
|
||
if line.startswith("```"): # блок кода
|
||
code = []
|
||
i += 1
|
||
while i < len(lines) and not lines[i].strip().startswith("```"):
|
||
code.append(lines[i])
|
||
i += 1
|
||
i += 1
|
||
p = doc.add_paragraph()
|
||
run = p.add_run("\n".join(code))
|
||
run.font.name = "Consolas"
|
||
run.font.size = Pt(9)
|
||
continue
|
||
|
||
if line.lower() in ("<pagebreak>", "\\pagebreak"):
|
||
doc.add_page_break()
|
||
i += 1
|
||
continue
|
||
|
||
if re.fullmatch(r"[-*_]{3,}", line): # горизонтальная линия — пропускаем
|
||
i += 1
|
||
continue
|
||
|
||
if line.startswith("|"): # таблица
|
||
block = []
|
||
while i < len(lines) and lines[i].strip().startswith("|"):
|
||
block.append(lines[i].strip())
|
||
i += 1
|
||
self._add_table(doc, block)
|
||
continue
|
||
|
||
m = re.match(r"^(#{1,6})\s+(.*)$", line)
|
||
if m:
|
||
doc.add_heading(
|
||
m.group(2).replace("**", "").strip(), level=len(m.group(1))
|
||
)
|
||
i += 1
|
||
continue
|
||
|
||
m = re.match(r"^(\s*)[-*+•]\s+(.*)$", raw)
|
||
if m:
|
||
lvl = min(len(m.group(1)) // 2, 2)
|
||
style = "List Bullet" if lvl == 0 else f"List Bullet {lvl + 1}"
|
||
p = doc.add_paragraph(style=style)
|
||
self._add_inline(p, m.group(2))
|
||
i += 1
|
||
continue
|
||
|
||
m = re.match(r"^(\s*)(\d+)[.)]\s+(.*)$", raw)
|
||
if m:
|
||
# Нумерация пишется текстом, чтобы разные списки не «склеивались»
|
||
lvl = min(len(m.group(1)) // 2, 2)
|
||
p = doc.add_paragraph()
|
||
p.paragraph_format.left_indent = DocxInches(0.25 * (lvl + 1))
|
||
p.paragraph_format.first_line_indent = DocxInches(-0.25)
|
||
self._add_inline(p, f"{m.group(2)}. {m.group(3)}")
|
||
i += 1
|
||
continue
|
||
|
||
p = doc.add_paragraph()
|
||
self._add_inline(p, line)
|
||
i += 1
|
||
|
||
def _add_page_number(self, doc):
|
||
p = doc.sections[0].footer.paragraphs[0]
|
||
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||
|
||
def fld(kind):
|
||
r = p.add_run()
|
||
el = OxmlElement("w:fldChar")
|
||
el.set(docx_qn("w:fldCharType"), kind)
|
||
r._r.append(el)
|
||
|
||
fld("begin")
|
||
r = p.add_run()
|
||
it = OxmlElement("w:instrText")
|
||
it.set(docx_qn("xml:space"), "preserve")
|
||
it.text = "PAGE"
|
||
r._r.append(it)
|
||
fld("end")
|
||
|
||
def _build_docx(self, title: str, content: str, path: str):
|
||
doc = Document()
|
||
try:
|
||
st = doc.styles["Normal"]
|
||
if self.valves.docx_font_name:
|
||
st.font.name = self.valves.docx_font_name
|
||
rpr = st.element.get_or_add_rPr()
|
||
rfonts = rpr.find(docx_qn("w:rFonts"))
|
||
if rfonts is not None:
|
||
rfonts.set(docx_qn("w:eastAsia"), self.valves.docx_font_name)
|
||
if self.valves.docx_font_size:
|
||
st.font.size = Pt(self.valves.docx_font_size)
|
||
except Exception:
|
||
logger.warning("Не удалось применить шрифт Word")
|
||
doc.add_heading(title, 0)
|
||
self._add_markdown_content_to_docx(doc, content)
|
||
if self.valves.docx_page_numbers:
|
||
self._add_page_number(doc)
|
||
doc.save(path)
|
||
|
||
# ------------------------------------------------------------------
|
||
# PPTX
|
||
# ------------------------------------------------------------------
|
||
_SLIDE_HDR = re.compile(
|
||
r"^\s*(?:(?:slide|слайд)\s*\d+\s*[:.\-–—)]*\s*(.*)|#{1,2}\s+(.*))$",
|
||
re.IGNORECASE,
|
||
)
|
||
_MD_IMG = re.compile(r"!\[[^\]]*\]\((\S+?)\)")
|
||
_DATA_URI_LINE = re.compile(
|
||
r"^\s*(data:image/[a-z+.\-]+;base64,[A-Za-z0-9+/=]+)\s*$", re.MULTILINE
|
||
)
|
||
_URL_LINE = re.compile(
|
||
r"^\s*((?:https?://\S+\.(?:png|jpe?g|webp|gif|bmp)(?:\?\S*)?)|(?:/api/v1/files/[\w-]+/content))\s*$",
|
||
re.IGNORECASE | re.MULTILINE,
|
||
)
|
||
|
||
def _parse_slides_from_text(self, text: str) -> list:
|
||
"""
|
||
Слайды из простого текста:
|
||
"Slide 1: Title" / "Слайд 1: Заголовок" / "# Заголовок" / "## Заголовок",
|
||
далее — пункты. Подзаголовки ### остаются внутри слайда.
|
||
"""
|
||
slides, cur = [], None
|
||
for line in str(text).replace("\r\n", "\n").split("\n"):
|
||
m = self._SLIDE_HDR.match(line)
|
||
if m:
|
||
title = m.group(1) if m.group(1) is not None else (m.group(2) or "")
|
||
title = title.strip().strip(":").strip().replace("**", "")
|
||
cur = {"title": title, "content": []}
|
||
slides.append(cur)
|
||
elif cur is not None:
|
||
cur["content"].append(line)
|
||
|
||
if not slides: # заголовков нет — делим по пустым строкам
|
||
for block in re.split(r"\n\s*\n", str(text).strip()):
|
||
ls = [l for l in block.split("\n") if l.strip()]
|
||
if ls:
|
||
slides.append(
|
||
{
|
||
"title": ls[0].strip().strip(":").replace("**", ""),
|
||
"content": ls[1:],
|
||
}
|
||
)
|
||
|
||
return [
|
||
{"title": s["title"], "content": "\n".join(s["content"]).strip()}
|
||
for s in slides
|
||
if s["title"] or any(x.strip() for x in s["content"])
|
||
]
|
||
|
||
@classmethod
|
||
def _flatten_list(cls, items, depth=0):
|
||
out = []
|
||
for it in items:
|
||
if isinstance(it, (list, tuple)):
|
||
out.extend(cls._flatten_list(it, depth + 1))
|
||
elif isinstance(it, dict):
|
||
out.append(" " * depth + json.dumps(it, ensure_ascii=False))
|
||
else:
|
||
out.append(" " * depth + str(it))
|
||
return out
|
||
|
||
def _split_image(self, item: dict):
|
||
"""Возвращает (источник_картинки, текст_без_картинки)."""
|
||
src = str(item.get("image") or "").strip()
|
||
content = str(item.get("content") or "")
|
||
|
||
m = self._MD_IMG.search(content)
|
||
if m:
|
||
src = src or m.group(1)
|
||
content = self._MD_IMG.sub("", content)
|
||
for pat in (self._DATA_URI_LINE, self._URL_LINE):
|
||
m = pat.search(content)
|
||
if m:
|
||
src = src or m.group(1)
|
||
content = pat.sub("", content)
|
||
|
||
lines = [l for l in content.split("\n") if l.strip() not in ("", "-", "*", "•")]
|
||
return src, "\n".join(lines).strip()
|
||
|
||
def _normalize_slides(self, raw: list) -> list:
|
||
out = []
|
||
for s in raw:
|
||
if isinstance(s, str):
|
||
s = {"title": s}
|
||
if not isinstance(s, dict):
|
||
continue
|
||
content = s.get("content", s.get("bullets", s.get("text", "")))
|
||
if isinstance(content, (list, tuple)):
|
||
content = "\n".join(self._flatten_list(content))
|
||
src, text = self._split_image({"image": s.get("image"), "content": content})
|
||
out.append(
|
||
{
|
||
"title": self._ILLEGAL.sub("", str(s.get("title", "")).strip()),
|
||
"text": self._ILLEGAL.sub("", text),
|
||
"image": src,
|
||
"notes": self._ILLEGAL.sub("", str(s.get("notes") or "").strip()),
|
||
}
|
||
)
|
||
return out
|
||
|
||
def _split_long_slides(self, slides: list) -> list:
|
||
limit = self.valves.max_bullets_per_slide
|
||
if not limit or limit < 1:
|
||
return slides
|
||
out = []
|
||
for s in slides:
|
||
lines = [l for l in s["text"].split("\n") if l.strip()]
|
||
if s["image"] or len(lines) <= limit:
|
||
out.append(s)
|
||
continue
|
||
for n, start in enumerate(range(0, len(lines), limit), 1):
|
||
part = dict(s)
|
||
part["text"] = "\n".join(lines[start : start + limit])
|
||
if n > 1:
|
||
part["title"] = f"{s['title']} ({n})"
|
||
part["notes"] = ""
|
||
out.append(part)
|
||
return out
|
||
|
||
@staticmethod
|
||
def _make_widescreen(prs):
|
||
"""4:3 -> 16:9 с пересчётом координат заголовков/плейсхолдеров шаблона."""
|
||
old = prs.slide_width
|
||
if abs(old - Inches(10)) > 10000:
|
||
return
|
||
new = int(Inches(13.333))
|
||
k = new / old
|
||
prs.slide_width = new
|
||
|
||
def scale(shapes, only_explicit):
|
||
for sh in shapes:
|
||
try:
|
||
if (
|
||
only_explicit
|
||
and sh._element.spPr.find(pptx_qn("a:xfrm")) is None
|
||
):
|
||
continue
|
||
if sh.left is None:
|
||
continue
|
||
sh.left = int(sh.left * k)
|
||
sh.width = int(sh.width * k)
|
||
except Exception:
|
||
pass
|
||
|
||
scale(prs.slide_master.shapes, False)
|
||
for lay in prs.slide_layouts:
|
||
scale(lay.shapes, True)
|
||
|
||
@staticmethod
|
||
def _drop_empty_placeholders(slide):
|
||
for sh in list(slide.placeholders):
|
||
if sh.has_text_frame and not sh.text_frame.text.strip():
|
||
sh._element.getparent().remove(sh._element)
|
||
|
||
@staticmethod
|
||
def _body_placeholder(slide):
|
||
fallback = None
|
||
for ph in slide.placeholders:
|
||
if ph.placeholder_format.idx == 1:
|
||
return ph
|
||
if (
|
||
ph.placeholder_format.idx != 0
|
||
and ph.has_text_frame
|
||
and fallback is None
|
||
):
|
||
fallback = ph
|
||
return fallback
|
||
|
||
@staticmethod
|
||
def _fill_bullets(ph, text: str):
|
||
tf = ph.text_frame
|
||
lines = [l for l in text.replace("\r\n", "\n").split("\n") if l.strip()]
|
||
for i, line in enumerate(lines):
|
||
line = line.expandtabs(4)
|
||
indent = len(line) - len(line.lstrip())
|
||
t = re.sub(r"^\s*(?:[-*•+]|\d+[.)])\s+", "", line)
|
||
t = re.sub(r"^#+\s*", "", t).replace("**", "").strip()
|
||
p = tf.paragraphs[0] if i == 0 else tf.add_paragraph()
|
||
p.text = t
|
||
p.level = min(indent // 2, 4)
|
||
|
||
@staticmethod
|
||
def _normalize_image(data: bytes):
|
||
"""Возвращает (bytes, w, h); форматы, которые pptx не читает (WEBP/TIFF и т.п.), конвертирует в PNG."""
|
||
if not _HAS_PIL:
|
||
raise ValueError("Pillow не установлен: `pip install pillow`")
|
||
with Image.open(io.BytesIO(data)) as im:
|
||
fmt, (w, h) = im.format, im.size
|
||
if fmt in ("PNG", "JPEG", "GIF", "BMP"):
|
||
return data, w, h
|
||
buf = io.BytesIO()
|
||
im.convert("RGBA").save(buf, "PNG")
|
||
return buf.getvalue(), w, h
|
||
|
||
@staticmethod
|
||
def _place_picture(slide, data, w, h, left, top, box_w, box_h):
|
||
k = min(box_w / w, box_h / h)
|
||
pw, ph = max(1, int(w * k)), max(1, int(h * k))
|
||
slide.shapes.add_picture(
|
||
io.BytesIO(data),
|
||
int(left + (box_w - pw) / 2),
|
||
int(top + (box_h - ph) / 2),
|
||
pw,
|
||
ph,
|
||
)
|
||
|
||
def _build_pptx(self, title: str, slides: list, images: dict, path: str):
|
||
tpl = self.valves.pptx_template_path.strip()
|
||
use_tpl = bool(tpl) and os.path.isfile(tpl)
|
||
prs = Presentation(tpl) if use_tpl else Presentation()
|
||
if not use_tpl and self.valves.pptx_widescreen:
|
||
self._make_widescreen(prs)
|
||
|
||
def layout(idx):
|
||
n = len(prs.slide_layouts)
|
||
return prs.slide_layouts[idx if idx < n else min(1, n - 1)]
|
||
|
||
# Титульный слайд
|
||
s = prs.slides.add_slide(layout(0))
|
||
if s.shapes.title is not None:
|
||
s.shapes.title.text = title
|
||
self._drop_empty_placeholders(s)
|
||
|
||
margin = Inches(0.5)
|
||
box_left, box_top = margin, Inches(1.5)
|
||
box_w = prs.slide_width - 2 * margin
|
||
box_h = prs.slide_height - Inches(2)
|
||
|
||
for idx, it in enumerate(slides):
|
||
has_body = bool(it["text"] or it["image"])
|
||
slide = prs.slides.add_slide(layout(1 if has_body else 2))
|
||
if slide.shapes.title is not None:
|
||
slide.shapes.title.text = it["title"]
|
||
body = self._body_placeholder(slide)
|
||
text = it["text"]
|
||
|
||
if it["image"]:
|
||
img = images.get(idx)
|
||
if isinstance(img, tuple):
|
||
data, w, h = img
|
||
if text and body is not None: # текст слева, картинка справа
|
||
half = int(box_w / 2)
|
||
body.left, body.top = box_left, box_top
|
||
body.width, body.height = half - Inches(0.1), box_h
|
||
self._fill_bullets(body, text)
|
||
self._place_picture(
|
||
slide,
|
||
data,
|
||
w,
|
||
h,
|
||
box_left + half + Inches(0.1),
|
||
box_top,
|
||
half - Inches(0.1),
|
||
box_h,
|
||
)
|
||
else:
|
||
self._place_picture(
|
||
slide, data, w, h, box_left, box_top, box_w, box_h
|
||
)
|
||
text = None
|
||
else:
|
||
reason = img if isinstance(img, str) else "неизвестная ошибка"
|
||
text = (
|
||
text + "\n" if text else ""
|
||
) + f"(изображение не загружено: {reason})"
|
||
|
||
if text and body is not None:
|
||
self._fill_bullets(body, text)
|
||
|
||
if it["notes"]:
|
||
slide.notes_slide.notes_text_frame.text = it["notes"]
|
||
self._drop_empty_placeholders(slide)
|
||
|
||
prs.save(path)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Загрузка в Open WebUI
|
||
# ------------------------------------------------------------------
|
||
def _public_base(self, request) -> str:
|
||
base = self.valves.public_domain.rstrip("/")
|
||
if "your-openwebui-domain" in base and request is not None:
|
||
try:
|
||
base = str(request.base_url).rstrip("/")
|
||
except Exception:
|
||
pass
|
||
return base
|
||
|
||
async def _upload_file(
|
||
self,
|
||
local_path: str,
|
||
filename: str,
|
||
api_key: str,
|
||
mime: str,
|
||
request=None,
|
||
note: str = "",
|
||
emitter=None,
|
||
) -> str:
|
||
url = f"{self.valves.internal_api_url.rstrip('/')}/api/v1/files/"
|
||
if not self.valves.process_uploaded_files:
|
||
url += "?process=false"
|
||
|
||
status, text, data = 0, "", {}
|
||
try:
|
||
with open(local_path, "rb") as f:
|
||
file_bytes = f.read()
|
||
|
||
for attempt in range(3):
|
||
try:
|
||
form = aiohttp.FormData()
|
||
form.add_field(
|
||
"file", file_bytes, filename=filename, content_type=mime
|
||
)
|
||
async with aiohttp.ClientSession(
|
||
timeout=aiohttp.ClientTimeout(total=60)
|
||
) as session:
|
||
async with session.post(
|
||
url,
|
||
headers={"Authorization": f"Bearer {api_key}"},
|
||
data=form,
|
||
) as resp:
|
||
status, text = resp.status, await resp.text()
|
||
if status < 500:
|
||
break
|
||
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
|
||
status, text = 0, str(e)
|
||
if attempt < 2: # после последней попытки не ждём
|
||
await asyncio.sleep(1.5 * (attempt + 1))
|
||
except Exception as e:
|
||
logger.exception("Upload failed for %s", filename)
|
||
return f"❌ Системная ошибка: {e}"
|
||
finally:
|
||
self._safe_unlink(local_path)
|
||
|
||
try:
|
||
data = json.loads(text)
|
||
if not isinstance(data, dict):
|
||
data = {}
|
||
except (json.JSONDecodeError, ValueError):
|
||
data = {}
|
||
|
||
if not 200 <= status < 300:
|
||
return f"❌ Ошибка загрузки {status}: {text[:300]}"
|
||
|
||
file_id = data.get("id") or data.get("uuid") or data.get("file_id")
|
||
if not file_id:
|
||
return f"❌ Ошибка загрузки: в ответе нет id файла: {text[:300]}"
|
||
|
||
# Прикрепляем файл к сообщению: скачивание из чата идёт с сессией пользователя,
|
||
# без отдельного входа в браузере (прямая ссылка требует авторизации)
|
||
if emitter:
|
||
try:
|
||
await emitter(
|
||
{
|
||
"type": "files",
|
||
"data": {
|
||
"files": [
|
||
{
|
||
"type": "file",
|
||
"id": file_id,
|
||
"name": filename,
|
||
"url": f"/api/v1/files/{file_id}/content",
|
||
"size": len(file_bytes),
|
||
}
|
||
]
|
||
},
|
||
}
|
||
)
|
||
except Exception:
|
||
logger.warning("Не удалось прикрепить файл к сообщению", exc_info=True)
|
||
|
||
download_url = f"{self._public_base(request)}/api/v1/files/{file_id}/content"
|
||
warn = f"\n⚠️ {note}" if note else ""
|
||
return f"""
|
||
|
||
---
|
||
### ✅ Документ создан
|
||
📁 **Файл:** `{filename}`
|
||
🔗 [Скачать]({download_url}){warn}
|
||
---
|
||
"""
|
||
|
||
def _check_input_size(self, s, limit=None) -> str:
|
||
limit = self.valves.max_content_chars if limit is None else limit
|
||
if len(str(s)) > limit:
|
||
return f"❌ Слишком большой ввод (лимит {limit} символов)."
|
||
return ""
|
||
|
||
# ------------------------------------------------------------------
|
||
# Реализации публичных функций
|
||
# ------------------------------------------------------------------
|
||
async def word(self, title, content, user, emitter, request) -> str:
|
||
api_key = self._get_api_key(user)
|
||
if not api_key:
|
||
return "⚠️ Не указан API Key (UserValves -> openwebui_api_key)."
|
||
if not _HAS_DOCX:
|
||
return "⚠️ Не установлен пакет `python-docx`. Установите: `pip install python-docx`."
|
||
err = self._check_input_size(content)
|
||
if err:
|
||
return err
|
||
|
||
title = str(title).strip() or "Document"
|
||
filename = self._safe_filename(title, ".docx")
|
||
content = self._ILLEGAL.sub("", str(content))
|
||
clean_title = self._ILLEGAL.sub("", title)
|
||
|
||
fd, local_path = tempfile.mkstemp(suffix=".docx")
|
||
os.close(fd)
|
||
await self._emit(emitter, "Генерирую Word-документ…")
|
||
try:
|
||
await asyncio.to_thread(self._build_docx, clean_title, content, local_path)
|
||
except Exception as e:
|
||
logger.exception("Word generation failed")
|
||
self._safe_unlink(local_path)
|
||
return f"❌ Ошибка Word: {e}"
|
||
|
||
await self._emit(emitter, "Загружаю файл…")
|
||
res = await self._upload_file(
|
||
local_path, filename, api_key, MIME_DOCX, request, emitter=emitter
|
||
)
|
||
await self._emit(emitter, "Готово", done=True)
|
||
return res
|
||
|
||
async def excel(self, filename, data_json, user, emitter, request) -> str:
|
||
api_key = self._get_api_key(user)
|
||
if not api_key:
|
||
return "⚠️ Не указан API Key (UserValves -> openwebui_api_key)."
|
||
if not _HAS_OPENPYXL:
|
||
return "⚠️ Не установлен пакет `openpyxl`. Установите: `pip install openpyxl`."
|
||
err = self._check_input_size(data_json, self.valves.max_excel_chars)
|
||
if err:
|
||
return err
|
||
|
||
filename = self._safe_filename(filename, ".xlsx")
|
||
|
||
try:
|
||
if isinstance(data_json, (list, dict)):
|
||
raw = data_json
|
||
else:
|
||
text = self._strip_fences(data_json)
|
||
try:
|
||
raw = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
raw = self._parse_plain_table(text) # CSV/Markdown -> [[..]]
|
||
tables = [(n, r) for n, r in self._find_tables(raw) if r]
|
||
except Exception as e:
|
||
return f"❌ Ошибка Excel: {e}"
|
||
|
||
if not tables:
|
||
return (
|
||
"❌ Ошибка Excel: не удалось найти таблицу в данных. "
|
||
f"Получено: {str(data_json)[:100]}..."
|
||
)
|
||
|
||
fd, local_path = tempfile.mkstemp(suffix=".xlsx")
|
||
os.close(fd)
|
||
await self._emit(emitter, "Генерирую Excel…")
|
||
try:
|
||
truncated = await asyncio.to_thread(self._build_xlsx, tables, local_path)
|
||
except Exception as e:
|
||
logger.exception("Excel generation failed")
|
||
self._safe_unlink(local_path)
|
||
return f"❌ Ошибка Excel: {e}"
|
||
|
||
note = ""
|
||
if truncated:
|
||
note = (
|
||
f"Строки обрезаны до {self.valves.max_rows} на листах: "
|
||
+ ", ".join(truncated)
|
||
)
|
||
|
||
await self._emit(emitter, "Загружаю файл…")
|
||
res = await self._upload_file(
|
||
local_path, filename, api_key, MIME_XLSX, request, note, emitter=emitter
|
||
)
|
||
await self._emit(emitter, "Готово", done=True)
|
||
return res
|
||
|
||
async def pptx(self, filename, title, slides_content, user, emitter, request) -> str:
|
||
api_key = self._get_api_key(user)
|
||
if not api_key:
|
||
return "⚠️ Не указан API Key (UserValves -> openwebui_api_key)."
|
||
if not _HAS_PPTX:
|
||
return "⚠️ Не установлен пакет `python-pptx`. Установите: `pip install python-pptx`."
|
||
err = self._check_input_size(slides_content)
|
||
if err:
|
||
return err
|
||
|
||
filename = self._safe_filename(filename, ".pptx")
|
||
title = self._ILLEGAL.sub("", str(title).strip()) or "Presentation"
|
||
|
||
# 1. Разбор входа: JSON (в т.ч. в ```-обёртке) -> иначе текст
|
||
text = self._strip_fences(slides_content)
|
||
try:
|
||
parsed = json.loads(text)
|
||
except (json.JSONDecodeError, ValueError):
|
||
parsed = None
|
||
|
||
if isinstance(parsed, dict):
|
||
if parsed.get("title") and not str(title).strip():
|
||
title = str(parsed["title"])
|
||
parsed = parsed.get("slides") or parsed.get("data")
|
||
if isinstance(parsed, list):
|
||
slides = self._normalize_slides(parsed)
|
||
else:
|
||
slides = self._normalize_slides(
|
||
self._parse_slides_from_text(str(slides_content))
|
||
)
|
||
|
||
if not slides:
|
||
return (
|
||
"❌ Ошибка PPTX: не найдено ни одного слайда. Формат:\n"
|
||
'- JSON: [{"title": "...", "content": "..."}]\n'
|
||
'- или текст: "Slide 1: Title\\n- content"'
|
||
)
|
||
slides = self._split_long_slides(slides)
|
||
if len(slides) > self.valves.max_slides:
|
||
return f"❌ Ошибка PPTX: слишком много слайдов (лимит {self.valves.max_slides})."
|
||
|
||
# 2. Картинки: загрузка асинхронно, до сборки файла
|
||
await self._emit(emitter, "Готовлю слайды…")
|
||
images = {}
|
||
for idx, it in enumerate(slides):
|
||
if not it["image"]:
|
||
continue
|
||
try:
|
||
raw_img = await self._fetch_image(it["image"], api_key)
|
||
images[idx] = await asyncio.to_thread(self._normalize_image, raw_img)
|
||
except Exception as e:
|
||
logger.warning("Image insert failed for slide %r: %s", it["title"], e)
|
||
images[idx] = str(e)[:120]
|
||
|
||
# 3. Сборка
|
||
fd, local_path = tempfile.mkstemp(suffix=".pptx")
|
||
os.close(fd)
|
||
await self._emit(emitter, "Генерирую PowerPoint…")
|
||
try:
|
||
await asyncio.to_thread(self._build_pptx, title, slides, images, local_path)
|
||
except Exception as e:
|
||
logger.exception("PPTX generation failed")
|
||
self._safe_unlink(local_path)
|
||
return f"❌ Ошибка PPTX: {e}"
|
||
|
||
await self._emit(emitter, "Загружаю файл…")
|
||
res = await self._upload_file(
|
||
local_path, filename, api_key, MIME_PPTX, request, emitter=emitter
|
||
)
|
||
await self._emit(emitter, "Готово", done=True)
|
||
return res
|
||
|
||
|
||
# ======================================================================
|
||
# Tools: только Valves, UserValves и три публичные функции (их и видит модель)
|
||
# ======================================================================
|
||
class Tools:
|
||
class Valves(BaseModel):
|
||
public_domain: str = Field(
|
||
default="https://your-openwebui-domain.com",
|
||
description="Публичный домен Open WebUI, БЕЗ слэша в конце. Если не задан — берётся из запроса.",
|
||
)
|
||
internal_api_url: str = Field(
|
||
default="http://127.0.0.1:8080",
|
||
description="Внутренний URL API (Docker по умолчанию: http://127.0.0.1:8080)",
|
||
)
|
||
process_uploaded_files: bool = Field(
|
||
default=False,
|
||
description="Прогонять сгенерированный файл через RAG-обработку при загрузке (обычно не нужно)",
|
||
)
|
||
max_rows: int = Field(
|
||
default=50000, description="Excel: максимум строк на лист"
|
||
)
|
||
max_slides: int = Field(default=60, description="PPTX: максимум слайдов")
|
||
max_content_chars: int = Field(
|
||
default=200000, description="Максимальная длина входного текста/JSON"
|
||
)
|
||
max_excel_chars: int = Field(
|
||
default=2000000,
|
||
description="Excel: отдельный лимит входного JSON (больше max_content_chars, т.к. крупные таблицы дают длинный JSON)",
|
||
)
|
||
max_bullets_per_slide: int = Field(
|
||
default=8,
|
||
description="PPTX: если в слайде больше пунктов — он делится на несколько (0 = не делить)",
|
||
)
|
||
max_image_mb: int = Field(
|
||
default=10, description="PPTX: максимум МБ на картинку"
|
||
)
|
||
allow_formulas: bool = Field(
|
||
default=False,
|
||
description="Excel: строки, начинающиеся с '=', считать формулами (иначе — текст, защита от formula injection)",
|
||
)
|
||
convert_numeric_strings: bool = Field(
|
||
default=False,
|
||
description="Excel: превращать строки вида '123' и '1 234,5' в числа",
|
||
)
|
||
docx_font_name: str = Field(default="Calibri", description="Word: шрифт")
|
||
docx_font_size: int = Field(default=11, description="Word: размер шрифта, pt")
|
||
docx_page_numbers: bool = Field(
|
||
default=True, description="Word: номера страниц"
|
||
)
|
||
pptx_widescreen: bool = Field(
|
||
default=True, description="PPTX: формат 16:9 (для встроенного шаблона)"
|
||
)
|
||
pptx_template_path: str = Field(
|
||
default="",
|
||
description="PPTX: путь к своему .pptx-шаблону на сервере (необязательно)",
|
||
)
|
||
|
||
class UserValves(BaseModel):
|
||
openwebui_api_key: str = Field(
|
||
default="",
|
||
description="🔴 ОБЯЗАТЕЛЬНО: API Key (Settings -> Account -> API Keys)",
|
||
json_schema_extra={"input": {"type": "password"}},
|
||
)
|
||
|
||
def __init__(self):
|
||
self.valves = self.Valves()
|
||
self.user_valves = self.UserValves()
|
||
# file_handler намеренно не выставляется: он отключает стандартную обработку вложений
|
||
|
||
async def generate_word_doc(
|
||
self,
|
||
title: str,
|
||
content: str,
|
||
__user__: dict = None,
|
||
__event_emitter__=None,
|
||
__request__=None,
|
||
) -> str:
|
||
"""
|
||
Generate a Word document (.docx).
|
||
|
||
Args:
|
||
title: Document title (also used for the file name)
|
||
content: Document body in Markdown: headings (# .. ######), **bold**, *italic*,
|
||
`code`, bullet and numbered lists (indent 2 spaces for nesting),
|
||
tables (| a | b |), code blocks, <pagebreak> for a page break
|
||
"""
|
||
# Impl создаётся на каждый вызов: self.valves может быть заменён из UI
|
||
return await _Impl(self.valves).word(
|
||
title, content, __user__, __event_emitter__, __request__
|
||
)
|
||
|
||
async def generate_excel_file(
|
||
self,
|
||
filename: str,
|
||
data_json: str,
|
||
__user__: dict = None,
|
||
__event_emitter__=None,
|
||
__request__=None,
|
||
) -> str:
|
||
"""
|
||
Generate an Excel file (.xlsx). Finds table data automatically in the JSON.
|
||
|
||
Args:
|
||
filename: Name of the Excel file
|
||
data_json: JSON with table data. Accepted: [[row], ...]; [{col: value}, ...];
|
||
{"Sheet1": [[...]], "Sheet2": [[...]]} or [{"name": "S1", "data": [[...]]}, ...]
|
||
for several sheets. CSV or a Markdown table also works.
|
||
"""
|
||
return await _Impl(self.valves).excel(
|
||
filename, data_json, __user__, __event_emitter__, __request__
|
||
)
|
||
|
||
async def generate_pptx_file(
|
||
self,
|
||
filename: str,
|
||
title: str,
|
||
slides_content: str,
|
||
__user__: dict = None,
|
||
__event_emitter__=None,
|
||
__request__=None,
|
||
) -> str:
|
||
"""
|
||
Generate a PowerPoint presentation (.pptx).
|
||
|
||
Args:
|
||
filename: Name of the PPTX file
|
||
title: Main title of the presentation
|
||
slides_content: JSON array [{"title": "...", "content": "text or list of bullets",
|
||
"image": "URL / /api/v1/files/<id>/content / data URI (optional)",
|
||
"notes": "speaker notes (optional)"}], or {"slides": [...]}, or plain text:
|
||
"Slide 1: Title\\n- bullet\\n\\nSlide 2: ...". Indent bullets with 2 spaces for sub-levels.
|
||
"""
|
||
return await _Impl(self.valves).pptx(
|
||
filename, title, slides_content, __user__, __event_emitter__, __request__
|
||
) |