crawl_docs_deep.py

# -*- coding: utf-8 -*-
"""
Pico XR 文档深度爬虫 - 基于已知 URL 列表
直接使用从 sitemap 和导航页提取的 URL 列表进行爬取
"""

import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')

import asyncio
import json
import re
import time
from pathlib import Path
from urllib.parse import urlparse

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode

DATA_DIR = Path(__file__).parent / "data"
DATA_DIR.mkdir(exist_ok=True)

# 已知的所有文档 URL（从 sitemap + 导航页提取）
KNOWN_URLS = [
    # === Unity SDK ===
    "https://developer-cn.picoxr.com/document/unity/overview/",
    "https://developer-cn.picoxr.com/document/unity/splash-screen/",
    "https://developer-cn.picoxr.com/document/unity/input-mapping/",
    "https://developer-cn.picoxr.com/document/unity/seethrough/",
    "https://developer-cn.picoxr.com/document/unity/spatial-audio/",
    "https://developer-cn.picoxr.com/document/unity/metrics-hud/",
    "https://developer-cn.picoxr.com/document/unity/242767/",
    "https://developer-cn.picoxr.com/document/unity/renderdoc-for-pico/",
    "https://developer-cn.picoxr.com/document/unity/command-line-utility/",
    # Unity 平台服务
    "https://developer-cn.picoxr.com/document/unity/accounts-and-friends/",
    "https://developer-cn.picoxr.com/document/unity/interaction/",
    "https://developer-cn.picoxr.com/document/unity/achievements/",
    "https://developer-cn.picoxr.com/document/unity/matchmaking/",
    "https://developer-cn.picoxr.com/document/unity/leaderboard/",
    "https://developer-cn.picoxr.com/document/unity/in-app-purchase/",
    "https://developer-cn.picoxr.com/document/unity/downloadable-content/",
    "https://developer-cn.picoxr.com/document/unity/subscription/",
    "https://developer-cn.picoxr.com/document/unity/cloud-storage/",
    # Unity Avatar
    "https://developer-cn.picoxr.com/document/unity-avatar/",
    # === Unreal SDK ===
    "https://developer-cn.picoxr.com/document/unreal/unreal-sdk-overview/",
    "https://developer-cn.picoxr.com/document/unreal/",
    # === Native SDK ===
    "https://developer-cn.picoxr.com/document/native/native-sdk-overview/",
    "https://developer-cn.picoxr.com/document/native/",
    # === OpenXR ===
    "https://developer-cn.picoxr.com/document/openxr/openxr-overview/",
    "https://developer-cn.picoxr.com/document/openxr/",
    # === 设备介绍 ===
    "https://developer-cn.picoxr.com/document/discover/xr-devices/",
    "https://developer-cn.picoxr.com/document/discover/",
    # === 设计指南 ===
    "https://developer-cn.picoxr.com/document/design/game-genres/",
    "https://developer-cn.picoxr.com/document/design/visual-design/",
    "https://developer-cn.picoxr.com/document/design/locomotion/",
    "https://developer-cn.picoxr.com/document/design/dock-icon-and-album-cover/",
    "https://developer-cn.picoxr.com/document/design/",
    # === 分发上架 ===
    "https://developer-cn.picoxr.com/document/distribute/app-distribution-overview/",
    "https://developer-cn.picoxr.com/document/distribute/app-review-overview/",
    "https://developer-cn.picoxr.com/document/distribute/create-a-developer-account/",
    "https://developer-cn.picoxr.com/document/distribute/create-an-app/",
    "https://developer-cn.picoxr.com/document/distribute/app-information-review/",
    "https://developer-cn.picoxr.com/document/distribute/formal-app/",
    "https://developer-cn.picoxr.com/document/distribute/analytics-IAP/",
    "https://developer-cn.picoxr.com/document/distribute/",
    # === 安全信息 ===
    "https://developer-cn.picoxr.com/document/safe-info/",
]


def get_category(url: str) -> str:
    path = urlparse(url).path
    if "/unity" in path:
        return "Unity SDK"
    elif "/unreal/" in path:
        return "Unreal SDK"
    elif "/native/" in path:
        return "Native SDK"
    elif "/openxr/" in path:
        return "OpenXR"
    elif "/discover/" in path:
        return "设备介绍"
    elif "/design/" in path:
        return "设计指南"
    elif "/distribute/" in path:
        return "分发上架"
    elif "/safe-info/" in path:
        return "安全信息"
    else:
        return "其他"


def clean_markdown(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"\n{4,}", "\n\n\n", text)
    return text.strip()


def save_docs(docs: list[dict]):
    with open(DATA_DIR / "pico_docs_full.json", "w", encoding="utf-8") as f:
        json.dump(docs, f, ensure_ascii=False, indent=2)
    
    index = [
        {
            "id": doc["id"],
            "url": doc["url"],
            "title": doc["title"],
            "category": doc["category"],
            "summary": doc["markdown"][:300] + "..." if len(doc["markdown"]) > 300 else doc["markdown"],
        }
        for doc in docs
    ]
    with open(DATA_DIR / "pico_docs_index.json", "w", encoding="utf-8") as f:
        json.dump(index, f, ensure_ascii=False, indent=2)
    
    print(f"\n[SAVE] {len(docs)} docs saved")


