一、拾枝杂谈

1.我用 WorkBuddy 完成了什么需求？

最近 WorkBuddy 是真的火，我看到很多B站的up主都出了 WorkBuddy 的视频，也不知道是商单还是真的好用。

我自己是程序员，也是一名内容创作者，写过几百篇文章了，接触过不少创作平台。正好这几天我有这么一个需求：先用 Playwright 爬取多平台的文章数据，并写入飞书表格，然后做成自动化。说的高大上一点，叫“搭建多平台数据矩阵”。

这样一来，如果我想统计每一个平台下面每一篇文章每一天的数据，就可以直接跑自动化任务就行了，效率的提升大的一塌糊涂。

我决定用 WorkBuddy 来帮我完成这个任务，顺便测试一下 WorkBuddy 的能力怎么样。

2.这篇文章讲什么？

因为不同的平台，统计的数据字段不一样，而且不同平台的后台展示方式也不尽相同，所以可想而且，不同平台的数据爬取一定会遇到不同的Bug和难题。

上一篇文章【https://blog.csdn.net/fastopAI/article/details/163295016?spm=1001.2014.3001.5502】，只是一篇总览性的文章，今天这篇文章就是实战篇。

这篇文章侧重于 CSDN 数据收集的完整打通，记录了up在此过程中踩过的坑。

本文会从0到1，手把手教你怎么收集CSDN上自己的文章数据，如果有幸帮到你了，就给up点个赞吧！也可以关注一下up-迅高智能。

---

二、最初的尝试

1.WorkBuddy的第一枪

在和 WorkBuddy 确定了 Python + Playwright 的技术选型之后，我就让 WorkBuddy 开始搭建项目框架和环境了。

WorkBuddy 首先加载了一个 Skill，叫做 lark-sheets，这个Skill是飞书（Lark）官方的一个能力套件，专门用来操作飞书电子表格。

WorkBuddy 要通过 lark-sheets 这个 Skill 来跟飞书的表格 API 打交道，比如创建子表、读写单元格、设置单元格格式，全都是通过它来完成的。

然后 WorkBuddy 去下载了 Playwright 和 Chromium。其中 Playwright 是一个浏览器自动化工具，它能让我们的Python爬虫脚本"操作"浏览器。但是由于它不信任电脑上随时会更新版本的 Chrome，因为万一 Chrome 自动升级，Playwright 的控制协议可能就不兼容了，所以 Playwright 的策略是：自带一个自己验证过的、版本锁定的 Chromium，确保每次运行行为一致。

Chromium 就是谷歌浏览器的"开源裸版"，我们可以理解为 Chrome 的技术内核，外观和功能几乎同谷歌浏览器一样，只是少了 Google 那些额外的服务集成。

接着，WorkBuddy 用一个测试子表的创建和写入，来验证是否可以连接到飞书，显然是可以的，因为我已经给它配置了飞书的MCP，在WorkBuddy 里叫做Connector（连接器）。

2.两种方式的选择

在知道了CSDN 的创作者后台链接后，WorkBuddy 给我提供了两种方案。

第一种方案，先关闭所有的 Chrome 窗口；再启动Chrome但是加了一个调试端口：chrome.exe --remote-debugging-port=9222；最后让脚本通过 CDP 协议连接到启动的 Chrome，这时候我们的所有登录态都会保留。

第二种方案，直接关闭 Chrome；脚本直接用我自己的 Chrome Profile 去启动浏览器；爬取数据完成后，再将 Chrome 自动关闭。

这两种方案都是将数据先存到了本地的 JSON 文件中，然后再写入飞书子表。

我当时也不懂这两种方案的区别，但是看到 WorkBuddy 推荐第一种，就直接让 WorkBuddy 按照第一种方案执行了。

3.第一种方式踩的坑

3.1 Chrome “单实例”机制 导致调试端口失效

WorkBuddy 推荐我用第一种方式，它还给出了我理由：只需要把我的所有的 Chrome 快捷方式都永久的加上 chrome.exe --remote-debugging-port=9222，之后我正常用CHrome，脚本随时能连上去，完全不干扰我的日常使用，可以说是一劳永逸。

我也算个懒人，一看到“一劳永逸”这个词，立刻轻哼了起来。

然后我本着“一懒到底”的核心信念，直接让 WorkBuddy 给我把快捷方式都改了。我一看，哟，啥活儿都能干呐，心里瞬间乐开了花儿。

随后 WorkBuddy 让我完全关闭 Chrome，然后重新打开。我照做了。结果 WorkBuddy 告诉我“Chrome的调试端口还没有激活”！

WorkBuddy 很快找到了原因： Chrome 有一个"单实例"机制：就是说，如果 Chrome 已经在运行（哪怕只是后台进程），我从快捷方式点击打开，它不会启动一个新进程，而是直接给已有进程发消息让它开一个新窗口。这意味着已有进程的启动参数（没有调试端口）决定了所有新窗口的行为。

