import json import os import subprocess import urllib.parse import uuid from io import BytesIO from typing import Any, Dict, List from concurrent.futures import ThreadPoolExecutor import threading 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") self.cache = {} self.cache_lock = threading.Lock() self.pre_cache_size = 300 # Pre-cache 300 articles ahead self.cache_executor = ThreadPoolExecutor(max_workers=2) # For background caching 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: # Check if we have cached posts for this range cache_key = f"posts_{start_index}_{count}" with self.cache_lock: if cache_key in self.cache: print(f"Cache hit for {cache_key}") return self.cache[cache_key] # 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] # Cache the results with self.cache_lock: self.cache[cache_key] = formatted_posts # Pre-cache the next batch in background if start_index + count < self.get_total_posts(): self._pre_cache_batch(start_index + count) 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 _pre_cache_batch(self, start_index: int): """Pre-cache the next batch of posts in the background.""" def cache_task(): try: # Pre-cache next 300 posts next_posts = self.get_posts(self.pre_cache_size, start_index) print(f"Pre-cached {len(next_posts)} posts starting from index {start_index}") except Exception as e: print(f"Error in pre-caching: {e}") # Submit to background thread self.cache_executor.submit(cache_task) def clear_cache(self): """Clear the cache.""" with self.cache_lock: self.cache.clear() 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