diff --git a/src/cli.py b/src/cli.py index 1c1a919..758bc2c 100644 --- a/src/cli.py +++ b/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) diff --git a/src/db.py b/src/db.py index c3cb3da..6fa3328 100644 --- a/src/db.py +++ b/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 ] diff --git a/src/rss.py b/src/rss.py index 8c75d4d..2e93ae5 100644 --- a/src/rss.py +++ b/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) diff --git a/tests/test_cli.py b/tests/test_cli.py index 1fc35dc..02e0f24 100644 --- a/tests/test_cli.py +++ b/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") diff --git a/tests/test_db.py b/tests/test_db.py index 46880a9..3ed3107 100644 --- a/tests/test_db.py +++ b/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: diff --git a/tests/test_rss.py b/tests/test_rss.py index 06407ef..a33a9d1 100644 --- a/tests/test_rss.py +++ b/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")