import json import os import subprocess import urllib.parse import uuid from io import BytesIO from typing import Any, Dict, List import requests from PIL import Image class ArchiveParser: def __init__(self): """Initialize the ArchiveParser with archive_dir from environment variable.""" self.archive_dir = os.environ.get("ARCHIVE_DIR", "/default/archive/path") def get_posts(self, count: int, start_index: int = 0) -> List[Dict[str, Any]]: """ Get posts from the archive using efficient shell commands. For pagination: - First call: get_posts(count=10, start_index=0) gets first 10 latest posts - Next call: get_posts(count=10, start_index=N) gets next 10 posts after index N Args: count (int): Number of posts to retrieve start_index (int): Starting index for retrieving posts Returns: List[Dict[str, Any]]: List of post dictionaries """ try: # Use ls with tail and head commands for efficient pagination # Get all directories, skip the first start_index, then get count number of entries cmd = f"ls -1 {self.archive_dir} | tail -{start_index + count + 1} | head -{count}" print("Running cmd: " + cmd) result = subprocess.run( cmd, shell=True, capture_output=True, text=True, check=True, timeout=300 ) # Parse the output (directory names) directories = [ line.strip() for line in result.stdout.split("\n") if line.strip() ] # Extract posts from each directory posts = [] for directory in directories: dir_path = os.path.join(self.archive_dir, directory) posts.extend(self._extract_posts_from_directory(dir_path)) # Convert extracted posts to final format with proper IDs formatted_posts = [self._create_post(post) for post in posts] return formatted_posts except subprocess.CalledProcessError as e: print(f"Error executing command: {e}") return [] except subprocess.TimeoutExpired: print("Command timed out") return [] except Exception as e: print(f"Error retrieving posts: {e}") return [] def _extract_posts_from_directory( self, directory_path: str ) -> List[Dict[str, Any]]: """Extract posts from a single directory.""" posts = [] # Check if there's an index.json in this directory index_file_path = os.path.join(directory_path, "index.json") if os.path.exists(index_file_path): try: with open(index_file_path, "r") as f: data = json.load(f) # Handle both single post and list of posts in the index.json if isinstance(data, list): posts.extend(data) else: posts.append(data) except Exception as e: print(f"Error reading index.json at {index_file_path}: {e}") else: # If no index.json, try to get individual post files (backward compatibility) try: for root, dirs, files in os.walk(directory_path): for file in files: if file.endswith(".json"): file_path = os.path.join(root, file) with open(file_path, "r") as f: data = json.load(f) posts.append(data) except Exception as e: print(f"Error reading post files: {e}") return posts def get_total_posts(self) -> int: """Get the total number of directories (post series) in the archive.""" try: # Use ls with wc -l to count directories efficiently cmd = f"ls -1 {self.archive_dir} | wc -l" result = subprocess.run( cmd, shell=True, capture_output=True, text=True, check=True, timeout=30 ) return int(result.stdout.strip()) except subprocess.CalledProcessError as e: print(f"Error executing count command: {e}") return 0 except subprocess.TimeoutExpired: print("Count command timed out") return 0 except Exception as e: print(f"Error getting total posts: {e}") return 0 def _extract_real_title_from_url(self, base_url: str) -> str: """Extract the real title from base_url when title is 'Reddit - Prove your humanity'.""" if not base_url or not isinstance(base_url, str): return "" # Check if the URL contains reddit pattern if "reddit.com/" in base_url: try: # Parse URL to extract path component parsed_url = urllib.parse.urlparse(base_url) path_parts = parsed_url.path.strip("/").split("/") if len(path_parts) >= 2: # Get the last part which should contain the title last_part = path_parts[-1] # Handle cases like "why_did_the_dday_beach_landing_soldiers_carry_all/#2025-10-14T23:33:16+00:00" if "#" in last_part: title_part = last_part.split("#")[0] else: title_part = last_part # Convert URL-encoded characters back to readable text title_part = urllib.parse.unquote(title_part) # Replace underscores with spaces and convert to title case title = title_part.replace("_", " ") # Capitalize first letter of each word for proper title formatting title = " ".join(word.capitalize() for word in title.split()) return title if title else "" except Exception as e: print(f"Error extracting real title from URL: {e}") return "" return "" def _extract_subreddit_from_url(self, base_url: str) -> str: """Extract the subreddit name from base_url.""" if not base_url or not isinstance(base_url, str): return "" # Check if the URL contains reddit pattern if "reddit.com/" in base_url: try: # Parse URL to extract path component parsed_url = urllib.parse.urlparse(base_url) path_parts = parsed_url.path.strip("/").split("/") # Find index of 'r' which marks the subreddit r_index = -1 for i, part in enumerate(path_parts): if part == "r": r_index = i break # If we found 'r', the next part should be the subreddit name if r_index != -1 and r_index + 1 < len(path_parts): subreddit = path_parts[r_index + 1] return subreddit except Exception as e: print(f"Error extracting subreddit from URL: {e}") return "" return "" def _create_post(self, index) -> Dict[str, Any]: """Convert an index entry to a properly formatted post with ID and default image.""" try: post = dict() # Ensure we have all required fields with defaults post["id"] = str(uuid.uuid4()) # Generate a unique ID for each post post["title"] = index.get("title", "Untitled Post") post["url"] = "http://archive2.home.ms/" + index.get("timestamp", "#") # Check if title is the placeholder and extract real title from base_url if post["title"] == "Reddit - Prove your humanity": base_url = index.get("base_url", "") real_title = self._extract_real_title_from_url(base_url) if real_title: # Extract subreddit name to format title properly subreddit = self._extract_subreddit_from_url(base_url) if subreddit: post["title"] = f"{subreddit} - {real_title}" else: post["title"] = real_title # Set default image to reddit_logo.webp if no image is provided post["image"] = ( "/Reddit_Logo.webp" # Default to the logo in website directory ) return post except Exception as e: print(f"Error converting index to post: {e}") # Return a minimal post if there's an error raise e