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 스레드 라우팅
437 lines
16 KiB
Python
437 lines
16 KiB
Python
r"""Nextcloud Contacts + 명함 스캔 모듈.
|
|
|
|
EasyOCR로 명함 이미지 텍스트 추출 → 정규식 필드 분류 → vCard 생성 → CardDAV 저장.
|
|
연락처 검색/조회 기능 포함.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
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, CARD_NS
|
|
|
|
logger = logging.getLogger("variet.tools.nc_contacts")
|
|
|
|
|
|
@dataclass
|
|
class Contact:
|
|
"""연락처 정보."""
|
|
uid: str = ""
|
|
name: str = ""
|
|
company: str = ""
|
|
title: str = "" # 직함
|
|
phone: list[str] = field(default_factory=list)
|
|
email: list[str] = field(default_factory=list)
|
|
address: str = ""
|
|
note: str = ""
|
|
href: str = "" # CardDAV 리소스 경로
|
|
|
|
def to_vcard(self) -> str:
|
|
"""vCard 3.0 포맷 생성."""
|
|
uid = self.uid or str(uuid.uuid4())
|
|
lines = [
|
|
"BEGIN:VCARD",
|
|
"VERSION:3.0",
|
|
f"UID:{uid}",
|
|
f"FN:{self.name}",
|
|
f"N:{self.name};;;;",
|
|
]
|
|
if self.company:
|
|
lines.append(f"ORG:{self.company}")
|
|
if self.title:
|
|
lines.append(f"TITLE:{self.title}")
|
|
for p in self.phone:
|
|
lines.append(f"TEL;TYPE=CELL:{p}")
|
|
for e in self.email:
|
|
lines.append(f"EMAIL:{e}")
|
|
if self.address:
|
|
lines.append(f"ADR;TYPE=WORK:;;{self.address};;;;")
|
|
if self.note:
|
|
lines.append(f"NOTE:{self.note}")
|
|
lines.append("END:VCARD")
|
|
return "\r\n".join(lines)
|
|
|
|
def to_display(self) -> str:
|
|
"""Discord embed용 표시."""
|
|
parts = [f"👤 **{self.name}**"]
|
|
if self.title and self.company:
|
|
parts.append(f" 🏢 {self.company} / {self.title}")
|
|
elif self.company:
|
|
parts.append(f" 🏢 {self.company}")
|
|
elif self.title:
|
|
parts.append(f" 💼 {self.title}")
|
|
for p in self.phone:
|
|
parts.append(f" 📞 {p}")
|
|
for e in self.email:
|
|
parts.append(f" 📧 {e}")
|
|
if self.address:
|
|
parts.append(f" 📍 {self.address}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
class NCContactsClient:
|
|
"""Nextcloud CardDAV 연락처 + 명함 스캔 클라이언트."""
|
|
|
|
DEFAULT_ADDRESSBOOK = "contacts"
|
|
|
|
def __init__(self, nc_client: NextcloudClient = None):
|
|
self.nc = nc_client or NextcloudClient()
|
|
self._ocr = None
|
|
|
|
def _addressbook_path(self, addressbook: str = None) -> str:
|
|
ab = addressbook or self.DEFAULT_ADDRESSBOOK
|
|
return f"addressbooks/users/{self.nc.username}/{ab}"
|
|
|
|
# ──────────────────────────────────────
|
|
# OCR (EasyOCR)
|
|
# ──────────────────────────────────────
|
|
|
|
def _get_ocr(self):
|
|
"""EasyOCR Reader 지연 로딩."""
|
|
if self._ocr is None:
|
|
try:
|
|
import easyocr
|
|
self._ocr = easyocr.Reader(["ko", "en", "ja"], gpu=False)
|
|
logger.info("EasyOCR 초기화 완료 (ko, en, ja)")
|
|
except ImportError:
|
|
raise RuntimeError(
|
|
"easyocr 미설치. pip install easyocr 실행 필요."
|
|
)
|
|
return self._ocr
|
|
|
|
async def scan_card(self, image_path: str) -> Contact:
|
|
"""명함 이미지 → 연락처 정보 추출.
|
|
|
|
Args:
|
|
image_path: 이미지 파일 경로 또는 URL
|
|
|
|
Returns:
|
|
추출된 Contact 객체
|
|
"""
|
|
def _ocr_extract():
|
|
reader = self._get_ocr()
|
|
results = reader.readtext(image_path)
|
|
# (bbox, text, confidence) 리스트
|
|
texts = [text for _, text, conf in results if conf > 0.3]
|
|
return texts
|
|
|
|
texts = await asyncio.to_thread(_ocr_extract)
|
|
logger.info(f"명함 OCR: {len(texts)}줄 추출")
|
|
|
|
return self._classify_fields(texts)
|
|
|
|
@staticmethod
|
|
def _classify_fields(texts: list[str]) -> Contact:
|
|
"""OCR 텍스트 → 구조화된 Contact로 분류.
|
|
|
|
정규식으로 이메일, 전화번호, URL 등을 먼저 분류하고
|
|
나머지 텍스트에서 이름/회사/직함을 추론.
|
|
"""
|
|
contact = Contact(uid=str(uuid.uuid4()))
|
|
remaining = []
|
|
|
|
for text in texts:
|
|
text = text.strip()
|
|
if not text:
|
|
continue
|
|
|
|
# 이메일
|
|
email_match = re.search(r"[\w.+-]+@[\w.-]+\.\w+", text)
|
|
if email_match:
|
|
contact.email.append(email_match.group())
|
|
continue
|
|
|
|
# 전화번호 (한국/국제)
|
|
phone_match = re.search(
|
|
r"(?:\+?\d{1,3}[-.\s]?)?"
|
|
r"(?:\(?\d{2,4}\)?[-.\s]?)?"
|
|
r"\d{3,4}[-.\s]?\d{3,4}",
|
|
text,
|
|
)
|
|
if phone_match and len(phone_match.group().replace("-", "").replace(" ", "").replace(".", "")) >= 8:
|
|
contact.phone.append(phone_match.group())
|
|
continue
|
|
|
|
# URL (웹사이트) → note에 저장
|
|
if re.match(r"(https?://|www\.)", text, re.IGNORECASE):
|
|
contact.note += f" {text}"
|
|
continue
|
|
|
|
# 주소 패턴 (번지, 로, 길, 동, 층 등)
|
|
if re.search(r"(시|구|동|로|길|층|호|번지|타워|빌딩|센터)", text):
|
|
contact.address = text
|
|
continue
|
|
|
|
remaining.append(text)
|
|
|
|
# 나머지 텍스트 분류 (이름, 회사, 직함)
|
|
if remaining:
|
|
# 보통 명함에서 가장 큰 텍스트(첫 번째)가 이름
|
|
contact.name = remaining[0]
|
|
|
|
for text in remaining[1:]:
|
|
# 직함 키워드
|
|
if re.search(r"(대표|이사|부장|과장|차장|팀장|사원|매니저|Director|Manager|CEO|CTO|VP|Engineer|Developer)", text, re.IGNORECASE):
|
|
contact.title = text
|
|
elif not contact.company:
|
|
# 회사명 (나머지 중 첫 번째)
|
|
contact.company = text
|
|
|
|
contact.note = contact.note.strip()
|
|
return contact
|
|
|
|
# ──────────────────────────────────────
|
|
# 연락처 저장 (CardDAV)
|
|
# ──────────────────────────────────────
|
|
|
|
async def create_contact(self, contact: Contact, addressbook: str = None) -> Contact:
|
|
"""연락처를 Nextcloud에 저장.
|
|
|
|
Args:
|
|
contact: Contact 객체
|
|
addressbook: 주소록 이름
|
|
|
|
Returns:
|
|
저장된 Contact (href 포함)
|
|
"""
|
|
if not contact.uid:
|
|
contact.uid = str(uuid.uuid4())
|
|
|
|
vcard = contact.to_vcard()
|
|
path = f"{self._addressbook_path(addressbook)}/{contact.uid}.vcf"
|
|
|
|
resp = await self.nc.dav_put(path, vcard, "text/vcard; charset=utf-8")
|
|
if resp.status_code in (200, 201, 204):
|
|
contact.href = path
|
|
logger.info(f"연락처 저장: {contact.name}")
|
|
return contact
|
|
else:
|
|
raise RuntimeError(f"연락처 저장 실패: {resp.status_code} {resp.text[:200]}")
|
|
|
|
# ──────────────────────────────────────
|
|
# 연락처 검색/조회
|
|
# ──────────────────────────────────────
|
|
|
|
async def search_contacts(self, query: str, addressbook: str = None) -> list[Contact]:
|
|
"""연락처 검색.
|
|
|
|
Args:
|
|
query: 검색어 (이름, 이메일, 전화번호)
|
|
"""
|
|
body = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<card:address-data/>
|
|
</d:prop>
|
|
<card:filter>
|
|
<card:prop-filter name="FN">
|
|
<card:text-match collation="i;unicode-casemap" match-type="contains">{query}</card:text-match>
|
|
</card:prop-filter>
|
|
</card:filter>
|
|
</card:addressbook-query>"""
|
|
|
|
path = self._addressbook_path(addressbook)
|
|
resp = await self.nc.dav_request(
|
|
"REPORT", path, body=body,
|
|
headers={"Depth": "1"},
|
|
)
|
|
|
|
contacts = []
|
|
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
|
|
addr_data = prop.find(f"{{{CARD_NS}}}address-data")
|
|
if addr_data is not None and addr_data.text:
|
|
contact = self._parse_vcard(addr_data.text, href)
|
|
if contact:
|
|
contacts.append(contact)
|
|
|
|
logger.info(f"연락처 검색 '{query}': {len(contacts)}건")
|
|
return contacts
|
|
|
|
async def list_contacts(self, addressbook: str = None) -> list[Contact]:
|
|
"""전체 연락처 목록."""
|
|
path = self._addressbook_path(addressbook)
|
|
body = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<card:address-data/>
|
|
</d:prop>
|
|
</card:addressbook-query>"""
|
|
|
|
resp = await self.nc.dav_request(
|
|
"REPORT", path, body=body,
|
|
headers={"Depth": "1"},
|
|
)
|
|
|
|
contacts = []
|
|
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
|
|
addr_data = prop.find(f"{{{CARD_NS}}}address-data")
|
|
if addr_data is not None and addr_data.text:
|
|
contact = self._parse_vcard(addr_data.text, href)
|
|
if contact:
|
|
contacts.append(contact)
|
|
|
|
contacts.sort(key=lambda c: c.name)
|
|
return contacts
|
|
|
|
# ──────────────────────────────────────
|
|
# vCard 파싱
|
|
# ──────────────────────────────────────
|
|
|
|
@staticmethod
|
|
def _parse_vcard(vcard_text: str, href: str = "") -> Optional[Contact]:
|
|
"""vCard 텍스트 → Contact 객체."""
|
|
def _get(text: str, field: str) -> str:
|
|
pattern = rf"^{field}[;:](.+)$"
|
|
m = re.search(pattern, text, re.MULTILINE)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
def _get_all(text: str, field: str) -> list[str]:
|
|
pattern = rf"^{field}[;:](.+)$"
|
|
return [m.group(1).strip() for m in re.finditer(pattern, text, re.MULTILINE)]
|
|
|
|
uid = _get(vcard_text, "UID")
|
|
fn = _get(vcard_text, "FN")
|
|
org = _get(vcard_text, "ORG")
|
|
title = _get(vcard_text, "TITLE")
|
|
note = _get(vcard_text, "NOTE")
|
|
|
|
phones = _get_all(vcard_text, "TEL")
|
|
# TEL;TYPE=CELL:010-... → 번호만 추출
|
|
phones = [re.sub(r"^.*:", "", p) if ":" in p else p for p in phones]
|
|
|
|
emails = _get_all(vcard_text, "EMAIL")
|
|
emails = [re.sub(r"^.*:", "", e) if ":" in e else e for e in emails]
|
|
|
|
adr = _get(vcard_text, "ADR")
|
|
# ADR;TYPE=WORK:;;... → 주소만 추출
|
|
if ";" in adr:
|
|
parts = adr.split(";")
|
|
adr = " ".join(p for p in parts if p).strip()
|
|
|
|
if not fn and not uid:
|
|
return None
|
|
|
|
return Contact(
|
|
uid=uid, name=fn, company=org, title=title,
|
|
phone=phones, email=emails, address=adr, note=note,
|
|
href=href,
|
|
)
|
|
|
|
# ──────────────────────────────────────
|
|
# 명함 스캔 + 저장 통합
|
|
# ──────────────────────────────────────
|
|
|
|
async def scan_and_save(
|
|
self,
|
|
image_path: str,
|
|
save_image: bool = True,
|
|
addressbook: str = None,
|
|
) -> Contact:
|
|
"""명함 스캔 → 연락처 저장 → 원본 이미지도 NC에 보관.
|
|
|
|
Args:
|
|
image_path: 이미지 파일 경로
|
|
save_image: 원본 이미지를 NC Files에도 저장할지
|
|
addressbook: 주소록 이름
|
|
"""
|
|
# 1. OCR
|
|
contact = await self.scan_card(image_path)
|
|
|
|
# 2. CardDAV 저장
|
|
contact = await self.create_contact(contact, addressbook)
|
|
|
|
# 3. 원본 이미지도 NC Files에 저장 (선택)
|
|
if save_image and os.path.exists(image_path):
|
|
try:
|
|
with open(image_path, "rb") as f:
|
|
img_data = f.read()
|
|
remote_name = f"BusinessCards/{contact.uid}.jpg"
|
|
await self.nc.upload_file(remote_name, img_data, "image/jpeg")
|
|
logger.info(f"명함 이미지 저장: {remote_name}")
|
|
except Exception as e:
|
|
logger.warning(f"명함 이미지 저장 실패: {e}")
|
|
|
|
return contact
|
|
|
|
|
|
# ──────────────────────────────────────
|
|
# 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 = NCContactsClient()
|
|
|
|
if not args:
|
|
print("사용법:")
|
|
print(' nc_contacts.py scan <이미지경로>')
|
|
print(' nc_contacts.py search "이름"')
|
|
print(' nc_contacts.py list')
|
|
return
|
|
|
|
if args[0] == "scan" and len(args) > 1:
|
|
image_path = args[1]
|
|
contact = await client.scan_card(image_path)
|
|
print(f"📇 명함 스캔 결과:\n")
|
|
print(contact.to_display())
|
|
print(f"\n💾 저장하려면: nc_contacts.py save <이미지경로>")
|
|
|
|
elif args[0] == "save" and len(args) > 1:
|
|
image_path = args[1]
|
|
contact = await client.scan_and_save(image_path)
|
|
print(f"✅ 연락처 저장 완료:\n")
|
|
print(contact.to_display())
|
|
|
|
elif args[0] == "search" and len(args) > 1:
|
|
query = " ".join(args[1:])
|
|
contacts = await client.search_contacts(query)
|
|
print(f"🔍 '{query}' 검색 결과: {len(contacts)}건\n")
|
|
for c in contacts:
|
|
print(c.to_display())
|
|
print()
|
|
|
|
elif args[0] == "list":
|
|
contacts = await client.list_contacts()
|
|
print(f"📇 전체 연락처: {len(contacts)}건\n")
|
|
for c in contacts:
|
|
print(f" 👤 {c.name} — {', '.join(c.phone) or '전화 없음'} — {', '.join(c.email) or '이메일 없음'}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_cli())
|