222 lines
8.6 KiB
Python
222 lines
8.6 KiB
Python
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 |