feat(heartbeat): добавить Product Hunt как источник трендов

tools/heartbeat/analyzer.py

@@ -8,7 +8,7 @@
 You are a technology strategist for a personal AI corporation project.
 
 Given:
-1. Raw trend data from Hacker News, GitHub Trending, Reddit, and X
+1. Raw trend data from Hacker News, GitHub Trending, Reddit, Product Hunt, and X
 2. The project context (goals, stack, current status)
 
 Produce a concise Heartbeat digest in this format:
@@ -28,6 +28,7 @@
 - HN stories analyzed: X
 - GitHub repos analyzed: Y
 - Reddit posts analyzed: Z
+- Product Hunt launches analyzed: P
 - X trends analyzed: W
 
 Rules:

tools/heartbeat/config.py

@@ -48,6 +48,10 @@
 REDDIT_SUBREDDITS = ["MachineLearning", "artificial", "LocalLLaMA", "ChatGPT"]
 REDDIT_RESULTS_LIMIT = 20
 
+# Product Hunt
+PH_FEED_URL = "https://www.producthunt.com/feed"
+PH_RESULTS_LIMIT = 5
+
 # X (Twitter)
 X_RESULTS_LIMIT = 10
 

tools/heartbeat/formatter.py

@@ -4,14 +4,15 @@
 
 from beartype import beartype
 