3.2 Chrome新版本的安全策略拒绝使用默认目录

随后 WorkBuddy 彻底杀掉了 Chrome 的进程，重新尝试远程调试，但是下一秒就报错了。

DevTools remote debugging requires a non-default data directory. Specify this using --user-data-dir.

WorkBuddy 告诉我，这是因为谷歌Chrome新版本的安全策略：开启远程调试时必须显式指定 --user-data-dir，不能用默认目录。这就是为什么调试端口一直不生效。

修复方案很简单，加上 --user-data-dir 参数指向一个指定的其他 Chrome 用户数据目录。

3.3 Chrome Profile 复制过去后登录态失效

加上指定参数后重新调试，调试窗口成功生效后，WorkBuddy 发现由于这个 Chrome 用的是空白 profile，没有我的登录状态，于是就把我的 Chrome 登录数据复制了过来，包括Cookies数据。

然后它打算用这个包含登录数据的 debug profile 启动 Chrome，用 Playwright 连接这个 Chrome 实例，并测试 CSDN 爬取。

WorkBuddy 先是更新了CSDN爬虫脚本，改用了connect_over_cdp连接方式，就是说通过 CDP 连接到一个已经运行的 Chrome 实例，然后它又更新了 config.py 配置文件，之后再运行爬虫脚本时，会从这个配置文件里面读取要访问的网址，以及要写入的飞书表格。

结果打开的页面根本没有登录！这是因为Chrome 用 DPAPI 加密方式保存登录凭证，这个 DPAPI 加密方式是与文件目录所在的路径绑定的，复制到新的 profile 目录后路径变了，解密失败，所以复制 profile 这个方案只能保留部分 Cookie，不能完全保留登录态！

因为我们通过 --user-data-dir 参数指定了非默认用户目录，指定了非默认目录，意味着是一个全新的空白 profile，所以没有任何登录状态。

4.第二种方式踩的坑

4.1 原生profile体积太大了，导致playwright不能完全加载

这时候我一看第一种方式没什么起色，就打了退堂鼓了，立马问了一下 WorkBuddy，现在切换到第二种方式中不中？

没想到 WorkBuddy 直接推荐我赶快转到第二种方式，第二种方式是脚本用 Playwright 的 launch_persistent_context 直接启动 Chrome，直接使用我的真实 Profile 目录。但是运行时需要我先关闭 Chrome，跑完再打开。

方案二 launch_persistent_context 本身支持直接指定 user_data_dir = 原 profile 路径，不像方案一会收受到安全策略限制。

但是WorkBuddy发现我的原生 profile 有超过 7个G，Playwright 直接加载会崩溃，所以才改成复制一个精简版 profile。

也就是说方案一和方案二 其实原本都不需要复制 profile，最初的设想是两种方案都可以使用我原生的自己的 profile 。

只不过方案一是因为谷歌新版本安全政策的问题，远程调试不能用默认目录，所以被迫复制，本来想的是直接用而不是复制；方案二不是远程调试，所以不受安全策略的限制，可以直接用我的 profile，也就是直接用原路径下的默认目录，但是 playwright 运行在内存中，加载不下臃肿的超过 7个G 的默认用户目录，所以主动采取复制策略，可以说最终遇到的问题和方案一一样。

4.2 登录窗口设置时间过短导致脚本失败

在明白了方案二也需要重新登录的原因后，我服软了，登录就登录吧！

这时候脚本跑起来，跳出了一个登录窗口，但是up当时正好有事，就错过了窗口期。后来才知道一开始的登录窗口期只有4分钟，脚本等待时间完成后我都一直没登录成功，所以爬取失败了。于是我让 WorkBuddy 把登录窗口期又改成了 10分钟。

---

三、有效的尝试

1.成功爬取数据但是白璧微瑕

在经历过窗口等待期限的调整后，WorkBuddy 告诉我：“ Chrome 窗口打开后，请在 10 分钟内完成 CSDN 登录。 登录成功后脚本会自动检测到文章列表并继续。”。

然后我成功登录了跳出的 Chrome 窗口中的CSDN，谁曾想脚本居然一次跑成功了，直接将数据写入了飞书表格，牛逼！

但是新的问题出现了，由于CSDN的创作者后台中，在“内容管理”界面是没有办法直接看到“展现量”的，所以这次的爬取结果就是，表格中的“展现量”一栏都为空，其他字段正常。

2.十全十美并不是奢望

我随即告诉 WorkBuddy，CSDN展现量可以在两个地方找到：一是在在“管理”-> “内容管理”里面，可以看到每篇文章的数据，但是没有直接看到展现量，不过在每篇文章右侧的“操作”里面，有一个“数据”选项，​点开后可以看到每篇文章当前的展现量；二是在“数据”->“作品数据”里面，在“博客数据”页面下，选择“单篇文章分析”，同样可以看到每篇文章的展现量​。

