Merge pull request 'Add per-repo cover image support for RSS feed items' (#1) from feature/rss-cover-image into master

Reviewed-on: #1

src/cli.py

@@ -301,6 +301,44 @@ def cmd_check_for_daemon(db_path: str, since: str = None) -> None:
         conn.close()
 
 
+def cmd_image_add(args: argparse.Namespace, db_path: str) -> None:
+    """Add an image URL for a repository."""
+    owner_repo = args.repo
+    if not REPO_PATTERN.match(owner_repo):
+        error(f"Invalid repo format '{owner_repo}'. Expected 'owner/repo'")
+
+    image_url = args.url
+    if not re.match(r"^https?://", image_url):
+        error("Image URL must start with http:// or https://")
+
+    conn = get_connection(db_path)
+    try:
+        from src.db import set_repo_image
+        if set_repo_image(conn, owner_repo, image_url):
+            print(f"Set image for {owner_repo}: {image_url}")
+        else:
+            error(f"Repository '{owner_repo}' not found")
+    finally:
+        conn.close()
+
+
+def cmd_image_remove(args: argparse.Namespace, db_path: str) -> None:
+    """Remove the image URL for a repository."""
+    owner_repo = args.repo
+    if not REPO_PATTERN.match(owner_repo):
+        error(f"Invalid repo format '{owner_repo}'. Expected 'owner/repo'")
+
+    conn = get_connection(db_path)
+    try:
+        from src.db import remove_repo_image
+        if remove_repo_image(conn, owner_repo):
+            print(f"Removed image for {owner_repo}")
+        else:
+            error(f"No image set for '{owner_repo}' (or repository not found)")
+    finally:
+        conn.close()
+
+
 def cmd_serve(args: argparse.Namespace, db_path: str) -> None:
     """Start the HTTP server."""
     host = args.host or DEFAULT_SERVER_HOST
@@ -367,6 +405,17 @@ def build_parser() -> argparse.ArgumentParser:
     p_serve.add_argument("--host", default=DEFAULT_SERVER_HOST, help="Server host")
     p_serve.add_argument("--base-url", default=None, help="Base URL for feed links. Port is auto-appended (e.g., http://192.168.1.100)")
 
+    # image
+    p_image = subparsers.add_parser("image", help="Manage repository cover images")
+    image_sub = p_image.add_subparsers(dest="image_command")
+
+    p_image_add = image_sub.add_parser("add", help="Set a cover image for a repository")
+    p_image_add.add_argument("repo", help="Repository in 'owner/repo' format")
+    p_image_add.add_argument("url", help="Image URL (http:// or https://)")
+
+    p_image_remove = image_sub.add_parser("remove", help="Remove the cover image for a repository")
+    p_image_remove.add_argument("repo", help="Repository in 'owner/repo' format")
+
     # daemon
     p_daemon = subparsers.add_parser("daemon", help="Run as background daemon")
     p_daemon.add_argument("--interval", default="30m",
@@ -399,6 +448,23 @@ def main() -> None:
         "daemon": cmd_daemon,
     }
 
+    # Handle image subcommand
+    if args.command == "image":
+        image_commands = {
+            "add": cmd_image_add,
+            "remove": cmd_image_remove,
+        }
+        if not getattr(args, "image_command", None):
+            parser.parse_args(["image", "--help"])
+            sys.exit(2)
+        cmd = image_commands.get(args.image_command)
+        if cmd:
+            cmd(args, db_path)
+        else:
+            parser.parse_args(["image", "--help"])
+            sys.exit(2)
+        return
+
     cmd = commands.get(args.command)
     if cmd:
         cmd(args, db_path)

src/db.py

@@ -31,6 +31,7 @@ def _create_tables(conn: sqlite3.Connection) -> None:
             id INTEGER PRIMARY KEY AUTOINCREMENT,
             owner_repo TEXT NOT NULL UNIQUE,
             mode TEXT NOT NULL DEFAULT 'release' CHECK(mode IN ('release', 'tag')),
+            image_url TEXT DEFAULT NULL,
             created_at DATETIME DEFAULT CURRENT_TIMESTAMP
         );
 
