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""" {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 = """ """ 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())