variet-agent/tools/community_scraper.py

r"""커뮤니티 트렌드 스크래퍼 — DCInside 갤러리 인기글 조회 + 본문 스크래핑.

사용법:
    # 인기글 랭킹 (기본 6시간, 상위 10개, 복합 스코어)
    python tools/community_scraper.py trend coffee

    # 조회수 기준 상위 5개
    python tools/community_scraper.py trend coffee --sort views --top 5

    # 추천 기준 3시간 이내
    python tools/community_scraper.py trend coffee --sort recommends --hours 3

    # 개별 글 본문 조회
    python tools/community_scraper.py view coffee 544618

    # 일반 갤러리 (마이너 아닌 경우)
    python tools/community_scraper.py trend programming --type gallery
"""

import argparse
import asyncio
import json
import logging
import re
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from typing import Optional

import httpx

try:
    from bs4 import BeautifulSoup
except ImportError:
    print("ERROR: beautifulsoup4 필요. pip install beautifulsoup4 lxml", file=sys.stderr)
    sys.exit(1)

logger = logging.getLogger("variet.tools.community")

# ──────────────────────────────────────────────
#  데이터 모델
# ──────────────────────────────────────────────

KST = timezone(timedelta(hours=9))

SORT_OPTIONS = ("score", "views", "recommends", "comments")


@dataclass
class CommunityPost:
    """스크래핑된 게시글."""

    no: int
    title: str
    author: str
    date: str  # ISO 8601
    views: int
    recommends: int
    comments: int
    url: str
    gallery_id: str
    category: str = ""
    score: float = 0.0

    def compute_score(self) -> float:
        """가중 복합 스코어 계산."""
        self.score = self.views + (self.recommends * 10) + (self.comments * 5)
        return self.score


# ──────────────────────────────────────────────
#  DCInside 스크래퍼
# ──────────────────────────────────────────────

# 요청 헤더 (차단 방지)
_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/131.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "ko-KR,ko;q=0.9,en;q=0.8",
    "Referer": "https://gall.dcinside.com/",
}

# 갤러리 타입별 URL 패턴
_BASE_URLS = {
    "mgallery": "https://gall.dcinside.com/mgallery/board",
    "mini": "https://gall.dcinside.com/mini/board",
    "gallery": "https://gall.dcinside.com/board",
}


