From 6bff87c98ce24e03e3e3858d6f79bd9b0eb6679d Mon Sep 17 00:00:00 2001 From: Richard Nixon Date: Mon, 1 Sep 2025 16:45:00 -0300 Subject: [PATCH] Add SQLite database layer for post tracking and deduplication --- src/database.py | 997 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 997 insertions(+) create mode 100644 src/database.py diff --git a/src/database.py b/src/database.py new file mode 100644 index 0000000..ba3d7b5 --- /dev/null +++ b/src/database.py @@ -0,0 +1,997 @@ +"""SQLite database for storing post metadata and tracking downloads.""" + +import sqlite3 +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional + + +@dataclass +class PostRecord: + id: str + subreddit: str + author: str + title: str + url: str + media_url: Optional[str] + media_type: Optional[str] + score: int + created_utc: float + downloaded_at: Optional[datetime] + local_path: Optional[str] + file_hash: Optional[str] + permalink: Optional[str] = None # Reddit permalink for Immich + source_type: Optional[str] = None # 'subreddit' or 'user' + flair: Optional[str] = None # Post flair for tagging + + +class Database: + """SQLite database wrapper for tracking downloaded posts.""" + + def __init__(self, db_path: str = "media.db"): + self.db_path = Path(db_path) + self._init_db() + + def _init_db(self): + """Initialize database schema.""" + with self._get_connection() as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS posts ( + id TEXT PRIMARY KEY, + subreddit TEXT NOT NULL, + author TEXT, + title TEXT, + url TEXT NOT NULL, + media_url TEXT, + media_type TEXT, + score INTEGER DEFAULT 0, + created_utc REAL, + downloaded_at TIMESTAMP, + local_path TEXT, + file_hash TEXT, + permalink TEXT, + source_type TEXT, + flair TEXT + ) + """) + # Favorites table + conn.execute(""" + CREATE TABLE IF NOT EXISTS favorites ( + post_id TEXT PRIMARY KEY, + favorited_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (post_id) REFERENCES posts(id) + ) + """) + # Add new columns if they don't exist (migration) + try: + conn.execute("ALTER TABLE posts ADD COLUMN permalink TEXT") + except sqlite3.OperationalError: + pass + try: + conn.execute("ALTER TABLE posts ADD COLUMN source_type TEXT") + except sqlite3.OperationalError: + pass + try: + conn.execute("ALTER TABLE posts ADD COLUMN flair TEXT") + except sqlite3.OperationalError: + pass + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_subreddit ON posts(subreddit)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_file_hash ON posts(file_hash)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_downloaded ON posts(downloaded_at)" + ) + # Scheduler history table + conn.execute(""" + CREATE TABLE IF NOT EXISTS scheduler_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + started_at TIMESTAMP, + finished_at TIMESTAMP, + status TEXT, + posts_processed INTEGER DEFAULT 0, + posts_downloaded INTEGER DEFAULT 0, + error_message TEXT + ) + """) + conn.commit() + + @contextmanager + def _get_connection(self): + """Context manager for database connections.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + def post_exists(self, post_id: str) -> bool: + """Check if a post has already been processed.""" + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT 1 FROM posts WHERE id = ?", (post_id,) + ) + return cursor.fetchone() is not None + + def hash_exists(self, file_hash: str) -> Optional[str]: + """Check if a file with this hash already exists. Returns local_path if found.""" + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT local_path FROM posts WHERE file_hash = ?", (file_hash,) + ) + row = cursor.fetchone() + return row["local_path"] if row else None + + def add_post(self, post: PostRecord) -> None: + """Add or update a post record.""" + with self._get_connection() as conn: + conn.execute( + """ + INSERT OR REPLACE INTO posts + (id, subreddit, author, title, url, media_url, media_type, + score, created_utc, downloaded_at, local_path, file_hash, + permalink, source_type, flair) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + post.id, + post.subreddit, + post.author, + post.title, + post.url, + post.media_url, + post.media_type, + post.score, + post.created_utc, + post.downloaded_at, + post.local_path, + post.file_hash, + post.permalink, + post.source_type, + post.flair, + ), + ) + conn.commit() + + def mark_downloaded( + self, post_id: str, local_path: str, file_hash: str + ) -> None: + """Mark a post as downloaded with file info.""" + with self._get_connection() as conn: + conn.execute( + """ + UPDATE posts + SET downloaded_at = ?, local_path = ?, file_hash = ? + WHERE id = ? + """, + (datetime.now(), local_path, file_hash, post_id), + ) + conn.commit() + + def update_media_type(self, post_id: str, media_type: str) -> None: + """Update the media_type for a post.""" + with self._get_connection() as conn: + conn.execute( + "UPDATE posts SET media_type = ? WHERE id = ?", + (media_type, post_id), + ) + conn.commit() + + def get_post(self, post_id: str) -> Optional[PostRecord]: + """Get a post record by ID.""" + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT * FROM posts WHERE id = ?", (post_id,) + ) + row = cursor.fetchone() + if not row: + return None + return PostRecord( + id=row["id"], + subreddit=row["subreddit"], + author=row["author"], + title=row["title"], + url=row["url"], + media_url=row["media_url"], + media_type=row["media_type"], + score=row["score"], + created_utc=row["created_utc"], + downloaded_at=row["downloaded_at"], + local_path=row["local_path"], + file_hash=row["file_hash"], + permalink=row["permalink"] if "permalink" in row.keys() else None, + source_type=row["source_type"] if "source_type" in row.keys() else None, + flair=row["flair"] if "flair" in row.keys() else None, + ) + + def get_all_downloaded(self) -> list[PostRecord]: + """Get all downloaded posts for migration.""" + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT * FROM posts WHERE downloaded_at IS NOT NULL" + ) + posts = [] + for row in cursor.fetchall(): + posts.append(PostRecord( + id=row["id"], + subreddit=row["subreddit"], + author=row["author"], + title=row["title"], + url=row["url"], + media_url=row["media_url"], + media_type=row["media_type"], + score=row["score"], + created_utc=row["created_utc"], + downloaded_at=row["downloaded_at"], + local_path=row["local_path"], + file_hash=row["file_hash"], + permalink=row["permalink"] if "permalink" in row.keys() else None, + source_type=row["source_type"] if "source_type" in row.keys() else None, + flair=row["flair"] if "flair" in row.keys() else None, + )) + return posts + + def update_local_path(self, post_id: str, new_path: str) -> None: + """Update local_path for a post (used in migration).""" + with self._get_connection() as conn: + conn.execute( + "UPDATE posts SET local_path = ? WHERE id = ?", + (new_path, post_id) + ) + conn.commit() + + def get_stats(self) -> dict: + """Get collection statistics.""" + with self._get_connection() as conn: + total = conn.execute("SELECT COUNT(*) FROM posts").fetchone()[0] + downloaded = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at IS NOT NULL" + ).fetchone()[0] + + # Group by source: subreddits as "r/name", users as "u/name" + by_source = {} + + # Downloads from subreddits + subreddit_counts = conn.execute( + """ + SELECT subreddit, COUNT(*) + FROM posts + WHERE downloaded_at IS NOT NULL + AND (source_type = 'subreddit' OR source_type IS NULL) + GROUP BY subreddit + """ + ).fetchall() + for name, count in subreddit_counts: + by_source[f"r/{name}"] = count + + # Downloads from users (group by author) + user_counts = conn.execute( + """ + SELECT author, COUNT(*) + FROM posts + WHERE downloaded_at IS NOT NULL + AND source_type = 'user' + AND author IS NOT NULL + GROUP BY author + """ + ).fetchall() + for name, count in user_counts: + by_source[f"u/{name}"] = count + + by_type = dict( + conn.execute( + """ + SELECT media_type, COUNT(*) + FROM posts + WHERE downloaded_at IS NOT NULL AND media_type IS NOT NULL + GROUP BY media_type + """ + ).fetchall() + ) + + return { + "total_posts": total, + "downloaded": downloaded, + "by_source": by_source, + "by_type": by_type, + } + + def get_enhanced_stats(self) -> dict: + """Get enhanced statistics for dashboard.""" + from datetime import datetime, timedelta + + with self._get_connection() as conn: + now = datetime.now() + today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) + week_start = today_start - timedelta(days=7) + month_start = today_start - timedelta(days=30) + + # Format dates for SQLite comparison (space separator, not 'T') + today_str = today_start.strftime('%Y-%m-%d %H:%M:%S') + week_str = week_start.strftime('%Y-%m-%d %H:%M:%S') + month_str = month_start.strftime('%Y-%m-%d %H:%M:%S') + + # Downloads by period + downloads_today = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at >= ?", + (today_str,) + ).fetchone()[0] + + downloads_week = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at >= ?", + (week_str,) + ).fetchone()[0] + + downloads_month = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at >= ?", + (month_str,) + ).fetchone()[0] + + # Average score + avg_score = conn.execute( + "SELECT AVG(score) FROM posts WHERE downloaded_at IS NOT NULL AND score IS NOT NULL" + ).fetchone()[0] or 0 + + # Unique authors + unique_authors = conn.execute( + "SELECT COUNT(DISTINCT author) FROM posts WHERE downloaded_at IS NOT NULL AND author IS NOT NULL" + ).fetchone()[0] + + # Favorites count + favorites_count = conn.execute( + "SELECT COUNT(*) FROM favorites" + ).fetchone()[0] + + # Last download + last_download = conn.execute( + "SELECT downloaded_at FROM posts WHERE downloaded_at IS NOT NULL ORDER BY downloaded_at DESC LIMIT 1" + ).fetchone() + last_download = last_download[0] if last_download else None + + # First download (to calculate avg/day) + first_download = conn.execute( + "SELECT downloaded_at FROM posts WHERE downloaded_at IS NOT NULL ORDER BY downloaded_at ASC LIMIT 1" + ).fetchone() + + total_downloaded = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at IS NOT NULL" + ).fetchone()[0] + + # Calculate average per day + avg_per_day = 0 + if first_download and first_download[0]: + try: + first_date = datetime.fromisoformat(first_download[0].replace('Z', '+00:00')) + days_active = max((now - first_date.replace(tzinfo=None)).days, 1) + avg_per_day = round(total_downloaded / days_active, 1) + except: + avg_per_day = 0 + + # Top 10 authors + top_authors = conn.execute( + """ + SELECT author, COUNT(*) as count + FROM posts + WHERE downloaded_at IS NOT NULL AND author IS NOT NULL AND author != 'deleted' AND author != '[deleted]' + GROUP BY author + ORDER BY count DESC + LIMIT 10 + """ + ).fetchall() + + # Downloads trend (last 14 days) + trend_data = [] + for i in range(13, -1, -1): + day = today_start - timedelta(days=i) + day_end = day + timedelta(days=1) + day_str = day.strftime('%Y-%m-%d %H:%M:%S') + day_end_str = day_end.strftime('%Y-%m-%d %H:%M:%S') + count = conn.execute( + "SELECT COUNT(*) FROM posts WHERE downloaded_at >= ? AND downloaded_at < ?", + (day_str, day_end_str) + ).fetchone()[0] + trend_data.append({ + "date": day.strftime("%m/%d"), + "count": count + }) + + return { + "downloads_today": downloads_today, + "downloads_week": downloads_week, + "downloads_month": downloads_month, + "avg_score": round(avg_score, 1), + "unique_authors": unique_authors, + "favorites_count": favorites_count, + "last_download": last_download, + "avg_per_day": avg_per_day, + "top_authors": [{"author": a, "count": c} for a, c in top_authors], + "trend": trend_data + } + + def get_recent_downloads(self, limit: int = 10) -> list[dict]: + """Get most recent downloads.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT id, subreddit, author, title, media_type, score, + local_path, downloaded_at, permalink + FROM posts + WHERE downloaded_at IS NOT NULL AND local_path IS NOT NULL + ORDER BY downloaded_at DESC + LIMIT ? + """, + (limit,) + ) + return [dict(row) for row in cursor.fetchall()] + + def get_media_files(self, limit: int = 50, offset: int = 0, + subreddit: str = None, media_type: str = None, + sort: str = "newest", author: str = None) -> list[dict]: + """Get media files with optional filtering and sorting. + + Args: + sort: 'newest' (default), 'oldest', 'score_high', 'score_low' + """ + with self._get_connection() as conn: + query = """ + SELECT id, subreddit, author, title, media_type, score, + local_path, downloaded_at, permalink, created_utc + FROM posts + WHERE downloaded_at IS NOT NULL AND local_path IS NOT NULL + """ + params = [] + + if subreddit: + query += " AND LOWER(subreddit) = LOWER(?)" + params.append(subreddit) + + if media_type: + query += " AND media_type = ?" + params.append(media_type) + + if author: + query += " AND LOWER(author) = LOWER(?)" + params.append(author) + + # Apply sorting + if sort == "oldest": + query += " ORDER BY created_utc ASC" + elif sort == "score_high": + query += " ORDER BY score DESC" + elif sort == "score_low": + query += " ORDER BY score ASC" + else: # newest (default) + query += " ORDER BY created_utc DESC" + + query += " LIMIT ? OFFSET ?" + params.extend([limit, offset]) + + cursor = conn.execute(query, params) + return [dict(row) for row in cursor.fetchall()] + + def get_total_media_count(self, subreddit: str = None, media_type: str = None, author: str = None) -> int: + """Get total count of media files with optional filtering.""" + with self._get_connection() as conn: + query = """ + SELECT COUNT(*) + FROM posts + WHERE downloaded_at IS NOT NULL AND local_path IS NOT NULL + """ + params = [] + + if subreddit: + query += " AND LOWER(subreddit) = LOWER(?)" + params.append(subreddit) + + if media_type: + query += " AND media_type = ?" + params.append(media_type) + + if author: + query += " AND LOWER(author) = LOWER(?)" + params.append(author) + + return conn.execute(query, params).fetchone()[0] + + def get_all_subreddits(self) -> list[str]: + """Get list of all subreddits with downloaded content.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT DISTINCT subreddit + FROM posts + WHERE downloaded_at IS NOT NULL + ORDER BY subreddit + """ + ) + return [row[0] for row in cursor.fetchall()] + + def get_all_authors(self) -> list[str]: + """Get list of all authors with downloaded content.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT DISTINCT author + FROM posts + WHERE downloaded_at IS NOT NULL + AND author IS NOT NULL + AND author != '' + AND author != 'deleted' + ORDER BY author + """ + ) + return [row[0] for row in cursor.fetchall()] + + def get_posts_by_authors(self, authors: list[str]) -> list[PostRecord]: + """Get all downloaded posts from specific authors (case-insensitive).""" + if not authors: + return [] + + with self._get_connection() as conn: + # Create placeholders for IN clause + placeholders = ','.join('?' * len(authors)) + # Convert authors to lowercase for case-insensitive matching + authors_lower = [a.lower() for a in authors] + + cursor = conn.execute( + f""" + SELECT * FROM posts + WHERE LOWER(author) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """, + authors_lower + ) + + posts = [] + for row in cursor.fetchall(): + posts.append(PostRecord( + id=row["id"], + subreddit=row["subreddit"], + author=row["author"], + title=row["title"], + url=row["url"], + media_url=row["media_url"], + media_type=row["media_type"], + score=row["score"], + created_utc=row["created_utc"], + downloaded_at=row["downloaded_at"], + local_path=row["local_path"], + file_hash=row["file_hash"], + permalink=row["permalink"] if "permalink" in row.keys() else None, + source_type=row["source_type"] if "source_type" in row.keys() else None, + flair=row["flair"] if "flair" in row.keys() else None, + )) + return posts + + def count_posts_by_authors(self, authors: list[str]) -> int: + """Count downloaded posts from specific authors (case-insensitive).""" + if not authors: + return 0 + + with self._get_connection() as conn: + placeholders = ','.join('?' * len(authors)) + authors_lower = [a.lower() for a in authors] + + cursor = conn.execute( + f""" + SELECT COUNT(*) FROM posts + WHERE LOWER(author) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """, + authors_lower + ) + return cursor.fetchone()[0] + + def get_posts_by_subreddits(self, subreddits: list[str]) -> list[PostRecord]: + """Get all downloaded posts from specific subreddits (case-insensitive).""" + if not subreddits: + return [] + + with self._get_connection() as conn: + placeholders = ','.join('?' * len(subreddits)) + subreddits_lower = [s.lower() for s in subreddits] + + cursor = conn.execute( + f""" + SELECT * FROM posts + WHERE LOWER(subreddit) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """, + subreddits_lower + ) + + posts = [] + for row in cursor.fetchall(): + posts.append(PostRecord( + id=row["id"], + subreddit=row["subreddit"], + author=row["author"], + title=row["title"], + url=row["url"], + media_url=row["media_url"], + media_type=row["media_type"], + score=row["score"], + created_utc=row["created_utc"], + downloaded_at=row["downloaded_at"], + local_path=row["local_path"], + file_hash=row["file_hash"], + permalink=row["permalink"] if "permalink" in row.keys() else None, + source_type=row["source_type"] if "source_type" in row.keys() else None, + flair=row["flair"] if "flair" in row.keys() else None, + )) + return posts + + def count_posts_by_subreddits(self, subreddits: list[str]) -> int: + """Count downloaded posts from specific subreddits (case-insensitive).""" + if not subreddits: + return 0 + + with self._get_connection() as conn: + placeholders = ','.join('?' * len(subreddits)) + subreddits_lower = [s.lower() for s in subreddits] + + cursor = conn.execute( + f""" + SELECT COUNT(*) FROM posts + WHERE LOWER(subreddit) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """, + subreddits_lower + ) + return cursor.fetchone()[0] + + # Favorites methods + + def add_favorite(self, post_id: str) -> bool: + """Add a post to favorites. Returns True if added, False if already exists.""" + with self._get_connection() as conn: + try: + conn.execute( + "INSERT INTO favorites (post_id) VALUES (?)", + (post_id,) + ) + conn.commit() + return True + except sqlite3.IntegrityError: + return False + + def remove_favorite(self, post_id: str) -> bool: + """Remove a post from favorites. Returns True if removed.""" + with self._get_connection() as conn: + cursor = conn.execute( + "DELETE FROM favorites WHERE post_id = ?", + (post_id,) + ) + conn.commit() + return cursor.rowcount > 0 + + def is_favorite(self, post_id: str) -> bool: + """Check if a post is favorited.""" + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT 1 FROM favorites WHERE post_id = ?", + (post_id,) + ) + return cursor.fetchone() is not None + + def get_favorites(self, limit: int = 50, offset: int = 0) -> list[dict]: + """Get all favorited posts with their info.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT p.id, p.subreddit, p.author, p.title, p.media_type, + p.score, p.local_path, p.downloaded_at, p.permalink, + p.created_utc, f.favorited_at + FROM favorites f + JOIN posts p ON f.post_id = p.id + ORDER BY f.favorited_at DESC + LIMIT ? OFFSET ? + """, + (limit, offset) + ) + return [dict(row) for row in cursor.fetchall()] + + def count_favorites(self) -> int: + """Count total favorites.""" + with self._get_connection() as conn: + cursor = conn.execute("SELECT COUNT(*) FROM favorites") + return cursor.fetchone()[0] + + def get_favorite_authors(self) -> list[str]: + """Get unique authors from favorited posts.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT DISTINCT p.author + FROM favorites f + JOIN posts p ON f.post_id = p.id + WHERE p.author IS NOT NULL + AND p.author != '[deleted]' + AND p.author != 'AutoModerator' + ORDER BY p.author + """ + ) + return [row[0] for row in cursor.fetchall()] + + def get_media_by_authors( + self, + authors: list[str], + limit: int = 50, + offset: int = 0, + subreddit: str = None, + media_type: str = None, + sort: str = "newest" + ) -> list[dict]: + """Get media files from specific authors.""" + if not authors: + return [] + + with self._get_connection() as conn: + placeholders = ','.join('?' * len(authors)) + authors_lower = [a.lower() for a in authors] + + query = f""" + SELECT id, subreddit, author, title, media_type, score, + local_path, downloaded_at, permalink, created_utc + FROM posts + WHERE LOWER(author) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """ + params = list(authors_lower) + + if subreddit: + query += " AND LOWER(subreddit) = ?" + params.append(subreddit.lower()) + + if media_type: + query += " AND media_type = ?" + params.append(media_type) + + # Add sorting + if sort == "oldest": + query += " ORDER BY created_utc ASC" + elif sort == "score_high": + query += " ORDER BY score DESC" + elif sort == "score_low": + query += " ORDER BY score ASC" + else: # newest + query += " ORDER BY created_utc DESC" + + query += " LIMIT ? OFFSET ?" + params.extend([limit, offset]) + + cursor = conn.execute(query, params) + return [dict(row) for row in cursor.fetchall()] + + def count_media_by_authors( + self, + authors: list[str], + subreddit: str = None, + media_type: str = None + ) -> int: + """Count media files from specific authors.""" + if not authors: + return 0 + + with self._get_connection() as conn: + placeholders = ','.join('?' * len(authors)) + authors_lower = [a.lower() for a in authors] + + query = f""" + SELECT COUNT(*) FROM posts + WHERE LOWER(author) IN ({placeholders}) + AND local_path IS NOT NULL + AND downloaded_at IS NOT NULL + """ + params = list(authors_lower) + + if subreddit: + query += " AND LOWER(subreddit) = ?" + params.append(subreddit.lower()) + + if media_type: + query += " AND media_type = ?" + params.append(media_type) + + cursor = conn.execute(query, params) + return cursor.fetchone()[0] + + def get_authors_with_stats( + self, + limit: int = 50, + offset: int = 0, + favorites_only: bool = False, + sort: str = "count" + ) -> list[dict]: + """Get list of authors with their media counts and a sample thumbnail. + + Args: + sort: 'count' (most media), 'name' (alphabetical), 'recent' (most recent) + """ + with self._get_connection() as conn: + # Build the query based on filters + if favorites_only: + # Only authors who have favorited posts + query = """ + SELECT + p.author, + COUNT(*) as media_count, + MAX(p.score) as max_score, + SUM(p.score) as total_score, + MAX(p.created_utc) as latest_post, + (SELECT local_path FROM posts p2 + WHERE p2.author = p.author + AND p2.local_path IS NOT NULL + ORDER BY p2.score DESC LIMIT 1) as thumb_path + FROM posts p + WHERE p.author IS NOT NULL + AND p.author != '[deleted]' + AND p.author != 'deleted' + AND p.local_path IS NOT NULL + AND p.downloaded_at IS NOT NULL + AND EXISTS ( + SELECT 1 FROM favorites f + JOIN posts fp ON f.post_id = fp.id + WHERE fp.author = p.author + ) + GROUP BY p.author + """ + else: + query = """ + SELECT + p.author, + COUNT(*) as media_count, + MAX(p.score) as max_score, + SUM(p.score) as total_score, + MAX(p.created_utc) as latest_post, + (SELECT local_path FROM posts p2 + WHERE p2.author = p.author + AND p2.local_path IS NOT NULL + ORDER BY p2.score DESC LIMIT 1) as thumb_path + FROM posts p + WHERE p.author IS NOT NULL + AND p.author != '[deleted]' + AND p.author != 'deleted' + AND p.local_path IS NOT NULL + AND p.downloaded_at IS NOT NULL + GROUP BY p.author + """ + + # Add sorting + if sort == "name": + query += " ORDER BY LOWER(p.author) ASC" + elif sort == "recent": + query += " ORDER BY latest_post DESC" + else: # count + query += " ORDER BY media_count DESC" + + query += " LIMIT ? OFFSET ?" + + cursor = conn.execute(query, (limit, offset)) + rows = cursor.fetchall() + + # Check if each author has favorited posts + authors = [] + for row in rows: + author_name = row[0] + # Check if author has any favorited posts + fav_cursor = conn.execute(""" + SELECT COUNT(*) FROM favorites f + JOIN posts p ON f.post_id = p.id + WHERE p.author = ? + """, (author_name,)) + fav_count = fav_cursor.fetchone()[0] + + authors.append({ + "author": author_name, + "media_count": row[1], + "max_score": row[2], + "total_score": row[3], + "latest_post": row[4], + "thumb_path": row[5], + "is_favorite": fav_count > 0 + }) + + return authors + + def count_authors(self, favorites_only: bool = False) -> int: + """Count total unique authors.""" + with self._get_connection() as conn: + if favorites_only: + query = """ + SELECT COUNT(DISTINCT p.author) + FROM posts p + WHERE p.author IS NOT NULL + AND p.author != '[deleted]' + AND p.author != 'deleted' + AND p.local_path IS NOT NULL + AND EXISTS ( + SELECT 1 FROM favorites f + JOIN posts fp ON f.post_id = fp.id + WHERE fp.author = p.author + ) + """ + else: + query = """ + SELECT COUNT(DISTINCT author) + FROM posts + WHERE author IS NOT NULL + AND author != '[deleted]' + AND author != 'deleted' + AND local_path IS NOT NULL + """ + cursor = conn.execute(query) + return cursor.fetchone()[0] + + # Scheduler history methods + + def add_scheduler_run(self, started_at: datetime) -> int: + """Start a new scheduler run. Returns the run ID.""" + with self._get_connection() as conn: + cursor = conn.execute( + "INSERT INTO scheduler_history (started_at, status) VALUES (?, 'running')", + (started_at,) + ) + conn.commit() + return cursor.lastrowid + + def finish_scheduler_run( + self, + run_id: int, + status: str, + posts_processed: int = 0, + posts_downloaded: int = 0, + error_message: str = None + ) -> None: + """Finish a scheduler run with results.""" + with self._get_connection() as conn: + conn.execute( + """ + UPDATE scheduler_history + SET finished_at = ?, status = ?, posts_processed = ?, + posts_downloaded = ?, error_message = ? + WHERE id = ? + """, + (datetime.now(), status, posts_processed, posts_downloaded, error_message, run_id) + ) + conn.commit() + + def get_scheduler_history(self, limit: int = 20) -> list[dict]: + """Get recent scheduler run history.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT id, started_at, finished_at, status, + posts_processed, posts_downloaded, error_message + FROM scheduler_history + ORDER BY started_at DESC + LIMIT ? + """, + (limit,) + ) + return [dict(row) for row in cursor.fetchall()] + + def get_last_scheduler_run(self) -> Optional[dict]: + """Get the most recent scheduler run.""" + with self._get_connection() as conn: + cursor = conn.execute( + """ + SELECT id, started_at, finished_at, status, + posts_processed, posts_downloaded, error_message + FROM scheduler_history + ORDER BY started_at DESC + LIMIT 1 + """ + ) + row = cursor.fetchone() + return dict(row) if row else None