diff --git a/AGENTS.md b/AGENTS.md
index 671f337..b864995 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -94,10 +94,45 @@ CLI (uv run skim ...) → skim_cli.cli → skim_core.crawlers.REGISTRY lookup
### 데이터 계약: DB는 소비 준비가 끝난 상태다
- `posts.content_markdown`은 **추출이 완료된 정본 본문**이다. 이 DB를 읽는 소비자(AI, digest, 데스크톱 앱, research)는 재추출 절차 없이 그대로 사용한다고 가정한다.
-- 따라서 추출 완결성은 크롤러의 책임이다. 저장 시점에 링크 원문 본문, 플랫폼 자체 본문(Ask/Show HN 텍스트, GeekNews 한국어 요약), 토론(HN 상위 댓글)까지 채워야 한다. "링크만 저장"은 계약 위반이다.
+- 따라서 추출 완결성은 크롤러의 책임이다. 저장 시점에 링크 원문 본문, 플랫폼 자체 본문(Ask/Show HN 텍스트, GeekNews 한국어 요약), 토론(댓글)까지 채워야 한다. "링크만 저장"은 계약 위반이다.
- 예외는 `--no-content` 명시 실행과 `youtube-history` 백필 행(임베드용 목록, 자막은 사용자가 요청할 때 `youtube-transcribe`로 채움)뿐이다.
- 크롤러가 본문에 합성하는 섹션 라벨은 항상 영어로 쓴다 (예: `## Hacker News Comments`, `## Original Article`). 가용한 메타데이터(작성자, 작성시각, 점수)는 텍스트에 함께 표기한다.
+#### 댓글 수집
+
+댓글은 `skim_core.comments`의 `Comment`로 정규화한 뒤 `render_comment_section()`으로 섹션을 만들고
+`append_comment_section()`으로 본문 뒤에 잇는다. 각 크롤러가 자기 포맷을 따로 만들지 않는다.
+
+| 플랫폼 | 섹션 라벨 | 추가 요청 |
+|--------|-----------|-----------|
+| hackernews | `## Hacker News Comments` | Algolia item API 1건 |
+| geeknews | `## GeekNews Comments` | 없음 (지표 수집이 받는 토픽 HTML 재사용) |
+| x | `## X Replies` | 스레드는 없음(TweetDetail 재사용). 단독 트윗은 답글 3개 이상인 것만, 회차당 20건까지 |
+| reddit | `## Reddit Comments` | 게시글당 1건 (초당 1요청 간격) |
+| linkedin | `## LinkedIn Comments` | 게시글당 1건 (Voyager `feed/comments`) |
+| youtube | `## YouTube Comments` | 영상당 yt-dlp 1회 |
+| producthunt | `## Product Hunt Comments` | 제품당 1건 (PH 제품 페이지) |
+| threads | `## Threads Replies` | 답글 1개 이상인 게시물 전부, 게시물당 1건 |
+
+- threads 답글은 타임라인 GraphQL이 주지 않는다. 대신 게시물 문서의 SSR 페이로드가
+ 답글까지 담고 있고 로그인도 필요 없어서, persisted query 좌표(`doc_id`)를 새로 들지 않는다.
+ 단 `threads.net`으로 요청하면 리다이렉트 뒤 페이로드가 빠진 셸이 오므로 `threads.com`으로 받는다.
+ 같은 URL이라도 페이로드가 빠진 문서가 간헐적으로 와서 한 번 재시도한다.
+- threads는 작성자 self-reply 연작을 답글과 같은 `edges`에 담는다. 그 연작은 이미 본문에
+ 있으므로 스레드 시작자가 원글 작성자면 통째로 건너뛴다. 대화 중 작성자가 남긴 답변은 남는다.
+- `comments`(= `direct_reply_count`)가 0보다 커도 답글 섹션이 안 붙을 수 있다. 삭제되거나
+ 비공개 계정이 단 답글까지 세는 값이라, 실제 노출되는 답글이 없는 게시물이 있다
+ (브라우저로 열어도 안 보인다). 이 불일치만으로 추출 실패로 판단하지 않는다.
+- 그래서 `comments`를 조회 임계로 높게 잡으면 안 된다. 반대 방향 오차도 있어서, `comments=1`인
+ 글에서 답글 2건이 나오기도 한다. 임계는 "0건만 거른다"로 둔다.
+- **threads 답글에 페이지네이션을 붙이지 않는다.** 문서가 한 번에 주는 만큼(실측 최대 24건)이
+ 전부이고, 그 이상은 `BarcelonaPostPageRefetchableDirectQuery`를 4건씩 반복 호출해야 한다.
+ 그 요청은 세션 쿠키와 `x-fb-lsd` 토큰을 요구해 **계정으로 식별된다**. 지금 방식은 로그인이
+ 필요 없어 계정이 노출되지 않으므로, 답글 수집량보다 계정 안전을 우선한 결정이다(2026-08-10).
+- 상한은 댓글당 1200자다. 개수 상한(`MAX_COMMENTS`)은 플랫폼마다 다르고 threads는 없다
+ (`None`이면 받은 만큼 전부). 15로 자르던 때는 문서에 24건이 와도 9건을 버렸다.
+- 댓글 수집 실패는 게시글 저장을 막지 않는다. 본문만 저장하고 경고만 남긴다.
+
### Crawler 유형과 패턴
모든 크롤러는 `packages/skim-core/src/skim_core/crawlers/base.py`의 `Crawler` Protocol을 구현하고, `packages/skim-core/src/skim_core/crawlers/__init__.py`의 `REGISTRY`에 등록된다.
@@ -117,7 +152,8 @@ CLI (uv run skim ...) → skim_cli.cli → skim_core.crawlers.REGISTRY lookup
- `packages/skim-core/src/skim_core/models.py`: `Post` Pydantic 모델
- `packages/skim-core/src/skim_core/db.py`: SQLite WAL 모드, `UNIQUE(platform, external_id)` 중복 제거
- `packages/skim-core/src/skim_core/enrichment.py`: `bunx defuddle`, `yt-dlp`, transcript 정리
-- `packages/skim-core/src/skim_core/feed_utils.py`: RSS/Atom 파싱, KST 변환
+- `packages/skim-core/src/skim_core/comments.py`: 플랫폼 중립 `Comment`와 본문 댓글 섹션 합성
+- `packages/skim-core/src/skim_core/feed_utils.py`: RSS/Atom 파싱, KST 변환. `FEED_HEADERS`의 Chrome 버전은 news.hada.io 차단선에 걸리므로 함부로 낮추지 않는다
- `packages/skim-core/src/skim_core/feed_config.py`: RSS URL, YouTube 채널 ID, API endpoint 설정
- `apps/desktop/`: SwiftUI desktop reader for local `data/skim.db`
diff --git a/packages/skim-core/src/skim_core/comments.py b/packages/skim-core/src/skim_core/comments.py
new file mode 100644
index 0000000..49bf336
--- /dev/null
+++ b/packages/skim-core/src/skim_core/comments.py
@@ -0,0 +1,91 @@
+"""크롤러가 수집한 댓글을 본문 마크다운 섹션으로 합성한다.
+
+`AGENTS.md`의 데이터 계약상 토론은 `content_markdown`에 함께 담겨야 한다.
+플랫폼마다 응답 구조는 다르지만 저장 형태는 같아야 하므로, 각 크롤러는
+자기 응답을 `Comment`로 정규화한 뒤 `render_comment_section()`에 넘긴다.
+
+섹션 라벨은 계약대로 항상 영어이고, 가용한 메타데이터(작성자, 점수, 작성시각)를
+텍스트에 함께 표기한다. hackernews가 이미 쓰던 출력 형태를 그대로 따른다.
+"""
+
+from dataclasses import dataclass
+from typing import Iterable, List, Optional
+
+DEFAULT_MAX_COMMENTS = 15
+DEFAULT_MAX_CHARS = 1200
+
+
+@dataclass
+class Comment:
+ """플랫폼 중립 댓글 1건."""
+
+ author: str
+ text: str
+ score: Optional[int] = None
+ created: Optional[str] = None
+ depth: int = 0
+
+
+def _clean(text: str, max_chars: int) -> str:
+ """줄바꿈을 접어 목록 항목 하나로 만든다. 들여쓰기가 깨지는 것을 막는다."""
+ collapsed = " ".join((text or "").split())
+ if len(collapsed) > max_chars:
+ collapsed = collapsed[:max_chars].rstrip() + "..."
+ return collapsed
+
+
+def _meta_suffix(comment: Comment, score_unit: str) -> str:
+ parts: List[str] = []
+ if comment.score is not None:
+ unit = score_unit if abs(comment.score) == 1 else f"{score_unit}s"
+ parts.append(f"{comment.score} {unit}")
+ if comment.created:
+ parts.append(comment.created)
+ return f" ({', '.join(parts)})" if parts else ""
+
+
+def render_comment_section(
+ label: str,
+ comments: Iterable[Comment],
+ *,
+ note: Optional[str] = None,
+ max_comments: Optional[int] = DEFAULT_MAX_COMMENTS,
+ max_chars: int = DEFAULT_MAX_CHARS,
+ score_unit: str = "point",
+) -> Optional[str]:
+ """댓글 목록을 `## {label}` 섹션으로 만든다. 유효한 댓글이 없으면 None.
+
+ `score_unit`은 플랫폼이 점수를 부르는 이름이다(HN/Reddit은 point, X/YouTube는 like).
+ `max_comments=None`이면 받은 만큼 전부 싣는다.
+ """
+ lines: List[str] = []
+ for comment in comments:
+ if max_comments is not None and len(lines) >= max_comments:
+ break
+ text = _clean(comment.text, max_chars)
+ if not text:
+ continue
+ indent = " " * max(0, comment.depth)
+ author = comment.author or "unknown"
+ lines.append(
+ f"{indent}- **{author}**{_meta_suffix(comment, score_unit)}: {text}"
+ )
+
+ if not lines:
+ return None
+
+ section = f"## {label}\n\n"
+ if note:
+ section += f"{note}\n\n"
+ return section + "\n".join(lines)
+
+
+def append_comment_section(
+ body: Optional[str], section: Optional[str]
+) -> Optional[str]:
+ """본문 뒤에 댓글 섹션을 잇는다. 구분자는 기존 본문 합성과 같은 `---`."""
+ if not section:
+ return body
+ if not body or not body.strip():
+ return section
+ return f"{body.rstrip()}\n\n---\n\n{section}"
diff --git a/packages/skim-core/src/skim_core/crawlers/api/linkedin.py b/packages/skim-core/src/skim_core/crawlers/api/linkedin.py
index 83e32f2..0e5e3fd 100644
--- a/packages/skim-core/src/skim_core/crawlers/api/linkedin.py
+++ b/packages/skim-core/src/skim_core/crawlers/api/linkedin.py
@@ -16,18 +16,27 @@
import requests
import typer
+from ...comments import Comment, append_comment_section, render_comment_section
from ...models import Post
from ...paths import DATA_DIR, SESSIONS_DIR
LINKEDIN_BASE_URL = "https://www.linkedin.com"
VOYAGER_FEED_URL = f"{LINKEDIN_BASE_URL}/voyager/api/feed/updatesV2"
+VOYAGER_COMMENTS_URL = f"{LINKEDIN_BASE_URL}/voyager/api/feed/comments"
+MAX_COMMENTS = 15
LINKEDIN_USER_AGENT = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36"
)
REQUIRED_COOKIE_NAMES = {"li_at", "JSESSIONID"}
REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
-SESSION_REJECTED_REASONS = {"login", "authwall", "checkpoint", "challenge", "self-redirect-loop"}
+SESSION_REJECTED_REASONS = {
+ "login",
+ "authwall",
+ "checkpoint",
+ "challenge",
+ "self-redirect-loop",
+}
def _classify_redirect(response: requests.Response) -> str:
@@ -83,19 +92,97 @@ def __init__(
async def crawl(self, **options) -> List[Post]:
count = options.get("count", 5)
+ no_content = options.get("no_content", False)
try:
- return self.fetch_feed(count=count)
+ posts = self.fetch_feed(count=count)
+ if not no_content:
+ self.attach_comments(posts)
+ return posts
finally:
# 크롤러 인스턴스는 1회성이다. 직접 만든 세션은 커넥션 풀을 정리한다.
if self._owns_session:
self.session.close()
+ def attach_comments(self, posts: List[Post]) -> None:
+ """게시글별 댓글을 정본 본문 뒤에 잇는다. 게시글당 요청 1건이 늘어난다."""
+ failures = 0
+ for post in posts:
+ section = self.fetch_comment_section(post.external_id)
+ if section is None and post.external_id:
+ failures += 1
+ continue
+ body = post.content_markdown or post.content
+ post.content_markdown = append_comment_section(body, section)
+
+ if failures:
+ typer.echo(f" [!] LinkedIn 댓글 수집 실패 {failures}건 (본문만 저장)")
+
+ def fetch_comment_section(self, external_id: Optional[str]) -> Optional[str]:
+ """activity id의 댓글을 본문용 마크다운 섹션으로 만든다."""
+ if not external_id:
+ return None
+ activity_id = self._extract_activity_id(str(external_id)) or str(external_id)
+ if not activity_id.isdigit():
+ return None
+
+ try:
+ response = self.session.get(
+ VOYAGER_COMMENTS_URL,
+ params={
+ "count": str(MAX_COMMENTS),
+ "q": "comments",
+ "sortOrder": "RELEVANCE",
+ "updateId": f"activity:{activity_id}",
+ },
+ allow_redirects=False,
+ timeout=20,
+ )
+ except Exception: # noqa: BLE001 - 댓글 실패가 게시글 저장을 막지 않는다
+ return None
+
+ if response.status_code != 200:
+ return None
+ try:
+ payload = response.json()
+ except ValueError:
+ return None
+
+ # Voyager는 정규화 응답이라 실데이터가 elements가 아닌 included에 온다.
+ collected: List[Comment] = []
+ for entry in payload.get("included") or []:
+ if not str(entry.get("$type", "")).endswith("feed.Comment"):
+ continue
+ # RELEVANCE 정렬은 답글을 부모보다 먼저 주기도 해서 그대로 담으면 순서가
+ # 뒤엉킨다. 트리를 재구성할 만한 값이 아니라 최상위 댓글만 담는다.
+ # spartan: 답글까지 필요해지면 parentCommentUrn으로 부모 뒤에 끼워 넣는다.
+ if entry.get("parentCommentUrn"):
+ continue
+ text = self._extract_path(entry, "commentV2.text") or ""
+ if not text:
+ continue
+ author = self._extract_path(entry, "commenterForDashConversion.title.text")
+ created_ms = entry.get("createdTime")
+ created = None
+ if isinstance(created_ms, (int, float)):
+ created = datetime.fromtimestamp(
+ created_ms / 1000, tz=timezone.utc
+ ).strftime("%Y-%m-%d %H:%M UTC")
+ collected.append(
+ Comment(author=author or "unknown", text=text, created=created)
+ )
+
+ return render_comment_section(
+ "LinkedIn Comments", collected, max_comments=MAX_COMMENTS
+ )
+
def fetch_feed(self, *, count: int) -> List[Post]:
"""LinkedIn 홈 피드 게시글을 수집합니다."""
typer.echo(f"[API 모드] LinkedIn 크롤링 시작... (게시글 {count}개)")
if not self._has_login_session:
- typer.echo("❌ 세션 파일이 없거나 LinkedIn 쿠키가 부족합니다. 먼저 로그인하세요:")
+ typer.echo(
+ "❌ 세션 파일이 없거나 LinkedIn 쿠키가 부족합니다. 먼저 로그인하세요:"
+ )
typer.echo(" uv run skim login linkedin")
return []
@@ -195,7 +282,9 @@ def _parse_response(self, data: dict) -> List[Post]:
posts.append(post)
return posts
- def _ordered_update_items(self, data: dict, urn_index: dict[str, dict]) -> list[dict]:
+ def _ordered_update_items(
+ self, data: dict, urn_index: dict[str, dict]
+ ) -> list[dict]:
element_urns = self._find_first_elements(data)
if not element_urns:
return []
@@ -244,7 +333,10 @@ def _is_update_item(item: Any) -> bool:
return (
"feed.Update" in item_type
or "fsd_update" in entity_urn
- or (isinstance(item.get("commentary"), dict) and isinstance(item.get("actor"), dict))
+ or (
+ isinstance(item.get("commentary"), dict)
+ and isinstance(item.get("actor"), dict)
+ )
)
def _extract_post(self, item: dict, urn_index: dict) -> Optional[Post]:
@@ -302,7 +394,9 @@ def _is_promoted(actor: dict) -> bool:
return False
for key in ("text", "accessibilityText"):
value = sub_desc.get(key)
- if isinstance(value, str) and ("promoted" in value.lower() or "광고" in value):
+ if isinstance(value, str) and (
+ "promoted" in value.lower() or "광고" in value
+ ):
return True
return False
@@ -372,7 +466,9 @@ def _coerce_linkedin_timestamp(self, raw_value) -> Optional[str]:
if isinstance(raw_value, (int, float)):
seconds = raw_value / 1000 if raw_value > 10_000_000_000 else raw_value
- return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat(timespec="seconds")
+ return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat(
+ timespec="seconds"
+ )
return None
@@ -502,7 +598,9 @@ def _extract_text(self, raw_value: Any) -> str:
return text
if isinstance(raw_value, list):
return " ".join(
- part for part in (self._extract_text(item) for item in raw_value) if part
+ part
+ for part in (self._extract_text(item) for item in raw_value)
+ if part
).strip()
return str(raw_value).strip()
diff --git a/packages/skim-core/src/skim_core/crawlers/api/reddit.py b/packages/skim-core/src/skim_core/crawlers/api/reddit.py
index e6d9ab0..bbcf5dc 100644
--- a/packages/skim-core/src/skim_core/crawlers/api/reddit.py
+++ b/packages/skim-core/src/skim_core/crawlers/api/reddit.py
@@ -9,6 +9,7 @@
import json
import re
+import time
from datetime import datetime, timezone
from html import unescape
from pathlib import Path
@@ -19,16 +20,30 @@
import requests
import typer
+from ...comments import Comment, append_comment_section, render_comment_section
from ...models import Post
from ...paths import SESSIONS_DIR
SESSION_PATH = SESSIONS_DIR / "reddit_session.json"
+# 링크 게시물은 본문이 제목뿐이라 토론이 사실상 본체다. 상한은 hackernews와 맞춘다.
+MAX_COMMENTS = 15
+# 댓글은 게시글당 요청 1건이라 기본 50건 크롤이면 연속 50요청이 된다.
+# Reddit 관례인 초당 1요청에 맞춰 간격을 둔다.
+COMMENT_REQUEST_INTERVAL_SECONDS = 1.0
+# 세션이 만료되면 전 건이 403이라 남은 요청이 전부 헛돈다. 초반에 끊는다.
+MAX_CONSECUTIVE_COMMENT_FAILURES = 3
REDDIT_BASE_URL = "https://www.reddit.com"
REDDIT_USER_AGENT = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
)
-HOME_FEED_COOKIE_NAMES = {"reddit_session", "token_v2", "session_tracker", "loid", "csrf_token"}
+HOME_FEED_COOKIE_NAMES = {
+ "reddit_session",
+ "token_v2",
+ "session_tracker",
+ "loid",
+ "csrf_token",
+}
VERIFICATION_TITLE = "please wait for verification"
@@ -61,8 +76,12 @@ async def crawl(self, **options) -> List[Post]:
count = options.get("count", 10)
subreddit = options.get("subreddit")
sort = options.get("sort", "hot")
+ no_content = options.get("no_content", False)
try:
- return self.fetch_listing(count=count, subreddit=subreddit, sort=sort)
+ posts = self.fetch_listing(count=count, subreddit=subreddit, sort=sort)
+ if not no_content:
+ self.attach_comments(posts)
+ return posts
finally:
# 크롤러 인스턴스는 1회성이다. 직접 만든 세션은 커넥션 풀을 정리한다.
if self._owns_session:
@@ -102,7 +121,9 @@ def fetch_listing(
try:
listing = self.fetch_listing_page(url)
except requests.HTTPError as exc:
- status_code = exc.response.status_code if exc.response is not None else None
+ status_code = (
+ exc.response.status_code if exc.response is not None else None
+ )
if normalized_subreddit and status_code == 403:
return self.fetch_subreddit_rss(
count=count,
@@ -119,6 +140,92 @@ def fetch_listing(
return posts[:count]
+ def attach_comments(self, posts: List[Post]) -> None:
+ """각 게시글의 상위 댓글을 정본 본문(content_markdown) 뒤에 잇는다.
+
+ 게시글당 요청 1건이 늘어난다. `--no-content`면 호출되지 않는다.
+ """
+ failures = 0
+ consecutive = 0
+ for index, post in enumerate(posts):
+ if consecutive >= MAX_CONSECUTIVE_COMMENT_FAILURES:
+ typer.echo(
+ f" [!] Reddit 댓글 수집을 중단합니다 "
+ f"(연속 {consecutive}건 실패, 남은 {len(posts) - index}건은 본문만 저장)"
+ )
+ break
+ if index:
+ time.sleep(COMMENT_REQUEST_INTERVAL_SECONDS)
+
+ section = self.fetch_comment_section(post.url)
+ if section is None and post.url:
+ failures += 1
+ consecutive += 1
+ continue
+ consecutive = 0
+ body = post.content_markdown or post.content
+ post.content_markdown = append_comment_section(body, section)
+
+ if failures:
+ typer.echo(f" [!] Reddit 댓글 수집 실패 {failures}건 (본문만 저장)")
+
+ def fetch_comment_section(self, url: Optional[str]) -> Optional[str]:
+ """permalink의 댓글 트리를 본문용 마크다운 섹션으로 만든다."""
+ if not url:
+ return None
+
+ comment_url = self.append_query_params(
+ url.rstrip("/") + ".json",
+ {"limit": str(MAX_COMMENTS), "sort": "top", "raw_json": "1"},
+ )
+ try:
+ payload = self.fetch_listing_page(comment_url)
+ except Exception: # noqa: BLE001 - 댓글 실패가 게시글 저장을 막지 않는다
+ return None
+
+ # 댓글 응답은 [게시글 listing, 댓글 listing] 2원소 배열이다.
+ if not isinstance(payload, list) or len(payload) < 2:
+ return None
+
+ collected: List[Comment] = []
+
+ def walk(listing: dict, depth: int) -> None:
+ children = (listing or {}).get("data", {}).get("children") or []
+ for child in children:
+ if len(collected) >= MAX_COMMENTS:
+ return
+ # "더 보기"(kind=more)와 삭제된 댓글은 본문이 없다.
+ if child.get("kind") != "t1":
+ continue
+ data = child.get("data") or {}
+ body = (data.get("body") or "").strip()
+ if not body or body in {"[deleted]", "[removed]"}:
+ continue
+ created = data.get("created_utc")
+ collected.append(
+ Comment(
+ author=f"u/{data.get('author') or 'unknown'}",
+ text=body,
+ score=data.get("score"),
+ created=(
+ datetime.fromtimestamp(created, tz=timezone.utc).strftime(
+ "%Y-%m-%d %H:%M UTC"
+ )
+ if created
+ else None
+ ),
+ depth=depth,
+ )
+ )
+ replies = data.get("replies")
+ if depth < 1 and isinstance(replies, dict):
+ walk(replies, depth + 1)
+
+ walk(payload[1], 0)
+ return render_comment_section(
+ "Reddit Comments", collected, max_comments=MAX_COMMENTS
+ )
+
def build_listing_url(
self,
*,
@@ -156,11 +263,15 @@ def fetch_listing_page(self, url: str) -> dict:
response.raise_for_status()
if not self.is_json_response(response):
- raise ValueError(f"Reddit listing JSON 응답을 받지 못했습니다: {response.url}")
+ raise ValueError(
+ f"Reddit listing JSON 응답을 받지 못했습니다: {response.url}"
+ )
return response.json()
- def fetch_subreddit_rss(self, *, count: int, subreddit: str, sort: str) -> List[Post]:
+ def fetch_subreddit_rss(
+ self, *, count: int, subreddit: str, sort: str
+ ) -> List[Post]:
"""비로그인 subreddit JSON endpoint가 403일 때 Atom feed로 fallback."""
url = self.build_subreddit_rss_url(subreddit=subreddit, sort=sort, limit=count)
response = self.session.get(url, timeout=15)
@@ -184,7 +295,9 @@ def is_verification_page(html: str) -> bool:
def is_json_response(response: requests.Response) -> bool:
"""JSON 응답 여부를 확인합니다."""
content_type = response.headers.get("content-type", "").lower()
- return "application/json" in content_type or response.text.lstrip().startswith("{")
+ return "application/json" in content_type or response.text.lstrip().startswith(
+ "{"
+ )
def solve_verification_challenge(self, url: str, html: str) -> None:
"""Reddit verification challenge를 풀어 세션 쿠키를 확보합니다."""
@@ -242,7 +355,11 @@ def _extract_image_urls(post_data: dict) -> List[str]:
"""직접 이미지 링크, preview 원본, 갤러리에서 이미지 CDN URL을 수집합니다."""
urls: List[str] = []
dest = post_data.get("url_overridden_by_dest") or ""
- if dest.split("?")[0].lower().endswith((".jpg", ".jpeg", ".png", ".gif", ".webp")):
+ if (
+ dest.split("?")[0]
+ .lower()
+ .endswith((".jpg", ".jpeg", ".png", ".gif", ".webp"))
+ ):
urls.append(dest)
for img in (post_data.get("preview") or {}).get("images") or []:
src = (img.get("source") or {}).get("url")
@@ -298,7 +415,9 @@ def parse_rss_entry(self, entry: dict, subreddit: str) -> Optional[Post]:
raw_id = entry.get("id", "")
external_id = raw_id.removeprefix("t3_") if raw_id else ""
title = (entry.get("title") or "").strip()
- content_html = entry.get("summary") or entry.get("content", [{}])[0].get("value", "")
+ content_html = entry.get("summary") or entry.get("content", [{}])[0].get(
+ "value", ""
+ )
content = self.strip_html(content_html) or title
link = entry.get("link") or ""
diff --git a/packages/skim-core/src/skim_core/crawlers/api/threads.py b/packages/skim-core/src/skim_core/crawlers/api/threads.py
index 1f233d0..cc858c2 100644
--- a/packages/skim-core/src/skim_core/crawlers/api/threads.py
+++ b/packages/skim-core/src/skim_core/crawlers/api/threads.py
@@ -23,7 +23,9 @@
import requests
import typer
+from bs4 import BeautifulSoup
+from ...comments import Comment, append_comment_section, render_comment_section
from ...models import Post
from ...paths import SESSIONS_DIR
@@ -36,6 +38,24 @@
"(KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36"
)
+# 답글은 문서가 실어 보내는 만큼 전부 담는다. 15개로 자르던 때는 문서에 24개가 와도
+# 9개를 버렸다. 상한이 없으므로 인기 게시물은 본문이 길어진다.
+MAX_REPLIES = None
+# 답글이 0건이라고 보고된 게시물만 건너뛴다. `comments`가 삭제·비공개 답글까지 세는
+# 부정확한 값이라 높게 잡으면 멀쩡한 글이 빠진다(3이던 때 답글 1~2개인 12.6%가 통째로 빠졌다).
+MIN_REPLIES_FOR_FETCH = 1
+# 게시물 페이지는 답글이 담긴 SSR 페이로드를 로그인 없이도 준다. 단 threads.net으로
+# 요청하면 리다이렉트 뒤 페이로드가 빠진 셸이 와서, threads.com으로 직접 받아야 한다.
+POST_PAGE_HEADERS = {
+ "User-Agent": WEB_USER_AGENT,
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+ "Accept-Language": "en-US,en;q=0.9,ko;q=0.8",
+ "Sec-Fetch-Dest": "document",
+ "Sec-Fetch-Mode": "navigate",
+ "Sec-Fetch-Site": "none",
+ "Upgrade-Insecure-Requests": "1",
+}
+
# persisted query 좌표. Meta가 웹앱을 재배포하면 doc_id가 바뀌어 execution error가 난다.
# 갱신하려면 브라우저로 threads.com을 열어 graphql/query 요청의 doc_id를 다시 읽는다.
TIMELINE_QUERY = {
@@ -153,12 +173,144 @@ def _load_cookies(self) -> Dict[str, str]:
async def crawl(self, **options) -> List[Post]:
count = options.get("count", 5)
user_id = options.get("user_id")
+ no_content = options.get("no_content", False)
try:
- return await self._crawl_impl(count, user_id)
+ posts = await self._crawl_impl(count, user_id)
+ if not no_content:
+ self.attach_replies(posts)
+ return posts
finally:
# 크롤러 인스턴스는 1회성이다. 세션 커넥션 풀을 정리한다.
self.session.close()
+ def attach_replies(self, posts: List[Post]) -> None:
+ """타인 답글을 정본 본문 뒤에 잇는다.
+
+ 답글이 있다고 보고된 게시물은 전부 조회한다(게시물당 요청 1건, 실측 1~4초).
+ `--count`를 크게 주면 그만큼 크롤이 길어진다.
+ """
+ for post in posts:
+ if (post.comments or 0) < MIN_REPLIES_FOR_FETCH:
+ continue
+ section = self.fetch_reply_section(post.url)
+ if section:
+ post.content_markdown = append_comment_section(
+ post.content_markdown or post.content, section
+ )
+
+ def fetch_reply_section(self, url: Optional[str]) -> Optional[str]:
+ """게시물 페이지의 SSR 페이로드에서 답글을 뽑아 마크다운 섹션으로 만든다.
+
+ 타임라인 GraphQL 응답에는 타인 답글이 오지 않는다. 게시물 문서는 답글까지 담고
+ 있고 로그인도 필요 없어서, persisted query 좌표(doc_id)를 따로 들지 않아도 된다.
+ """
+ if not url:
+ return None
+ # threads.net으로 요청하면 리다이렉트 뒤 페이로드가 빠진 셸이 온다.
+ page_url = url.replace("threads.net", "threads.com")
+
+ # 같은 URL이라도 답글 페이로드가 빠진 문서가 간헐적으로 온다(실측). 한 번 더 받아본다.
+ root_author, threads = "", []
+ for _ in range(2):
+ try:
+ response = requests.get(page_url, headers=POST_PAGE_HEADERS, timeout=25)
+ response.raise_for_status()
+ except Exception: # noqa: BLE001 - 답글 실패가 게시물 저장을 막지 않는다
+ return None
+ root_author, threads = self._extract_reply_threads(response.text)
+ if threads:
+ break
+
+ collected: List[Comment] = []
+ for thread in threads:
+ items = thread.get("thread_items") or []
+ if not items:
+ continue
+ # 작성자가 스스로 시작한 스레드는 self-reply 연작이라 이미 본문에 담겨 있다.
+ # 대화 도중 작성자가 남의 답글에 단 답변은 depth>0이라 그대로 살아남는다.
+ starter = ((items[0].get("post") or {}).get("user") or {}).get("username")
+ if root_author and starter == root_author:
+ continue
+ for depth, item in enumerate(items):
+ post_data = item.get("post") or {}
+ text = ((post_data.get("caption") or {}) or {}).get("text") or ""
+ if not text:
+ continue
+ user = (post_data.get("user") or {}).get("username") or "unknown"
+ taken_at = post_data.get("taken_at")
+ collected.append(
+ Comment(
+ author=f"@{user}",
+ text=text,
+ score=post_data.get("like_count"),
+ created=(
+ datetime.fromtimestamp(taken_at, tz=timezone.utc).strftime(
+ "%Y-%m-%d %H:%M UTC"
+ )
+ if taken_at
+ else None
+ ),
+ depth=min(depth, 1),
+ )
+ )
+
+ return render_comment_section(
+ "Threads Replies", collected, max_comments=MAX_REPLIES, score_unit="like"
+ )
+
+ @staticmethod
+ def _extract_reply_threads(html: str) -> tuple[str, List[Dict[str, Any]]]:
+ """문서에 심긴 JSON에서 (루트 작성자, 답글 스레드 목록)을 찾는다.
+
+ 페이로드는 `data.data.edges`에 [루트 게시물, 답글, 답글...] 순으로 담긴다.
+ 같은 문서의 `relatedPosts`도 thread_items를 갖지만 그건 무관한 추천 게시물이라,
+ edges 배열만 집어 첫 항목(루트)을 버린다.
+ """
+ found: List[Dict[str, Any]] = []
+ root_author = ""
+
+ def thread_author(thread: Dict[str, Any]) -> str:
+ items = thread.get("thread_items") or []
+ if not items:
+ return ""
+ return ((items[0].get("post") or {}).get("user") or {}).get(
+ "username"
+ ) or ""
+
+ def walk(node: Any) -> None:
+ nonlocal root_author
+ if isinstance(node, dict):
+ edges = node.get("edges")
+ if isinstance(edges, list) and len(edges) > 1:
+ threads = [
+ edge.get("node")
+ for edge in edges
+ if isinstance(edge, dict) and isinstance(edge.get("node"), dict)
+ ]
+ if threads and all(
+ t.get("thread_items") is not None for t in threads
+ ):
+ if not root_author:
+ root_author = thread_author(threads[0])
+ found.extend(threads[1:]) # [0]은 본문에 이미 담긴 루트 게시물
+ return
+ for value in node.values():
+ walk(value)
+ elif isinstance(node, list):
+ for value in node:
+ walk(value)
+
+ soup = BeautifulSoup(html, "html.parser")
+ for tag in soup.find_all("script", attrs={"type": "application/json"}):
+ raw = tag.string or ""
+ if "thread_items" not in raw:
+ continue
+ try:
+ walk(json.loads(raw))
+ except ValueError:
+ continue
+ return root_author, found
+
async def _crawl_impl(
self, count: int = 5, user_id: Optional[str] = None
) -> List[Post]:
diff --git a/packages/skim-core/src/skim_core/crawlers/api/x.py b/packages/skim-core/src/skim_core/crawlers/api/x.py
index 8a1dafb..a5750c8 100644
--- a/packages/skim-core/src/skim_core/crawlers/api/x.py
+++ b/packages/skim-core/src/skim_core/crawlers/api/x.py
@@ -24,10 +24,16 @@
import typer
+from ...comments import Comment, append_comment_section, render_comment_section
from ...models import Post
from ...paths import SESSIONS_DIR
SESSION_PATH = SESSIONS_DIR / "x_session.json"
+MAX_REPLIES = 15
+# 스레드는 TweetDetail을 이미 부르므로 답글이 공짜다. 단독 트윗은 요청이 추가로 드는데
+# 실측 3~9초/건이라 무제한으로 두면 크롤이 몇 배 길어진다. 반응이 있는 것만, 회차 상한 안에서.
+MIN_REPLIES_FOR_FETCH = 3
+MAX_REPLY_FETCHES_PER_RUN = 20
class XAPICrawler:
@@ -39,6 +45,9 @@ class XAPICrawler:
"""
platform = "x"
+ # 단독 트윗 답글은 요청이 추가로 드는 유일한 경로라 --no-content로 끌 수 있다.
+ # 클래스 속성으로 둬야 생성자를 우회해 만든 인스턴스(테스트)에서도 정의돼 있다.
+ _skip_extra_reply_fetch = False
def __init__(self, debug_mode: bool = False):
self.platform_name = "X"
@@ -100,9 +109,12 @@ def _load_cookies(self) -> Dict[str, str]:
async def crawl(self, **options) -> List[Post]:
count = options.get("count", 10)
user_id = options.get("user_id")
+ self._skip_extra_reply_fetch = options.get("no_content", False)
return await self._crawl_impl(count, user_id)
- async def _crawl_impl(self, count: int = 10, user_id: Optional[str] = None) -> List[Post]:
+ async def _crawl_impl(
+ self, count: int = 10, user_id: Optional[str] = None
+ ) -> List[Post]:
"""
X 게시글 크롤링
@@ -114,7 +126,9 @@ async def _crawl_impl(self, count: int = 10, user_id: Optional[str] = None) -> L
크롤링된 게시글 목록
"""
mode = f"사용자 @{user_id}" if user_id else "For You 타임라인"
- typer.echo(f"[API 모드] {self.platform_name} 크롤링 시작 - {mode} (게시글 {count}개)")
+ typer.echo(
+ f"[API 모드] {self.platform_name} 크롤링 시작 - {mode} (게시글 {count}개)"
+ )
try:
if user_id:
@@ -232,6 +246,7 @@ def _parse_tweets(self, raw_data: list[dict], count: int) -> List[Post]:
posts: List[Post] = []
done_conv: set[str] = set()
+ reply_fetches = 0
for tweet, meta in parsed:
if len(posts) >= count:
break
@@ -241,13 +256,27 @@ def _parse_tweets(self, raw_data: list[dict], count: int) -> List[Post]:
group = conv_groups[conv]
# 그룹 안에 작성자 자신에게 단 답글이 있으면 self-reply 스레드다
is_thread = any(
- gm["reply_to_user"] and gm["reply_to_user"] == gm["author_id"] for _, gm in group
+ gm["reply_to_user"] and gm["reply_to_user"] == gm["author_id"]
+ for _, gm in group
)
if is_thread:
done_conv.add(conv)
post = self._build_thread_post(conv, group)
else:
post = self._parse_single_tweet(tweet)
+ # 단독 트윗은 스레드와 달리 TweetDetail을 부르지 않았다. 반응이 있는 것만
+ # 회차 상한 안에서 재조회해 답글을 채운다.
+ if (
+ post
+ and not self._skip_extra_reply_fetch
+ and reply_fetches < MAX_REPLY_FETCHES_PER_RUN
+ and (post.comments or 0) >= MIN_REPLIES_FOR_FETCH
+ ):
+ reply_fetches += 1
+ post.content_markdown = append_comment_section(
+ post.content_markdown or post.content,
+ self._reply_section(conv, meta["author_id"]),
+ )
if post:
posts.append(post)
if self.debug_mode:
@@ -273,6 +302,58 @@ def _tweet_meta(self, tweet: dict) -> Optional[dict]:
"created_at": legacy.get("created_at", ""),
}
+ def _reply_section(
+ self,
+ conv_id: str,
+ root_author_id: Optional[str],
+ source_tweets: Optional[list[dict]] = None,
+ ) -> Optional[str]:
+ """conversation에 달린 타인 답글을 본문용 섹션으로 만든다.
+
+ `source_tweets`가 오면 이미 받은 TweetDetail 응답을 재사용해 요청이 늘지 않는다.
+ 스레드 병합이 작성자 본인 트윗만 쓰고 나머지를 버리던 자리를 그대로 활용한다.
+ """
+ tweets = (
+ source_tweets
+ if source_tweets is not None
+ else self._fetch_thread_detail(conv_id)
+ )
+ if not tweets:
+ return None
+
+ collected: list[Comment] = []
+ for tweet in tweets:
+ meta = self._tweet_meta(tweet)
+ if not meta or meta["is_rt"]:
+ continue
+ # TweetDetail 응답에는 이 대화와 무관한 추천/광고 트윗도 섞여 온다.
+ # conversation_id가 다르면 답글이 아니다.
+ if meta["conv_id"] != conv_id:
+ continue
+ if meta["id"] == conv_id or meta["author_id"] == root_author_id:
+ continue
+ content, _, _ = self._clean_content_and_media(tweet)
+ if not content:
+ continue
+ legacy = tweet.get("legacy", {})
+ collected.append(
+ Comment(
+ author=f"@{meta['author_screen']}",
+ text=content,
+ score=legacy.get("favorite_count"),
+ created=(self._parse_timestamp(meta["created_at"]) or "")[
+ :16
+ ].replace("T", " ")
+ or None,
+ )
+ )
+
+ # API는 시간순으로 주지만 읽는 쪽에는 반응이 큰 답글이 먼저인 편이 낫다.
+ collected.sort(key=lambda comment: comment.score or 0, reverse=True)
+ return render_comment_section(
+ "X Replies", collected, max_comments=MAX_REPLIES, score_unit="like"
+ )
+
def _build_thread_post(
self, conv_id: str, fallback_group: list[tuple[dict, dict]]
) -> Optional[Post]:
@@ -294,14 +375,23 @@ def _build_thread_post(
return None
# 스레드 주인 = 루트 트윗 작성자(없으면 가장 이른 트윗 작성자)
- root = indexed.get(conv_id) or min(indexed.values(), key=lambda x: int(x[1]["id"]))
+ root = indexed.get(conv_id) or min(
+ indexed.values(), key=lambda x: int(x[1]["id"])
+ )
root_author = root[1]["author_id"]
own = [pair for pair in indexed.values() if pair[1]["author_id"] == root_author]
own.sort(key=lambda pair: int(pair[1]["id"]))
if len(own) <= 1:
- # 이어지는 self-reply가 실제로 없으면 단독 트윗으로 처리
- return self._parse_single_tweet(root[0])
+ # 이어지는 self-reply가 실제로 없으면 단독 트윗으로 처리.
+ # 재조회는 이미 끝났으니 답글은 추가 요청 없이 붙인다.
+ post = self._parse_single_tweet(root[0])
+ if post:
+ post.content_markdown = append_comment_section(
+ post.content_markdown or post.content,
+ self._reply_section(conv_id, root_author, source),
+ )
+ return post
texts: list[str] = []
image_urls: list[str] = []
@@ -323,6 +413,9 @@ def _build_thread_post(
platform="x",
author=author,
content=content,
+ content_markdown=append_comment_section(
+ content, self._reply_section(conv_id, root_author, source)
+ ),
timestamp=self._parse_timestamp(root_meta["created_at"]),
url=f"https://x.com/{author}/status/{tweet_id}" if tweet_id else None,
likes=root_legacy.get("favorite_count", 0),
@@ -489,7 +582,11 @@ def _clean_content_and_media(
# 사진 첨부의 CDN 원본 URL(pbs.twimg.com)을 보존한다
if m.get("type") == "photo" and m.get("media_url_https"):
image_urls.append(m["media_url_https"])
- media_link = m.get("expanded_url") or m.get("display_url") or m.get("media_url_https")
+ media_link = (
+ m.get("expanded_url")
+ or m.get("display_url")
+ or m.get("media_url_https")
+ )
if media_link:
media_fallbacks.append(media_link)
if media_url:
diff --git a/packages/skim-core/src/skim_core/crawlers/feed/geeknews.py b/packages/skim-core/src/skim_core/crawlers/feed/geeknews.py
index 88757b9..fef089e 100644
--- a/packages/skim-core/src/skim_core/crawlers/feed/geeknews.py
+++ b/packages/skim-core/src/skim_core/crawlers/feed/geeknews.py
@@ -10,6 +10,7 @@
import typer
from bs4 import BeautifulSoup
+from ...comments import Comment, append_comment_section, render_comment_section
from ...enrichment import enrich_with_content
from ...feed_config import GEEKNEWS_RSS
from ...feed_utils import FEED_HEADERS, fetch_feed
@@ -18,6 +19,7 @@
GEEKNEWS_URL = "https://news.hada.io/"
_TOPIC_ID = re.compile(r"topic\?id=(\d+)")
+MAX_COMMENTS = 15
def topic_id_from_url(url: str) -> Optional[str]:
@@ -31,11 +33,49 @@ def _digits_to_int(raw: str) -> Optional[int]:
return int(digits) if digits else None
+def _parse_comment_section(soup: BeautifulSoup) -> Optional[str]:
+ """토픽 페이지의 댓글 목록을 본문용 마크다운 섹션으로 만든다.
+
+ 지표를 받으려고 어차피 토픽 HTML을 통째로 받으므로 추가 요청이 들지 않는다.
+ GeekNews는 댓글 점수를 노출하지 않아 작성자와 작성시각까지가 가용 메타데이터다.
+ """
+ collected: List[Comment] = []
+ for row in soup.select(".comment_row"):
+ body_el = row.select_one(".comment_contents")
+ if not body_el:
+ continue
+ text = body_el.get_text(" ", strip=True)
+ if not text:
+ continue
+
+ author_el = row.select_one('.commentinfo a[href^="/@"]')
+ author = author_el.get_text(strip=True) if author_el else "unknown"
+
+ # 상대시각("14시간전")보다 title의 절대시각이 안정적이다.
+ time_el = row.select_one(".commentinfo time")
+ created = time_el.get("title") if time_el else None
+
+ # 들여쓰기는 `style="--depth:N"`으로만 노출된다.
+ depth = 0
+ depth_match = re.search(r"--depth:\s*(\d+)", row.get("style") or "")
+ if depth_match:
+ depth = min(int(depth_match.group(1)), 3)
+
+ collected.append(
+ Comment(author=author, text=text, created=created, depth=depth)
+ )
+
+ return render_comment_section(
+ "GeekNews Comments", collected, max_comments=MAX_COMMENTS
+ )
+
+
def fetch_geeknews_metrics(topic_id: str) -> Optional[dict]:
- """토픽 페이지에서 포인트와 댓글 수를 가져온다.
+ """토픽 페이지에서 포인트, 댓글 수, 댓글 본문을 가져온다.
RSS는 지표를 싣지 않아 `--days` 경로로 저장한 행은 likes/comments가 비어 있었다.
홈페이지 스크래핑 경로만 지표를 채우던 비대칭을 없앤다.
+ 같은 응답에 댓글 본문도 들어 있으므로 `comment_section`으로 함께 돌려준다.
"""
try:
resp = requests.get(
@@ -56,11 +96,17 @@ def fetch_geeknews_metrics(topic_id: str) -> Optional[dict]:
# 댓글 수는 링크 텍스트("댓글 3개")보다 data 속성이 안정적이다.
comment_el = soup.select_one("[data-topic-comment-count]")
- comments = _digits_to_int(comment_el.get("data-topic-comment-count")) if comment_el else None
+ comments = (
+ _digits_to_int(comment_el.get("data-topic-comment-count"))
+ if comment_el
+ else None
+ )
- if likes is None and comments is None:
+ comment_section = _parse_comment_section(soup)
+
+ if likes is None and comments is None and comment_section is None:
return None
- return {"likes": likes, "comments": comments}
+ return {"likes": likes, "comments": comments, "comment_section": comment_section}
class GeekNewsCrawler:
@@ -82,6 +128,9 @@ async def crawl(self, **options: Any) -> List[Post]:
if metrics:
item["likes"] = metrics["likes"]
item["comments"] = metrics["comments"]
+ item["content_markdown"] = append_comment_section(
+ item.get("content_markdown"), metrics.get("comment_section")
+ )
return [self._item_to_post(item) for item in items]
else:
return self._scrape_homepage(count)
@@ -114,7 +163,9 @@ def _scrape_homepage(self, count: int) -> List[Post]:
# 요약 텍스트
desc_el = row.select_one(".topicdesc a")
- description = desc_el.get_text(strip=True).lstrip("- ") if desc_el else ""
+ description = (
+ desc_el.get_text(strip=True).lstrip("- ") if desc_el else ""
+ )
# 메타 정보 (.topicinfo: 포인트, 작성자, 시간, 댓글)
topicinfo = row.select_one(".topicinfo")
diff --git a/packages/skim-core/src/skim_core/crawlers/feed/producthunt.py b/packages/skim-core/src/skim_core/crawlers/feed/producthunt.py
index ecd0a5c..c1337af 100644
--- a/packages/skim-core/src/skim_core/crawlers/feed/producthunt.py
+++ b/packages/skim-core/src/skim_core/crawlers/feed/producthunt.py
@@ -5,13 +5,22 @@
import re
from datetime import datetime, timedelta, timezone
-from typing import Any, List
+from typing import Any, List, Optional
+import requests
+from bs4 import BeautifulSoup
+
+from ...comments import Comment, append_comment_section, render_comment_section
from ...enrichment import enrich_with_content
from ...feed_config import PRODUCTHUNT_RSS
-from ...feed_utils import fetch_feed
+from ...feed_utils import FEED_HEADERS, fetch_feed
from ...models import Post
+MAX_COMMENTS = 15
+# 개별 댓글은 `data-test="comment-{id}"`로만 안정적으로 잡힌다.
+# 같은 접두사의 `comment-form`, `comments-feed`는 컨테이너라 제외해야 한다.
+_COMMENT_ID = re.compile(r"^comment-\d+$")
+
# PH 피드 content에는 제품 외부 사이트로 가는 리다이렉트(/r/p/{id})가 들어있다.
# /products/{slug} 페이지는 JS SPA + 403이라 본문 추출이 불가하므로, 이 링크를 enrich 대상으로 쓴다.
_RP_HREF = re.compile(r'href="(https://www\.producthunt\.com/r/p/[^"]+)"')
@@ -19,6 +28,65 @@
_TAGLINE = re.compile(r"\s*(.*?)\s*
", re.DOTALL)
+def fetch_comment_section(product_url: str) -> Optional[str]:
+ """제품 페이지의 댓글을 본문용 마크다운 섹션으로 만든다.
+
+ 본문 enrich는 외부 제품 사이트를 보므로 PH 페이지를 따로 받아야 한다(제품당 요청 1건).
+ """
+ if not product_url or "producthunt.com/products/" not in product_url:
+ return None
+ try:
+ response = requests.get(product_url, headers=FEED_HEADERS, timeout=20)
+ response.raise_for_status()
+ except Exception: # pylint: disable=broad-except
+ return None
+
+ soup = BeautifulSoup(response.text, "html.parser")
+ collected: List[Comment] = []
+ for node in soup.select("[data-test]"):
+ if not _COMMENT_ID.match(node.get("data-test") or ""):
+ continue
+ inner = node.select_one("div.flex.flex-1")
+ if not inner:
+ continue
+
+ # 자식 블록은 [작성자 헤더, 본문, 액션] 순이다. 액션은 time/Upvote로 확실히
+ # 걸러지지만 헤더는 아바타가 있을 때만 표시가 나서(없으면 이름만 든 텍스트 블록)
+ # 남은 블록의 첫 줄을 헤더로 본다.
+ created = None
+ blocks: List[str] = []
+ for block in inner.find_all(recursive=False):
+ text = " ".join(block.get_text(" ", strip=True).split())
+ if not text:
+ continue
+ time_el = block.select_one("time")
+ if time_el or text.startswith("Upvote"):
+ if time_el:
+ created = time_el.get("datetime")
+ continue
+ blocks.append(text)
+
+ if len(blocks) < 2: # 헤더만 있고 본문이 없는 노드
+ continue
+
+ # 아바타 alt가 가장 정확하다. 헤더 텍스트는 "이름 제품명 Maker" 꼴로 섞여 온다.
+ avatar = node.select_one("img[alt]")
+ author = (avatar.get("alt") if avatar else "") or blocks[0]
+ body = " ".join(blocks[1:]).strip()
+ if body:
+ collected.append(
+ Comment(
+ author=author or "unknown",
+ text=body,
+ created=(created or "")[:16].replace("T", " ") or None,
+ )
+ )
+
+ return render_comment_section(
+ "Product Hunt Comments", collected, max_comments=MAX_COMMENTS
+ )
+
+
def _redirect_url(item: dict) -> str:
"""피드 항목에서 제품 외부 사이트 리다이렉트 URL을 추출 (없으면 원 URL)."""
match = _RP_HREF.search(item.get("content_html", ""))
@@ -98,5 +166,10 @@ async def crawl(self, **options: Any) -> List[Post]:
if not no_content:
enrich_with_content(items)
+ for item in items:
+ item["content_markdown"] = append_comment_section(
+ item.get("content_markdown"),
+ fetch_comment_section(item.get("url", "")),
+ )
return [_item_to_post(item) for item in items]
diff --git a/packages/skim-core/src/skim_core/enrichment.py b/packages/skim-core/src/skim_core/enrichment.py
index f637867..7475e62 100644
--- a/packages/skim-core/src/skim_core/enrichment.py
+++ b/packages/skim-core/src/skim_core/enrichment.py
@@ -27,9 +27,11 @@
except ImportError: # pragma: no cover — optional dependency
fitz = None # type: ignore[assignment]
+from .comments import Comment, append_comment_section, render_comment_section
from .paths import workspace_root
SRT_TO_TXT = str(workspace_root() / "scripts" / "srt_to_txt.sh")
+YOUTUBE_MAX_COMMENTS = 15
def _run_group(cmd: list, timeout: int) -> subprocess.CompletedProcess:
@@ -243,7 +245,9 @@ def _looks_like_placeholder_content(content: str) -> bool:
return False
if normalized.lower() in _PLACEHOLDER_EXACT:
return True
- return bool(_PLACEHOLDER_START.search(normalized) or _PLACEHOLDER_ANY.search(normalized))
+ return bool(
+ _PLACEHOLDER_START.search(normalized) or _PLACEHOLDER_ANY.search(normalized)
+ )
def _is_content_usable(data: Optional[dict], title: str, min_words: int = 60) -> bool:
@@ -264,7 +268,9 @@ def _is_content_usable(data: Optional[dict], title: str, min_words: int = 60) ->
return True
-def extract_article_content(url: str, title: str) -> tuple[Optional[dict], str, Optional[str]]:
+def extract_article_content(
+ url: str, title: str
+) -> tuple[Optional[dict], str, Optional[str]]:
"""
품질 게이트가 붙은 본문 추출 — Python 내부에서 전부 처리.
@@ -284,7 +290,9 @@ def extract_article_content(url: str, title: str) -> tuple[Optional[dict], str,
thin_reason_1 = (
"http fetch failed"
if not html
- else "trafilatura empty" if not data else f"thin (words={data.get('word_count', 0)})"
+ else "trafilatura empty"
+ if not data
+ else f"thin (words={data.get('word_count', 0)})"
)
# 2) Playwright 렌더 + trafilatura
@@ -326,7 +334,13 @@ def extract_article_content(url: str, title: str) -> tuple[Optional[dict], str,
print(f" [!] defuddle fallback 예외: {e}")
fallback = rendered_data or data
- best_method = "playwright+trafilatura" if rendered_data else "trafilatura" if data else "failed"
+ best_method = (
+ "playwright+trafilatura"
+ if rendered_data
+ else "trafilatura"
+ if data
+ else "failed"
+ )
return fallback, best_method, f"{thin_reason_1}; {thin_reason_2}"
@@ -443,6 +457,52 @@ def _pref_rank(path: Path) -> int:
}
+def extract_youtube_comments(
+ url: str, limit: int = YOUTUBE_MAX_COMMENTS, timeout: int = 90
+) -> Optional[str]:
+ """yt-dlp로 상위 댓글을 받아 본문용 마크다운 섹션으로 만든다.
+
+ 자막과 달리 요청이 따로 들어가므로 실패해도 자막 추출을 막지 않는다.
+ 답글은 받지 않는다(max_comments의 마지막 인자 0) — 영상 댓글의 답글은
+ 대개 잡담이라 본문 신호 대비 길이만 늘린다.
+ """
+ with tempfile.TemporaryDirectory() as tmpdir:
+ _run_group(
+ [
+ "yt-dlp",
+ "--skip-download",
+ "--write-comments",
+ "--write-info-json",
+ "--extractor-args",
+ f"youtube:comment_sort=top;max_comments={limit},all,{limit},0",
+ "-o",
+ f"{tmpdir}/%(id)s.%(ext)s",
+ url,
+ ],
+ timeout=timeout,
+ )
+
+ info_files = list(Path(tmpdir).glob("*.info.json"))
+ if not info_files:
+ return None
+ try:
+ info = json.loads(info_files[0].read_text(encoding="utf-8"))
+ except (OSError, ValueError):
+ return None
+
+ collected = [
+ Comment(
+ author=str(raw.get("author") or "unknown"),
+ text=str(raw.get("text") or ""),
+ score=raw.get("like_count"),
+ )
+ for raw in (info.get("comments") or [])
+ ]
+ return render_comment_section(
+ "YouTube Comments", collected, max_comments=limit, score_unit="like"
+ )
+
+
def _parse_subtitle_codes(section_text: str) -> List[str]:
"""yt-dlp --list-subs 출력에서 자막 코드 목록을 추출합니다."""
codes: List[str] = []
@@ -490,9 +550,13 @@ def _resolve_preferred_subtitle_codes(available_codes: List[str]) -> List[str]:
if not available_codes:
return []
- if any(code == "en-orig" or code.startswith("en-orig-") for code in available_codes):
+ if any(
+ code == "en-orig" or code.startswith("en-orig-") for code in available_codes
+ ):
prefixes = ["en-orig", "en", "ko"]
- elif any(code == "ko-orig" or code.startswith("ko-orig-") for code in available_codes):
+ elif any(
+ code == "ko-orig" or code.startswith("ko-orig-") for code in available_codes
+ ):
prefixes = ["ko-orig", "ko", "en"]
else:
prefixes = ["en", "ko"]
@@ -593,6 +657,21 @@ def _enrich_article_item(item: dict, url: str) -> Optional[dict]:
return data
+def _apply_youtube_comments(item: dict, url: str) -> None:
+ """영상 댓글을 본문 뒤에 잇는다. 자막과 별개 요청이라 자막이 없어도 남길 가치가 있다."""
+ try:
+ section = extract_youtube_comments(url)
+ except Exception as e: # noqa: BLE001 - 댓글 실패가 영상 저장을 막지 않는다
+ print(f" [!] 댓글 추출 실패: {e}")
+ return
+ if not section:
+ return
+ item["content_markdown"] = append_comment_section(
+ item.get("content_markdown"), section
+ )
+ print(f" -> 댓글 {section.count(chr(10) + '- ') + 1}건")
+
+
def enrich_with_content(items: List[dict]) -> List[dict]:
"""각 항목에 defuddle로 원문 콘텐츠 추가"""
targets = list(items)
@@ -631,6 +710,8 @@ def enrich_with_content(items: List[dict]) -> List[dict]:
item["content_markdown"] = ""
item["word_count"] = 0
print(f" [!] 자막 추출 실패: {e}")
+
+ _apply_youtube_comments(item, url)
continue
# GeekNews: 토픽 페이지의 한국어 요약을 1차 본문으로 삼고, 원문 추출은 뒤에 붙인다.
@@ -715,7 +796,9 @@ def _pdf_page_text(page) -> str:
return "\n".join(b[4].strip() for b in left + right)
-def extract_pdf_text(pdf_url: str, timeout: int = 30, min_words: int = 100) -> Optional[dict]:
+def extract_pdf_text(
+ pdf_url: str, timeout: int = 30, min_words: int = 100
+) -> Optional[dict]:
"""arXiv PDF를 내려받아 본문 텍스트를 추출 (2단 레이아웃 대응)."""
if fitz is None:
return None
@@ -788,7 +871,9 @@ def enrich_papers_with_content(items: List[dict]) -> List[dict]:
item["content_markdown"] = abstract
item["word_count"] = len(abstract.split())
item["enrichment_method"] = "failed"
- print(f" -> HTML/PDF 없음, abstract 폴백 ({item['word_count']} words)")
+ print(
+ f" -> HTML/PDF 없음, abstract 폴백 ({item['word_count']} words)"
+ )
else:
item["content_markdown"] = ""
item["word_count"] = 0
@@ -796,7 +881,9 @@ def enrich_papers_with_content(items: List[dict]) -> List[dict]:
print(" -> HTML/PDF/abstract 모두 없음")
html_n = sum(
- 1 for it in targets if it.get("enrichment_method") is None and it.get("word_count", 0) > 0
+ 1
+ for it in targets
+ if it.get("enrichment_method") is None and it.get("word_count", 0) > 0
)
pdf_n = sum(1 for it in targets if it.get("enrichment_method") == "pdf")
abstract_n = sum(
diff --git a/packages/skim-core/src/skim_core/feed_utils.py b/packages/skim-core/src/skim_core/feed_utils.py
index f06860c..09d1aad 100644
--- a/packages/skim-core/src/skim_core/feed_utils.py
+++ b/packages/skim-core/src/skim_core/feed_utils.py
@@ -14,10 +14,13 @@
# 저장 측은 UTC ISO 8601 로 강제 (fetch_feed `published` 필드).
KST = timezone(timedelta(hours=9))
FEED_TIMEOUT_SECONDS = 15
+# news.hada.io는 브라우저 토큰뿐 아니라 Chrome 메이저 버전도 본다. 2026-08-09부터
+# Chrome/124가 403으로 막혀 그날 지표 수집이 절반 실패했다(128 이상은 통과).
+# 차단선이 다시 올라가면 이 버전을 올린다.
FEED_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
- "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
+ "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
)
}
@@ -36,7 +39,9 @@ def is_within_range(entry_dt: Optional[datetime], since: datetime) -> bool:
return entry_dt >= since
-def fetch_feed(url: str, source_name: str, since: datetime, quiet: bool = False) -> List[dict]:
+def fetch_feed(
+ url: str, source_name: str, since: datetime, quiet: bool = False
+) -> List[dict]:
"""RSS/Atom 피드를 가져와서 since 이후 항목만 반환"""
try:
response = requests.get(url, headers=FEED_HEADERS, timeout=FEED_TIMEOUT_SECONDS)
@@ -90,9 +95,9 @@ def fetch_feed(url: str, source_name: str, since: datetime, quiet: bool = False)
"external_id": entry.get("id", ""),
"published": entry_dt.isoformat() if entry_dt else "",
"summary": (
- re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", entry.get("summary") or "")).strip()[
- :300
- ]
+ re.sub(
+ r"\s+", " ", re.sub(r"<[^>]+>", "", entry.get("summary") or "")
+ ).strip()[:300]
),
}
)
diff --git a/tests/test_comment_sections.py b/tests/test_comment_sections.py
new file mode 100644
index 0000000..3dd7018
--- /dev/null
+++ b/tests/test_comment_sections.py
@@ -0,0 +1,446 @@
+"""댓글 본문 합성 회귀 테스트.
+
+`AGENTS.md`의 데이터 계약상 토론은 content_markdown에 함께 담긴다. 플랫폼마다
+응답 구조가 달라 파서가 조용히 틀리기 쉬우므로, 실제 응답 모양을 픽스처로 고정한다.
+"""
+
+import json
+import unittest
+from unittest.mock import MagicMock, patch
+
+from bs4 import BeautifulSoup
+
+from skim_core.comments import (
+ Comment,
+ append_comment_section,
+ render_comment_section,
+)
+from skim_core.crawlers.api.linkedin import LinkedInAPICrawler
+from skim_core.crawlers.api.reddit import RedditAPICrawler
+from skim_core.crawlers.api.threads import ThreadsAPICrawler
+from skim_core.crawlers.feed.geeknews import _parse_comment_section
+from skim_core.crawlers.feed.producthunt import fetch_comment_section as ph_comments
+
+
+class RenderTests(unittest.TestCase):
+ def test_renders_author_score_and_time(self):
+ section = render_comment_section(
+ "Demo Comments",
+ [
+ Comment(
+ author="alice", text="hello", score=3, created="2026-08-09 10:00"
+ )
+ ],
+ )
+ self.assertIn("## Demo Comments", section)
+ self.assertIn("- **alice** (3 points, 2026-08-09 10:00): hello", section)
+
+ def test_score_unit_is_configurable_and_singular(self):
+ section = render_comment_section(
+ "Demo", [Comment(author="a", text="t", score=1)], score_unit="like"
+ )
+ self.assertIn("(1 like)", section)
+
+ def test_depth_indents_and_newlines_are_folded(self):
+ section = render_comment_section(
+ "Demo",
+ [
+ Comment(author="a", text="parent"),
+ Comment(author="b", text="line1\nline2", depth=1),
+ ],
+ )
+ # 줄바꿈이 남으면 목록 항목이 깨진다.
+ self.assertIn(" - **b**: line1 line2", section)
+
+ def test_returns_none_when_nothing_usable(self):
+ self.assertIsNone(render_comment_section("Demo", []))
+ self.assertIsNone(
+ render_comment_section("Demo", [Comment(author="a", text=" ")])
+ )
+
+ def test_long_text_is_truncated(self):
+ section = render_comment_section(
+ "Demo", [Comment(author="a", text="x" * 5000)], max_chars=100
+ )
+ self.assertIn("...", section)
+ self.assertLess(len(section), 300)
+
+ def test_max_comments_caps_output(self):
+ section = render_comment_section(
+ "Demo",
+ [Comment(author=f"u{i}", text=f"t{i}") for i in range(50)],
+ max_comments=3,
+ )
+ self.assertEqual(section.count("\n- **"), 3)
+
+ def test_append_keeps_body_and_separator(self):
+ self.assertEqual(append_comment_section("body", None), "body")
+ self.assertEqual(append_comment_section(None, "## S\n\n- a"), "## S\n\n- a")
+ self.assertEqual(append_comment_section("body", "## S"), "body\n\n---\n\n## S")
+
+
+GEEKNEWS_HTML = """
+
+
+
+
+"""
+
+
+class GeekNewsCommentTests(unittest.TestCase):
+ def test_parses_author_time_and_depth(self):
+ section = _parse_comment_section(BeautifulSoup(GEEKNEWS_HTML, "html.parser"))
+ self.assertIn("- **princox** (2026-08-09 17:23): 본문 하나", section)
+ self.assertIn(" - **xguru**: 답글", section)
+
+ def test_no_comments_returns_none(self):
+ self.assertIsNone(
+ _parse_comment_section(BeautifulSoup("", "html.parser"))
+ )
+
+
+REDDIT_PAYLOAD = [
+ {"kind": "Listing", "data": {"children": []}},
+ {
+ "kind": "Listing",
+ "data": {
+ "children": [
+ {
+ "kind": "t1",
+ "data": {
+ "author": "alice",
+ "body": "top comment",
+ "score": 12,
+ "created_utc": 1786000000,
+ "replies": {
+ "data": {
+ "children": [
+ {
+ "kind": "t1",
+ "data": {
+ "author": "bob",
+ "body": "a reply",
+ "score": 3,
+ "created_utc": 1786000100,
+ },
+ }
+ ]
+ }
+ },
+ },
+ },
+ {
+ "kind": "t1",
+ "data": {"author": "x", "body": "[deleted]", "score": 1},
+ },
+ {"kind": "more", "data": {"count": 20}},
+ ]
+ },
+ },
+]
+
+
+class RedditCommentTests(unittest.TestCase):
+ def _crawler(self):
+ with patch.object(RedditAPICrawler, "_load_session_cookies", return_value=None):
+ return RedditAPICrawler()
+
+ def test_parses_tree_and_skips_deleted_and_more(self):
+ crawler = self._crawler()
+ with patch.object(crawler, "fetch_listing_page", return_value=REDDIT_PAYLOAD):
+ section = crawler.fetch_comment_section(
+ "https://www.reddit.com/r/a/comments/b/c/"
+ )
+
+ self.assertIn("- **u/alice** (12 points,", section)
+ self.assertIn(" - **u/bob** (3 points,", section)
+ self.assertNotIn("[deleted]", section)
+
+ def test_request_failure_returns_none(self):
+ crawler = self._crawler()
+ with patch.object(crawler, "fetch_listing_page", side_effect=Exception("boom")):
+ self.assertIsNone(
+ crawler.fetch_comment_section(
+ "https://www.reddit.com/r/a/comments/b/c/"
+ )
+ )
+
+ def test_attach_stops_after_consecutive_failures(self):
+ """세션이 죽으면 남은 게시글까지 헛 요청하지 않는다."""
+ crawler = self._crawler()
+ posts = [
+ MagicMock(
+ url=f"https://www.reddit.com/r/a/comments/{i}/x/",
+ content="b",
+ content_markdown=None,
+ )
+ for i in range(10)
+ ]
+ with (
+ patch.object(crawler, "fetch_comment_section", return_value=None) as fetch,
+ patch("skim_core.crawlers.api.reddit.time.sleep"),
+ ):
+ crawler.attach_comments(posts)
+
+ self.assertEqual(fetch.call_count, 3)
+
+
+LINKEDIN_PAYLOAD = {
+ "included": [
+ {
+ "$type": "com.linkedin.voyager.feed.Comment",
+ "commentV2": {"text": "top level"},
+ "commenterForDashConversion": {"title": {"text": "Jane Doe"}},
+ "createdTime": 1786054696447,
+ "parentCommentUrn": None,
+ },
+ {
+ "$type": "com.linkedin.voyager.feed.Comment",
+ "commentV2": {"text": "a reply"},
+ "commenterForDashConversion": {"title": {"text": "John Roe"}},
+ "createdTime": 1786054796447,
+ "parentCommentUrn": "urn:li:comment:(1,2)",
+ },
+ {
+ "$type": "com.linkedin.voyager.identity.shared.MiniProfile",
+ "lastName": "Doe",
+ },
+ ]
+}
+
+
+class LinkedInCommentTests(unittest.TestCase):
+ def _crawler(self):
+ with patch.object(
+ LinkedInAPICrawler, "_load_session_cookies", return_value=None
+ ):
+ return LinkedInAPICrawler()
+
+ def test_reads_included_and_skips_replies(self):
+ crawler = self._crawler()
+ response = MagicMock(status_code=200)
+ response.json.return_value = LINKEDIN_PAYLOAD
+ with patch.object(crawler.session, "get", return_value=response):
+ section = crawler.fetch_comment_section("7492224454731714560")
+
+ self.assertIn("- **Jane Doe**", section)
+ self.assertIn("top level", section)
+ # RELEVANCE 정렬이 답글을 부모보다 먼저 주기도 해서 최상위만 담는다.
+ self.assertNotIn("a reply", section)
+
+ def test_non_200_returns_none(self):
+ crawler = self._crawler()
+ with patch.object(
+ crawler.session, "get", return_value=MagicMock(status_code=403)
+ ):
+ self.assertIsNone(crawler.fetch_comment_section("7492224454731714560"))
+
+
+PRODUCTHUNT_HTML = """
+
+
+
+
+
+
Chirag Chopra Workflo Maker
+
실제 본문입니다
+
Upvote (2)
+
+
+
+
+
Abdullah Javaid
+
아바타 없는 댓글
+
+
+
+
Login to comment
+
본문 하나