1164 lines
45 KiB
Python
1164 lines
45 KiB
Python
"""
|
|
title: Document Generator 3.7
|
|
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.7
|
|
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"
|
|
|
|
|
|
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_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 намеренно не выставляется: он отключает стандартную обработку вложений
|
|
|
|
# ------------------------------------------------------------------
|
|
# Утилиты
|
|
# ------------------------------------------------------------------
|
|
_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 _assert_public_url(self, url: str):
|
|
p = urlparse(url)
|
|
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)
|
|
loop = asyncio.get_running_loop()
|
|
infos = await loop.getaddrinfo(p.hostname, port, type=socket.SOCK_STREAM)
|
|
for info in infos:
|
|
ip = ipaddress.ip_address(info[4][0].split("%")[0])
|
|
if not ip.is_global:
|
|
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) :]
|
|
|
|
if src.startswith("/") and not src.startswith("//"):
|
|
if not src.startswith("/api/v1/files/"):
|
|
raise ValueError("Разрешены только /api/v1/files/ пути")
|
|
url = self.valves.internal_api_url.rstrip("/") + src
|
|
headers["Authorization"] = f"Bearer {api_key}"
|
|
else:
|
|
await self._assert_public_url(src)
|
|
url = src
|
|
|
|
timeout = aiohttp.ClientTimeout(total=30)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.get(url, headers=headers, allow_redirects=False) 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):
|
|
rows = rows[: max(1, self.valves.max_rows)]
|
|
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
|
|
|
|
def _build_xlsx(self, tables, path):
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
used = set()
|
|
for name, rows in tables:
|
|
ws = wb.create_sheet(self._sheet_title(name, used))
|
|
self._write_sheet(ws, rows)
|
|
wb.save(path)
|
|
|
|
# ------------------------------------------------------------------
|
|
# WORD: Markdown -> docx
|
|
# ------------------------------------------------------------------
|
|
_INLINE_RE = re.compile(
|
|
r"(\*\*.+?\*\*|`[^`]+`|\*[^*\s][^*]*?\*|\[[^\]]+\]\([^)\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:
|
|
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 и т.п.), конвертирует в PNG."""
|
|
if not _HAS_PIL:
|
|
raise ValueError("Pillow не установлен")
|
|
with Image.open(io.BytesIO(data)) as im:
|
|
fmt, (w, h) = im.format, im.size
|
|
if fmt in ("PNG", "JPEG", "GIF", "BMP", "TIFF"):
|
|
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
|
|
) -> 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)
|
|
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]}"
|
|
|
|
download_url = f"{self._public_base(request)}/api/v1/files/{file_id}/content"
|
|
return f"""
|
|
|
|
---
|
|
### ✅ Документ создан
|
|
📁 **Файл:** `{filename}`
|
|
🔗 [Скачать]({download_url})
|
|
---
|
|
"""
|
|
|
|
def _check_input_size(self, s) -> str:
|
|
if len(str(s)) > self.valves.max_content_chars:
|
|
return f"❌ Слишком большой ввод (лимит {self.valves.max_content_chars} символов)."
|
|
return ""
|
|
|
|
# ------------------------------------------------------------------
|
|
# ПУБЛИЧНЫЕ ФУНКЦИИ
|
|
# ------------------------------------------------------------------
|
|
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
|
|
"""
|
|
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`."
|
|
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(__event_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(__event_emitter__, "Загружаю файл…")
|
|
res = await self._upload_file(
|
|
local_path, filename, api_key, MIME_DOCX, __request__
|
|
)
|
|
await self._emit(__event_emitter__, "Готово", done=True)
|
|
return res
|
|
|
|
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.
|
|
"""
|
|
api_key = self._get_api_key(__user__)
|
|
if not api_key:
|
|
return "⚠️ Не указан API Key (UserValves -> openwebui_api_key)."
|
|
if not _HAS_OPENPYXL:
|
|
return "⚠️ Не установлен пакет `openpyxl`."
|
|
err = self._check_input_size(data_json)
|
|
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):
|
|
rows = self._parse_plain_table(text)
|
|
raw = rows # 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(__event_emitter__, "Генерирую Excel…")
|
|
try:
|
|
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}"
|
|
|
|
await self._emit(__event_emitter__, "Загружаю файл…")
|
|
res = await self._upload_file(
|
|
local_path, filename, api_key, MIME_XLSX, __request__
|
|
)
|
|
await self._emit(__event_emitter__, "Готово", done=True)
|
|
return res
|
|
|
|
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.
|
|
"""
|
|
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`."
|
|
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(__event_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(__event_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(__event_emitter__, "Загружаю файл…")
|
|
res = await self._upload_file(
|
|
local_path, filename, api_key, MIME_PPTX, __request__
|
|
)
|
|
await self._emit(__event_emitter__, "Готово", done=True)
|
|
return res
|