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 스레드 라우팅
455 lines
16 KiB
Python
455 lines
16 KiB
Python
r"""Nextcloud Calendar (CalDAV) 모듈 — 일정 CRUD.
|
|
|
|
사용자 명령 기반 일정 조회/추가/수정/삭제.
|
|
CalDAV 프로토콜로 iCalendar (VEVENT) 객체를 관리.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
import config
|
|
from tools.nextcloud_client import NextcloudClient, DAV_NS, CAL_NS
|
|
|
|
logger = logging.getLogger("variet.tools.nc_calendar")
|
|
|
|
|
|
@dataclass
|
|
class CalEvent:
|
|
"""캘린더 이벤트."""
|
|
uid: str
|
|
summary: str
|
|
dtstart: datetime
|
|
dtend: Optional[datetime] = None
|
|
location: str = ""
|
|
description: str = ""
|
|
href: str = "" # CalDAV 리소스 경로
|
|
calendar: str = "" # 소속 캘린더 이름
|
|
raw_ics: str = "" # 원본 ICS 데이터
|
|
|
|
@property
|
|
def duration_str(self) -> str:
|
|
if self.dtend and self.dtstart:
|
|
delta = self.dtend - self.dtstart
|
|
hours = delta.seconds // 3600
|
|
mins = (delta.seconds % 3600) // 60
|
|
if hours > 0:
|
|
return f"{hours}시간 {mins}분" if mins else f"{hours}시간"
|
|
return f"{mins}분"
|
|
return ""
|
|
|
|
@property
|
|
def date_str(self) -> str:
|
|
return self.dtstart.strftime("%Y-%m-%d")
|
|
|
|
@property
|
|
def time_str(self) -> str:
|
|
return self.dtstart.strftime("%H:%M")
|
|
|
|
def to_display(self) -> str:
|
|
"""Discord embed용 표시 문자열."""
|
|
time = self.time_str
|
|
dur = f" ({self.duration_str})" if self.duration_str else ""
|
|
loc = f" 📍{self.location}" if self.location else ""
|
|
return f"⏰ {time}{dur} — **{self.summary}**{loc}"
|
|
|
|
|
|
class NCCalendarClient:
|
|
"""Nextcloud CalDAV 캘린더 클라이언트."""
|
|
|
|
DEFAULT_CALENDAR = "personal"
|
|
|
|
def __init__(self, nc_client: NextcloudClient = None):
|
|
self.nc = nc_client or NextcloudClient()
|
|
|
|
def _calendar_path(self, calendar: str = None) -> str:
|
|
cal = calendar or self.DEFAULT_CALENDAR
|
|
return f"calendars/{self.nc.username}/{cal}"
|
|
|
|
# ──────────────────────────────────────
|
|
# 캘린더 목록
|
|
# ──────────────────────────────────────
|
|
|
|
async def list_calendars(self) -> list[str]:
|
|
"""사용자 캘린더 목록 조회."""
|
|
path = f"calendars/{self.nc.username}/"
|
|
resp = await self.nc.dav_request(
|
|
"PROPFIND", path,
|
|
body="""<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:propfind xmlns:d="DAV:" xmlns:cal="urn:ietf:params:xml:ns:caldav">
|
|
<d:prop>
|
|
<d:displayname/>
|
|
<d:resourcetype/>
|
|
</d:prop>
|
|
</d:propfind>""",
|
|
headers={"Depth": "1"},
|
|
)
|
|
calendars = []
|
|
if resp.status_code == 207:
|
|
from xml.etree import ElementTree as ET
|
|
root = ET.fromstring(resp.text)
|
|
for r in root.findall(f"{{{DAV_NS}}}response"):
|
|
href = r.find(f"{{{DAV_NS}}}href")
|
|
propstat = r.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")
|
|
if rtype is not None and rtype.find(f"{{{CAL_NS}}}calendar") is not None:
|
|
dn = prop.find(f"{{{DAV_NS}}}displayname")
|
|
name = dn.text if dn is not None and dn.text else href.text.rstrip("/").split("/")[-1]
|
|
calendars.append(name)
|
|
return calendars
|
|
|
|
# ──────────────────────────────────────
|
|
# 이벤트 조회
|
|
# ──────────────────────────────────────
|
|
|
|
async def get_events(
|
|
self,
|
|
start: datetime = None,
|
|
end: datetime = None,
|
|
calendar: str = None,
|
|
) -> list[CalEvent]:
|
|
"""기간 내 이벤트 조회.
|
|
|
|
Args:
|
|
start: 시작일 (기본: 오늘 00:00)
|
|
end: 종료일 (기본: start + 7일)
|
|
calendar: 캘린더 이름 (기본: personal)
|
|
"""
|
|
if start is None:
|
|
start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
if end is None:
|
|
end = start + timedelta(days=7)
|
|
|
|
start_str = start.strftime("%Y%m%dT%H%M%SZ")
|
|
end_str = end.strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
body = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<cal:calendar-query xmlns:d="DAV:" xmlns:cal="urn:ietf:params:xml:ns:caldav">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<cal:calendar-data/>
|
|
</d:prop>
|
|
<cal:filter>
|
|
<cal:comp-filter name="VCALENDAR">
|
|
<cal:comp-filter name="VEVENT">
|
|
<cal:time-range start="{start_str}" end="{end_str}"/>
|
|
</cal:comp-filter>
|
|
</cal:comp-filter>
|
|
</cal:filter>
|
|
</cal:calendar-query>"""
|
|
|
|
path = self._calendar_path(calendar)
|
|
resp = await self.nc.dav_request(
|
|
"REPORT", path, body=body,
|
|
headers={"Depth": "1"},
|
|
)
|
|
|
|
events = []
|
|
if resp.status_code == 207:
|
|
from xml.etree import ElementTree as ET
|
|
root = ET.fromstring(resp.text)
|
|
for r in root.findall(f"{{{DAV_NS}}}response"):
|
|
href_el = r.find(f"{{{DAV_NS}}}href")
|
|
href = href_el.text if href_el is not None else ""
|
|
propstat = r.find(f"{{{DAV_NS}}}propstat")
|
|
if propstat is None:
|
|
continue
|
|
prop = propstat.find(f"{{{DAV_NS}}}prop")
|
|
if prop is None:
|
|
continue
|
|
cal_data = prop.find(f"{{{CAL_NS}}}calendar-data")
|
|
if cal_data is not None and cal_data.text:
|
|
event = self._parse_ics(cal_data.text, href, calendar or self.DEFAULT_CALENDAR)
|
|
if event:
|
|
events.append(event)
|
|
|
|
events.sort(key=lambda e: e.dtstart)
|
|
logger.info(f"캘린더 이벤트 조회: {len(events)}건 ({start.date()} ~ {end.date()})")
|
|
return events
|
|
|
|
async def get_today(self) -> list[CalEvent]:
|
|
"""오늘 일정 조회."""
|
|
now = datetime.now()
|
|
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end = start + timedelta(days=1)
|
|
return await self.get_events(start, end)
|
|
|
|
async def get_week(self) -> list[CalEvent]:
|
|
"""이번주 일정 조회."""
|
|
now = datetime.now()
|
|
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
# 이번주 월요일부터
|
|
start = start - timedelta(days=start.weekday())
|
|
end = start + timedelta(days=7)
|
|
return await self.get_events(start, end)
|
|
|
|
# ──────────────────────────────────────
|
|
# 이벤트 생성
|
|
# ──────────────────────────────────────
|
|
|
|
async def create_event(
|
|
self,
|
|
summary: str,
|
|
dtstart: datetime,
|
|
dtend: datetime = None,
|
|
location: str = "",
|
|
description: str = "",
|
|
calendar: str = None,
|
|
) -> CalEvent:
|
|
"""이벤트 생성.
|
|
|
|
Args:
|
|
summary: 제목
|
|
dtstart: 시작 시간
|
|
dtend: 종료 시간 (기본: 시작 + 1시간)
|
|
location: 장소
|
|
description: 설명
|
|
calendar: 캘린더 이름
|
|
"""
|
|
uid = f"{uuid.uuid4()}@variet-agent"
|
|
if dtend is None:
|
|
dtend = dtstart + timedelta(hours=1)
|
|
|
|
ics = self._build_ics(uid, summary, dtstart, dtend, location, description)
|
|
path = f"{self._calendar_path(calendar)}/{uid}.ics"
|
|
|
|
resp = await self.nc.dav_put(path, ics, "text/calendar; charset=utf-8")
|
|
if resp.status_code in (200, 201, 204):
|
|
logger.info(f"일정 생성: {summary} ({dtstart})")
|
|
return CalEvent(
|
|
uid=uid, summary=summary, dtstart=dtstart, dtend=dtend,
|
|
location=location, description=description, href=path,
|
|
calendar=calendar or self.DEFAULT_CALENDAR, raw_ics=ics,
|
|
)
|
|
else:
|
|
raise RuntimeError(f"일정 생성 실패: {resp.status_code} {resp.text[:200]}")
|
|
|
|
# ──────────────────────────────────────
|
|
# 이벤트 수정
|
|
# ──────────────────────────────────────
|
|
|
|
async def update_event(
|
|
self,
|
|
uid: str,
|
|
calendar: str = None,
|
|
summary: str = None,
|
|
dtstart: datetime = None,
|
|
dtend: datetime = None,
|
|
location: str = None,
|
|
description: str = None,
|
|
) -> CalEvent:
|
|
"""이벤트 수정 — 기존 이벤트를 찾아서 변경."""
|
|
# 먼저 기존 이벤트를 찾음
|
|
events = await self.get_events(
|
|
start=datetime(2020, 1, 1),
|
|
end=datetime(2030, 12, 31),
|
|
calendar=calendar,
|
|
)
|
|
target = None
|
|
for e in events:
|
|
if e.uid == uid:
|
|
target = e
|
|
break
|
|
|
|
if target is None:
|
|
raise RuntimeError(f"이벤트 {uid}를 찾을 수 없습니다.")
|
|
|
|
# 변경 적용
|
|
new_summary = summary if summary is not None else target.summary
|
|
new_dtstart = dtstart if dtstart is not None else target.dtstart
|
|
new_dtend = dtend if dtend is not None else target.dtend
|
|
new_location = location if location is not None else target.location
|
|
new_description = description if description is not None else target.description
|
|
|
|
ics = self._build_ics(uid, new_summary, new_dtstart, new_dtend, new_location, new_description)
|
|
path = target.href or f"{self._calendar_path(calendar)}/{uid}.ics"
|
|
|
|
resp = await self.nc.dav_put(path, ics, "text/calendar; charset=utf-8")
|
|
if resp.status_code in (200, 201, 204):
|
|
logger.info(f"일정 수정: {new_summary}")
|
|
return CalEvent(
|
|
uid=uid, summary=new_summary, dtstart=new_dtstart, dtend=new_dtend,
|
|
location=new_location, description=new_description, href=path,
|
|
calendar=calendar or self.DEFAULT_CALENDAR,
|
|
)
|
|
else:
|
|
raise RuntimeError(f"일정 수정 실패: {resp.status_code}")
|
|
|
|
# ──────────────────────────────────────
|
|
# 이벤트 삭제
|
|
# ──────────────────────────────────────
|
|
|
|
async def delete_event(self, uid: str, calendar: str = None) -> bool:
|
|
"""이벤트 삭제."""
|
|
# href 직접 삭제 시도
|
|
path = f"{self._calendar_path(calendar)}/{uid}.ics"
|
|
ok = await self.nc.dav_delete(path)
|
|
if ok:
|
|
logger.info(f"일정 삭제: {uid}")
|
|
else:
|
|
logger.warning(f"일정 삭제 실패: {uid}")
|
|
return ok
|
|
|
|
# ──────────────────────────────────────
|
|
# ICS 빌드/파싱
|
|
# ──────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _build_ics(
|
|
uid: str, summary: str,
|
|
dtstart: datetime, dtend: datetime,
|
|
location: str = "", description: str = "",
|
|
) -> str:
|
|
"""VEVENT ICS 문자열 생성."""
|
|
now = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
|
|
ds = dtstart.strftime("%Y%m%dT%H%M%S")
|
|
de = dtend.strftime("%Y%m%dT%H%M%S")
|
|
|
|
lines = [
|
|
"BEGIN:VCALENDAR",
|
|
"VERSION:2.0",
|
|
"PRODID:-//Variet-Agent//NC Calendar//KO",
|
|
"BEGIN:VEVENT",
|
|
f"UID:{uid}",
|
|
f"DTSTAMP:{now}",
|
|
f"DTSTART:{ds}",
|
|
f"DTEND:{de}",
|
|
f"SUMMARY:{summary}",
|
|
]
|
|
if location:
|
|
lines.append(f"LOCATION:{location}")
|
|
if description:
|
|
lines.append(f"DESCRIPTION:{description}")
|
|
lines.extend([
|
|
"END:VEVENT",
|
|
"END:VCALENDAR",
|
|
])
|
|
return "\r\n".join(lines)
|
|
|
|
@staticmethod
|
|
def _parse_ics(ics_text: str, href: str = "", calendar: str = "") -> Optional[CalEvent]:
|
|
"""ICS 텍스트에서 CalEvent 추출 (간단 파서)."""
|
|
def _get_field(text: str, field: str) -> str:
|
|
pattern = rf"^{field}[;:](.+)$"
|
|
m = re.search(pattern, text, re.MULTILINE)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
def _parse_dt(val: str) -> Optional[datetime]:
|
|
if not val:
|
|
return None
|
|
# VALUE=DATE:20260318 형식 처리
|
|
if "VALUE=DATE:" in val:
|
|
val = val.split(":")[-1]
|
|
elif ":" in val:
|
|
val = val.split(":")[-1]
|
|
val = val.replace("Z", "")
|
|
for fmt in ("%Y%m%dT%H%M%S", "%Y%m%d"):
|
|
try:
|
|
return datetime.strptime(val, fmt)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
uid = _get_field(ics_text, "UID")
|
|
summary = _get_field(ics_text, "SUMMARY")
|
|
dtstart = _parse_dt(_get_field(ics_text, "DTSTART"))
|
|
dtend = _parse_dt(_get_field(ics_text, "DTEND"))
|
|
location = _get_field(ics_text, "LOCATION")
|
|
description = _get_field(ics_text, "DESCRIPTION")
|
|
|
|
if not uid or not dtstart:
|
|
return None
|
|
|
|
return CalEvent(
|
|
uid=uid, summary=summary, dtstart=dtstart, dtend=dtend,
|
|
location=location, description=description,
|
|
href=href, calendar=calendar, raw_ics=ics_text,
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────
|
|
# 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 = NCCalendarClient()
|
|
|
|
if not args:
|
|
print("사용법:")
|
|
print(" nc_calendar.py calendars")
|
|
print(" nc_calendar.py today")
|
|
print(" nc_calendar.py week")
|
|
print(' nc_calendar.py add "제목" 2026-03-20 14:00 [1h]')
|
|
print(' nc_calendar.py delete <uid>')
|
|
return
|
|
|
|
if args[0] == "calendars":
|
|
cals = await client.list_calendars()
|
|
print(f"📅 캘린더 {len(cals)}개:")
|
|
for c in cals:
|
|
print(f" • {c}")
|
|
|
|
elif args[0] == "today":
|
|
events = await client.get_today()
|
|
print(f"📅 오늘 일정 ({datetime.now().strftime('%Y-%m-%d')}):\n")
|
|
if not events:
|
|
print(" 일정 없음")
|
|
for e in events:
|
|
print(f" {e.to_display()}")
|
|
|
|
elif args[0] == "week":
|
|
events = await client.get_week()
|
|
print(f"📅 이번주 일정:\n")
|
|
current_date = ""
|
|
for e in events:
|
|
if e.date_str != current_date:
|
|
current_date = e.date_str
|
|
print(f"\n 📆 {current_date}")
|
|
print(f" {e.to_display()}")
|
|
|
|
elif args[0] == "add" and len(args) >= 3:
|
|
summary = args[1]
|
|
date_str = args[2]
|
|
time_str = args[3] if len(args) > 3 else "09:00"
|
|
duration = args[4] if len(args) > 4 else "1h"
|
|
|
|
dtstart = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M")
|
|
# duration 파싱
|
|
dur_match = re.match(r"(\d+)([hm])", duration)
|
|
if dur_match:
|
|
val, unit = int(dur_match.group(1)), dur_match.group(2)
|
|
delta = timedelta(hours=val) if unit == "h" else timedelta(minutes=val)
|
|
else:
|
|
delta = timedelta(hours=1)
|
|
dtend = dtstart + delta
|
|
|
|
event = await client.create_event(summary, dtstart, dtend)
|
|
print(f"✅ 일정 생성: {event.to_display()}")
|
|
print(f" UID: {event.uid}")
|
|
|
|
elif args[0] == "delete" and len(args) > 1:
|
|
uid = args[1]
|
|
ok = await client.delete_event(uid)
|
|
print(f"{'✅ 삭제 완료' if ok else '❌ 삭제 실패'}: {uid}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_cli())
|