Spaces:
Build error
Build error
File size: 3,429 Bytes
fcfa3d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """
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")
}
|