318 lines
11 KiB
Python
318 lines
11 KiB
Python
import os
|
|
import subprocess
|
|
import json
|
|
from typing import List, Dict, Any
|
|
import uuid
|
|
import urllib.parse
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from io import BytesIO
|
|
|
|
|
|
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_dir = os.path.join(os.path.dirname(__file__), "..", "..", "cache")
|
|
os.makedirs(self.cache_dir, exist_ok=True)
|
|
|
|
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=30
|
|
)
|
|
|
|
# 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 _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", "#")
|
|
|
|
# Set default image to reddit_logo.webp if no image is provided
|
|
post["image"] = (
|
|
"/Reddit_Logo.webp" # Default to the logo in website directory
|
|
)
|
|
|
|
# Generate HTML preview for the post URL
|
|
post["image"] = self._generate_preview_image(post["url"], post["id"])
|
|
|
|
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
|
|
|
|
def _generate_preview_image(self, url: str, post_id: str) -> str:
|
|
"""
|
|
Generate a preview image for the HTML page at the given URL.
|
|
|
|
This implementation extracts the largest image from the HTML page content,
|
|
falling back to a placeholder if no suitable images are found.
|
|
|
|
Args:
|
|
url (str): The URL to generate preview for
|
|
post_id (str): The ID of the post
|
|
|
|
Returns:
|
|
str: Path to the generated preview image
|
|
"""
|
|
try:
|
|
# Validate the URL
|
|
parsed_url = urllib.parse.urlparse(url)
|
|
if not parsed_url.scheme or not parsed_url.netloc:
|
|
return "/Reddit_Logo.webp"
|
|
|
|
# Generate filename based on post ID
|
|
filename = f"{post_id}.webp"
|
|
image_path = os.path.join(self.cache_dir, filename)
|
|
|
|
# If image already exists, return its path
|
|
if os.path.exists(image_path):
|
|
return f"/cache/{filename}"
|
|
|
|
# Try to extract and use the largest image from the webpage content
|
|
image_extracted = self._extract_largest_image_from_webpage(url, image_path)
|
|
|
|
if not image_extracted:
|
|
# Fall back to creating a placeholder image
|
|
self._create_placeholder_image(url, image_path)
|
|
|
|
return f"/cache/{filename}"
|
|
|
|
except Exception as e:
|
|
print(f"Error generating preview for URL {url}: {e}")
|
|
# Return default fallback image in case of any error
|
|
return "/Reddit_Logo.webp"
|
|
|
|
def _create_placeholder_image(self, url: str, save_path: str):
|
|
"""
|
|
Create a simple placeholder image when real preview isn't possible.
|
|
|
|
Args:
|
|
url (str): The URL that was attempted
|
|
save_path (str): Path to save the image
|
|
"""
|
|
try:
|
|
# Create a simple placeholder image
|
|
img = Image.new("RGB", (800, 600), color=(73, 109, 137))
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# Try to get default font or use None (which will fall back to default)
|
|
try:
|
|
font = ImageFont.load_default()
|
|
except Exception:
|
|
font = None
|
|
|
|
# Draw some text on the image
|
|
draw.text(
|
|
(10, 10), "HTML Preview Placeholder", fill=(255, 255, 0), font=font
|
|
)
|
|
draw.text(
|
|
(10, 40),
|
|
f"URL: {url[:50]}{'...' if len(url) > 50 else ''}",
|
|
fill=(255, 255, 255),
|
|
font=font,
|
|
)
|
|
|
|
# Add instructions
|
|
draw.text(
|
|
(10, 80),
|
|
"Largest image from webpage would appear here",
|
|
fill=(255, 255, 255),
|
|
font=font,
|
|
)
|
|
draw.text(
|
|
(10, 100),
|
|
"This is a placeholder image",
|
|
fill=(255, 255, 255),
|
|
font=font,
|
|
)
|
|
|
|
# Save as webp image
|
|
img.save(save_path, "WEBP", quality=80)
|
|
except Exception as e:
|
|
print(f"Error creating placeholder image: {e}")
|
|
# If everything fails, we still have the default fallback
|
|
return
|
|
|
|
def _extract_largest_image_from_webpage(self, url: str, save_path: str) -> bool:
|
|
"""
|
|
Extract the largest image from a webpage and save it.
|
|
|
|
Args:
|
|
url (str): URL of the webpage to extract images from
|
|
save_path (str): Path where to save the extracted image
|
|
|
|
Returns:
|
|
bool: True if successful, False otherwise
|
|
"""
|
|
try:
|
|
# Fetch the webpage content
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
}
|
|
response = requests.get(url, timeout=10, headers=headers)
|
|
response.raise_for_status()
|
|
|
|
# Parse HTML with BeautifulSoup
|
|
soup = BeautifulSoup(response.content, "html.parser")
|
|
|
|
# Find all img tags
|
|
img_tags = soup.find_all("img")
|
|
|
|
if not img_tags:
|
|
return False
|
|
|
|
# Get the largest image by size (height * width)
|
|
largest_img = None
|
|
max_area = 0
|
|
largest_img_src = None
|
|
|
|
for img in img_tags:
|
|
src = img.get("src") or img.get("data-src")
|
|
if not src:
|
|
continue
|
|
|
|
# Resolve relative URLs
|
|
if not src.startswith(("http://", "https://")):
|
|
from urllib.parse import urljoin
|
|
|
|
src = urljoin(url, src)
|
|
|
|
try:
|
|
# Get image dimensions by downloading it
|
|
img_response = requests.get(src, timeout=10)
|
|
img_response.raise_for_status()
|
|
|
|
# Open image with PIL to get dimensions
|
|
img_data = Image.open(BytesIO(img_response.content))
|
|
area = img_data.width * img_data.height
|
|
|
|
if area > max_area:
|
|
max_area = area
|
|
largest_img_src = src
|
|
|
|
except Exception as e:
|
|
print(f"Could not process image from {src}: {e}")
|
|
continue
|
|
|
|
# If we found a suitable image, download and save it
|
|
if largest_img_src:
|
|
img_response = requests.get(largest_img_src, timeout=10)
|
|
img_response.raise_for_status()
|
|
|
|
# Save as webp with quality 80
|
|
img = Image.open(BytesIO(img_response.content))
|
|
img.save(save_path, "WEBP", quality=80)
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error extracting image from webpage: {e}")
|
|
return False
|
|
|
|
return False
|