async def crawl_all():
    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1920,
        viewport_height=1080,
    )
    
    # 爬取配置 - 等待内容渲染
    content_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        page_timeout=25000,
        remove_overlay_elements=True,
        exclude_external_links=True,
        excluded_tags=["nav", "footer", "header"],
        # 等待文档内容区域加载
        wait_for="css:.doc-content, main, article, #content, .content",
    )
    
    # 分类首页用不同的配置（需要等 JS 渲染导航树）
    index_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        page_timeout=30000,
        remove_overlay_elements=True,
        # 等待更长时间让 JS 完成渲染
        wait_for="css:a[href*='/document/']",
    )
    
    all_docs = []
    failed_urls = []
    
    # 去重 URL
    seen = set()
    unique_urls = []
    for url in KNOWN_URLS:
        norm = url.rstrip("/")
        if norm not in seen:
            seen.add(norm)
            unique_urls.append(url)
    
    print("=" * 60)
    print(f"[START] Pico XR Doc Deep Crawler")
    print(f"[INFO] {len(unique_urls)} URLs to crawl")
    print("=" * 60)
    
    async with AsyncWebCrawler(config=browser_config) as crawler:
        batch_size = 3  # 降低并发避免被封
        
        for i in range(0, len(unique_urls), batch_size):
            batch = unique_urls[i : i + batch_size]
            batch_num = i // batch_size + 1
            total_batches = (len(unique_urls) + batch_size - 1) // batch_size
            print(f"\n[BATCH] {batch_num}/{total_batches}: {len(batch)} pages")
            
            # 判断是否是分类首页（URL 路径层级较浅的）
            is_index_page = any(url.rstrip("/").endswith(("/unity", "/unreal", "/native", "/openxr", "/discover", "/design", "/distribute", "/safe-info")) for url in batch)
            config = index_config if is_index_page else content_config
            
            try:
                results = await crawler.arun_many(urls=batch, config=config)
                
                for result in results:
                    url = getattr(result, 'url', '?').split("#")[0].rstrip("/")
                    
                    if not result.success:
                        failed_urls.append(url)
                        print(f"  [ERR] {url}")
                        continue
                    
                    title = result.metadata.get("title", "") or ""
                    title = re.sub(r"\s*[-|—]\s*PICO.*$", "", title, flags=re.IGNORECASE).strip()
                    title = re.sub(r"\s*[-|—]\s*Pico.*$", "", title, flags=re.IGNORECASE).strip()
                    
                    markdown = ""
                    if hasattr(result.markdown, "fit_markdown") and result.markdown.fit_markdown:
                        markdown = result.markdown.fit_markdown
                    elif hasattr(result.markdown, "raw_markdown"):
                        markdown = result.markdown.raw_markdown
                    else:
                        markdown = str(result.markdown)
                    markdown = clean_markdown(markdown)
                    
                    if len(markdown) < 100:
                        print(f"  [SKIP] too short: {url} ({len(markdown)} chars)")
                        continue
                    
                    # 也从内容中提取新链接
                    new_links = set()
                    for link in result.links.get("internal", []):
                        href = link.get("href", "")
                        if href and "/document/" in href:
                            norm_href = href.split("#")[0].rstrip("/")
                            if norm_href.startswith("/"):
                                norm_href = "https://developer-cn.picoxr.com" + norm_href
                            if norm_href not in seen and "picoxr.com/document/" in norm_href:
                                new_links.add(norm_href)
                    
                    if new_links:
                        print(f"  [NEW] found {len(new_links)} new URLs")
                        for nl in new_links:
                            if nl not in seen:
                                seen.add(nl)
                                unique_urls.append(nl)
                    
                    path_parts = [p for p in urlparse(url).path.strip("/").split("/") if p]
                    doc_id = "_".join(path_parts[-3:]) if len(path_parts) >= 3 else "_".join(path_parts)
                    
                    doc = {
                        "id": doc_id,
                        "url": url,
                        "title": title,
                        "category": get_category(url),
                        "markdown": markdown,
                        "crawled_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
                    }
                    all_docs.append(doc)
                    print(f"  [OK] [{len(all_docs)}] {title[:60]} ({len(markdown)} chars)")
                
            except Exception as e:
                print(f"  [ERR] batch error: {e}")
                failed_urls.extend(batch)
            
            # 保存中间结果
            if all_docs:
                save_docs(all_docs)
            
            await asyncio.sleep(1)
    
    # 最终保存
    save_docs(all_docs)
    
    if failed_urls:
        with open(DATA_DIR / "failed_urls.json", "w", encoding="utf-8") as f:
            json.dump(failed_urls, f, ensure_ascii=False, indent=2)
        print(f"\n[WARN] {len(failed_urls)} pages failed")
    
    print("\n" + "=" * 60)
    print(f"[DONE] Total: {len(all_docs)} docs crawled")
    print(f"[SAVE] Data: {DATA_DIR}")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(crawl_all())