Upload cache_db.py

cache_db.py

@@ -0,0 +1,510 @@
+"""
+과제 공고 벡터 DB 캐시 시스템
+- ChromaDB를 사용한 로컬 캐시
+- 매일 KST 10:00, 22:00 자동 동기화
+- Hugging Face Space 영구 스토리지 활용 (/data)
+"""
+import os
+import json
+import hashlib
+import threading
+import logging
+from datetime import datetime, timedelta
+from typing import List, Dict, Tuple, Optional
+from pathlib import Path
+import pytz
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# 영구 스토리지 경로 (HF Space)
+PERSISTENT_DIR = Path("/data") if os.path.exists("/data") else Path("./data")
+CACHE_DIR = PERSISTENT_DIR / "announcement_cache"
+DB_PATH = CACHE_DIR / "chroma_db"
+METADATA_FILE = CACHE_DIR / "sync_metadata.json"
+
+# 디렉토리 생성
+CACHE_DIR.mkdir(parents=True, exist_ok=True)
+
+# ChromaDB 사용 가능 여부
+try:
+    import chromadb
+    from chromadb.config import Settings
+    CHROMADB_AVAILABLE = True
+except ImportError:
+    CHROMADB_AVAILABLE = False
+    logger.warning("ChromaDB not available. Using JSON fallback.")
+
+# APScheduler 사용 가능 여부
+try:
+    from apscheduler.schedulers.background import BackgroundScheduler
+    from apscheduler.triggers.cron import CronTrigger
+    SCHEDULER_AVAILABLE = True
+except ImportError:
+    SCHEDULER_AVAILABLE = False
+    logger.warning("APScheduler not available. Auto-sync disabled.")
+
+# 한국 시간대
+KST = pytz.timezone('Asia/Seoul')
+
+
+class AnnouncementCache:
+    """공고 캐시 관리 클래스"""
+    
+    def __init__(self):
+        self.collection = None
+        self.client = None
+        self._init_db()
+        self._load_metadata()
+    
+    def _init_db(self):
+        """ChromaDB 초기화"""
+        if CHROMADB_AVAILABLE:
+            try:
+                self.client = chromadb.PersistentClient(
+                    path=str(DB_PATH),
+                    settings=Settings(anonymized_telemetry=False)
+                )
+                self.collection = self.client.get_or_create_collection(
+                    name="announcements",
+                    metadata={"description": "기업마당 과제 공고 캐시"}
+                )
+                logger.info(f"ChromaDB initialized at {DB_PATH}")
+            except Exception as e:
+                logger.error(f"ChromaDB init error: {e}")
+                self.collection = None
+        else:
+            logger.info("Using JSON fallback storage")
+    
+    def _load_metadata(self):
+        """동기화 메타데이터 로드"""
+        self.metadata = {
+            "last_sync": None,
+            "total_count": 0,
+            "sync_history": []
+        }
+        if METADATA_FILE.exists():
+            try:
+                with open(METADATA_FILE, 'r', encoding='utf-8') as f:
+                    self.metadata = json.load(f)
+            except Exception as e:
+                logger.error(f"Metadata load error: {e}")
+    
+    def _save_metadata(self):
+        """동기화 메타데이터 저장"""
+        try:
+            with open(METADATA_FILE, 'w', encoding='utf-8') as f:
+                json.dump(self.metadata, f, ensure_ascii=False, indent=2)
+        except Exception as e:
+            logger.error(f"Metadata save error: {e}")
+    
+    def _generate_id(self, item: Dict) -> str:
+        """공고 고유 ID 생성"""
+        unique_str = f"{item.get('pblancId', '')}-{item.get('title', '')}-{item.get('pubDate', '')}"
+        return hashlib.md5(unique_str.encode()).hexdigest()
+    
+    def _item_to_document(self, item: Dict) -> Tuple[str, str, Dict]:
+        """API 아이템을 ChromaDB 문서로 변환"""
+        doc_id = self._generate_id(item)
+        
+        # 검색 가능한 텍스트 생성
+        searchable_text = " ".join(filter(None, [
+            item.get("title", ""),
+            item.get("author", ""),
+            item.get("description", ""),
+            item.get("hashTags", ""),
+            item.get("lcategory", ""),
+            item.get("trgetNm", ""),
+            item.get("excInsttNm", ""),
+        ]))
+        
+        # 메타데이터
+        metadata = {
+            "pblancId": str(item.get("pblancId", item.get("seq", ""))),
+            "title": item.get("title", item.get("pblancNm", ""))[:500],
+            "author": item.get("author", item.get("jrsdInsttNm", ""))[:200],
+            "reqstDt": item.get("reqstDt", item.get("reqstBeginEndDe", ""))[:100],
+            "pubDate": item.get("pubDate", item.get("creatPnttm", ""))[:20],
+            "link": item.get("link", item.get("pblancUrl", ""))[:500],
+            "lcategory": item.get("lcategory", item.get("pldirSportRealmLclasCodeNm", ""))[:50],
+            "hashTags": item.get("hashTags", "")[:200],
+            "description": item.get("description", item.get("bsnsSumryCn", ""))[:1000],
+            "trgetNm": item.get("trgetNm", "")[:200],
+            "excInsttNm": item.get("excInsttNm", "")[:200],
+            "inqireCo": str(item.get("inqireCo", "0")),
+            "flpthNm": item.get("flpthNm", "")[:500],
+            "fileNm": item.get("fileNm", "")[:200],
+            "refrncNm": item.get("refrncNm", "")[:200],
+            "rceptEngnHmpgUrl": item.get("rceptEngnHmpgUrl", "")[:500],
+            "cached_at": datetime.now(KST).isoformat(),
+        }
+        
+        return doc_id, searchable_text, metadata
+    
+    def get_count(self) -> int:
+        """캐시된 공고 수 반환"""
+        if self.collection:
+            try:
+                return self.collection.count()
+            except:
+                return 0
+        return self.metadata.get("total_count", 0)
+    
+    def get_all(self) -> List[Dict]:
+        """모든 캐시된 공고 반환"""
+        if not self.collection:
+            return self._get_all_from_json()
+        
+        try:
+            result = self.collection.get(include=["metadatas", "documents"])
+            items = []
+            for i, meta in enumerate(result.get("metadatas", [])):
+                item = self._metadata_to_item(meta)
+                items.append(item)
+            return items
+        except Exception as e:
+            logger.error(f"Get all error: {e}")
+            return []
+    
+    def _metadata_to_item(self, meta: Dict) -> Dict:
+        """메타데이터를 원본 아이템 형식으로 변환"""
+        return {
+            "pblancId": meta.get("pblancId", ""),
+            "seq": meta.get("pblancId", ""),
+            "title": meta.get("title", ""),
+            "pblancNm": meta.get("title", ""),
+            "author": meta.get("author", ""),
+            "jrsdInsttNm": meta.get("author", ""),
+            "reqstDt": meta.get("reqstDt", ""),
+            "reqstBeginEndDe": meta.get("reqstDt", ""),
+            "pubDate": meta.get("pubDate", ""),
+            "creatPnttm": meta.get("pubDate", ""),
+            "link": meta.get("link", ""),
+            "pblancUrl": meta.get("link", ""),
+            "lcategory": meta.get("lcategory", ""),
+            "pldirSportRealmLclasCodeNm": meta.get("lcategory", ""),
+            "hashTags": meta.get("hashTags", ""),
+            "description": meta.get("description", ""),
+            "bsnsSumryCn": meta.get("description", ""),
+            "trgetNm": meta.get("trgetNm", ""),
+            "excInsttNm": meta.get("excInsttNm", ""),
+            "inqireCo": meta.get("inqireCo", "0"),
+            "flpthNm": meta.get("flpthNm", ""),
+            "fileNm": meta.get("fileNm", ""),
+            "refrncNm": meta.get("refrncNm", ""),
+            "rceptEngnHmpgUrl": meta.get("rceptEngnHmpgUrl", ""),
+        }
+    
+    def _get_all_from_json(self) -> List[Dict]:
+        """JSON 폴백에서 모든 공고 로드"""
+        json_file = CACHE_DIR / "announcements.json"
+        if json_file.exists():
+            try:
+                with open(json_file, 'r', encoding='utf-8') as f:
+                    return json.load(f)
+            except:
+                return []
+        return []
+    
+    def _save_to_json(self, items: List[Dict]):
+        """JSON 폴백으로 저장"""
+        json_file = CACHE_DIR / "announcements.json"
+        try:
+            with open(json_file, 'w', encoding='utf-8') as f:
+                json.dump(items, f, ensure_ascii=False, indent=2)
+        except Exception as e:
+            logger.error(f"JSON save error: {e}")
+    
+    def bulk_upsert(self, items: List[Dict]) -> Tuple[int, int]:
+        """대량 삽입/업데이트"""
+        if not items:
+            return 0, 0
+        
+        added = 0
+        updated = 0
+        
+        if self.collection:
+            try:
+                # 기존 ID 목록 가져오기
+                existing = set()
+                try:
+                    result = self.collection.get()
+                    existing = set(result.get("ids", []))
+                except:
+                    pass
+                
+                ids = []
+                documents = []
+                metadatas = []
+                
+                for item in items:
+                    doc_id, doc_text, meta = self._item_to_document(item)
+                    ids.append(doc_id)
+                    documents.append(doc_text)
+                    metadatas.append(meta)
+                    
+                    if doc_id in existing:
+                        updated += 1
+                    else:
+                        added += 1
+                
+                # upsert
+                self.collection.upsert(
+                    ids=ids,
+                    documents=documents,
+                    metadatas=metadatas
+                )
+                
+                logger.info(f"Bulk upsert: {added} added, {updated} updated")
+                
+            except Exception as e:
+                logger.error(f"Bulk upsert error: {e}")
+                # JSON 폴백
+                self._save_to_json(items)
+                added = len(items)
+        else:
+            # JSON 폴백
+            self._save_to_json(items)
+            added = len(items)
+        
+        # 메타데이터 업데이트
+        self.metadata["total_count"] = self.get_count()
+        self.metadata["last_sync"] = datetime.now(KST).isoformat()
+        self._save_metadata()
+        
+        return added, updated
+    
+    def search(self, query: str, n_results: int = 20) -> List[Dict]:
+        """텍스트 검색"""
+        if not self.collection or not query.strip():
+            return self.get_all()[:n_results]
+        
+        try:
+            result = self.collection.query(
+                query_texts=[query],
+                n_results=n_results,
+                include=["metadatas", "documents", "distances"]
+            )
+            
+            items = []
+            for meta in result.get("metadatas", [[]])[0]:
+                items.append(self._metadata_to_item(meta))
+            return items
+            
+        except Exception as e:
+            logger.error(f"Search error: {e}")
+            return []
+    
+    def get_existing_ids(self) -> set:
+        """기존 공고 ID 집합 반환"""
+        if self.collection:
+            try:
+                result = self.collection.get(include=["metadatas"])
+                return {meta.get("pblancId", "") for meta in result.get("metadatas", [])}
+            except:
+                return set()
+        
+        items = self._get_all_from_json()
+        return {item.get("pblancId", item.get("seq", "")) for item in items}
+    
+    def remove_expired(self, days: int = 90) -> int:
+        """만료된 공고 삭제 (선택적)"""
+        # TODO: 필요시 구현
+        return 0
+
+
+# 글로벌 캐시 인스턴스
+_cache_instance = None
+
+
+def get_cache() -> AnnouncementCache:
+    """싱글톤 캐시 인스턴스 반환"""
+    global _cache_instance
+    if _cache_instance is None:
+        _cache_instance = AnnouncementCache()
+    return _cache_instance
+
+
+# ============================================================
+# 동기화 함수
+# ============================================================
+def sync_from_api() -> Tuple[int, int, str]:
+    """
+    API에서 공고를 가져와 캐시에 동기화
+    Returns: (added_count, updated_count, status_message)
+    """
+    from file_api import fetch_all_from_api
+    
+    cache = get_cache()
+    sync_time = datetime.now(KST)
+    
+    logger.info(f"Starting sync at {sync_time.strftime('%Y-%m-%d %H:%M:%S')} KST")
+    
+    try:
+        # API에서 전체 데이터 가져오기
+        items, error = fetch_all_from_api(category="전체", region="전체(지역)", keyword="")
+        
+        if error and not items:
+            msg = f"❌ API 오류: {error}"
+            logger.error(msg)
+            return 0, 0, msg
+        
+        if not items:
+            msg = "⚠️ API에서 데이터를 가져올 수 없습니다."
+            logger.warning(msg)
+            return 0, 0, msg
+        
+        # 캐시에 저장
+        added, updated = cache.bulk_upsert(items)
+        
+        # 동기화 이력 저장
+        sync_record = {
+            "timestamp": sync_time.isoformat(),
+            "api_count": len(items),
+            "added": added,
+            "updated": updated,
+            "total_cached": cache.get_count()
+        }
+        
+        cache.metadata.setdefault("sync_history", []).append(sync_record)
+        # 최근 100개 이력만 유지
+        cache.metadata["sync_history"] = cache.metadata["sync_history"][-100:]
+        cache._save_metadata()
+        
+        msg = f"✅ 동기화 완료: API {len(items)}건 → 신규 {added}건, 업데이트 {updated}건 (총 {cache.get_count()}건)"
+        logger.info(msg)
+        return added, updated, msg
+        
+    except Exception as e:
+        msg = f"❌ 동기화 오류: {str(e)}"
+        logger.error(msg)
+        return 0, 0, msg
+
+
+def get_cached_announcements() -> Tuple[List[Dict], str]:
+    """
+    캐시에서 공고 목록 반환 (캐시가 비어있으면 API에서 로드)
+    Returns: (items, status_message)
+    """
+    cache = get_cache()
+    count = cache.get_count()
+    
+    if count == 0:
+        # 초기 로드
+        logger.info("Cache empty, performing initial sync...")
+        added, updated, msg = sync_from_api()
+        if added == 0 and updated == 0:
+            return [], msg
+    
+    items = cache.get_all()
+    last_sync = cache.metadata.get("last_sync", "알 수 없음")
+    
+    status = f"📦 캐시에서 {len(items)}건 로드 (마지막 동기화: {last_sync})"
+    return items, status
+
+
+def get_sync_status() -> Dict:
+    """동기화 상태 정보 반환"""
+    cache = get_cache()
+    return {
+        "total_count": cache.get_count(),
+        "last_sync": cache.metadata.get("last_sync"),
+        "sync_history": cache.metadata.get("sync_history", [])[-5:],
+        "db_path": str(DB_PATH),
+        "chromadb_available": CHROMADB_AVAILABLE,
+        "scheduler_available": SCHEDULER_AVAILABLE,
+    }
+
+
+# ============================================================
+# 스케줄러
+# ============================================================
+_scheduler = None
+
+
+def start_scheduler():
+    """백그라운드 스케줄러 시작 (KST 10:00, 22:00)"""
+    global _scheduler
+    
+    if not SCHEDULER_AVAILABLE:
+        logger.warning("Scheduler not available")
+        return False
+    
+    if _scheduler is not None:
+        logger.info("Scheduler already running")
+        return True
+    
+    try:
+        _scheduler = BackgroundScheduler(timezone=KST)
+        
+        # 매일 오전 10시 (KST)
+        _scheduler.add_job(
+            sync_from_api,
+            CronTrigger(hour=10, minute=0, timezone=KST),
+            id='sync_10am',
+            name='Daily sync at 10:00 KST',
+            replace_existing=True
+        )
+        
+        # 매일 오후 10시 (KST)
+        _scheduler.add_job(
+            sync_from_api,
+            CronTrigger(hour=22, minute=0, timezone=KST),
+            id='sync_10pm',
+            name='Daily sync at 22:00 KST',
+            replace_existing=True
+        )
+        
+        _scheduler.start()
+        logger.info("Scheduler started: sync at 10:00 and 22:00 KST")
+        return True
+        
+    except Exception as e:
+        logger.error(f"Scheduler start error: {e}")
+        return False
+
+
+def stop_scheduler():
+    """스케줄러 중지"""
+    global _scheduler
+    if _scheduler:
+        _scheduler.shutdown()
+        _scheduler = None
+        logger.info("Scheduler stopped")
+
+
+def manual_sync() -> str:
+    """수동 동기화 실행"""
+    added, updated, msg = sync_from_api()
+    return msg
+
+
+# ============================================================
+# 앱 시작 시 초기화
+# ============================================================
+def initialize_cache_system():
+    """캐시 시스템 초기화 (앱 시작 시 호출)"""
+    logger.info("Initializing cache system...")
+    
+    # 캐시 초기화
+    cache = get_cache()
+    count = cache.get_count()
+    
+    if count == 0:
+        logger.info("Cache is empty, performing initial sync...")
+        sync_from_api()
+    else:
+        logger.info(f"Cache loaded with {count} announcements")
+    
+    # 스케줄러 시작
+    start_scheduler()
+    
+    return get_sync_status()
+
+
+if __name__ == "__main__":
+    # 테스트
+    print("Testing cache system...")
+    status = initialize_cache_system()
+    print(f"Status: {json.dumps(status, ensure_ascii=False, indent=2)}")