""" Comment Library Atomic functions for comment processing and tree manipulation. """ import json from pathlib import Path from typing import List, Dict, Optional class comment_lib: """Atomic comment processing functions""" @staticmethod def build_comment_tree(flat_comments: List[Dict]) -> List[Dict]: """ Convert flat array of comments to nested tree structure. Returns list of root-level comments with nested children. """ if not flat_comments: return [] # Create lookup dict comment_map = {c['uuid']: {**c, 'children': []} for c in flat_comments} # Build tree roots = [] for comment in flat_comments: parent_uuid = comment.get('parent_comment_uuid') if parent_uuid and parent_uuid in comment_map: comment_map[parent_uuid]['children'].append(comment_map[comment['uuid']]) else: roots.append(comment_map[comment['uuid']]) return roots @staticmethod def flatten_comment_tree(tree: List[Dict]) -> List[Dict]: """ Convert nested tree structure to flat array. Removes 'children' key from each comment. """ flat = [] def traverse(nodes): for node in nodes: children = node.pop('children', []) flat.append(node) if children: traverse(children) traverse(tree) return flat @staticmethod def load_comments_for_post(post_uuid: str, data_dir: str) -> List[Dict]: """ Load all comment files linked to a post. Scans comment directory for comments with matching post_uuid. """ comments_dir = Path(data_dir) / 'comments' if not comments_dir.exists(): return [] comments = [] for comment_file in comments_dir.glob('*.json'): with open(comment_file, 'r') as f: comment = json.load(f) if comment.get('post_uuid') == post_uuid: comments.append(comment) return comments @staticmethod def sort_comments(comments: List[Dict], by: str = 'score', order: str = 'desc') -> List[Dict]: """ Sort comments by specified field. Args: comments: List of comment dicts by: Field to sort by ('score', 'timestamp', 'depth', 'author') order: 'asc' or 'desc' Returns: Sorted list of comments """ reverse = (order == 'desc') return sorted(comments, key=lambda c: c.get(by, 0), reverse=reverse) @staticmethod def get_comment_depth(comment: Dict, comment_map: Dict) -> int: """ Calculate actual depth of a comment by traversing up parent chain. Useful for recalculating depth after filtering. """ depth = 0 current_uuid = comment.get('parent_comment_uuid') while current_uuid and current_uuid in comment_map: depth += 1 current_uuid = comment_map[current_uuid].get('parent_comment_uuid') return depth @staticmethod def get_comment_stats(comments: List[Dict]) -> Dict: """ Get statistics about a comment list. Returns: Dict with total, max_depth, avg_score, etc. """ if not comments: return { 'total': 0, 'max_depth': 0, 'avg_score': 0, 'total_score': 0 } depths = [c.get('depth', 0) for c in comments] scores = [c.get('score', 0) for c in comments] return { 'total': len(comments), 'max_depth': max(depths) if depths else 0, 'avg_score': sum(scores) / len(scores) if scores else 0, 'total_score': sum(scores) } @staticmethod def filter_by_depth(comments: List[Dict], max_depth: int) -> List[Dict]: """ Filter comments to only include those at or below max_depth. """ return [c for c in comments if c.get('depth', 0) <= max_depth] @staticmethod def get_top_level_comments(comments: List[Dict]) -> List[Dict]: """ Get only top-level comments (depth 0, no parent). """ return [c for c in comments if c.get('depth', 0) == 0 or not c.get('parent_comment_uuid')] @staticmethod def count_replies(comment_uuid: str, comments: List[Dict]) -> int: """ Count total number of replies (direct and nested) for a comment. """ count = 0 for comment in comments: if comment.get('parent_comment_uuid') == comment_uuid: count += 1 # Recursively count this comment's replies count += comment_lib.count_replies(comment['uuid'], comments) return count