Variet Agent
d22493125c
- tools/nextcloud_client.py: WebDAV/OCS/CalDAV/CardDAV 공통 클라이언트 - tools/nc_files.py: 파일 검색/목록/최근/공유링크 - tools/nc_calendar.py: CalDAV 일정 CRUD + ICS 빌더 - tools/nc_mail.py: IMAP 메일 조회 (PLAIN auth for Mailcow) - tools/nc_contacts.py: CardDAV 연락처 + EasyOCR 명함 스캔 - handlers/nc_handler.py: 자연어→NC도구 자동 라우팅 - core/foreman.py: 목표 분해 + 상담 세션 + Vikunja 등록 - prompts/foreman.md: Foreman 시스템 프롬프트 - prompts/unified.md: nextcloud 모드 분류 추가 - config.py: .env 따옴표 파싱 버그 수정 - api/discord_bot.py: /goal 커맨드 + Foreman 스레드 라우팅
246 lines
8.7 KiB
Python
246 lines
8.7 KiB
Python
r"""Nextcloud Files 검색/관리 모듈.
|
|
|
|
파일 검색, 최근 파일 조회, 공유 링크 생성 등.
|
|
업로드는 NC 앱에서 직접 수행 — 봇은 검색/관리/링크 역할.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
import config
|
|
from tools.nextcloud_client import NextcloudClient
|
|
|
|
logger = logging.getLogger("variet.tools.nc_files")
|
|
|
|
|
|
@dataclass
|
|
class NCFile:
|
|
"""Nextcloud 파일 정보."""
|
|
name: str
|
|
path: str # 사용자 상대 경로 (예: Documents/세금/...)
|
|
size: int = 0
|
|
lastmod: str = ""
|
|
content_type: str = ""
|
|
is_dir: bool = False
|
|
|
|
@property
|
|
def size_human(self) -> str:
|
|
"""사람이 읽기 쉬운 크기 표시."""
|
|
if self.size < 1024:
|
|
return f"{self.size}B"
|
|
elif self.size < 1024 * 1024:
|
|
return f"{self.size / 1024:.1f}KB"
|
|
elif self.size < 1024 * 1024 * 1024:
|
|
return f"{self.size / (1024 * 1024):.1f}MB"
|
|
else:
|
|
return f"{self.size / (1024 * 1024 * 1024):.1f}GB"
|
|
|
|
@property
|
|
def icon(self) -> str:
|
|
if self.is_dir:
|
|
return "📁"
|
|
ct = self.content_type or ""
|
|
if "image" in ct:
|
|
return "🖼️"
|
|
elif "pdf" in ct:
|
|
return "📕"
|
|
elif "video" in ct:
|
|
return "🎬"
|
|
elif "audio" in ct:
|
|
return "🎵"
|
|
elif "zip" in ct or "tar" in ct or "rar" in ct:
|
|
return "📦"
|
|
elif "spreadsheet" in ct or "excel" in ct:
|
|
return "📊"
|
|
elif "document" in ct or "word" in ct:
|
|
return "📝"
|
|
return "📄"
|
|
|
|
|
|
class NCFilesClient:
|
|
"""Nextcloud Files 검색/관리 클라이언트."""
|
|
|
|
def __init__(self, nc_client: NextcloudClient = None):
|
|
self.nc = nc_client or NextcloudClient()
|
|
|
|
def _href_to_user_path(self, href: str) -> str:
|
|
"""WebDAV href → 사용자 상대 경로."""
|
|
# href: /remote.php/dav/files/username/Documents/...
|
|
marker = f"/files/{self.nc.username}/"
|
|
idx = href.find(marker)
|
|
if idx >= 0:
|
|
return href[idx + len(marker):].rstrip("/")
|
|
return href.rstrip("/").split("/")[-1]
|
|
|
|
def _to_ncfile(self, item: dict) -> NCFile:
|
|
"""propfind 결과 dict → NCFile."""
|
|
return NCFile(
|
|
name=item.get("name", ""),
|
|
path=self._href_to_user_path(item.get("href", "")),
|
|
size=item.get("size", 0),
|
|
lastmod=item.get("lastmod", ""),
|
|
content_type=item.get("content_type", ""),
|
|
is_dir=item.get("is_dir", False),
|
|
)
|
|
|
|
# ──────────────────────────────────────
|
|
# 검색
|
|
# ──────────────────────────────────────
|
|
|
|
async def search(self, query: str) -> list[NCFile]:
|
|
"""파일 검색 (이름 기준).
|
|
|
|
Args:
|
|
query: 검색어 (부분 일치)
|
|
|
|
Returns:
|
|
매칭된 NCFile 리스트
|
|
"""
|
|
results = await self.nc.webdav_search(query)
|
|
files = [self._to_ncfile(r) for r in results]
|
|
# 디렉토리보다 파일 우선, 최신순
|
|
files.sort(key=lambda f: (f.is_dir, f.name))
|
|
logger.info(f"파일 검색 '{query}': {len(files)}건")
|
|
return files
|
|
|
|
# ──────────────────────────────────────
|
|
# 목록
|
|
# ──────────────────────────────────────
|
|
|
|
async def list_dir(self, path: str = "") -> list[NCFile]:
|
|
"""디렉토리 목록 조회.
|
|
|
|
Args:
|
|
path: 사용자 상대 경로 (빈 문자열 = 루트)
|
|
"""
|
|
dav_path = f"files/{self.nc.username}/{path.lstrip('/')}"
|
|
results = await self.nc.webdav_propfind(dav_path)
|
|
# 첫 번째는 자기 자신 → 제외
|
|
files = [self._to_ncfile(r) for r in results[1:]]
|
|
files.sort(key=lambda f: (not f.is_dir, f.name.lower()))
|
|
return files
|
|
|
|
async def list_recent(self, limit: int = 10) -> list[NCFile]:
|
|
"""최근 수정된 파일 목록.
|
|
|
|
PROPFIND depth=infinity는 부하가 크므로
|
|
OCS activity API나 search를 활용.
|
|
"""
|
|
# 대안: WebDAV SEARCH로 최근 수정 파일 검색
|
|
# 현재는 루트 1단계만 조회하여 lastmod 정렬
|
|
dav_path = f"files/{self.nc.username}"
|
|
# depth=infinity는 서버 부하 → 대신 OCS favorite 또는 search 활용
|
|
# 간단하게: search로 '*' 전체 검색 후 정렬
|
|
results = await self.nc.webdav_search("*")
|
|
files = [self._to_ncfile(r) for r in results if not r.get("is_dir")]
|
|
# lastmod 기준 정렬 (최신 먼저)
|
|
files.sort(key=lambda f: f.lastmod, reverse=True)
|
|
return files[:limit]
|
|
|
|
# ──────────────────────────────────────
|
|
# 공유 링크
|
|
# ──────────────────────────────────────
|
|
|
|
async def create_link(self, path: str, expire_days: int = 7) -> Optional[str]:
|
|
"""파일/폴더의 공유 링크 생성.
|
|
|
|
Args:
|
|
path: 사용자 상대 경로
|
|
expire_days: 만료 일수 (0 = 만료 없음)
|
|
|
|
Returns:
|
|
공유 URL 또는 None
|
|
"""
|
|
# OCS Share API는 / 시작 경로 필요
|
|
share_path = f"/{path.lstrip('/')}"
|
|
return await self.nc.create_share_link(share_path, expire_days)
|
|
|
|
# ──────────────────────────────────────
|
|
# 용량 분석
|
|
# ──────────────────────────────────────
|
|
|
|
async def get_quota(self) -> dict:
|
|
"""사용자 스토리지 용량 정보.
|
|
|
|
Returns:
|
|
{used, total, free, percent} (바이트 기준)
|
|
"""
|
|
try:
|
|
dav_path = f"files/{self.nc.username}"
|
|
results = await self.nc.webdav_propfind(
|
|
dav_path,
|
|
props=["d:quota-used-bytes", "d:quota-available-bytes"],
|
|
depth=0,
|
|
)
|
|
if results:
|
|
prop = results[0]
|
|
# propfind 커스텀 필드는 직접 파싱 필요
|
|
# 기본 구현에서는 OCS 활용
|
|
return {}
|
|
except Exception as e:
|
|
logger.warning(f"용량 조회 실패: {e}")
|
|
return {}
|
|
|
|
|
|
# ──────────────────────────────────────
|
|
# CLI
|
|
# ──────────────────────────────────────
|
|
|
|
async def _cli():
|
|
import io
|
|
if sys.stdout.encoding != "utf-8":
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
|
|
args = sys.argv[1:]
|
|
client = NCFilesClient()
|
|
|
|
if not args:
|
|
print("사용법:")
|
|
print(" nc_files.py search <검색어>")
|
|
print(" nc_files.py ls [경로]")
|
|
print(" nc_files.py recent [갯수]")
|
|
print(" nc_files.py link <경로>")
|
|
return
|
|
|
|
if args[0] == "search" and len(args) > 1:
|
|
query = " ".join(args[1:])
|
|
files = await client.search(query)
|
|
print(f"🔍 '{query}' 검색 결과: {len(files)}건\n")
|
|
for i, f in enumerate(files, 1):
|
|
print(f" {i}. {f.icon} {f.name} ({f.size_human})")
|
|
print(f" 📂 {f.path}")
|
|
|
|
elif args[0] == "ls":
|
|
path = args[1] if len(args) > 1 else ""
|
|
files = await client.list_dir(path)
|
|
print(f"📂 /{path or '(루트)'} — {len(files)}건\n")
|
|
for f in files:
|
|
size = f" ({f.size_human})" if not f.is_dir else ""
|
|
print(f" {f.icon} {f.name}{size}")
|
|
|
|
elif args[0] == "recent":
|
|
limit = int(args[1]) if len(args) > 1 else 10
|
|
files = await client.list_recent(limit)
|
|
print(f"🕐 최근 파일 {len(files)}건:\n")
|
|
for i, f in enumerate(files, 1):
|
|
print(f" {i}. {f.icon} {f.name} ({f.size_human})")
|
|
print(f" 📂 {f.path} | 📅 {f.lastmod}")
|
|
|
|
elif args[0] == "link" and len(args) > 1:
|
|
path = args[1]
|
|
url = await client.create_link(path)
|
|
if url:
|
|
print(f"🔗 {url}")
|
|
else:
|
|
print("❌ 공유 링크 생성 실패")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_cli())
|