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 스레드 라우팅
431 lines
15 KiB
Python
431 lines
15 KiB
Python
r"""Nextcloud 공통 클라이언트 — WebDAV / OCS / CalDAV / CardDAV.
|
|
|
|
모든 Nextcloud 연동 모듈(nc_files, nc_calendar, nc_mail, nc_contacts)의 기반.
|
|
|
|
인증: App Password + Basic Auth
|
|
API:
|
|
- WebDAV: remote.php/dav/files/{user}/
|
|
- OCS: ocs/v2.php/apps/...
|
|
- CalDAV: remote.php/dav/calendars/{user}/
|
|
- CardDAV: remote.php/dav/addressbooks/users/{user}/
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
import os
|
|
from typing import Optional
|
|
from xml.etree import ElementTree as ET
|
|
|
|
import httpx
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
import config
|
|
|
|
logger = logging.getLogger("variet.tools.nextcloud")
|
|
|
|
# XML 네임스페이스
|
|
DAV_NS = "DAV:"
|
|
OC_NS = "http://owncloud.org/ns"
|
|
NC_NS = "http://nextcloud.org/ns"
|
|
CAL_NS = "urn:ietf:params:xml:ns:caldav"
|
|
CARD_NS = "urn:ietf:params:xml:ns:carddav"
|
|
|
|
NS_MAP = {
|
|
"d": DAV_NS,
|
|
"oc": OC_NS,
|
|
"nc": NC_NS,
|
|
"cal": CAL_NS,
|
|
"card": CARD_NS,
|
|
}
|
|
|
|
|
|
class NextcloudClient:
|
|
"""Nextcloud 공통 HTTP 클라이언트.
|
|
|
|
WebDAV, OCS, CalDAV, CardDAV 공통 기능 제공.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str = None,
|
|
username: str = None,
|
|
app_password: str = None,
|
|
timeout: float = 30.0,
|
|
):
|
|
self.base_url = (base_url or config.NEXTCLOUD_URL).rstrip("/")
|
|
self.username = username or config.NEXTCLOUD_USER
|
|
self.app_password = app_password or config.NEXTCLOUD_APP_PASSWORD
|
|
self.timeout = timeout
|
|
self._auth = (self.username, self.app_password)
|
|
|
|
def _client(self) -> httpx.AsyncClient:
|
|
"""새 httpx 클라이언트 생성."""
|
|
return httpx.AsyncClient(
|
|
base_url=self.base_url,
|
|
auth=self._auth,
|
|
timeout=self.timeout,
|
|
headers={
|
|
"OCS-APIRequest": "true",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
|
|
# ──────────────────────────────────────
|
|
# OCS API (JSON)
|
|
# ──────────────────────────────────────
|
|
|
|
async def ocs_get(self, endpoint: str, params: dict = None) -> dict:
|
|
"""OCS GET 요청 (JSON 응답)."""
|
|
url = f"/ocs/v2.php/{endpoint}"
|
|
if "format=json" not in url:
|
|
params = params or {}
|
|
params["format"] = "json"
|
|
async with self._client() as client:
|
|
resp = await client.get(url, params=params)
|
|
resp.raise_for_status()
|
|
return resp.json().get("ocs", {}).get("data", {})
|
|
|
|
async def ocs_post(self, endpoint: str, data: dict = None) -> dict:
|
|
"""OCS POST 요청."""
|
|
url = f"/ocs/v2.php/{endpoint}"
|
|
async with self._client() as client:
|
|
resp = await client.post(
|
|
url,
|
|
data={**(data or {}), "format": "json"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json().get("ocs", {}).get("data", {})
|
|
|
|
async def ocs_delete(self, endpoint: str) -> bool:
|
|
"""OCS DELETE 요청."""
|
|
url = f"/ocs/v2.php/{endpoint}"
|
|
async with self._client() as client:
|
|
resp = await client.delete(url, params={"format": "json"})
|
|
return resp.status_code in (200, 204)
|
|
|
|
# ──────────────────────────────────────
|
|
# WebDAV (XML)
|
|
# ──────────────────────────────────────
|
|
|
|
async def webdav_request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
body: str = None,
|
|
headers: dict = None,
|
|
) -> httpx.Response:
|
|
"""WebDAV 요청 (PROPFIND, REPORT, SEARCH, PUT, DELETE 등)."""
|
|
url = f"/remote.php/dav/{path}"
|
|
req_headers = {"Content-Type": "application/xml; charset=utf-8"}
|
|
if headers:
|
|
req_headers.update(headers)
|
|
|
|
async with self._client() as client:
|
|
resp = await client.request(
|
|
method, url,
|
|
content=body.encode("utf-8") if body else None,
|
|
headers=req_headers,
|
|
)
|
|
return resp
|
|
|
|
async def webdav_propfind(
|
|
self, path: str, props: list[str] = None, depth: int = 1,
|
|
) -> list[dict]:
|
|
"""PROPFIND — 파일/디렉토리 속성 조회.
|
|
|
|
Returns:
|
|
[{href, name, size, lastmod, content_type, is_dir}, ...]
|
|
"""
|
|
if props is None:
|
|
props = [
|
|
"d:getlastmodified",
|
|
"d:getcontentlength",
|
|
"d:getcontenttype",
|
|
"d:resourcetype",
|
|
"d:displayname",
|
|
"oc:fileid",
|
|
]
|
|
props_xml = "\n".join(f"<{p}/>" for p in props)
|
|
body = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns" xmlns:nc="http://nextcloud.org/ns">
|
|
<d:prop>
|
|
{props_xml}
|
|
</d:prop>
|
|
</d:propfind>"""
|
|
|
|
resp = await self.webdav_request(
|
|
"PROPFIND", path, body=body,
|
|
headers={"Depth": str(depth)},
|
|
)
|
|
resp.raise_for_status()
|
|
return self._parse_propfind(resp.text)
|
|
|
|
async def webdav_search(
|
|
self, query: str, path: str = None,
|
|
) -> list[dict]:
|
|
"""WebDAV SEARCH — 파일 검색.
|
|
|
|
Args:
|
|
query: 검색어 (파일명 매칭)
|
|
path: 검색 시작 경로 (기본: 사용자 루트)
|
|
"""
|
|
search_path = path or f"files/{self.username}"
|
|
body = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:searchrequest xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
|
|
<d:basicsearch>
|
|
<d:select>
|
|
<d:prop>
|
|
<d:displayname/>
|
|
<d:getcontentlength/>
|
|
<d:getlastmodified/>
|
|
<d:getcontenttype/>
|
|
<d:resourcetype/>
|
|
<oc:fileid/>
|
|
</d:prop>
|
|
</d:select>
|
|
<d:from>
|
|
<d:scope>
|
|
<d:href>/remote.php/dav/{search_path}</d:href>
|
|
<d:depth>infinity</d:depth>
|
|
</d:scope>
|
|
</d:from>
|
|
<d:where>
|
|
<d:like>
|
|
<d:prop><d:displayname/></d:prop>
|
|
<d:literal>%{query}%</d:literal>
|
|
</d:like>
|
|
</d:where>
|
|
<d:limit>
|
|
<d:nresults>50</d:nresults>
|
|
</d:limit>
|
|
</d:basicsearch>
|
|
</d:searchrequest>"""
|
|
|
|
resp = await self.webdav_request("SEARCH", search_path, body=body)
|
|
if resp.status_code == 207:
|
|
return self._parse_propfind(resp.text)
|
|
logger.warning(f"WebDAV SEARCH 실패: {resp.status_code}")
|
|
return []
|
|
|
|
# ──────────────────────────────────────
|
|
# 파일 조작
|
|
# ──────────────────────────────────────
|
|
|
|
async def upload_file(
|
|
self, remote_path: str, data: bytes, content_type: str = "application/octet-stream",
|
|
) -> bool:
|
|
"""WebDAV PUT — 파일 업로드."""
|
|
url = f"files/{self.username}/{remote_path.lstrip('/')}"
|
|
async with self._client() as client:
|
|
resp = await client.put(
|
|
f"/remote.php/dav/{url}",
|
|
content=data,
|
|
headers={"Content-Type": content_type},
|
|
)
|
|
ok = resp.status_code in (200, 201, 204)
|
|
if ok:
|
|
logger.info(f"파일 업로드: {remote_path}")
|
|
else:
|
|
logger.error(f"업로드 실패: {resp.status_code} {resp.text[:200]}")
|
|
return ok
|
|
|
|
async def download_file(self, remote_path: str) -> Optional[bytes]:
|
|
"""WebDAV GET — 파일 다운로드."""
|
|
url = f"files/{self.username}/{remote_path.lstrip('/')}"
|
|
async with self._client() as client:
|
|
resp = await client.get(f"/remote.php/dav/{url}")
|
|
if resp.status_code == 200:
|
|
return resp.content
|
|
logger.warning(f"다운로드 실패: {resp.status_code}")
|
|
return None
|
|
|
|
# ──────────────────────────────────────
|
|
# 공유 링크 (OCS Share API)
|
|
# ──────────────────────────────────────
|
|
|
|
async def create_share_link(
|
|
self, path: str, expire_days: int = 7,
|
|
) -> Optional[str]:
|
|
"""OCS 공유 링크 생성.
|
|
|
|
Returns:
|
|
공유 URL 또는 None
|
|
"""
|
|
data = {
|
|
"shareType": "3", # 3 = 공개 링크
|
|
"path": path,
|
|
}
|
|
if expire_days:
|
|
from datetime import datetime, timedelta
|
|
expire = (datetime.now() + timedelta(days=expire_days)).strftime("%Y-%m-%d")
|
|
data["expireDate"] = expire
|
|
|
|
result = await self.ocs_post("apps/files_sharing/api/v1/shares", data)
|
|
url = result.get("url")
|
|
if url:
|
|
logger.info(f"공유 링크 생성: {path} → {url}")
|
|
return url
|
|
|
|
async def delete_share(self, share_id: int) -> bool:
|
|
"""OCS 공유 링크 삭제."""
|
|
return await self.ocs_delete(f"apps/files_sharing/api/v1/shares/{share_id}")
|
|
|
|
# ──────────────────────────────────────
|
|
# DAV (CalDAV / CardDAV 공통)
|
|
# ──────────────────────────────────────
|
|
|
|
async def dav_request(
|
|
self, method: str, path: str, body: str = None, headers: dict = None,
|
|
) -> httpx.Response:
|
|
"""CalDAV/CardDAV 요청 래퍼."""
|
|
return await self.webdav_request(method, path, body=body, headers=headers)
|
|
|
|
async def dav_put(
|
|
self, path: str, data: str, content_type: str,
|
|
) -> httpx.Response:
|
|
"""CalDAV/CardDAV PUT — 리소스 생성/수정."""
|
|
async with self._client() as client:
|
|
resp = await client.put(
|
|
f"/remote.php/dav/{path}",
|
|
content=data.encode("utf-8"),
|
|
headers={"Content-Type": content_type},
|
|
)
|
|
return resp
|
|
|
|
async def dav_delete(self, path: str) -> bool:
|
|
"""CalDAV/CardDAV DELETE."""
|
|
async with self._client() as client:
|
|
resp = await client.delete(f"/remote.php/dav/{path}")
|
|
return resp.status_code in (200, 204)
|
|
|
|
# ──────────────────────────────────────
|
|
# XML 파싱 유틸
|
|
# ──────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _parse_propfind(xml_text: str) -> list[dict]:
|
|
"""PROPFIND 응답 XML → dict 리스트."""
|
|
results = []
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
except ET.ParseError:
|
|
logger.error("PROPFIND XML 파싱 실패")
|
|
return results
|
|
|
|
for response in root.findall(f"{{{DAV_NS}}}response"):
|
|
href_el = response.find(f"{{{DAV_NS}}}href")
|
|
if href_el is None:
|
|
continue
|
|
href = href_el.text or ""
|
|
|
|
propstat = response.find(f"{{{DAV_NS}}}propstat")
|
|
if propstat is None:
|
|
continue
|
|
prop = propstat.find(f"{{{DAV_NS}}}prop")
|
|
if prop is None:
|
|
continue
|
|
|
|
# 디렉토리 여부
|
|
rtype = prop.find(f"{{{DAV_NS}}}resourcetype")
|
|
is_dir = rtype is not None and rtype.find(f"{{{DAV_NS}}}collection") is not None
|
|
|
|
# 이름 추출
|
|
displayname = prop.find(f"{{{DAV_NS}}}displayname")
|
|
name = displayname.text if displayname is not None and displayname.text else href.rstrip("/").split("/")[-1]
|
|
|
|
# 크기
|
|
size_el = prop.find(f"{{{DAV_NS}}}getcontentlength")
|
|
size = int(size_el.text) if size_el is not None and size_el.text else 0
|
|
|
|
# 수정시간
|
|
lastmod_el = prop.find(f"{{{DAV_NS}}}getlastmodified")
|
|
lastmod = lastmod_el.text if lastmod_el is not None else ""
|
|
|
|
# 콘텐츠 타입
|
|
ctype_el = prop.find(f"{{{DAV_NS}}}getcontenttype")
|
|
content_type = ctype_el.text if ctype_el is not None else ""
|
|
|
|
results.append({
|
|
"href": href,
|
|
"name": name,
|
|
"size": size,
|
|
"lastmod": lastmod,
|
|
"content_type": content_type,
|
|
"is_dir": is_dir,
|
|
})
|
|
|
|
return results
|
|
|
|
# ──────────────────────────────────────
|
|
# 헬스체크
|
|
# ──────────────────────────────────────
|
|
|
|
async def health_check(self) -> bool:
|
|
"""Nextcloud 연결 확인."""
|
|
try:
|
|
path = f"files/{self.username}"
|
|
results = await self.webdav_propfind(path, depth=0)
|
|
ok = len(results) > 0
|
|
if ok:
|
|
logger.info("Nextcloud 연결 확인 OK")
|
|
return ok
|
|
except Exception as e:
|
|
logger.error(f"Nextcloud 연결 실패: {e}")
|
|
return False
|
|
|
|
|
|
# ──────────────────────────────────────
|
|
# 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 = NextcloudClient()
|
|
|
|
if not args or args[0] == "check":
|
|
ok = await client.health_check()
|
|
print(f"{'✅ 연결 성공' if ok else '❌ 연결 실패'}")
|
|
|
|
elif args[0] == "ls" and len(args) > 1:
|
|
path = f"files/{client.username}/{args[1].lstrip('/')}"
|
|
items = await client.webdav_propfind(path)
|
|
print(f"📂 {args[1]} — {len(items)}건:")
|
|
for item in items[1:]: # 첫 번째는 자기 자신
|
|
icon = "📁" if item["is_dir"] else "📄"
|
|
size = f" ({item['size']:,}B)" if not item["is_dir"] else ""
|
|
print(f" {icon} {item['name']}{size}")
|
|
|
|
elif args[0] == "search" and len(args) > 1:
|
|
query = " ".join(args[1:])
|
|
results = await client.webdav_search(query)
|
|
print(f"🔍 '{query}' 검색 결과: {len(results)}건")
|
|
for item in results:
|
|
icon = "📁" if item["is_dir"] else "📄"
|
|
size = f" ({item['size']:,}B)" if not item["is_dir"] else ""
|
|
print(f" {icon} {item['name']}{size}")
|
|
print(f" {item['href']}")
|
|
|
|
elif args[0] == "share" and len(args) > 1:
|
|
path = args[1]
|
|
url = await client.create_share_link(path)
|
|
if url:
|
|
print(f"🔗 {url}")
|
|
else:
|
|
print("❌ 공유 링크 생성 실패")
|
|
|
|
else:
|
|
print("사용법:")
|
|
print(" nextcloud_client.py check")
|
|
print(" nextcloud_client.py ls <path>")
|
|
print(" nextcloud_client.py search <query>")
|
|
print(" nextcloud_client.py share <path>")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_cli())
|