""" Hacker News Plugin for Funguy Bot Fetches top stories from Hacker News using the Firebase API. No API key required - completely free. Commands: !hn - Show top 5 stories (default) !hn top - Top stories !hn new - Newest stories !hn best - Best stories !hn ask - Ask HN threads !hn show - Show HN posts !hn job - Job postings !hn story - Get details of a specific story !hn comments - Show comments for a story !hn search - Search stories (via Algolia) """ import logging import aiohttp import re import asyncio from typing import Optional, Dict, Any, List from datetime import datetime, timezone # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- DEFAULT_STORIES = 5 MAX_STORIES = 15 # Story types STORY_TYPES = { "top": "topstories", "new": "newstories", "best": "beststories", "ask": "askstories", "show": "showstories", "job": "jobstories" } # --------------------------------------------------------------------------- # Helper Functions # --------------------------------------------------------------------------- def _format_collapsible(title: str, content: str, expanded: bool = False) -> str: """Format content in a collapsible details/summary block.""" open_attr = ' open' if expanded else '' return f"\n📰 {title}\n\n{content}\n\n" def _format_time(timestamp: int) -> str: """Format Unix timestamp to relative time string.""" if not timestamp: return "unknown time" try: dt = datetime.fromtimestamp(timestamp, tz=timezone.utc) now = datetime.now(timezone.utc) diff = now - dt if diff.days > 0: return f"{diff.days}d ago" elif diff.seconds > 3600: return f"{diff.seconds // 3600}h ago" else: return f"{diff.seconds // 60}m ago" except: return "unknown time" def _format_story(story: Dict, index: int, show_points: bool = True) -> str: """Format a story for display using HTML list elements.""" title = story.get("title", "No title") url = story.get("url", "#") score = story.get("score", 0) by = story.get("by", "unknown") descendants = story.get("descendants", 0) time_str = _format_time(story.get("time", 0)) story_id = story.get("id", 0) # Build story HTML story_html = f"
  • {index}. {title}
    " if url and url != "#": story_html += f"🔗 {url}
    " else: hn_url = f"https://news.ycombinator.com/item?id={story_id}" story_html += f"💬 Discussion on Hacker News
    " story_html += f"👤 {by} | 🕐 {time_str}" if show_points: story_html += f" | ⭐ {score} points | 💬 {descendants} comments" story_html += "
    • " return story_html def _format_comment(comment: Dict, depth: int = 0) -> str: """Format a comment for display.""" if comment.get("deleted") or comment.get("dead"): return None text = comment.get("text", "") if not text: return None text = re.sub(r'<[^>]+>', '', text) if len(text) > 300: text = text[:297] + "..." by = comment.get("by", "unknown") indent = " " * depth return f"{indent}
  • {by}: {text}
    • " def _format_search_result(hit: Dict, index: int) -> str: """Format a search result using HTML list elements.""" title = hit.get("title", "No title") url = hit.get("url", "#") points = hit.get("points", 0) author = hit.get("author", "unknown") comment_count = hit.get("num_comments", 0) story_id = hit.get("objectID") result_html = f"
  • {index}. {title}
    " if url and url != "#": result_html += f"🔗 {url}
    " else: hn_url = f"https://news.ycombinator.com/item?id={story_id}" result_html += f"💬 Discussion on Hacker News
    " result_html += f"👤 {author} | ⭐ {points} points | 💬 {comment_count} comments" result_html += "
    • " return result_html async def _fetch_story_ids(story_type: str) -> Optional[List[int]]: """Fetch story IDs from Hacker News API.""" if story_type not in STORY_TYPES.values(): return None url = f"https://hacker-news.firebaseio.com/v0/{story_type}.json" try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: data = await response.json() if data and isinstance(data, list): logging.info(f"Fetched {len(data)} story IDs for {story_type}") return data return None return None except Exception as e: logging.error(f"Error fetching story IDs: {e}") return None async def _fetch_item(item_id: int) -> Optional[Dict]: """Fetch a single item by ID.""" url = f"https://hacker-news.firebaseio.com/v0/item/{item_id}.json" try: async with aiohttp.ClientSession() as session: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: return await response.json() return None except Exception as e: logging.error(f"Error fetching item {item_id}: {e}") return None async def _search_hackernews(query: str, limit: int = DEFAULT_STORIES) -> Optional[List[Dict]]: """Search Hacker News using Algolia API.""" url = "https://hn.algolia.com/api/v1/search" params = { "query": query, "tags": "story", "hitsPerPage": limit } try: async with aiohttp.ClientSession() as session: async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: data = await response.json() return data.get("hits", []) return None except Exception as e: logging.error(f"Error searching HN: {e}") return None # --------------------------------------------------------------------------- # Command Handler # --------------------------------------------------------------------------- async def handle_command(room, message, bot, prefix, config): """Handle !hn commands.""" import simplematrixbotlib as botlib match = botlib.MessageMatch(room, message, bot, prefix) if not (match.is_not_from_this_bot() and match.prefix() and match.command("hn")): return args = match.args() limit = DEFAULT_STORIES if args and args[-1].isdigit(): limit = min(int(args[-1]), MAX_STORIES) args = args[:-1] # No arguments - show top stories if not args: story_type = "top" type_name = "Top" await bot.api.send_text_message(room.room_id, "📰 Fetching top Hacker News stories...") # Handle story ID elif args[0].lower() == "story" and len(args) >= 2: try: story_id = int(args[1]) await bot.api.send_text_message(room.room_id, f"📖 Fetching story {story_id}...") story = await _fetch_item(story_id) if not story or story.get("deleted"): await bot.api.send_text_message(room.room_id, f"❌ Story {story_id} not found.") return content = f"
      \n{_format_story(story, 1, show_points=True)}\n
    " if len(args) >= 3 and args[2].lower() == "comments": content += "\nTop Comments:\n
      \n" if story.get("kids"): comment_ids = story["kids"][:5] for comment_id in comment_ids: comment = await _fetch_item(comment_id) if comment: formatted = _format_comment(comment) if formatted: content += f"{formatted}\n" content += "
    " response = _format_collapsible(f"Story: {story.get('title', 'Unknown')[:50]}", content, expanded=True) await bot.api.send_markdown_message(room.room_id, response) except ValueError: await bot.api.send_text_message(room.room_id, "❌ Invalid story ID.") return # Handle comments elif args[0].lower() == "comments" and len(args) >= 2: try: story_id = int(args[1]) await bot.api.send_text_message(room.room_id, f"💬 Fetching comments...") story = await _fetch_item(story_id) if not story: await bot.api.send_text_message(room.room_id, f"❌ Story {story_id} not found.") return content = f"{story.get('title', 'Unknown')}\n
      \n" if story.get("kids"): comment_ids = story["kids"][:limit] count = 0 for comment_id in comment_ids: comment = await _fetch_item(comment_id) if comment and not comment.get("deleted"): formatted = _format_comment(comment) if formatted: content += f"{formatted}\n" count += 1 if count >= limit: break if count == 0: content += "
    • No comments found.
      • " else: content += f"\n
    \nShowing {count} comments" else: content += "
  • This story has no comments yet.
    • \n" response = _format_collapsible(f"Comments for Story {story_id}", content, expanded=True) await bot.api.send_markdown_message(room.room_id, response) except ValueError: await bot.api.send_text_message(room.room_id, "❌ Invalid story ID.") return # Handle search elif args[0].lower() == "search" and len(args) >= 2: query = " ".join(args[1:]) await bot.api.send_text_message(room.room_id, f"🔍 Searching for: *{query}*...") results = await _search_hackernews(query, limit) if not results: await bot.api.send_text_message(room.room_id, "❌ No results found.") return content = "
      \n" for i, hit in enumerate(results[:limit], 1): content += _format_search_result(hit, i) + "\n" content += f"
    \n\nFound {len(results[:limit])} results" response = _format_collapsible(f"Search: '{query}'", content, expanded=False) await bot.api.send_markdown_message(room.room_id, response) return # Handle story type (top, new, best, etc.) else: cmd = args[0].lower() if cmd in STORY_TYPES: story_type = STORY_TYPES[cmd] type_name = cmd.capitalize() await bot.api.send_text_message(room.room_id, f"📰 Fetching {type_name} stories...") else: # Assume it's a story ID try: story_id = int(cmd) story = await _fetch_item(story_id) if story: content = f"
      \n{_format_story(story, 1, show_points=True)}\n
    " response = _format_collapsible(f"Story {story_id}", content, expanded=True) await bot.api.send_markdown_message(room.room_id, response) else: await bot.api.send_text_message(room.room_id, f"❌ Story {story_id} not found.") return except ValueError: await bot.api.send_text_message(room.room_id, f"❌ Unknown command. Use !hn for help.") return # Fetch and display stories story_ids = await _fetch_story_ids(story_type) if not story_ids: await bot.api.send_text_message(room.room_id, f"❌ Failed to fetch {type_name} stories.") return story_ids = story_ids[:limit] stories = [] for story_id in story_ids: story = await _fetch_item(story_id) if story and not story.get("deleted"): stories.append(story) await asyncio.sleep(0.1) if not stories: await bot.api.send_text_message(room.room_id, f"❌ Could not fetch any {type_name} stories.") return content = "
      \n" for i, story in enumerate(stories, 1): content += _format_story(story, i, show_points=True) + "\n" content += f"
    \n\nFetched {len(stories)} stories" response = _format_collapsible(f"Hacker News - {type_name} Stories", content, expanded=False) await bot.api.send_markdown_message(room.room_id, response) logging.info(f"Sent Hacker News to {room.room_id}: type={story_type}") # --------------------------------------------------------------------------- # Plugin Setup # --------------------------------------------------------------------------- def setup(bot): """Initialize plugin with bot instance.""" logging.info("Hacker News plugin loaded") # --------------------------------------------------------------------------- # Plugin Metadata # --------------------------------------------------------------------------- __version__ = "1.0.0" __author__ = "Funguy Bot" __description__ = "Hacker News integration"