Initial commit: Document Generator 3.7
This commit is contained in:
host committed
commit
f0caa8431f
1 file changed
+888
@@ -0,0 +1,888 @@
|
||||
"""
|
||||
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_openwebui_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
|
||||
|
||||
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:-2]).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 ph in slide.placeholders:
|
||||
if ph.has_text_frame and not ph.text_frame.text.strip():
|
||||
ph._element.getparent().remove(ph._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
|
||||
Reference in new issue
Block a user