feat: implement Comment Tracker Service to manage posted video IDs and prevent duplicate comments

app/routes/commend/comment_tracker.py

@@ -0,0 +1,117 @@
+"""
+Comment Tracker Service for Crown Commend
+Tracks posted video IDs to prevent duplicate comments (anti-spam/bot protection).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, Dict, Any
+from threading import Lock
+
+from ...services.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+# Thread-safe lock for file operations
+_file_lock = Lock()
+
+# Storage path - use /data for HF Spaces persistence, fallback to local
+def _get_storage_path() -> Path:
+    """Get the path for storing comment history."""
+    # HF Spaces persistent storage
+    if os.path.exists('/data'):
+        return Path('/data/commend_history.json')
+    # Local development
+    return Path(__file__).parent / 'commend_history.json'
+
+
+def _load_history() -> Dict[str, Any]:
+    """Load comment history from file."""
+    path = _get_storage_path()
+    try:
+        if path.exists():
+            with open(path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+    except (json.JSONDecodeError, IOError) as e:
+        logger.warning(f"Could not load history: {e}")
+    return {"posted_videos": {}}
+
+
+def _save_history(history: Dict[str, Any]) -> bool:
+    """Save comment history to file."""
+    path = _get_storage_path()
+    try:
+        # Ensure directory exists
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, 'w', encoding='utf-8') as f:
+            json.dump(history, f, ensure_ascii=False, indent=2)
+        return True
+    except IOError as e:
+        logger.error(f"Could not save history: {e}")
+        return False
+
+
+def is_video_commented(video_id: str) -> bool:
+    """Check if a comment has already been posted to this video."""
+    with _file_lock:
+        history = _load_history()
+        return video_id in history.get("posted_videos", {})
+
+
+def get_video_comment_info(video_id: str) -> Optional[Dict[str, Any]]:
+    """Get comment info for a video if it exists."""
+    with _file_lock:
+        history = _load_history()
+        return history.get("posted_videos", {}).get(video_id)
+
+
+def mark_video_commented(
+    video_id: str,
+    comment_id: str,
+    video_title: Optional[str] = None
+) -> bool:
+    """
+    Mark a video as commented.
+
+    Args:
+        video_id: YouTube video ID
+        comment_id: The posted comment's ID from YouTube
+        video_title: Optional video title for reference
+
+    Returns:
+        True if successfully saved
+    """
+    with _file_lock:
+        history = _load_history()
+
+        if "posted_videos" not in history:
+            history["posted_videos"] = {}
+
+        history["posted_videos"][video_id] = {
+            "comment_id": comment_id,
+            "video_title": video_title,
+            "posted_at": datetime.utcnow().isoformat() + "Z",
+        }
+
+        # Keep stats
+        history["total_comments"] = len(history["posted_videos"])
+        history["last_updated"] = datetime.utcnow().isoformat() + "Z"
+
+        success = _save_history(history)
+        if success:
+            logger.info(f"Marked video {video_id} as commented")
+        return success
+
+
+def get_comment_stats() -> Dict[str, Any]:
+    """Get comment statistics."""
+    with _file_lock:
+        history = _load_history()
+        return {
+            "total_comments": len(history.get("posted_videos", {})),
+            "last_updated": history.get("last_updated")
+        }