我让 WorkBuddy 自行选择使用哪种方式获取展现量。WorkBuddy很聪明，选择了第二种方法，它给的理由也很有意思：因为一张表就能看到所有文章的展现量，不需要每篇文章点一次。

WorkBuddy 随后又小刀窄斧地修改了对应的CSDN脚本，开始进行爬取，直接搞定！这时候我先让 WorkBuddy 把第一种方案下失效的文件和一些残留的数据清理干净。

给大家看一下最终的效果，CSDN的飞书表格结构图如下所示：

CSDN第二篇文章的统计子表如下图所示：

---

四、相关代码及说明

1.配置信息的代码

最最最开始，第一次测试时，WorkBuddy 就创建了一个配置文件 config.py。CSDN 的入口网址、飞书表格 ID、字段定义（展现量/阅读量/点赞/收藏/评论），都写在这里。

但实际上 config.py 是一个通用的“集中控制面板”，它不但记录了每个平台的入口地址，还记录了对应的飞书子表的位置，这样每个爬虫脚本启动时，不用自己硬记"去哪个网址、写到哪个飞书表"，直接从 config.py 读。

其他脚本启动时会直接 import config，把它都进来当词典用。

config.py 中与CSDN相关的代码如下，大家可以参考：（我已经替换了具体的ID）

    # 采集起始日期（全局，影响所有平台包括 CSDN）
    START_DATE = "自己选个具体日期"

    # 飞书知识空间 ID（全局，CSDN 的表格也在这个空间下）
    FEISHU_SPACE_ID = "你的飞书知识库的ID"

    # ============================================
    # CSDN 专属配置
    # ============================================
    CSDN = {
        # 平台名称
        "name": "CSDN",

        # CSDN 创作者后台入口
        "creator_url": "https://mp.csdn.net/",

        # 文章管理页面
        "manage_url": "https://mp.csdn.net/mp_blog/manage/article",

        # 数据中心 → 内容分析 → 单篇分析
        "data_center_url": "https://mp.csdn.net/mp_blog/analysis/article/all",

        # 飞书父表格 ID（总览表）
        "feishu_sheet_token": "迅高智能",

        # 飞书父节点 ID（wiki 节点）
        "feishu_wiki_node_token": "大家给我点个赞吧！",

        # 所属飞书知识空间
        "feishu_space_id": FEISHU_SPACE_ID,

        # 是否启用采集
        "enabled": True,

        # CSDN 的真实统计字段（6 个）
        "data_fields": ["日期", "展现量", "阅读量", "点赞", "收藏", "评论"],
    }

    # ============================================
    # 全局命名规则（CSDN 也遵循）
    # ============================================
    # 子表命名：GEO-001：文章标题（全角冒号）
    TOPIC_PREFIX = "GEO"
    TOPIC_COLON = "："   # 中文全角冒号

2.具体的实现爬取的代码