class DCInsideScraper:
    """DCInside 갤러리 스크래퍼."""

    def __init__(self, request_delay: float = 0.5):
        self.delay = request_delay

    def _list_url(self, gallery_id: str, gtype: str, page: int) -> str:
        base = _BASE_URLS.get(gtype, _BASE_URLS["mgallery"])
        return f"{base}/lists/?id={gallery_id}&page={page}"

    def _view_url(self, gallery_id: str, gtype: str, post_no: int) -> str:
        base = _BASE_URLS.get(gtype, _BASE_URLS["mgallery"])
        return f"{base}/view/?id={gallery_id}&no={post_no}"

    async def fetch_posts(
        self,
        gallery_id: str,
        gallery_type: str = "mgallery",
        pages: int = 3,
    ) -> list[CommunityPost]:
        """갤러리 목록 스크래핑 → 게시글 리스트 반환."""
        posts: list[CommunityPost] = []
        seen_nos: set[int] = set()

        async with httpx.AsyncClient(
            headers=_HEADERS, follow_redirects=True, timeout=15.0
        ) as client:
            for page in range(1, pages + 1):
                url = self._list_url(gallery_id, gallery_type, page)
                try:
                    resp = await client.get(url)
                    resp.raise_for_status()
                except httpx.HTTPError as e:
                    logger.warning(f"페이지 {page} 요청 실패: {e}")
                    continue

                page_posts = self._parse_list_page(
                    resp.text, gallery_id, gallery_type
                )
                for p in page_posts:
                    if p.no not in seen_nos:
                        seen_nos.add(p.no)
                        posts.append(p)

                if page < pages:
                    await asyncio.sleep(self.delay)

        return posts

    def _parse_list_page(
        self, html: str, gallery_id: str, gallery_type: str
    ) -> list[CommunityPost]:
        """HTML 파싱 → CommunityPost 리스트."""
        soup = BeautifulSoup(html, "lxml")
        rows = soup.select("tr.ub-content")
        posts: list[CommunityPost] = []

        now = datetime.now(KST)

        for row in rows:
            try:
                # 공지 스킵
                num_cell = row.select_one("td.gall_num")
                if not num_cell:
                    continue
                num_text = num_cell.get_text(strip=True)
                if num_text in ("공지", "설문", "AD"):
                    continue

                # 글 번호
                post_no = int(row.get("data-no", 0) or num_text)
                if post_no == 0:
                    continue

                # 제목
                title_link = row.select_one("td.gall_tit a:first-of-type")
                title = title_link.get_text(strip=True) if title_link else ""

                # 댓글 수
                reply_span = row.select_one("td.gall_tit span.reply_num")
                comments = 0
                if reply_span:
                    m = re.search(r"\d+", reply_span.get_text())
                    comments = int(m.group()) if m else 0

                # 카테고리
                subject_cell = row.select_one("td.gall_subject")
                category = subject_cell.get_text(strip=True) if subject_cell else ""

                # 작성자
                writer_cell = row.select_one("td.gall_writer")
                author = ""
                if writer_cell:
                    nick = writer_cell.select_one(".nickname")
                    author = nick.get_text(strip=True) if nick else ""

                # 날짜
                date_cell = row.select_one("td.gall_date")
                date_str = ""
                if date_cell:
                    # title 속성에 전체 날짜가 있을 수 있음
                    full_date = date_cell.get("title", "")
                    if full_date:
                        date_str = full_date
                    else:
                        raw = date_cell.get_text(strip=True)
                        # HH:mm 형식이면 오늘 날짜 붙이기
                        if re.match(r"^\d{2}:\d{2}$", raw):
                            date_str = f"{now.strftime('%Y-%m-%d')} {raw}:00"
                        elif re.match(r"^\d{2}\.\d{2}$", raw):
                            date_str = f"{now.year}-{raw.replace('.', '-')} 00:00:00"
                        else:
                            date_str = raw

                # 조회수
                count_cell = row.select_one("td.gall_count")
                views = 0
                if count_cell:
                    v_text = count_cell.get_text(strip=True).replace(",", "")
                    views = int(v_text) if v_text.isdigit() else 0

                # 추천수
                rec_cell = row.select_one("td.gall_recommend")
                recommends = 0
                if rec_cell:
                    r_text = rec_cell.get_text(strip=True).replace(",", "")
                    recommends = int(r_text) if r_text.isdigit() else 0

                # URL
                url = self._view_url(gallery_id, gallery_type, post_no)

                post = CommunityPost(
                    no=post_no,
                    title=title,
                    author=author,
                    date=date_str,
                    views=views,
                    recommends=recommends,
                    comments=comments,
                    url=url,
                    gallery_id=gallery_id,
                    category=category,
                )
                post.compute_score()
                posts.append(post)

            except Exception as e:
                logger.debug(f"행 파싱 오류: {e}")
                continue

        return posts

    async def fetch_post_content(
        self,
        gallery_id: str,
        post_no: int,
        gallery_type: str = "mgallery",
    ) -> dict:
        """개별 게시글 본문 조회.

        Returns:
            {title, author, date, views, recommends, comments, content, url}
        """
        url = self._view_url(gallery_id, gallery_type, post_no)

        async with httpx.AsyncClient(
            headers=_HEADERS, follow_redirects=True, timeout=15.0
        ) as client:
            resp = await client.get(url)
            resp.raise_for_status()

        soup = BeautifulSoup(resp.text, "lxml")

        # 제목
        title_el = soup.select_one("span.title_subject")
        title = title_el.get_text(strip=True) if title_el else ""

        # 작성자
        nick_el = soup.select_one("div.gall_writer .nickname")
        author = nick_el.get_text(strip=True) if nick_el else ""

        # 날짜
        date_el = soup.select_one("span.gall_date")
        date_str = date_el.get_text(strip=True) if date_el else ""

        # 본문
        body_el = soup.select_one("div.write_div")
        content = ""
        if body_el:
            # 이미지 alt 텍스트 보존, 나머지 HTML 태그 제거
            for img in body_el.find_all("img"):
                alt = img.get("alt", "")
                if alt:
                    img.replace_with(f"[이미지: {alt}]")
                else:
                    img.replace_with("[이미지]")
            # 줄바꿈 보존
            for br in body_el.find_all("br"):
                br.replace_with("\n")
            for p in body_el.find_all("p"):
                p.insert_after("\n")
            content = body_el.get_text().strip()
            # 연속 줄바꿈 정리
            content = re.sub(r"\n{3,}", "\n\n", content)

        # 조회수: <span class="gall_count">조회 685</span>
        views = 0
        views_el = soup.select_one("span.gall_count")
        if views_el:
            m = re.search(r"\d[\d,]*", views_el.get_text())
            views = int(m.group().replace(",", "")) if m else 0

        # 추천수: <span class="up_num">추천 24</span> 또는 기타 위치
        rec = 0
        rec_el = soup.select_one("span.up_num, p.up_num span, .btn_recommend_box .up_num")
        if rec_el:
            m = re.search(r"\d[\d,]*", rec_el.get_text())
            rec = int(m.group().replace(",", "")) if m else 0

        # 댓글수: <a href="#focus_cmt">댓글 35</a>
        cmt = 0
        cmt_el = soup.select_one("a[href='#focus_cmt'], span.gall_comment")
        if cmt_el:
            m = re.search(r"\d[\d,]*", cmt_el.get_text())
            cmt = int(m.group().replace(",", "")) if m else 0

        return {
            "title": title,
            "author": author,
            "date": date_str,
            "views": views,
            "recommends": rec,
            "comments": cmt,
            "content": content,
            "url": url,
        }


