"""Post/comment data service. Owns the short-lived in-memory cache of posts and comments. Phase 3 makes Postgres the primary source of truth; the legacy ``data/*.json`` reader remains as a fallback for local/dev environments before the backfill has run. """ import json import logging import time from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path from sqlalchemy import func from models import Comment, Post logger = logging.getLogger(__name__) _CACHE_DURATION = 300 # 5 minutes def load_platform_config(): """Load platform configuration from ``platform_config.json``. Returns a safe default (empty platforms, no targets) on any error so callers can iterate without extra guarding. """ try: with open("platform_config.json", "r", encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError, IOError) as e: logger.warning(f"Could not load platform config: {e}") return {"platforms": {}, "collection_targets": []} def get_display_name_for_source(platform, source, platform_config): """Get a human display name for a (platform, source) pair.""" if not platform_config or "platforms" not in platform_config: return source platform_info = platform_config["platforms"].get(platform, {}) if platform_info.get("supports_communities"): for community in platform_info.get("communities", []): if community["id"] == source: return community["display_name"] prefix = platform_info.get("prefix", "") return f"{prefix}{source}" if source else platform_info.get("name", platform) return platform_info.get("name", platform) class PostService: """Cache and serve posts/comments from Postgres, plus derived views.""" def __init__(self, cache_duration=_CACHE_DURATION): self.post_cache = {} self.comment_cache = defaultdict(list) self.cache_timestamp = 0 self.cache_duration = cache_duration self.cache_source = None def load(self): """Return (post_cache, comment_cache), refreshing from Postgres if stale.""" current_time = time.time() if current_time - self.cache_timestamp < self.cache_duration and self.post_cache: return self.post_cache, self.comment_cache self.post_cache.clear() self.comment_cache.clear() loaded_from_db = self._load_from_db() if not loaded_from_db: self._load_from_disk() self.cache_source = "disk" else: self.cache_source = "db" self.cache_timestamp = current_time logger.info( f"Cache refreshed from {self.cache_source}: {len(self.post_cache)} posts, " f"{len(self.comment_cache)} comment groups" ) return self.post_cache, self.comment_cache def _load_from_db(self): """Populate caches from Postgres. Return False if unavailable or empty.""" try: posts = Post.query.order_by(Post.timestamp.desc()).all() if not posts: return False for post in posts: self.post_cache[post.uuid] = self._post_to_dict(post) comments = Comment.query.order_by(Comment.timestamp.asc()).all() for comment in comments: self.comment_cache[comment.post_uuid].append(self._comment_to_dict(comment)) return True except Exception as e: logger.warning(f"Postgres content load unavailable; falling back to disk: {e}") self.post_cache.clear() self.comment_cache.clear() return False def _load_from_disk(self): posts_dir = Path("data/posts") comments_dir = Path("data/comments") if posts_dir.exists(): for post_file in posts_dir.glob("*.json"): try: with open(post_file, "r", encoding="utf-8") as f: post_data = json.load(f) post_uuid = post_data.get("uuid") if post_uuid: self.post_cache[post_uuid] = post_data except (json.JSONDecodeError, IOError) as e: logger.debug(f"Error reading post file {post_file}: {e}") if comments_dir.exists(): for comment_file in comments_dir.glob("*.json"): try: with open(comment_file, "r", encoding="utf-8") as f: comment_data = json.load(f) post_uuid = comment_data.get("post_uuid") if post_uuid: self.comment_cache[post_uuid].append(comment_data) except (json.JSONDecodeError, IOError) as e: logger.debug(f"Error reading comment file {comment_file}: {e}") @staticmethod def _post_to_dict(post): return { "uuid": post.uuid, "id": post.external_id or post.uuid, "platform": post.platform, "source": post.source, "title": post.title, "author": post.author, "url": post.url, "content": post.content, "score": post.score, "timestamp": post.timestamp, "tags": post.tags or [], "moderation_uuid": post.moderation_uuid, } @staticmethod def _comment_to_dict(comment): return { "uuid": comment.uuid, "post_uuid": comment.post_uuid, "platform": comment.platform, "parent_comment_uuid": comment.parent_comment_uuid, "id": comment.comment_id or comment.uuid, "comment_id": comment.comment_id, "author": comment.author, "content": comment.content, "score": comment.score, "timestamp": comment.timestamp, "depth": comment.depth, "moderation_uuid": comment.moderation_uuid, } def invalidate(self): """Force the next ``load()`` to refresh content.""" self.cache_timestamp = 0 def quick_stats(self): """Return {posts_today, total_posts} for the dashboard.""" cached_posts, _ = self.load() now = datetime.utcnow() today_timestamp = (now - timedelta(hours=24)).timestamp() posts_today = sum( 1 for post in cached_posts.values() if post.get("timestamp", 0) >= today_timestamp ) return {"posts_today": posts_today, "total_posts": len(cached_posts)} @staticmethod def build_comment_tree(comments): """Build a hierarchical comment tree from a flat comment list.""" comment_dict = {c["uuid"]: {**c, "replies": []} for c in comments} root_comments = [] for comment in comments: parent_uuid = comment.get("parent_comment_uuid") if parent_uuid and parent_uuid in comment_dict: comment_dict[parent_uuid]["replies"].append( comment_dict[comment["uuid"]] ) else: root_comments.append(comment_dict[comment["uuid"]]) def sort_tree(comments_list): comments_list.sort(key=lambda x: x.get("timestamp", 0)) for comment in comments_list: if comment.get("replies"): sort_tree(comment["replies"]) sort_tree(root_comments) return root_comments def latest_content_mtime(self): """Latest content update timestamp for client auto-refresh polling.""" try: latest_created = Post.query.with_entities(func.max(Post.created_at)).scalar() if latest_created: return latest_created.timestamp() except Exception as e: logger.debug(f"Could not read latest content timestamp from DB: {e}") posts_dir = Path("data/posts") if not posts_dir.exists(): return 0 latest = 0 for post_file in posts_dir.glob("*.json"): mtime = post_file.stat().st_mtime if mtime > latest: latest = mtime return latest def source_counts(self): """Count posts per ``platform:source`` for the platforms API.""" try: rows = ( Post.query.with_entities(Post.platform, Post.source, func.count(Post.uuid)) .group_by(Post.platform, Post.source) .all() ) if rows: return {f"{platform}:{source}": count for platform, source, count in rows} except Exception as e: logger.debug(f"Could not read source counts from DB: {e}") counts = {} posts_dir = Path("data/posts") if not posts_dir.exists(): return counts for post_file in posts_dir.glob("*.json"): try: with open(post_file, "r", encoding="utf-8") as f: post_data = json.load(f) key = f"{post_data.get('platform', 'unknown')}:{post_data.get('source', '')}" counts[key] = counts.get(key, 0) + 1 except (json.JSONDecodeError, IOError): continue return counts post_service = PostService()