309 lines
14 KiB
Python
309 lines
14 KiB
Python
"""
|
||
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 |