-from sources import GitHubRepo, HNStory, RedditPost, XTrend
+from sources import GitHubRepo, HNStory, ProductHuntLaunch, RedditPost, XTrend
 
 
 @beartype
 def format_raw_digest(
     hn_stories: list[HNStory],
     github_repos: list[GitHubRepo],
     reddit_posts: list[RedditPost],
+    ph_launches: list[ProductHuntLaunch],
     x_trends: list[XTrend],
 ) -> str:
     lines: list[str] = [
@@ -57,6 +58,18 @@ def format_raw_digest(
         lines.append(f"   {post.url}")
         lines.append("")
 
+    lines.extend([
+        f"## Product Hunt Top Launches ({len(ph_launches)} found)",
+        "",
+    ])
+
+    for i, launch in enumerate(ph_launches, 1):
+        lines.append(f"{i}. **{launch.title}** (by {launch.author})")
+        if launch.description:
+            lines.append(f"   {launch.description}")
+        lines.append(f"   {launch.url}")
+        lines.append("")
+
     if x_trends:
         lines.extend([
             f"## X Trending Topics ({len(x_trends)} found)",

tools/heartbeat/main.py

@@ -27,7 +27,7 @@ def cmd_fetch() -> None:
     print("Fetching trends...", file=sys.stderr)
     data = fetch_all()
     raw_md = format_raw_digest(
-        data["hn"], data["github"], data["reddit"], data["x"]  # type: ignore[arg-type]
+        data["hn"], data["github"], data["reddit"], data["ph"], data["x"]  # type: ignore[arg-type]
     )
     print(raw_md)
 
@@ -41,7 +41,7 @@ def cmd_digest() -> None:
     print("Fetching trends...", file=sys.stderr)
     data = fetch_all()
     raw_md = format_raw_digest(
-        data["hn"], data["github"], data["reddit"], data["x"]  # type: ignore[arg-type]
+        data["hn"], data["github"], data["reddit"], data["ph"], data["x"]  # type: ignore[arg-type]
     )
 
     print("Analyzing with Claude...", file=sys.stderr)

tools/heartbeat/sources.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import json
+import re
 import urllib.parse
 import urllib.request
+import xml.etree.ElementTree as ET
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
@@ -21,6 +23,8 @@
     HN_FETCH_WORKERS,
     HN_RESULTS_LIMIT,
     HN_TOP_STORIES_LIMIT,
+    PH_FEED_URL,
+    PH_RESULTS_LIMIT,
     REDDIT_RESULTS_LIMIT,
     REDDIT_SUBREDDITS,
     X_RESULTS_LIMIT,
@@ -68,6 +72,17 @@ class XTrend:
     volume: int | None
 
 
+@dataclass
+class ProductHuntLaunch:
+    title: str
+    description: str
+    url: str
+    votes: int
+    comments: int
+    author: str
+    published_at: datetime
+
+
 @beartype
 def fetch_reddit_posts() -> list[RedditPost]:
     subreddits = "+".join(REDDIT_SUBREDDITS)
@@ -115,7 +130,7 @@ def fetch_x_trends() -> list[XTrend]:
 
 
 @beartype
-def _http_get_json(url: str) -> dict | list:  # type: ignore[type-arg]
+def _http_get(url: str) -> bytes:
     # Reddit and other APIs often block default Python/urllib User-Agents.
     # Using a common browser-like User-Agent to ensure better compatibility.
     req = urllib.request.Request(
@@ -129,7 +144,12 @@ def _http_get_json(url: str) -> dict | list:  # type: ignore[type-arg]
         },
     )
     with urllib.request.urlopen(req, timeout=15) as resp:
-        return json.loads(resp.read().decode())
+        return resp.read()
+
+
+@beartype
+def _http_get_json(url: str) -> dict | list:  # type: ignore[type-arg]
+    return json.loads(_http_get(url).decode())
 
 
 @beartype
@@ -182,6 +202,53 @@ def fetch_hn_stories() -> list[HNStory]:
     return stories[:HN_RESULTS_LIMIT]
 
 
+@beartype
+def fetch_product_hunt_launches() -> list[ProductHuntLaunch]:
+    try:
+        content = _http_get(PH_FEED_URL).decode()
+        root = ET.fromstring(content)
+    except Exception:
+        return []
+
+    # Atom namespace
+    ns = {"atom": "http://www.w3.org/2005/Atom"}
+    launches: list[ProductHuntLaunch] = []
+
+    for entry in root.findall("atom:entry", ns):
+        title = entry.findtext("atom:title", "", ns)
+        link_elem = entry.find("atom:link[@rel='alternate']", ns)
+        url = link_elem.get("href", "") if link_elem is not None else ""
+
+        content_html = entry.findtext("atom:content", "", ns)
+        # Extract description from the first <p>
+        desc_match = re.search(r"<p>(.*?)</p>", content_html, re.DOTALL)
+        description = desc_match.group(1).strip() if desc_match else ""
+
+        author_elem = entry.find("atom:author/atom:name", ns)
+        author = author_elem.text if author_elem is not None else "unknown"
+
+        published_str = entry.findtext("atom:published", "", ns)
+        try:
+            # Format: 2026-02-24T14:37:19-08:00
+            # Python's fromisoformat handles this in 3.11+
+            published_at = datetime.fromisoformat(published_str)
+        except Exception:
+            published_at = datetime.now(timezone.utc)
+
+        if _matches_keywords(title) or _matches_keywords(description):
+            launches.append(ProductHuntLaunch(
+                title=title,
+                description=description,
+                url=url,
+                votes=0,  # RSS doesn't provide votes
+                comments=0,
+                author=author,
+                published_at=published_at,
+            ))
+
+    return launches[:PH_RESULTS_LIMIT]
+
+
 @beartype
 def fetch_github_trending() -> list[GitHubRepo]:
     cutoff = (
@@ -227,10 +294,13 @@ def fetch_github_trending() -> list[GitHubRepo]:
 
 
 @beartype
-def fetch_all() -> dict[str, list[HNStory] | list[GitHubRepo] | list[RedditPost] | list[XTrend]]:
+def fetch_all() -> dict[str, (
+    list[HNStory] | list[GitHubRepo] | list[RedditPost] | list[XTrend] | list[ProductHuntLaunch]
+)]:
     return {
         "hn": fetch_hn_stories(),
         "github": fetch_github_trending(),
         "reddit": fetch_reddit_posts(),
+        "ph": fetch_product_hunt_launches(),
         "x": fetch_x_trends(),
     }

tools/heartbeat/tests/test_formatter.py

@@ -5,7 +5,7 @@
 from beartype import beartype
 
 from formatter import format_raw_digest
-from sources import GitHubRepo, HNStory, RedditPost, XTrend
+from sources import GitHubRepo, HNStory, ProductHuntLaunch, RedditPost, XTrend
 
 
 @beartype
@@ -47,8 +47,21 @@ def test_format_raw_digest() -> None:
     x_trends = [
         XTrend(name="Trend", url="http://x.com/trend", volume=1000)
     ]
+    ph_launches = [
+        ProductHuntLaunch(
+            title="PH Launch",
+            description="desc",
+            url="http://ph.com",
+            votes=100,
+            comments=10,
+            author="author",
+            published_at=datetime.now(timezone.utc),
+        )
+    ]
 
-    digest = format_raw_digest(hn_stories, github_repos, reddit_posts, x_trends)
+    digest = format_raw_digest(
+        hn_stories, github_repos, reddit_posts, ph_launches, x_trends
+    )
 
     assert "Heartbeat Raw Data" in digest
     assert "Hacker News Top Stories (1 relevant)" in digest
@@ -57,5 +70,7 @@ def test_format_raw_digest() -> None:
     assert "org/repo" in digest
     assert "Reddit Top Posts (1 found)" in digest
     assert "Reddit Post" in digest
+    assert "Product Hunt Top Launches (1 found)" in digest
+    assert "PH Launch" in digest
     assert "X Trending Topics (1 found)" in digest
     assert "Trend" in digest

tools/heartbeat/tests/test_main.py

@@ -36,7 +36,7 @@ def test_main_default(mock_fetch: MagicMock) -> None:
 @patch("sources.fetch_all")
 @patch("sys.stdout", new_callable=MagicMock)
 def test_cmd_fetch(mock_stdout: MagicMock, mock_fetch_all: MagicMock, mock_format: MagicMock) -> None:
-    mock_fetch_all.return_value = {"hn": [], "github": [], "reddit": [], "x": []}
+    mock_fetch_all.return_value = {"hn": [], "github": [], "reddit": [], "ph": [], "x": []}
     mock_format.return_value = "Mocked Raw Digest"
 
     cmd_fetch()
@@ -52,7 +52,7 @@ def test_cmd_fetch(mock_stdout: MagicMock, mock_fetch_all: MagicMock, mock_forma
 def test_cmd_digest(
     mock_stdout: MagicMock, mock_fetch_all: MagicMock, mock_format: MagicMock, mock_analyze: MagicMock
 ) -> None:
-    mock_fetch_all.return_value = {"hn": [], "github": [], "reddit": [], "x": []}
+    mock_fetch_all.return_value = {"hn": [], "github": [], "reddit": [], "ph": [], "x": []}
     mock_format.return_value = "Mocked Raw Digest"
     mock_analyze.return_value = "Mocked Digest Analysis"
 

tools/heartbeat/tests/test_sources.py

@@ -7,12 +7,15 @@
 from sources import (
     GitHubRepo,
     HNStory,
+    ProductHuntLaunch,
     RedditPost,
+    _http_get,
     _http_get_json,
     _matches_keywords,
     fetch_all,
     fetch_github_trending,
     fetch_hn_stories,
+    fetch_product_hunt_launches,
     fetch_reddit_posts,
 )
 
@@ -125,24 +128,74 @@ def test_fetch_reddit_posts_error(mock_get: MagicMock) -> None:
     assert posts == []
 
 
+@beartype
+@patch("sources._http_get")
+def test_fetch_product_hunt_launches(mock_get: MagicMock) -> None:
+    xml_content = """<?xml version="1.0" encoding="UTF-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <entry>
+    <title>AI Agent Tool</title>
+    <link rel="alternate" href="https://example.com/ai-agent"/>
+    <content type="html">&lt;p&gt;A cool AI agent for coding&lt;/p&gt;</content>
+    <author><name>John Doe</name></author>
+    <published>2026-02-25T12:00:00Z</published>
+  </entry>
+  <entry>
+    <title>Unrelated Product</title>
+    <link rel="alternate" href="https://example.com/unrelated"/>
+    <content type="html">&lt;p&gt;Nothing to do with AI&lt;/p&gt;</content>
+    <author><name>Jane Doe</name></author>
+    <published>2026-02-25T13:00:00Z</published>
+  </entry>
+</feed>
+"""
+    mock_get.return_value = xml_content.encode()
+
+    launches = fetch_product_hunt_launches()
+    # Only 1 launch matches keywords (AI agent)
+    assert len(launches) == 1
+    assert isinstance(launches[0], ProductHuntLaunch)
+    assert launches[0].title == "AI Agent Tool"
+    assert launches[0].author == "John Doe"
+
+
+@beartype
+@patch("sources._http_get")
+def test_fetch_product_hunt_launches_error(mock_get: MagicMock) -> None:
+    mock_get.side_effect = Exception("PH error")
+    launches = fetch_product_hunt_launches()
+    assert launches == []
+
+
 @beartype
 @patch("sources._http_get_json")
-def test_fetch_all(mock_get: MagicMock) -> None:
+@patch("sources.fetch_product_hunt_launches")
+def test_fetch_all(mock_ph: MagicMock, mock_get: MagicMock) -> None:
     mock_get.return_value = {}
+    mock_ph.return_value = []
     res = fetch_all()
     assert "hn" in res
     assert "github" in res
     assert "reddit" in res
+    assert "ph" in res
     assert "x" in res
 
 
 @beartype
 @patch("urllib.request.urlopen")
-def test_http_get_json(mock_urlopen: MagicMock) -> None:
+def test_http_get(mock_urlopen: MagicMock) -> None:
     mock_response = MagicMock()
-    mock_response.read.return_value = b'{"key": "value"}'
+    mock_response.read.return_value = b"raw data"
     mock_response.__enter__.return_value = mock_response
     mock_urlopen.return_value = mock_response
 
+    result = _http_get("http://example.com")
+    assert result == b"raw data"
+
+
+@beartype
+@patch("sources._http_get")
+def test_http_get_json(mock_get: MagicMock) -> None:
+    mock_get.return_value = b'{"key": "value"}'
     result = _http_get_json("http://example.com")
     assert result == {"key": "value"}