From 965d98e938139a503932dc96b78dbd071ff5bf6a Mon Sep 17 00:00:00 2001 From: host Date: Sun, 18 Oct 2026 15:00:17 +0300 Subject: [PATCH] Add Nextcloud WebDAV tools (list, create, move, delete, sort photos, read file contents) --- nextcloud_tools.py | 309 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 309 insertions(+) create mode 100644 nextcloud_tools.py diff --git a/nextcloud_tools.py b/nextcloud_tools.py new file mode 100644 index 0000000..3636e40 --- /dev/null +++ b/nextcloud_tools.py @@ -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/ + 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 \ No newline at end of file