feat(crawlers): 댓글과 답글을 정본 본문에 담는다

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`
 

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}"

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()