@@ -48,9 +49,20 @@ def _create_tables(conn: sqlite3.Connection) -> None:
             UNIQUE(repo_id, tag_name)
         );
     """)
+    _migrate_add_image_url(conn)
     conn.commit()
 
 
+def _migrate_add_image_url(conn: sqlite3.Connection) -> None:
+    """Add image_url column to repos if it doesn't exist (migration)."""
+    cursor = conn.execute(
+        "PRAGMA table_info(repos)"
+    ).fetchall()
+    column_names = [col[1] for col in cursor]
+    if "image_url" not in column_names:
+        conn.execute("ALTER TABLE repos ADD COLUMN image_url TEXT DEFAULT NULL")
+
+
 # ── Repos ──
 
 
@@ -114,6 +126,25 @@ def get_repos_with_mode(conn: sqlite3.Connection) -> list[tuple[int, str, str]]:
     return conn.execute("SELECT id, owner_repo, mode FROM repos").fetchall()
 
 
+def set_repo_image(conn: sqlite3.Connection, owner_repo: str, image_url: str) -> bool:
+    """Set an image URL for a repo. Returns True if a row was updated."""
+    cursor = conn.execute(
+        "UPDATE repos SET image_url = ? WHERE owner_repo = ?", (image_url, owner_repo)
+    )
+    conn.commit()
+    return cursor.rowcount > 0
+
+
+def remove_repo_image(conn: sqlite3.Connection, owner_repo: str) -> bool:
+    """Remove the image URL for a repo. Returns True if a row was updated."""
+    cursor = conn.execute(
+        "UPDATE repos SET image_url = NULL WHERE owner_repo = ? AND image_url IS NOT NULL",
+        (owner_repo,),
+    )
+    conn.commit()
+    return cursor.rowcount > 0
+
+
 # ── Entries ──
 
 