# ──────────────────────────────────────────────
#  랭킹 & 포매팅
# ──────────────────────────────────────────────

def rank_posts(
    posts: list[CommunityPost],
    hours: float = 6.0,
    top_n: int = 10,
    sort_by: str = "score",
) -> list[CommunityPost]:
    """시간 필터링 + 정렬.

    Args:
        posts: 게시글 리스트
        hours: 최근 N시간 이내 필터 (0이면 필터 없음)
        top_n: 상위 N개
        sort_by: score | views | recommends | comments
    """
    now = datetime.now(KST)

    if hours > 0:
        cutoff = now - timedelta(hours=hours)
        filtered = []
        for p in posts:
            try:
                # 다양한 날짜 포맷 파싱
                for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y.%m.%d %H:%M:%S", "%Y.%m.%d %H:%M"):
                    try:
                        dt = datetime.strptime(p.date, fmt).replace(tzinfo=KST)
                        break
                    except ValueError:
                        continue
                else:
                    # 파싱 실패 시 포함 (데이터 유실 방지)
                    filtered.append(p)
                    continue

                if dt >= cutoff:
                    filtered.append(p)
            except Exception:
                filtered.append(p)
    else:
        filtered = list(posts)

    # 스코어 재계산
    for p in filtered:
        p.compute_score()

    # 정렬
    key_map = {
        "score": lambda p: p.score,
        "views": lambda p: p.views,
        "recommends": lambda p: p.recommends,
        "comments": lambda p: p.comments,
    }
    key_fn = key_map.get(sort_by, key_map["score"])
    filtered.sort(key=key_fn, reverse=True)

    return filtered[:top_n]


def format_ranking(posts: list[CommunityPost], sort_by: str = "score") -> str:
    """랭킹 결과를 마크다운으로 포맷."""
    if not posts:
        return "해당 기간에 게시글이 없습니다."

    sort_labels = {
        "score": "종합점수",
        "views": "조회수",
        "recommends": "추천수",
        "comments": "댓글수",
    }
    label = sort_labels.get(sort_by, "종합점수")

    lines = [f"## 커뮤니티 인기글 ({label} 기준)\n"]
    lines.append("| # | 제목 | 조회 | 추천 | 댓글 | 점수 |")
    lines.append("|---|------|------|------|------|------|")

    for i, p in enumerate(posts, 1):
        # 제목 길이 제한
        title = p.title[:40] + ("…" if len(p.title) > 40 else "")
        lines.append(
            f"| {i} | {title} | {p.views} | {p.recommends} | {p.comments} | {p.score:.0f} |"
        )

    lines.append(f"\n> 갤러리: {posts[0].gallery_id} | "
                 f"조회 기준 시각: {datetime.now(KST).strftime('%Y-%m-%d %H:%M')}")

    return "\n".join(lines)


