Add Nextcloud WebDAV tools (list, create, move, delete, sort photos, read file contents)
This commit is contained in:
host committed
1 parent
6a1085572d
commit
965d98e938
1 file changed
+309
@@ -0,0 +1,309 @@
|
||||
"""
|
||||
title: Nextcloud Files
|
||||
description: Управление файлами Nextcloud через WebDAV: список, папки, перемещение, удаление, сортировка фото по месяцам, чтение текста из .pptx/.docx/.xlsx/.pdf/.txt.
|
||||
requirements: requests, pydantic, python-pptx, python-docx, openpyxl, pypdf
|
||||
"""
|
||||
import posixpath
|
||||
import xml.etree.ElementTree as ET
|
||||
from email.utils import parsedate_to_datetime
|
||||
from urllib.parse import quote, unquote, urlparse
|
||||
import requests
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
DAV_NS = {"d": "DAV:"}
|
||||
PHOTO_EXT = (".jpg", ".jpeg", ".png", ".heic", ".webp", ".gif")
|
||||
|
||||
# Магия OLE/CFB — устаревший бинарный формат Office 97-2003 (.ppt/.doc/.xls)
|
||||
OLE_MAGIC = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"
|
||||
|
||||
|
||||
class Tools:
|
||||
class Valves(BaseModel):
|
||||
NEXTCLOUD_URL: str = Field(
|
||||
"https://nc.nasweb.ru", description="Базовый URL Nextcloud"
|
||||
)
|
||||
USERNAME: str = Field("", description="Логин Nextcloud (не email)")
|
||||
APP_PASSWORD: str = Field("", description="App password, а не основной пароль")
|
||||
|
||||
def __init__(self):
|
||||
self.valves = self.Valves()
|
||||
|
||||
# ---------- внутренние хелперы ----------
|
||||
def _session(self) -> requests.Session:
|
||||
s = requests.Session()
|
||||
s.auth = (self.valves.USERNAME, self.valves.APP_PASSWORD)
|
||||
return s
|
||||
|
||||
def _base(self) -> str:
|
||||
return f"{self.valves.NEXTCLOUD_URL.rstrip('/')}/remote.php/dav/files/{self.valves.USERNAME}"
|
||||
|
||||
def _url(self, path: str) -> str:
|
||||
# quote() экранирует пробелы, кириллицу и прочие спецсимволы
|
||||
return f"{self._base()}/{quote(path.strip('/'))}"
|
||||
|
||||
def _list(self, path: str) -> list[dict]:
|
||||
"""PROPFIND с Depth: 1 — содержимое папки, без самой папки."""
|
||||
r = self._session().request(
|
||||
"PROPFIND", self._url(path), headers={"Depth": "1"}, timeout=30
|
||||
)
|
||||
r.raise_for_status()
|
||||
prefix = urlparse(self._base()).path # /remote.php/dav/files/<user>
|
||||
root = ET.fromstring(r.content)
|
||||
own = path.strip("/")
|
||||
items = []
|
||||
for resp in root.findall("d:response", DAV_NS):
|
||||
href = unquote(resp.findtext("d:href", namespaces=DAV_NS) or "")
|
||||
rel = (
|
||||
href[len(prefix):].strip("/")
|
||||
if href.startswith(prefix)
|
||||
else href.strip("/")
|
||||
)
|
||||
if rel == own:
|
||||
continue
|
||||
prop = resp.find("d:propstat/d:prop", DAV_NS)
|
||||
items.append(
|
||||
{
|
||||
"path": rel,
|
||||
"name": posixpath.basename(rel),
|
||||
"is_dir": prop.find("d:resourcetype/d:collection", DAV_NS)
|
||||
is not None,
|
||||
"modified": prop.findtext("d:getlastmodified", namespaces=DAV_NS),
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
def _ensure_folder(self, path: str) -> None:
|
||||
"""MKCOL не умеет рекурсию, поэтому создаём папки по одной."""
|
||||
current = ""
|
||||
for part in path.strip("/").split("/"):
|
||||
current = f"{current}/{part}".strip("/")
|
||||
r = self._session().request("MKCOL", self._url(current), timeout=30)
|
||||
if r.status_code not in (201, 405): # 405 — папка уже существует
|
||||
r.raise_for_status()
|
||||
|
||||
# ---------- чтение содержимого файлов ----------
|
||||
def read_file(self, path: str, max_chars: int = 8000, offset: int = 0) -> str:
|
||||
"""
|
||||
Прочитать текст из файла Nextcloud (.pptx, .docx, .xlsx, .pdf, .txt, .md, .csv).
|
||||
Сначала проверьте реальный путь через list_folder — не выдумывайте его.
|
||||
:param path: путь к файлу внутри аккаунта, например "Documents/report.pptx"
|
||||
:param max_chars: сколько символов вернуть (по умолчанию 8000)
|
||||
:param offset: с какого символа читать — для длинных файлов по частям
|
||||
:return: извлечённый текст
|
||||
"""
|
||||
try:
|
||||
r = self._session().get(self._url(path), timeout=120)
|
||||
r.raise_for_status()
|
||||
except requests.RequestException as e:
|
||||
return f"Ошибка загрузки '{path}': {e}. Проверьте путь через list_folder."
|
||||
|
||||
# Устаревший бинарный Office 97-2003 — его парсеры не понимают
|
||||
if r.content[:8] == OLE_MAGIC:
|
||||
return (
|
||||
"Файл в устаревшем бинарном формате (OLE/CFB, .ppt/.doc/.xls 97-2003). "
|
||||
"Не поддерживается напрямую — нужна конвертация, например: "
|
||||
"libreoffice --headless --convert-to pptx"
|
||||
)
|
||||
|
||||
ext = posixpath.splitext(path)[1].lower()
|
||||
try:
|
||||
text = self._extract_text(r.content, ext)
|
||||
except ImportError as e:
|
||||
return f"Не установлена библиотека: {e}"
|
||||
except Exception as e:
|
||||
return f"Не удалось разобрать файл '{path}' ({ext}): {e}"
|
||||
|
||||
if offset > 0:
|
||||
text = text[offset:]
|
||||
if len(text) > max_chars:
|
||||
return (
|
||||
text[:max_chars]
|
||||
+ f'\n...[обрезано. Продолжение: read_file(path="{path}", offset={offset + max_chars})]'
|
||||
)
|
||||
return text
|
||||
|
||||
def _iter_shape_texts(self, shape) -> list[str]:
|
||||
"""Рекурсивный сбор текста из фигур python-pptx, включая GroupShape."""
|
||||
from pptx.shapes.group import GroupShape
|
||||
if isinstance(shape, GroupShape):
|
||||
out = []
|
||||
for child in shape.shapes:
|
||||
out.extend(self._iter_shape_texts(child))
|
||||
return out
|
||||
out = []
|
||||
if shape.has_text_frame:
|
||||
out.append(shape.text_frame.text)
|
||||
if getattr(shape, "has_table", False) and shape.has_table:
|
||||
for row in shape.table.rows:
|
||||
out.append(" | ".join(c.text for c in row.cells))
|
||||
return out
|
||||
|
||||
def _extract_text(self, data: bytes, ext: str) -> str:
|
||||
"""Извлечь текст из байтов файла в зависимости от расширения."""
|
||||
import io
|
||||
|
||||
if ext == ".pptx":
|
||||
from pptx import Presentation
|
||||
prs = Presentation(io.BytesIO(data))
|
||||
parts = []
|
||||
for i, slide in enumerate(prs.slides, 1):
|
||||
parts.append(f"--- Слайд {i} ---")
|
||||
for shape in slide.shapes:
|
||||
parts.extend(self._iter_shape_texts(shape))
|
||||
# Заметки спикера
|
||||
if slide.has_notes_slide:
|
||||
notes = slide.notes_slide.notes_text_frame.text.strip()
|
||||
if notes:
|
||||
parts.append(f"[заметки спикера: {notes}]")
|
||||
return "\n".join(p for p in parts if p.strip())
|
||||
|
||||
if ext == ".docx":
|
||||
from docx import Document
|
||||
doc = Document(io.BytesIO(data))
|
||||
parts = [p.text for p in doc.paragraphs]
|
||||
# Таблицы doc.paragraphs не покрывает — читаем отдельно
|
||||
for table in doc.tables:
|
||||
for row in table.rows:
|
||||
parts.append(" | ".join(c.text for c in row.cells))
|
||||
return "\n".join(parts)
|
||||
|
||||
if ext == ".xlsx":
|
||||
from openpyxl import load_workbook
|
||||
wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
||||
out = []
|
||||
for ws in wb.worksheets:
|
||||
out.append(f"--- Лист {ws.title} ---")
|
||||
for row in ws.iter_rows(values_only=True):
|
||||
out.append(" | ".join("" if v is None else str(v) for v in row))
|
||||
return "\n".join(out)
|
||||
|
||||
if ext == ".pdf":
|
||||
from pypdf import PdfReader
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
return "\n".join(page.extract_text() or "" for page in reader.pages)
|
||||
|
||||
# Текстовые форматы (.txt, .md, .csv)
|
||||
return data.decode("utf-8", errors="replace")
|
||||
|
||||
# ---------- публичные функции для модели ----------
|
||||
def list_folder(self, path: str = "") -> str:
|
||||
"""
|
||||
Показать содержимое папки Nextcloud.
|
||||
:param path: путь внутри аккаунта, например "media" или "документы"; пусто — корень
|
||||
:return: список файлов и папок
|
||||
"""
|
||||
try:
|
||||
items = self._list(path)
|
||||
except requests.RequestException as e:
|
||||
return f"Ошибка: {e}"
|
||||
if not items:
|
||||
return "Папка пуста"
|
||||
return "\n".join(("📁 " if i["is_dir"] else "📄 ") + i["path"] for i in items)
|
||||
|
||||
def create_folder(self, path: str) -> str:
|
||||
"""
|
||||
Создать папку вместе с недостающими родительскими.
|
||||
:param path: путь новой папки, например "media/2026/03"
|
||||
:return: подтверждение
|
||||
"""
|
||||
try:
|
||||
self._ensure_folder(path)
|
||||
except requests.RequestException as e:
|
||||
return f"Ошибка: {e}"
|
||||
return f"Папка готова: {path}"
|
||||
|
||||
def move(self, source: str, destination: str, overwrite: bool = False) -> str:
|
||||
"""
|
||||
Переместить или переименовать файл/папку.
|
||||
:param source: текущий путь, например "media/photo.jpg"
|
||||
:param destination: новый полный путь с именем файла, например "media/2026/03/photo.jpg"
|
||||
:param overwrite: перезаписывать ли файл, если он уже есть в цели (по умолчанию False)
|
||||
:return: подтверждение или текст ошибки
|
||||
"""
|
||||
headers = {
|
||||
"Destination": self._url(destination),
|
||||
"Overwrite": "T" if overwrite else "F",
|
||||
}
|
||||
r = self._session().request(
|
||||
"MOVE", self._url(source), headers=headers, timeout=120
|
||||
)
|
||||
if r.status_code == 412:
|
||||
return "Отмена: файл с таким именем уже есть в целевой папке. Передайте overwrite=True для замены."
|
||||
if r.status_code not in (201, 204):
|
||||
return f"Ошибка MOVE: {r.status_code} — {r.text[:300]}"
|
||||
return f"Перемещено: {source} → {destination}"
|
||||
|
||||
def delete_file(self, path: str) -> str:
|
||||
"""
|
||||
Удалить файл или папку из Nextcloud. Деструктивная операция — только по явной просьбе пользователя.
|
||||
:param path: путь к файлу или папке, например "media/old.jpg"
|
||||
:return: подтверждение или текст ошибки
|
||||
"""
|
||||
r = self._session().delete(self._url(path), timeout=60)
|
||||
if r.status_code not in (200, 204):
|
||||
return f"Ошибка DELETE: {r.status_code} — {r.text[:300]}"
|
||||
return f"Удалено: {path}"
|
||||
|
||||
def sort_photos_by_month(self, folder: str, dry_run: bool = True) -> str:
|
||||
"""
|
||||
Разложить фото из папки по подпапкам вида ГГГГ-ММ (по дате изменения файла).
|
||||
:param folder: папка с фото, например "media"
|
||||
:param dry_run: True — только показать план; False — выполнить перемещение
|
||||
:return: план или отчёт
|
||||
"""
|
||||
try:
|
||||
photos = [
|
||||
i
|
||||
for i in self._list(folder)
|
||||
if not i["is_dir"] and i["name"].lower().endswith(PHOTO_EXT)
|
||||
]
|
||||
except requests.RequestException as e:
|
||||
return f"Ошибка: {e}"
|
||||
if not photos:
|
||||
return f"Фото не найдены в '{folder}'"
|
||||
|
||||
plan = []
|
||||
for p in photos:
|
||||
if not p["modified"]:
|
||||
continue
|
||||
try:
|
||||
month = parsedate_to_datetime(p["modified"]).strftime("%Y-%m")
|
||||
except Exception:
|
||||
continue
|
||||
dest = f"{folder.strip('/')}/{month}/{p['name']}"
|
||||
if dest != p["path"]:
|
||||
plan.append((p["path"], dest))
|
||||
|
||||
if not plan:
|
||||
return "Все фото уже разложены по месяцам"
|
||||
|
||||
if dry_run:
|
||||
preview = "\n".join(f" {s} → {d}" for s, d in plan[:50])
|
||||
return (
|
||||
f"План: {len(plan)} файлов (показаны первые 50):\n{preview}\n"
|
||||
f"Подтвердите — повторите вызов с dry_run=False."
|
||||
)
|
||||
|
||||
# Создаём все месячные папки
|
||||
try:
|
||||
for month_dir in {posixpath.dirname(d) for _, d in plan}:
|
||||
self._ensure_folder(month_dir)
|
||||
except requests.RequestException as e:
|
||||
return f"Ошибка при создании папок: {e}"
|
||||
|
||||
# Перемещаем
|
||||
moved, errors = 0, []
|
||||
for src, dst in plan:
|
||||
try:
|
||||
res = self.move(src, dst)
|
||||
if res.startswith("Перемещено"):
|
||||
moved += 1
|
||||
else:
|
||||
errors.append(f"{src}: {res}")
|
||||
except requests.RequestException as e:
|
||||
errors.append(f"{src}: {e}")
|
||||
|
||||
report = f"Перемещено: {moved}, ошибок: {len(errors)}"
|
||||
if errors:
|
||||
report += "\n" + "\n".join(errors[:20])
|
||||
return report
|
||||
Reference in new issue
Block a user