@@ -149,7 +180,7 @@ def get_all_entries(conn: sqlite3.Connection, limit: int = 50) -> list[dict]:
     """Get all entries sorted by published_at descending, capped at *limit*."""
     rows = conn.execute("""
         SELECT e.tag_name, e.title, e.body, e.published_at, e.html_url,
-               e.kind, r.owner_repo
+               e.kind, r.owner_repo, r.image_url
         FROM entries e
         JOIN repos r ON r.id = e.repo_id
         ORDER BY e.published_at DESC
@@ -164,6 +195,7 @@ def get_all_entries(conn: sqlite3.Connection, limit: int = 50) -> list[dict]:
             "html_url": row[4],
             "kind": row[5],
             "owner_repo": row[6],
+            "image_url": row[7],
         }
         for row in rows
     ]

src/rss.py

@@ -9,6 +9,8 @@ import markdown
 
 from src.config import RSS_MAX_ITEMS
 
+ET.register_namespace("media", "http://search.yahoo.com/mrss/")
+
 
 def strip_markdown(text: str) -> str:
     """Convert markdown to HTML, then strip tags to produce plain text."""
@@ -29,6 +31,7 @@ def generate_feed(entries: list[dict], base_url: str = "http://127.0.0.1:8080")
     """Generate an RSS 2.0 XML feed from a list of entry dicts.
 
     Each entry should have: tag_name, title, body, published_at, html_url, owner_repo.
+    Optionally: image_url (for Media RSS cover image).
     """
     root = ET.Element("rss")
     root.set("version", "2.0")
@@ -62,6 +65,12 @@ def generate_feed(entries: list[dict], base_url: str = "http://127.0.0.1:8080")
             f"github.com/{owner_repo}/releases/tag/{entry['tag_name']}"
         )
 
+        image_url = entry.get("image_url")
+        if image_url:
+            media_content = ET.SubElement(item, "{http://search.yahoo.com/mrss/}content")
+            media_content.set("url", image_url)
+            media_content.set("medium", "image")
+
     return ET.tostring(root, encoding="unicode", xml_declaration=False)
 
 

tests/test_cli.py

@@ -130,6 +130,89 @@ class TestCliDaemon:
         cmd_check_for_daemon(db_path)  # Should not raise
 
 
+class TestCliImage:
+    def _setup_db(self, tmp_path):
+        db_path = str(tmp_path / "test.db")
+        conn = sqlite3.connect(db_path)
+        _create_tables(conn)
+        add_repo(conn, "owner/repo")
+        conn.close()
+        return db_path
+
+    def test_image_add(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "add", "owner/repo", "https://example.com/logo.png"],
+            capture_output=True, text=True
+        )
+        assert result.returncode == 0
+        assert "Set image" in result.stdout
+
+    def test_image_add_invalid_url(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "add", "owner/repo", "not-a-url"],
+            capture_output=True, text=True
+        )
+        assert result.returncode != 0
+        assert "http:// or https://" in result.stderr
+
+    def test_image_add_invalid_repo(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "add", "invalid", "https://example.com/logo.png"],
+            capture_output=True, text=True
+        )
+        assert result.returncode != 0
+        assert "Invalid repo format" in result.stderr
+
+    def test_image_add_nonexistent_repo(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "add", "other/repo", "https://example.com/logo.png"],
+            capture_output=True, text=True
+        )
+        assert result.returncode != 0
+        assert "not found" in result.stderr
+
+    def test_image_remove(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "add", "owner/repo", "https://example.com/logo.png"],
+            capture_output=True, text=True
+        )
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "remove", "owner/repo"],
+            capture_output=True, text=True
+        )
+        assert result.returncode == 0
+        assert "Removed image" in result.stdout
+
+    def test_image_remove_no_image(self, tmp_path):
+        db_path = self._setup_db(tmp_path)
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "--db-path", db_path,
+             "image", "remove", "owner/repo"],
+            capture_output=True, text=True
+        )
+        assert result.returncode != 0
+        assert "No image set" in result.stderr
+
+    def test_image_help(self):
+        result = subprocess.run(
+            [sys.executable, "-m", "src.cli", "image", "--help"],
+            capture_output=True, text=True
+        )
+        assert result.returncode == 0
+        assert "cover image" in result.stdout.lower()
+
+
 class TestCliAddRemove:
     def test_add_invalid_format(self, tmp_path):
         db_path = str(tmp_path / "test.db")

tests/test_db.py

@@ -131,6 +131,81 @@ class TestGetAllEntries:
         assert len(entries) == 3
 
 
+class TestRepoImage:
+    def test_set_image(self, conn):
+        db.add_repo(conn, "owner/repo")
+        assert db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png") is True
+        row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
+        assert row[0] == "https://example.com/logo.png"
+
+    def test_set_image_nonexistent(self, conn):
+        assert db.set_repo_image(conn, "no/one", "https://example.com/logo.png") is False
+
+    def test_set_image_overwrites(self, conn):
+        db.add_repo(conn, "owner/repo")
+        db.set_repo_image(conn, "owner/repo", "https://example.com/old.png")
+        db.set_repo_image(conn, "owner/repo", "https://example.com/new.png")
+        row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
+        assert row[0] == "https://example.com/new.png"
+
+    def test_remove_image(self, conn):
+        db.add_repo(conn, "owner/repo")
+        db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png")
+        assert db.remove_repo_image(conn, "owner/repo") is True
+        row = conn.execute("SELECT image_url FROM repos WHERE owner_repo = 'owner/repo'").fetchone()
+        assert row[0] is None
+
+    def test_remove_image_no_image_set(self, conn):
+        db.add_repo(conn, "owner/repo")
+        assert db.remove_repo_image(conn, "owner/repo") is False
+
+    def test_image_url_in_entries(self, conn):
+        repo_id = db.add_repo(conn, "owner/repo")
+        db.set_repo_image(conn, "owner/repo", "https://example.com/logo.png")
+        db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
+        entries = db.get_all_entries(conn)
+        assert len(entries) == 1
+        assert entries[0]["image_url"] == "https://example.com/logo.png"
+
+    def test_image_url_none_by_default(self, conn):
+        repo_id = db.add_repo(conn, "owner/repo")
+        db.upsert_entry(conn, repo_id, "release", "v1", "v1", "", "2025-01-01", "http://x")
+        entries = db.get_all_entries(conn)
+        assert entries[0]["image_url"] is None
+
+    def test_migration_adds_column(self):
+        """Test that _migrate_add_image_url adds the column if missing."""
+        c = sqlite3.connect(":memory:")
+        c.execute("PRAGMA foreign_keys=ON")
+        c.execute("""
+            CREATE TABLE repos (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                owner_repo TEXT NOT NULL UNIQUE,
+                mode TEXT NOT NULL DEFAULT 'release'
+            )
+        """)
+        c.execute("""
+            CREATE TABLE entries (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                repo_id INTEGER NOT NULL,
+                kind TEXT NOT NULL CHECK(kind IN ('release', 'tag')),
+                tag_name TEXT NOT NULL,
+                title TEXT NOT NULL,
+                body TEXT DEFAULT '',
+                published_at DATETIME NOT NULL,
+                html_url TEXT NOT NULL,
+                fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (repo_id) REFERENCES repos(id) ON DELETE CASCADE,
+                UNIQUE(repo_id, tag_name)
+            )
+        """)
+        c.commit()
+        db._migrate_add_image_url(c)
+        columns = [col[1] for col in c.execute("PRAGMA table_info(repos)").fetchall()]
+        assert "image_url" in columns
+        c.close()
+
+
 class TestGetConnection:
     def test_creates_file(self):
         with tempfile.TemporaryDirectory() as tmp:

tests/test_rss.py

@@ -132,3 +132,53 @@ class TestGenerateFeed:
         xml_str = generate_feed(entries, base_url="http://example.com:9000")
         root = ET.fromstring(xml_str)
         assert root[0].find("link").text == "http://example.com:9000/"
+
+    def test_media_content_with_image(self):
+        entries = self._entries(1)
+        entries[0]["image_url"] = "https://example.com/logo.png"
+        xml_str = generate_feed(entries)
+        root = ET.fromstring(xml_str)
+        assert root is not None
+        media_el = root[0].find("item").find("{http://search.yahoo.com/mrss/}content")
+        assert media_el is not None
+        assert media_el.get("url") == "https://example.com/logo.png"
+        assert media_el.get("medium") == "image"
+        assert "media" in xml_str or "search.yahoo.com/mrss" in xml_str
+
+    def test_no_media_content_without_image(self):
+        entries = self._entries(1)
+        entries[0]["image_url"] = None
+        xml_str = generate_feed(entries)
+        root = ET.fromstring(xml_str)
+        media_el = root[0].find("item").find("{http://search.yahoo.com/mrss/}content")
+        assert media_el is None
+
+    def test_no_media_content_when_key_missing(self):
+        entries = self._entries(1)
+        entries[0]["image_url"] = None
+        xml_str = generate_feed(entries)
+        root = ET.fromstring(xml_str)
+        media_el = root[0].find("item").find("{http://search.yahoo.com/mrss/}content")
+        assert media_el is None
+
+    def test_media_content_multiple_items_selective(self):
+        entries = self._entries(2)
+        entries[0]["image_url"] = "https://example.com/logo1.png"
+        entries[1]["image_url"] = None
+        xml_str = generate_feed(entries)
+        root = ET.fromstring(xml_str)
+        items = root[0].findall("item")
+        media_1 = items[0].find("{http://search.yahoo.com/mrss/}content")
+        media_2 = items[1].find("{http://search.yahoo.com/mrss/}content")
+        assert media_1 is not None
+        assert media_1.get("url") == "https://example.com/logo1.png"
+        assert media_2 is None
+
+    def test_image_with_special_chars_in_url(self):
+        entries = self._entries(1)
+        entries[0]["image_url"] = "https://example.com/logo.png?w=200&h=200"
+        xml_str = generate_feed(entries)
+        root = ET.fromstring(xml_str)
+        assert root is not None
+        media_el = root[0].find("item").find("{http://search.yahoo.com/mrss/}content")
+        assert "w=200" in media_el.get("url")