csdn.py 代码如下：

    """CSDN creator backend scraper using Playwright persistent context.

    Architecture:
    1. Check that Chrome is NOT running (profile must not be locked)
    2. Refresh the small auto-profile from the user's real Chrome profile
    3. Launch Chrome via Playwright's launch_persistent_context using the auto-profile
    4. This preserves login sessions while avoiding the 7 GB profile crash
    5. Navigate to CSDN article management page
    6. Detect login state; if login dialog blocks, ask user to log in
    7. Extract article list (titles, publish dates, stats)
    8. Save data locally as JSON
    9. Close Chrome (user can reopen normally)

    PREREQUISITE: User must close ALL Chrome windows and background processes
    before running this script.
    """

    import asyncio
    import json
    import os
    import re
    import sys
    import subprocess
    from datetime import datetime, date

    from playwright.async_api import async_playwright

    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    from config import CSDN, START_DATE, CHROME, DATA_FIELDS
    from chrome_profile import refresh_auto_profile, get_profile_size


    class CSDNScraper:
        """Scraper for CSDN creator backend using Playwright persistent context."""

        MANAGE_URL = CSDN["manage_url"]
        DATA_CENTER_URL = CSDN["data_center_url"]

        def __init__(self, output_dir: str = None):
            self.name = CSDN["name"]
            self.output_dir = output_dir or os.path.join(
                os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                "data"
            )
            self._playwright = None
            self.context = None
            self.page = None
            self.start_date = date.fromisoformat(START_DATE)

        def _check_chrome_not_running(self):
            """Verify Chrome is not running (profile must be unlocked)."""
            result = subprocess.run(
                ["tasklist", "/FI", "IMAGENAME eq chrome.exe", "/FO", "CSV"],
                capture_output=True, encoding="gbk", errors="replace"
            )
            lines = result.stdout.strip().split("\n")
            chrome_lines = [l for l in lines if "chrome.exe" in l.lower()]
            if chrome_lines:
                print(f"[{self.name}] ERROR: Chrome is still running "
                    f"({len(chrome_lines)} processes).")
                print(f"[{self.name}] Please close ALL Chrome windows and "
                    f"background processes, then re-run.")
                return False
            print(f"[{self.name}] Chrome is not running - profile is unlocked.")
            return True

        def _refresh_profile(self):
            """Copy essential login files from real profile to auto profile.

            Only copies if the auto-profile doesn't exist yet. After the user
            logs in once, the auto-profile keeps that session, so we don't
            overwrite it on subsequent runs.
            """
            import os as _os
            if _os.path.exists(CHROME["user_data_dir"]) and \
            _os.path.isdir(CHROME["user_data_dir"]):
                size = get_profile_size(CHROME["user_data_dir"])
                print(f"[{self.name}] Using existing auto profile "
                    f"({size / 1024 / 1024:.1f} MB). Login session should persist.")
                return

            print(f"[{self.name}] Auto profile not found. "
                f"Refreshing from real Chrome profile...")
            refresh_auto_profile(CHROME["source_user_data_dir"], CHROME["user_data_dir"])
            size = get_profile_size(CHROME["user_data_dir"])
            print(f"[{self.name}] Auto profile size: {size / 1024 / 1024:.1f} MB")

        async def launch_browser(self):
            """Launch Chrome with the small auto-profile."""
            if not self._check_chrome_not_running():
                return False

            self._refresh_profile()

            print(f"[{self.name}] Launching Chrome with auto profile...")
            self._playwright = await async_playwright().start()

            self.context = await self._playwright.chromium.launch_persistent_context(
                user_data_dir=CHROME["user_data_dir"],
                channel=CHROME["channel"],
                headless=False,
                viewport={"width": 1920, "height": 1080},
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--no-first-run",
                    "--start-maximized",
                ],
            )

            # Use the first page (Chrome opens with a default tab)
            if self.context.pages:
                self.page = self.context.pages[0]
            else:
                self.page = await self.context.new_page()

            print(f"[{self.name}] Chrome launched")
            return True

        async def take_screenshot(self, filename: str):
            """Take a screenshot and save to output dir."""
            path = os.path.join(self.output_dir, filename)
            os.makedirs(self.output_dir, exist_ok=True)
            await self.page.screenshot(path=path, full_page=True)
            print(f"[{self.name}] Screenshot saved: {path}")
            return path

        async def handle_login_dialog(self):
            """Check for and dismiss login promo dialogs.

            CSDN sometimes shows a '登录可享更多权益' promo dialog. We try
            several close selectors and keyboard shortcuts.
            """
            close_selectors = [
                ".modal .close",
                ".login-dialog .close",
                ".login-modal .close",
                "button[aria-label='Close']",
                ".csdn-modal__close",
                ".layui-layer-close",
                ".dialog-close",
                ".passport-login-box .close",
                ".passport-login-box3 .close",
                "img[src*='close']",
            ]

            for selector in close_selectors:
                try:
                    close_btn = await self.page.query_selector(selector)
                    if close_btn:
                        await close_btn.click()
                        print(f"[{self.name}] Closed login promo dialog with: {selector}")
                        await self.page.wait_for_timeout(1000)
                        return True
                except Exception:
                    pass

            try:
                await self.page.keyboard.press("Escape")
                await self.page.wait_for_timeout(1000)
            except Exception:
                pass

            return False

        async def wait_for_login(self, timeout_seconds: int = 600):
            """Wait for user to log in using the visible Chrome window.

            Polls the page until the login dialog/page disappears or an
            article list appears. Prints clear instructions for the user.
            """
            print(f"[{self.name}] -------------------------------------------------")
            print(f"[{self.name}] CSDN requires login to access the article list.")
            print(f"[{self.name}] Please log in using the Chrome window that just opened.")
            print(f"[{self.name}] Waiting up to {timeout_seconds} seconds for login...")
            print(f"[{self.name}] -------------------------------------------------")

            for i in range(timeout_seconds // 5):
                await self.page.wait_for_timeout(5000)

                current_url = self.page.url
                if "login" not in current_url and "passport" not in current_url:
                    # Try to detect if article list loaded
                    article_items = await self.page.query_selector_all(
                        ".article-list-item-mp, .article_manage_list .item"
                    )
                    if article_items:
                        print(f"[{self.name}] Login complete - article list detected.")
                        return True

                # Try dismissing any remaining promo dialogs
                await self.handle_login_dialog()

                if i % 6 == 0 and i > 0:  # Remind every 30 seconds
                    print(f"[{self.name}] Still waiting for login... "
                        f"({i * 5}/{timeout_seconds}s)")

            print(f"[{self.name}] Timeout waiting for login.")
            return False

        async def scrape_article_list(self):
            """Navigate to CSDN article management and extract article data."""
            print(f"[{self.name}] Navigating to article management page...")
            await self.page.goto(self.MANAGE_URL, wait_until="networkidle")
            await self.page.wait_for_timeout(5000)

            # Take screenshot
            await self.take_screenshot("csdn_article_list.png")

            # Try to dismiss any login promo dialogs
            await self.handle_login_dialog()

            # Re-check URL and login state
            current_url = self.page.url
            print(f"[{self.name}] Current URL: {current_url}")

            # Check if we're blocked by login page or dialog
            needs_login = ("login" in current_url or "passport" in current_url)

            if not needs_login:
                # Also check for a blocking login dialog inside the management page
                login_dialog = await self.page.query_selector(
                    ".passport-login-container, .passport-login-box, .csdn-login-box"
                )
                if login_dialog:
                    print(f"[{self.name}] Login promo dialog detected.")
                    closed = await self.handle_login_dialog()
                    if not closed:
                        needs_login = True

            if needs_login:
                await self.take_screenshot("csdn_login_required.png")
                if not await self.wait_for_login(timeout_seconds=600):
                    return []
                await self.page.goto(self.MANAGE_URL, wait_until="networkidle")
                await self.page.wait_for_timeout(5000)
                await self.take_screenshot("csdn_article_list_after_login.png")

            # Save page HTML for analysis
            content = await self.page.content()
            html_path = os.path.join(self.output_dir, "csdn_article_list_page.html")
            with open(html_path, "w", encoding="utf-8") as f:
                f.write(content)
            print(f"[{self.name}] Saved page HTML for analysis")

            # Extract articles using JavaScript for robustness (Vue-rendered DOM)
            articles = await self._extract_csdn_articles_js()

            # Filter by start date
            filtered = []
            for a in articles:
                if not a["publish_date"]:
                    filtered.append(a)
                    continue
                try:
                    parsed = _parse_date(a["publish_date"])
                    if parsed is None or parsed >= self.start_date:
                        filtered.append(a)
                except Exception:
                    filtered.append(a)

            print(f"[{self.name}] Found {len(articles)} total, "
                f"{len(filtered)} after {START_DATE}")
            return filtered

        async def _extract_csdn_articles_js(self) -> list:
            """Extract article list from CSDN content management page via JS.

            The page is rendered by Vue.js; static HTML does not contain the list.
            We evaluate in the browser to read the actual DOM and numeric stats.

            Column order on the management page (from screenshot):
                阅读 (reads), 点赞 (likes), 评论 (comments), 收藏 (collections)
            展现量 is not shown here and will be left as 0.
            """
            try:
                js_result = await self.page.evaluate("""
                    () => {
                        const results = [];

                        // Try to get the CSDN username from the toolbar avatar link
                        let username = '';
                        const avatarLink = document.querySelector('.toolbar-inside .hasAvatar');
                        if (avatarLink && avatarLink.href) {
                            const m = avatarLink.href.match(/blog\\.csdn\\.net\\/([^/]+)/);
                            if (m) username = m[1];
                        }
                        // Fallback: any link to blog.csdn.net/username
                        if (!username) {
                            document.querySelectorAll('a[href*="blog.csdn.net/"]').forEach(a => {
                                if (!username) {
                                    const m = a.href.match(/blog\\.csdn\\.net\\/([^/?#]+)/);
                                    if (m && !['www','mp','editor','i'].includes(m[1])) username = m[1];
                                }
                            });
                        }

                        const items = document.querySelectorAll('.article-list-item-mp');
                        items.forEach(item => {
                            // Title + editor link
                            const titleEl = item.querySelector('.article-list-item-txt a, .article-title a, a.title, h4 a');
                            let title = '';
                            let editorUrl = '';
                            let articleId = '';
                            if (titleEl) {
                                title = titleEl.innerText.trim();
                                editorUrl = titleEl.href || '';
                                const idMatch = editorUrl.match(/articleId=(\\d+)/);
                                if (idMatch) articleId = idMatch[1];
                            }

                            // Publish date
                            const dateEl = item.querySelector('.article-list-item-time, .date, time, .pub-time');
                            const pubDate = dateEl ? dateEl.innerText.trim() : '';

                            // Stats: management page shows 阅读, 点赞, 评论, 收藏.
                            // The innerText layout is: title, 原创, date, read, like, comment, collect, actions.
                            let reads = 0, likes = 0, comments = 0, collections = 0;
                            const parts = item.innerText.split(/\\n\\n+/).map(s => s.trim()).filter(s => s);
                            // Find the date part index, then next 4 numeric parts are stats
                            for (let i = 0; i < parts.length; i++) {
                                if (/\\d{4}[-/]\\d{1,2}[-/]\\d{1,2}/.test(parts[i])) {
                                    const nums = [];
                                    for (let j = i + 1; j < parts.length && nums.length < 4; j++) {
                                        const lines = parts[j].split('\\n').map(s => s.trim()).filter(s => /^-?\\d+$/.test(s));
                                        lines.forEach(n => nums.push(n));
                                    }
                                    if (nums.length >= 4) {
                                        reads = parseInt(nums[0], 10) || 0;
                                        likes = parseInt(nums[1], 10) || 0;
                                        comments = parseInt(nums[2], 10) || 0;
                                        collections = parseInt(nums[3], 10) || 0;
                                    }
                                    break;
                                }
                            }

                            // Build public article URL from articleId + username
                            let publicUrl = editorUrl;
                            if (articleId && username) {
                                publicUrl = `https://blog.csdn.net/${username}/article/details/${articleId}`;
                            }

                            if (title) {
                                results.push({
                                    title,
                                    editorUrl,
                                    publicUrl,
                                    articleId,
                                    publishDate: pubDate,
                                    reads,
                                    likes,
                                    comments,
                                    collections,
                                    impressions: 0,  // Filled later from data center
                                    statsText: item.innerText
                                });
                            }
                        });

                        return { username, articles: results };
                    }
                """)

                username = js_result.get("username", "")
                raw_articles = js_result.get("articles", [])
                print(f"[{self.name}] JS extraction: username={username}, "
                    f"articles={len(raw_articles)}")

                articles = []
                for a in raw_articles:
                    articles.append({
                        "title": a["title"],
                        "url": a["publicUrl"],
                        "editor_url": a["editorUrl"],
                        "article_id": a["articleId"],
                        "publish_date": a["publishDate"],
                        "stats": {
                            "展现量": a["impressions"],
                            "阅读量": a["reads"],
                            "点赞": a["likes"],
                            "评论": a["comments"],
                            "收藏": a["collections"],
                        },
                        "stats_text": a["statsText"],
                    })
                return articles

            except Exception as e:
                print(f"[{self.name}] JS extraction failed: {e}")
                return []

        async def scrape_impressions(self) -> list:
            """Navigate to CSDN data center and extract per-article impressions.

            The data center's "单篇文章分析" (single-article analysis) table
            includes a 展现量 column that is not shown on the content management
            page. We collect: title, publish_date, impressions, reads, comments,
            collections. Likes are not available here and remain from the article
            list scrape.

            Returns:
                List of dicts: [{title, publish_date, impressions, reads,
                                comments, collections}, ...]
            """
            print(f"[{self.name}] Navigating to data center for impressions...")
            await self.page.goto(self.DATA_CENTER_URL, wait_until="networkidle")
            await self.page.wait_for_timeout(5000)
            await self.take_screenshot("csdn_data_center.png")

            # Try to activate the "单篇文章分析" tab if it exists and is not active
            tab_clicked = False
            for sel in [
                "text=单篇文章分析",
                ".tab-item:has-text('单篇文章分析')",
                ".analysis-tab:has-text('单篇文章分析')",
                "[data-tab='single']",
                "//*[contains(text(), '单篇文章分析')]",
            ]:
                try:
                    tab = await self.page.query_selector(sel)
                    if tab:
                        cls = await tab.get_attribute("class") or ""
                        if "active" not in cls.lower() and "current" not in cls.lower():
                            await tab.click()
                            tab_clicked = True
                            print(f"[{self.name}] Clicked '单篇文章分析' tab")
                            await self.page.wait_for_timeout(3000)
                        break
                except Exception as e:
                    print(f"[{self.name}] Tab selector {sel} failed: {e}")

            # Save HTML for debugging
            content = await self.page.content()
            html_path = os.path.join(self.output_dir, "csdn_data_center_page.html")
            with open(html_path, "w", encoding="utf-8") as f:
                f.write(content)
            print(f"[{self.name}] Saved data center HTML for analysis")

            # Extract impression table data
            articles = await self._extract_impressions_js()
            print(f"[{self.name}] Extracted impressions for {len(articles)} "
                f"articles from data center")
            for a in articles:
                print(f"  - [{a.get('publish_date', '?')}] {a['title']}: "
                    f"impressions={a.get('impressions', 0)}")

            return articles

        async def _extract_impressions_js(self) -> list:
            """Extract single-article analysis table via JavaScript.

            The CSDN data-center table uses Element UI classes:
            - rows: .el_mcm-table__row
            - cells: .el_mcm-table__cell
            - inner text wrapper: .cell
            Column order (from observed DOM): title, date, impressions, reads,
            comments, collections, follows, ...
            """
            try:
                js_result = await self.page.evaluate("""
                    () => {
                        const results = [];

                        function parseNum(text) {
                            if (!text) return 0;
                            const cleaned = text.replace(/,/g, '').replace(/\\s/g, '').trim();
                            const m = cleaned.match(/-?\\d+/);
                            return m ? parseInt(m[0], 10) : 0;
                        }

                        // CSDN Vue table uses .el_mcm-table__row rows
                        const rows = document.querySelectorAll('.el_mcm-table__row');
                        rows.forEach(row => {
                            const cells = row.querySelectorAll('.el_mcm-table__cell');
                            if (cells.length < 6) return;

                            // Title is in the first (fixed-left) cell, usually inside an <a>
                            const titleEl = cells[0].querySelector('a') || cells[0];
                            const title = titleEl.innerText.trim();
                            if (!title) return;

                            const publishDate = cells[1] ? cells[1].innerText.trim() : '';
                            const impressions = parseNum(cells[2] ? cells[2].innerText : '0');
                            const reads = parseNum(cells[3] ? cells[3].innerText : '0');
                            const comments = parseNum(cells[4] ? cells[4].innerText : '0');
                            const collections = parseNum(cells[5] ? cells[5].innerText : '0');

                            results.push({title, publish_date: publishDate, impressions, reads, comments, collections});
                        });

                        // Fallback: if the specific class isn't present, try generic selectors
                        if (results.length === 0) {
                            const tables = document.querySelectorAll('table');
                            for (const table of tables) {
                                const headerCells = table.querySelectorAll('thead th, tr:first-child th, tr:first-child td');
                                if (headerCells.length < 4) continue;

                                let impressionIdx = -1, readIdx = -1, commentIdx = -1,
                                    collectIdx = -1, dateIdx = -1;
                                headerCells.forEach((th, idx) => {
                                    const text = th.innerText.trim();
                                    if (text.includes('展现量') || text.includes('曝光')) impressionIdx = idx;
                                    if (text.includes('阅读量')) readIdx = idx;
                                    if (text.includes('评论')) commentIdx = idx;
                                    if (text.includes('收藏')) collectIdx = idx;
                                    if (text.includes('创建日期') || text.includes('日期')) dateIdx = idx;
                                });

                                if (impressionIdx < 0) continue;

                                const bodyRows = table.querySelectorAll('tbody tr, tr');
                                for (const row of bodyRows) {
                                    const tds = row.querySelectorAll('td');
                                    if (tds.length < headerCells.length) continue;
                                    const titleEl = row.querySelector('a') || tds[0];
                                    const title = titleEl.innerText.trim();
                                    if (!title) continue;
                                    results.push({
                                        title,
                                        publish_date: dateIdx >= 0 ? tds[dateIdx].innerText.trim() : '',
                                        impressions: parseNum(tds[impressionIdx].innerText),
                                        reads: parseNum(readIdx >= 0 ? tds[readIdx].innerText : '0'),
                                        comments: parseNum(commentIdx >= 0 ? tds[commentIdx].innerText : '0'),
                                        collections: parseNum(collectIdx >= 0 ? tds[collectIdx].innerText : '0'),
                                    });
                                }
                                if (results.length > 0) break;
                            }
                        }

                        return results;
                    }
                """)
                return js_result or []
            except Exception as e:
                print(f"[{self.name}] Impression JS extraction failed: {e}")
                return []

        async def scrape_all(self):
            """Full scraping pipeline."""
            if not await self.launch_browser():
                return []

            articles = await self.scrape_article_list()

            # Get impressions from data center and merge into articles
            impression_articles = await self.scrape_impressions()

            # Build normalized map of content-management articles
            article_map = {}
            for a in articles:
                norm = _normalize_title(a["title"])
                article_map[norm] = a

            # Merge impressions into articles. The data center may contain extra
            # tables/demos with placeholder rows; only merge by title match or
            # add as a new article if the row has a plausible publish date.
            for imp in impression_articles:
                title = imp["title"]
                norm = _normalize_title(title)
                if norm in article_map:
                    article_map[norm]["stats"]["展现量"] = imp.get("impressions", 0)
                else:
                    pub_date = imp.get("publish_date", "")
                    if re.match(r"\d{4}-\d{2}-\d{2}", pub_date):
                        articles.append({
                            "title": title,
                            "url": "",
                            "editor_url": "",
                            "article_id": "",
                            "publish_date": pub_date,
                            "stats": {
                                "展现量": imp.get("impressions", 0),
                                "阅读量": imp.get("reads", 0),
                                "点赞": 0,  # Likes not available from data center
                                "收藏": imp.get("collections", 0),
                                "评论": imp.get("comments", 0),
                            },
                            "stats_text": "",
                        })
                        article_map[norm] = articles[-1]
                    else:
                        print(f"[{self.name}] Skipping data-center placeholder row: "
                            f"'{title}' (date='{pub_date}')")

            # Sort by publish date ascending (oldest first) so that GEO sequence
            # numbers match chronological publish order
            articles.sort(key=lambda a: _parse_date(a.get("publish_date", "")) or date.min)
            print(f"[{self.name}] Sorted {len(articles)} articles by publish date (oldest first)")
            for a in articles:
                stats = a.get("stats", {})
                print(f"  - [{a.get('publish_date', '?')}] {a['title']} "
                    f"(impressions={stats.get('展现量', 0)})")

            all_data = []
            today = datetime.now().strftime("%Y-%m-%d")
            for article in articles:
                stats = article.get("stats", {})
                article_data = {
                    "platform": self.name,
                    "article_title": article["title"],
                    "url": article.get("url", ""),
                    "editor_url": article.get("editor_url", ""),
                    "article_id": article.get("article_id", ""),
                    "publish_date": article.get("publish_date", ""),
                    "headers": DATA_FIELDS,
                    "daily_stats": [{
                        "日期": today,
                        "展现量": stats.get("展现量", 0),
                        "阅读量": stats.get("阅读量", 0),
                        "点赞": stats.get("点赞", 0),
                        "收藏": stats.get("收藏", 0),
                        "评论": stats.get("评论", 0),
                    }],
                    "stats_text": article.get("stats_text", ""),
                    "scraped_at": datetime.now().isoformat(),
                }
                all_data.append(article_data)

            # Save raw data locally
            output_path = os.path.join(self.output_dir, f"{self.name}_data.json")
            os.makedirs(self.output_dir, exist_ok=True)
            with open(output_path, "w", encoding="utf-8") as f:
                json.dump(all_data, f, ensure_ascii=False, indent=2)
            print(f"[{self.name}] Data saved to {output_path}")

            return all_data

        async def close(self):
            """Close browser context and Playwright."""
            if self.context:
                await self.context.close()
                print(f"[{self.name}] Chrome context closed")
            if self._playwright:
                await self._playwright.stop()
                print(f"[{self.name}] Playwright stopped")


    def _normalize_title(title: str) -> str:
        """Normalize a title for comparison (matches feishu_writer logic)."""
        title = re.sub(r'["\u201c\u201d\u2018\u2019]', '"', title)
        title = title.strip().lower()
        title = re.sub(r'\s+', ' ', title)
        return title


    def _parse_date(date_str: str):
        """Parse various date formats."""
        date_str = date_str.strip()
        try:
            return date.fromisoformat(date_str)
        except ValueError:
            pass

        match = re.match(r"(\d{4}).(\d{1,2}).(\d{1,2})", date_str)
        if match:
            return date(int(match.group(1)), int(match.group(2)), int(match.group(3)))

        match = re.match(r"(\d{4})年(\d+)月(\d+)日", date_str)
        if match:
            return date(int(match.group(1)), int(match.group(2)), int(match.group(3)))

        match = re.match(r"(\d+)月(\d+)日", date_str)
        if match:
            return date(2026, int(match.group(1)), int(match.group(2)))

        try:
            parts = date_str.split(".")
            if len(parts) == 3:
                return date(int(parts[0]), int(parts[1]), int(parts[2]))
        except ValueError:
            pass

        return None


    async def run_scraper():
        """Run the CSDN scraper end-to-end."""
        scraper = CSDNScraper()

        try:
            data = await scraper.scrape_all()

            if data:
                print(f"\n[{scraper.name}] Scraped {len(data)} articles")
                for a in data:
                    print(f"  - {a['article_title']}")
            else:
                print(f"\n[{scraper.name}] No articles found or login required")

            return data
        finally:
            await scraper.close()


    if __name__ == "__main__":
        result = asyncio.run(run_scraper())
        print(f"\nDone. Scraped {len(result) if result else 0} articles.")

相关代码说明如下表所示：

行数范围	内容	干什么的
1-16	文件说明（注释）	解释整体架构和前置条件
18-30	导入依赖	引用 Playwright、datetime、config.py 等
33-48	__init__ 初始化	设定输出目录、启动日期等
50-65	检测 Chrome 是否关闭	爬之前确保 Chrome 没在跑，否则 profile 锁住会报错
67-86	刷新 Chrome profile	从真实 Chrome 复制登录状态到 auto-profile，实现免登
88-117	启动浏览器	用 Playwright 打开 Chrome，挂载 auto-profile
127-163	处理登录弹窗	CSDN 偶尔弹出"登录享权益"广告，自动关闭
165-198	等待用户登录	如果需要登录，等用户在 Chrome 窗口里手动操作，最长等 10 分钟
200-263	爬取文章列表	导航到文章管理页 → 截图 → 保存 HTML → 提取文章标题/日期/数据 → 按起始日期过滤
265-389	JS 提取文章数据	在浏览器里执行 JavaScript，从 Vue 渲染的 DOM 中扒出每篇文章的阅读/点赞/评论/收藏
392-537	爬取展现量	导航到数据中心 → 切到"单篇文章分析" → 提取展现量（文章列表页没有）
539-627	合并 + 保存	把文章列表数据 + 展现量合并 → 按发布日期排序 → 存为 CSDN_data.json
629-636	关闭浏览器	清理资源，释放 Chrome
639-698	辅助函数 + 启动入口	标题标准化、日期解析、一键运行

---

Δ总结

• CSDN 是 up 开始尝试的第一个平台，所以可谓是大费周章。其中其实还有很多细节 up 没有往上放，比如飞书表格的识别问题，字段匹配问题，等等。后续 up 也会出其他文章进行讲解。感谢阅读！
• 良工不示人以朴，up所有文章都会适时补充完善。大家如果有问题都可以在评论区进行交流或者私信up。感谢阅读！