lljz66 commited on
Commit
403bb3d
·
verified ·
1 Parent(s): 5153b0c

devnet: hackernews source

Browse files
Files changed (1) hide show
  1. src/devnet/sources/hackernews.py +61 -0
src/devnet/sources/hackernews.py ADDED
@@ -0,0 +1,61 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Hacker News via the public Algolia HN Search API. No key, no auth."""
2
+ from __future__ import annotations
3
+
4
+ from ..core import Resource, UpstreamError, get, strip_html
5
+
6
+ API = "https://hn.algolia.com/api/v1/search"
7
+ _KIND_TAGS = {"story": "story", "comment": "comment", "show_hn": "show_hn", "ask_hn": "ask_hn"}
8
+
9
+
10
+ def search(query: str, limit: int = 20, kind: str = "all") -> dict:
11
+ """Search HN stories or comments. kind: all, story, comment, show_hn, ask_hn."""
12
+ query = (query or "").strip()
13
+ if not query:
14
+ raise UpstreamError("query is required")
15
+ if kind != "all" and kind not in _KIND_TAGS:
16
+ raise UpstreamError("kind must be all, story, comment, show_hn or ask_hn")
17
+ limit = max(1, min(int(limit), 50))
18
+
19
+ params: dict = {"query": query, "hitsPerPage": limit}
20
+ if kind != "all":
21
+ params["tags"] = _KIND_TAGS[kind]
22
+ payload = get(API, params, ttl=300)
23
+
24
+ items: list[dict] = []
25
+ for hit in payload.get("hits") or []:
26
+ object_id = hit.get("objectID")
27
+ url = hit.get("url") or (
28
+ f"https://news.ycombinator.com/item?id={object_id}" if object_id else None
29
+ )
30
+ items.append(
31
+ Resource(
32
+ id=f"hn:{object_id}",
33
+ type="comment" if kind == "comment" else "story",
34
+ source="hackernews",
35
+ title=hit.get("title") or hit.get("story_title") or f"HN {object_id}",
36
+ url=url,
37
+ description=strip_html(hit.get("story_text") or hit.get("comment_text")),
38
+ author={"name": hit.get("author")},
39
+ timestamps={"published_at": hit.get("created_at")},
40
+ metrics={"points": hit.get("points"), "comments": hit.get("num_comments")},
41
+ tags=[t for t in (hit.get("_tags") or []) if isinstance(t, str)][:8],
42
+ metadata={
43
+ "object_id": object_id,
44
+ "story_id": hit.get("story_id"),
45
+ "kind": kind,
46
+ },
47
+ provenance={"provider": "hn_algolia"},
48
+ ).dict()
49
+ )
50
+
51
+ total = payload.get("nbHits") or len(items)
52
+ return {
53
+ "kind": "search",
54
+ "source": "hackernews",
55
+ "query": query,
56
+ "filter": kind,
57
+ "count": len(items),
58
+ "total_hits": total,
59
+ "pagination": {"has_more": len(items) < total, "next_cursor": None},
60
+ "items": items,
61
+ }