def format_post_detail(data: dict) -> str:
    """게시글 본문을 마크다운으로 포맷."""
    lines = [
        f"## {data['title']}",
        f"**작성자**: {data['author']} | **날짜**: {data['date']}",
        f"**조회** {data['views']} | **추천** {data['recommends']}",
        "",
        "---",
        "",
        data["content"][:3000] if data["content"] else "(본문 없음)",
        "",
        f"> 원문: {data['url']}",
    ]
    return "\n".join(lines)


# ──────────────────────────────────────────────
#  JSON 출력 (Gemini CLI 연동용)
# ──────────────────────────────────────────────

def output_json(data):
    """JSON 출력 (Gemini CLI가 파싱하기 쉽도록)."""
    print(json.dumps(data, ensure_ascii=False, indent=2, default=str))


# ──────────────────────────────────────────────
#  CLI 진입점
# ──────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="커뮤니티 인기글 조회 도구 (DCInside)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""예시:
  trend coffee                       # 커피갤 인기글 (6시간, 상위 10개)
  trend coffee --hours 3 --top 5     # 3시간, 상위 5개
  trend coffee --sort views --top 5  # 조회수 기준 상위 5개
  trend electricguitar --hours 12    # 일렉기타갤 12시간
  view coffee 544618                 # 글 본문 조회
""",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    # trend 커맨드
    trend_p = sub.add_parser("trend", help="인기글 랭킹")
    trend_p.add_argument("gallery_id", help="갤러리 ID (예: coffee, electricguitar)")
    trend_p.add_argument("--type", dest="gallery_type", default="mgallery",
                         choices=["mgallery", "mini", "gallery"],
                         help="갤러리 유형 (기본: mgallery)")
    trend_p.add_argument("--hours", type=float, default=6.0,
                         help="최근 N시간 이내 (기본: 6)")
    trend_p.add_argument("--top", type=int, default=10,
                         help="상위 N개 (기본: 10)")
    trend_p.add_argument("--sort", default="score",
                         choices=SORT_OPTIONS,
                         help="정렬 기준: score(종합), views(조회수), recommends(추천), comments(댓글)")
    trend_p.add_argument("--pages", type=int, default=5,
                         help="스크래핑할 페이지 수 (기본: 5)")
    trend_p.add_argument("--format", dest="out_format", default="markdown",
                         choices=["markdown", "json"],
                         help="출력 형식 (기본: markdown)")

    # view 커맨드
    view_p = sub.add_parser("view", help="글 본문 조회")
    view_p.add_argument("gallery_id", help="갤러리 ID")
    view_p.add_argument("post_no", type=int, help="글 번호")
    view_p.add_argument("--type", dest="gallery_type", default="mgallery",
                         choices=["mgallery", "mini", "gallery"],
                         help="갤러리 유형 (기본: mgallery)")
    view_p.add_argument("--format", dest="out_format", default="markdown",
                         choices=["markdown", "json"],
                         help="출력 형식 (기본: markdown)")

    args = parser.parse_args()

    asyncio.run(_async_main(args))


async def _async_main(args):
    scraper = DCInsideScraper()

    if args.command == "trend":
        posts = await scraper.fetch_posts(
            args.gallery_id,
            gallery_type=args.gallery_type,
            pages=args.pages,
        )

        ranked = rank_posts(
            posts,
            hours=args.hours,
            top_n=args.top,
            sort_by=args.sort,
        )

        if args.out_format == "json":
            output_json([asdict(p) for p in ranked])
        else:
            print(format_ranking(ranked, sort_by=args.sort))

    elif args.command == "view":
        data = await scraper.fetch_post_content(
            args.gallery_id,
            args.post_no,
            gallery_type=args.gallery_type,
        )

        if args.out_format == "json":
            output_json(data)
        else:
            print(format_post_detail(data))


if __name__ == "__main__":
    main()