From dadd5f4b19dd883a198d1c71aabb667475be0afd Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 20:19:38 +0800
Subject: [PATCH 01/12] fix: limit bilibili sub-comment fetching

---
 docs/plans/2026-07-01-bili-article-support.md | 931 ++++++++++++++++++
 media_platform/bilibili/client.py             |  14 +-
 tests/test_bilibili_client_comments.py        |  53 +
 3 files changed, 990 insertions(+), 8 deletions(-)
 create mode 100644 docs/plans/2026-07-01-bili-article-support.md
 create mode 100644 tests/test_bilibili_client_comments.py

diff --git a/docs/plans/2026-07-01-bili-article-support.md b/docs/plans/2026-07-01-bili-article-support.md
new file mode 100644
index 000000000..cd1261cfc
--- /dev/null
+++ b/docs/plans/2026-07-01-bili-article-support.md
@@ -0,0 +1,931 @@
+# Bilibili Article Support Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Add complete Bilibili article support, including article URL parsing, article detail crawling, first-level comments, optional second-level comments, and dedicated storage.
+
+**Architecture:** Keep existing video public APIs compatible while extracting Bilibili comment crawling into a generic `type + oid` implementation. Article crawling is added as a first-class content flow with dedicated parsing, client methods, core orchestration, and storage models. The video comment pagination semantics have already been corrected: first-level comments are trimmed by `max_count` before saving and before second-level comment fetching, and retained first-level comments are always counted even when second-level comments are enabled.
+
+**Tech Stack:** Python async/await, Playwright-backed Bilibili API client, SQLAlchemy models, existing MediaCrawler store abstractions, pytest.
+
+---
+
+## Context
+
+Current Bilibili code is video-centered:
+
+- `media_platform/bilibili/client.py` calls Bilibili comment APIs with `type=1`, which means video comments.
+- Bilibili article comments require the same comment API family but with `type=12` and `oid=<article cvid>`.
+- `media_platform/bilibili/help.py` only parses video BV URLs and creator URLs.
+- `store/bilibili` and `database/models.py` only have video and video comment storage models.
+
+Completed prerequisite: video comment crawling now trims first-level comments before fetching second-level comments and extends the local result in both modes. This behavior is covered by `tests/test_bilibili_client_comments.py`.
+
+The implementation should preserve existing video methods such as `get_video_comments()` as compatibility wrappers. The new generic implementation should become the only place that knows how to page comments and second-level comments, while preserving the corrected video semantics above.
+
+## External API Notes
+
+Use Bilibili's common comment model:
+
+- Video comments: `type=1`, `oid=<aid>`
+- Article comments: `type=12`, `oid=<cvid>`
+- First-level comments endpoint: `/x/v2/reply/wbi/main`
+- Second-level comments endpoint: `/x/v2/reply/reply`
+
+Before implementing article detail mapping, verify a real article detail response with one known `cv` article ID. Do not guess the final response shape for title/content/stat fields.
+
+---
+
+### Task 1: Add Bilibili Article URL Model And Parser
+
+**Files:**
+- Modify: `model/m_bilibili.py`
+- Modify: `media_platform/bilibili/help.py`
+- Test: `tests/media_platform/bilibili/test_help.py`
+
+**Step 1: Write failing parser tests**
+
+Create or extend `tests/media_platform/bilibili/test_help.py`:
+
+```python
+import pytest
+
+from media_platform.bilibili.help import parse_article_info_from_url
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("cv123456", "123456"),
+        ("123456", "123456"),
+        ("https://www.bilibili.com/read/cv123456", "123456"),
+        ("https://www.bilibili.com/read/cv123456?spm_id_from=333.999.0.0", "123456"),
+    ],
+)
+def test_parse_article_info_from_url(raw, expected):
+    article_info = parse_article_info_from_url(raw)
+    assert article_info.article_id == expected
+    assert article_info.article_type == "article"
+
+
+def test_parse_article_info_from_url_invalid():
+    with pytest.raises(ValueError):
+        parse_article_info_from_url("https://www.bilibili.com/video/BV1d54y1g7db")
+```
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_help.py -v
+```
+
+Expected: FAIL because `parse_article_info_from_url` and `ArticleUrlInfo` do not exist.
+
+**Step 3: Add model and parser**
+
+In `model/m_bilibili.py`, add:
+
+```python
+class ArticleUrlInfo(BaseModel):
+    """Bilibili article URL information"""
+    article_id: str = Field(title="article id (cv id without cv prefix)")
+    article_type: str = Field(default="article", title="article type")
+```
+
+In `media_platform/bilibili/help.py`, import `ArticleUrlInfo` and add `parse_article_info_from_url(url: str) -> ArticleUrlInfo`.
+
+Parsing rules:
+
+- If input is digits, return it directly.
+- If input starts with `cv` followed by digits, strip `cv`.
+- If URL contains `/read/cv<digits>`, extract digits.
+- Otherwise raise `ValueError`.
+
+**Step 4: Run test to verify it passes**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_help.py -v
+```
+
+Expected: PASS.
+
+**Step 5: Commit**
+
+```bash
+git add model/m_bilibili.py media_platform/bilibili/help.py tests/media_platform/bilibili/test_help.py
+git commit -m "feat: parse bilibili article urls"
+```
+
+---
+
+### Task 2: Add Generic Bilibili Comment Type Support
+
+**Files:**
+- Modify: `media_platform/bilibili/field.py`
+- Modify: `media_platform/bilibili/client.py`
+- Test: `tests/media_platform/bilibili/test_client_comments.py`
+
+**Step 1: Write failing client tests**
+
+Create `tests/media_platform/bilibili/test_client_comments.py` with a minimal client fixture that bypasses network by monkeypatching `get`.
+
+Test these cases:
+
+```python
+import pytest
+
+from media_platform.bilibili.field import BilibiliCommentType, CommentOrderType
+
+
+@pytest.mark.asyncio
+async def test_get_comments_uses_comment_type(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["uri"] = uri
+        captured["params"] = params
+        return {}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    await bili_client.get_comments(
+        oid="123456",
+        comment_type=BilibiliCommentType.ARTICLE,
+        order_mode=CommentOrderType.DEFAULT,
+        next=0,
+    )
+
+    assert captured["uri"] == "/x/v2/reply/wbi/main"
+    assert captured["params"]["oid"] == "123456"
+    assert captured["params"]["type"] == 12
+```
+
+Also add a compatibility test:
+
+```python
+@pytest.mark.asyncio
+async def test_get_video_comments_keeps_type_1(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["params"] = params
+        return {}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    await bili_client.get_video_comments("998877")
+
+    assert captured["params"]["oid"] == "998877"
+    assert captured["params"]["type"] == 1
+```
+
+If there is no reusable `bili_client` fixture, create a local fixture with dummy headers, page, and cookies. Monkeypatched tests should not touch network or Playwright.
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: FAIL because `BilibiliCommentType` and `get_comments()` do not exist.
+
+**Step 3: Implement comment type enum and generic first-level method**
+
+In `media_platform/bilibili/field.py`, add:
+
+```python
+class BilibiliCommentType(Enum):
+    VIDEO = 1
+    ARTICLE = 12
+```
+
+In `media_platform/bilibili/client.py`, add:
+
+```python
+async def get_comments(
+    self,
+    oid: str,
+    comment_type: BilibiliCommentType,
+    order_mode: CommentOrderType = CommentOrderType.DEFAULT,
+    next: int = 0,
+) -> Dict:
+    uri = "/x/v2/reply/wbi/main"
+    post_data = {
+        "oid": oid,
+        "mode": order_mode.value,
+        "type": comment_type.value,
+        "ps": 20,
+        "next": next,
+    }
+    return await self.get(uri, post_data)
+```
+
+Change `get_video_comments()` to call `get_comments(..., BilibiliCommentType.VIDEO, ...)`.
+
+**Step 4: Run test to verify it passes**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: PASS.
+
+**Step 5: Commit**
+
+```bash
+git add media_platform/bilibili/field.py media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
+git commit -m "feat: add generic bilibili comment type support"
+```
+
+---
+
+### Task 3: Generalize Corrected Comment Pagination And Second-Level Comments
+
+**Files:**
+- Modify: `media_platform/bilibili/client.py`
+- Test: `tests/media_platform/bilibili/test_client_comments.py`
+- Existing regression: `tests/test_bilibili_client_comments.py`
+
+**Step 1: Write failing pagination tests**
+
+Add tests that monkeypatch first-level and second-level methods:
+
+```python
+@pytest.mark.asyncio
+async def test_get_all_comments_limits_first_level_before_fetching_sub_comments(monkeypatch, bili_client):
+    fetched_roots = []
+    saved_batches = []
+
+    async def fake_get_comments(oid, comment_type, order_mode, next):
+        return {
+            "cursor": {"is_end": True, "next": 0},
+            "replies": [
+                {"rpid": 1, "rcount": 1, "content": {"message": "a"}, "member": {"mid": 1, "uname": "u1"}},
+                {"rpid": 2, "rcount": 1, "content": {"message": "b"}, "member": {"mid": 2, "uname": "u2"}},
+            ],
+        }
+
+    async def fake_get_all_level_two_comments(oid, comment_type, level_one_comment_id, order_mode, ps, crawl_interval, callback):
+        fetched_roots.append(level_one_comment_id)
+
+    async def fake_callback(oid, comments):
+        saved_batches.append(comments)
+
+    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
+    monkeypatch.setattr(bili_client, "get_all_level_two_comments", fake_get_all_level_two_comments)
+
+    await bili_client.get_all_comments(
+        oid="123",
+        comment_type=BilibiliCommentType.ARTICLE,
+        crawl_interval=0,
+        is_fetch_sub_comments=True,
+        callback=fake_callback,
+        max_count=1,
+    )
+
+    assert len(saved_batches[0]) == 1
+    assert fetched_roots == [1]
+```
+
+Also add or preserve a video-wrapper regression equivalent to `tests/test_bilibili_client_comments.py`:
+
+- `get_video_all_comments(..., is_fetch_sub_comments=True, max_count=1)` saves and returns only one retained first-level comment.
+- Second-level comments are fetched only for that retained first-level comment.
+- This regression should continue to pass after video methods delegate to generic methods.
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: FAIL because `get_all_comments()` and generic second-level methods do not exist yet. The existing video-specific regression should already pass before this task and must continue to pass after the refactor.
+
+**Step 3: Implement generic pagination**
+
+In `media_platform/bilibili/client.py`, add:
+
+- `get_all_comments(oid, comment_type, crawl_interval, is_fetch_sub_comments, callback, max_count)`
+- `get_all_level_two_comments(oid, comment_type, level_one_comment_id, order_mode, ps, crawl_interval, callback)`
+- `get_level_two_comments(oid, comment_type, level_one_comment_id, pn, ps, order_mode)`
+
+Required behavior:
+
+1. Fetch a page of first-level comments.
+2. Validate `cursor.is_end` and `cursor.next`.
+3. Trim `comment_list` before saving and before fetching sub-comments.
+4. Save first-level comments with `callback(oid, comment_list)`.
+5. If `is_fetch_sub_comments` is true, fetch second-level comments only for retained first-level comments where `rcount > 0`.
+6. Always extend the local result with retained first-level comments so `max_count` works in both modes.
+
+Important: `max_count` limits retained first-level comments. It does not impose a global cap on second-level comments; second-level pagination remains controlled by the second-level API loop for each retained first-level comment.
+
+Update old video methods as wrappers:
+
+- `get_video_all_comments()` calls `get_all_comments(..., BilibiliCommentType.VIDEO, ...)`.
+- `get_video_all_level_two_comments()` calls generic second-level method with `VIDEO`.
+- `get_video_level_two_comments()` calls generic method with `VIDEO`.
+
+**Step 4: Run tests**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py tests/test_bilibili_client_comments.py -v
+```
+
+Expected: PASS.
+
+**Step 5: Commit**
+
+```bash
+git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
+git commit -m "refactor: generalize bilibili comment pagination"
+```
+
+---
+
+### Task 4: Verify And Implement Article Detail Client
+
+**Files:**
+- Modify: `media_platform/bilibili/client.py`
+- Test: `tests/media_platform/bilibili/test_client_article.py`
+
+**Step 1: Manually verify article detail endpoint**
+
+Use one known public article ID and run a small script or interactive call through existing client authentication if needed. Candidate API endpoints to verify include Bilibili article/read APIs. Record the response fields used for:
+
+- article ID
+- title
+- summary or content
+- creator ID and name
+- publish timestamp
+- stats: likes, favorites, shares, comments
+
+Do not commit captured personal cookies or full response dumps.
+
+**Step 2: Write failing mapping test**
+
+Create `tests/media_platform/bilibili/test_client_article.py`:
+
+```python
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_get_article_info_calls_expected_endpoint(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["uri"] = uri
+        captured["params"] = params
+        return {"id": 123456, "title": "article title"}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    result = await bili_client.get_article_info("123456")
+
+    assert captured["params"]
+    assert result["title"] == "article title"
+```
+
+Adjust exact assertions after endpoint verification.
+
+**Step 3: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_article.py -v
+```
+
+Expected: FAIL because `get_article_info()` does not exist.
+
+**Step 4: Implement `get_article_info()`**
+
+Add a method to `BilibiliClient` that accepts article ID without the `cv` prefix and returns raw article detail data.
+
+Keep response normalization out of the client unless existing project style clearly normalizes in client. Current Bilibili video code stores raw-ish API detail and maps it in store layer, so follow that pattern.
+
+**Step 5: Run tests**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_article.py tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: PASS.
+
+**Step 6: Commit**
+
+```bash
+git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_article.py
+git commit -m "feat: add bilibili article detail client"
+```
+
+---
+
+### Task 5: Add Article Database Models
+
+**Files:**
+- Modify: `database/models.py`
+- Test: `tests/database/test_bilibili_article_models.py`
+
+**Step 1: Write model metadata test**
+
+Create `tests/database/test_bilibili_article_models.py`:
+
+```python
+from database.models import BilibiliArticle, BilibiliArticleComment
+
+
+def test_bilibili_article_table_name():
+    assert BilibiliArticle.__tablename__ == "bilibili_article"
+
+
+def test_bilibili_article_comment_table_name():
+    assert BilibiliArticleComment.__tablename__ == "bilibili_article_comment"
+```
+
+Add assertions for key columns if existing database model tests follow that style.
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/database/test_bilibili_article_models.py -v
+```
+
+Expected: FAIL because models do not exist.
+
+**Step 3: Add models**
+
+In `database/models.py`, add:
+
+`BilibiliArticle` fields:
+
+- `id`
+- `article_id`
+- `article_url`
+- `title`
+- `desc`
+- `content`
+- `creator_hash`
+- `nickname`
+- `liked_count`
+- `favorite_count`
+- `share_count`
+- `comment_count`
+- `create_time`
+- `source_keyword`
+- `add_ts`
+- `last_modify_ts`
+
+`BilibiliArticleComment` fields:
+
+- `id`
+- `creator_hash`
+- `nickname`
+- `add_ts`
+- `last_modify_ts`
+- `comment_id`
+- `article_id`
+- `content`
+- `create_time`
+- `sub_comment_count`
+- `parent_comment_id`
+- `like_count`
+
+Follow existing type choices from `BilibiliVideo` and `BilibiliVideoComment`.
+
+**Step 4: Run test**
+
+Run:
+
+```bash
+pytest tests/database/test_bilibili_article_models.py -v
+```
+
+Expected: PASS.
+
+**Step 5: Commit**
+
+```bash
+git add database/models.py tests/database/test_bilibili_article_models.py
+git commit -m "feat: add bilibili article database models"
+```
+
+---
+
+### Task 6: Add Article Store Methods
+
+**Files:**
+- Modify: `store/bilibili/__init__.py`
+- Modify: `store/bilibili/_store_impl.py`
+- Test: `tests/store/bilibili/test_article_store.py`
+
+**Step 1: Write failing store mapping tests**
+
+Create `tests/store/bilibili/test_article_store.py`.
+
+Use monkeypatching to avoid filesystem and database writes:
+
+```python
+import pytest
+
+from store import bilibili as bilibili_store
+
+
+@pytest.mark.asyncio
+async def test_update_bilibili_article_maps_article_fields(monkeypatch):
+    saved = {}
+
+    class FakeStore:
+        async def store_content(self, content_item):
+            saved.update(content_item)
+
+    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
+
+    await bilibili_store.update_bilibili_article({
+        "id": 123456,
+        "title": "title",
+        "summary": "summary",
+        "content": "content",
+        "publish_time": 1710000000,
+        "author": {"mid": 100, "name": "author"},
+        "stats": {"like": 1, "favorite": 2, "share": 3, "reply": 4},
+    })
+
+    assert saved["article_id"] == "123456"
+    assert saved["article_url"] == "https://www.bilibili.com/read/cv123456"
+    assert saved["title"] == "title"
+```
+
+Adjust raw field names after Task 4 endpoint verification.
+
+Add a comment mapping test:
+
+```python
+@pytest.mark.asyncio
+async def test_update_bilibili_article_comment_maps_common_reply_fields(monkeypatch):
+    saved = {}
+
+    class FakeStore:
+        async def store_article_comment(self, comment_item):
+            saved.update(comment_item)
+
+    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
+
+    await bilibili_store.update_bilibili_article_comment("123456", {
+        "rpid": 9,
+        "parent": 0,
+        "ctime": 1710000000,
+        "content": {"message": "hello"},
+        "member": {"mid": 10, "uname": "user"},
+        "like": 5,
+        "rcount": 1,
+    })
+
+    assert saved["article_id"] == "123456"
+    assert saved["comment_id"] == "9"
+    assert saved["content"] == "hello"
+```
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/store/bilibili/test_article_store.py -v
+```
+
+Expected: FAIL because article store methods do not exist.
+
+**Step 3: Add public store functions**
+
+In `store/bilibili/__init__.py`, add:
+
+- `update_bilibili_article(article_item: Dict)`
+- `batch_update_bilibili_article_comments(article_id: str, comments: List[Dict])`
+- `update_bilibili_article_comment(article_id: str, comment_item: Dict)`
+
+Use `anonymize_user_id()` and `mask_nickname()` consistently with video storage.
+
+**Step 4: Add store implementation methods**
+
+In `store/bilibili/_store_impl.py`:
+
+- Import `BilibiliArticle` and `BilibiliArticleComment`.
+- Add `store_article_comment()` to store classes that need a separate comment collection/table.
+- For file-based stores, write article content with `item_type="articles"` and article comments with `item_type="article_comments"`.
+- For DB store, upsert `BilibiliArticle` by `article_id` and `BilibiliArticleComment` by `comment_id`.
+- For Mongo store, use collection suffixes `articles` and `article_comments`.
+
+If changing the abstract store interface would affect every platform, avoid adding abstract methods. Use Bilibili store implementation methods directly where needed, or fall back to existing `store_comment()` only if the implementation can route by `item_type`.
+
+**Step 5: Run tests**
+
+Run:
+
+```bash
+pytest tests/store/bilibili/test_article_store.py tests/database/test_bilibili_article_models.py -v
+```
+
+Expected: PASS.
+
+**Step 6: Commit**
+
+```bash
+git add store/bilibili/__init__.py store/bilibili/_store_impl.py tests/store/bilibili/test_article_store.py
+git commit -m "feat: add bilibili article storage"
+```
+
+---
+
+### Task 7: Wire Article Detail Flow Into Bilibili Core
+
+**Files:**
+- Modify: `media_platform/bilibili/core.py`
+- Modify: `config/bilibili_config.py`
+- Test: `tests/media_platform/bilibili/test_core_article_flow.py`
+
+**Step 1: Write failing orchestration tests**
+
+Create `tests/media_platform/bilibili/test_core_article_flow.py`.
+
+Test that article inputs are parsed and passed to article detail/comment methods. Use a `BilibiliCrawler` instance with monkeypatched `bili_client` and store functions.
+
+Important cases:
+
+- `get_specified_articles(["cv123456"])` fetches article detail for `123456`.
+- If comments are enabled, article comments are fetched.
+- Existing `get_specified_videos()` remains unchanged.
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_core_article_flow.py -v
+```
+
+Expected: FAIL because article core methods do not exist.
+
+**Step 3: Add config documentation**
+
+In `config/bilibili_config.py`, update `BILI_SPECIFIED_ID_LIST` comments to say it supports:
+
+- Bilibili video URL
+- BV number
+- Bilibili article URL
+- `cv` article ID
+- numeric article ID, when treated as article ID in article-specific config
+
+Prefer adding `BILI_SPECIFIED_ARTICLE_ID_LIST = []` if mixed numeric IDs would be ambiguous. A pure numeric input could be a creator ID or article ID depending on context, so article-specific config is safer.
+
+Recommended approach:
+
+- Keep `BILI_SPECIFIED_ID_LIST` for videos.
+- Add `BILI_SPECIFIED_ARTICLE_ID_LIST` for articles.
+- In `detail` mode, crawl both lists.
+
+**Step 4: Add core methods**
+
+In `media_platform/bilibili/core.py`, add:
+
+- `get_specified_articles(article_url_list: List[str])`
+- `batch_get_article_comments(article_id_list: List[str])`
+- `get_article_comments(article_id: str, semaphore: asyncio.Semaphore)`
+
+Behavior:
+
+1. Parse article IDs.
+2. Fetch article detail with `self.bili_client.get_article_info(article_id)`.
+3. Store with `bilibili_store.update_bilibili_article(article_detail)`.
+4. If `ENABLE_GET_COMMENTS`, fetch article comments with:
+   - `self.bili_client.get_all_comments(...)`
+   - `comment_type=BilibiliCommentType.ARTICLE`
+   - `callback=bilibili_store.batch_update_bilibili_article_comments`
+   - `max_count=config.CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES`
+
+In `start()`, under `CRAWLER_TYPE == "detail"`, call both:
+
+- `get_specified_videos(config.BILI_SPECIFIED_ID_LIST)`
+- `get_specified_articles(config.BILI_SPECIFIED_ARTICLE_ID_LIST)`
+
+Only call article flow if the list exists and is non-empty.
+
+**Step 5: Run tests**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_core_article_flow.py tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: PASS.
+
+**Step 6: Commit**
+
+```bash
+git add media_platform/bilibili/core.py config/bilibili_config.py tests/media_platform/bilibili/test_core_article_flow.py
+git commit -m "feat: wire bilibili article detail crawling"
+```
+
+---
+
+### Task 8: Add Article Comment Wrapper Methods
+
+**Files:**
+- Modify: `media_platform/bilibili/client.py`
+- Test: `tests/media_platform/bilibili/test_client_comments.py`
+
+**Step 1: Write failing wrapper tests**
+
+Add tests:
+
+```python
+@pytest.mark.asyncio
+async def test_get_article_comments_uses_type_12(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get_comments(oid, comment_type, order_mode, next):
+        captured["oid"] = oid
+        captured["type"] = comment_type.value
+        return {}
+
+    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
+
+    await bili_client.get_article_comments("123456")
+
+    assert captured == {"oid": "123456", "type": 12}
+```
+
+**Step 2: Run test to verify it fails**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: FAIL because article wrappers do not exist.
+
+**Step 3: Add wrappers**
+
+Add:
+
+- `get_article_comments(article_id, order_mode=CommentOrderType.DEFAULT, next=0)`
+- `get_article_all_comments(article_id, crawl_interval=1.0, is_fetch_sub_comments=False, callback=None, max_count=10)`
+
+Both should delegate to generic comment methods with `BilibiliCommentType.ARTICLE`.
+
+**Step 4: Run tests**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili/test_client_comments.py -v
+```
+
+Expected: PASS.
+
+**Step 5: Commit**
+
+```bash
+git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
+git commit -m "feat: add bilibili article comment wrappers"
+```
+
+---
+
+### Task 9: Documentation And User Configuration
+
+**Files:**
+- Modify: `README.md`
+- Modify: `docs/index.md`
+- Modify: `docs/项目架构文档.md`
+
+**Step 1: Update user-facing docs**
+
+Document Bilibili article support:
+
+- `BILI_SPECIFIED_ARTICLE_ID_LIST`
+- Supported formats:
+  - `https://www.bilibili.com/read/cv123456`
+  - `cv123456`
+  - `123456`
+- Comment flags:
+  - `ENABLE_GET_COMMENTS`
+  - `ENABLE_GET_SUB_COMMENTS`
+  - `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES`
+- Data outputs:
+  - `articles`
+  - `article_comments`
+
+**Step 2: Run docs grep**
+
+Run:
+
+```bash
+rg -n "BILI_SPECIFIED_ARTICLE_ID_LIST|article_comments|专栏" README.md docs config/bilibili_config.py
+```
+
+Expected: New config and docs are discoverable.
+
+**Step 3: Commit**
+
+```bash
+git add README.md docs/index.md docs/项目架构文档.md
+git commit -m "docs: document bilibili article crawling"
+```
+
+---
+
+### Task 10: End-To-End Verification
+
+**Files:**
+- No code changes expected unless verification finds defects.
+
+**Step 1: Run focused tests**
+
+Run:
+
+```bash
+pytest tests/media_platform/bilibili tests/store/bilibili tests/database/test_bilibili_article_models.py -v
+```
+
+Expected: PASS.
+
+**Step 2: Run broader test suite**
+
+Run:
+
+```bash
+pytest tests -v
+```
+
+Expected: PASS, or document unrelated existing failures with exact failing tests.
+
+**Step 3: Run a dry manual detail crawl**
+
+Set a local config or command-line args for:
+
+- `PLATFORM=bili`
+- `CRAWLER_TYPE=detail`
+- one valid `BILI_SPECIFIED_ARTICLE_ID_LIST`
+- `ENABLE_GET_COMMENTS=True`
+- `ENABLE_GET_SUB_COMMENTS=False`
+- `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES=2`
+- `SAVE_DATA_OPTION=jsonl`
+
+Run the crawler and verify:
+
+- Article detail output exists.
+- Article comment output exists.
+- No video output is created unless video IDs are configured.
+
+**Step 4: Run optional second-level comment verification**
+
+Repeat with:
+
+- `ENABLE_GET_SUB_COMMENTS=True`
+- `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES=1`
+
+Verify:
+
+- For both video and article flows, only retained first-level comments trigger second-level comment fetching.
+- Saved second-level comments have non-zero `parent_comment_id`.
+
+**Step 5: Final status**
+
+Run:
+
+```bash
+git status --short --branch
+```
+
+Expected: clean working tree after final commit.
+
+---
+
+## Non-Goals For This Plan
+
+- Bilibili article search by keyword.
+- Crawling all articles from a creator homepage.
+- Downloading images embedded in article content.
+- Changing comment crawling behavior for non-Bilibili platforms.
+
+These can be added later once the specified-article flow is stable.
diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py
index a9c326231..f04c7f781 100644
--- a/media_platform/bilibili/client.py
+++ b/media_platform/bilibili/client.py
@@ -315,19 +315,17 @@ async def get_video_all_comments(
             if not isinstance(is_end, bool):
                 utils.logger.warning(f"[BilibiliClient.get_video_all_comments] 'is_end' is not a boolean for video_id: {video_id}. Assuming end of comments.")
                 is_end = True
-            if is_fetch_sub_comments:
-                for comment in comment_list:
-                    comment_id = comment['rpid']
-                    if (comment.get("rcount", 0) > 0):
-                        {await self.get_video_all_level_two_comments(video_id, comment_id, CommentOrderType.DEFAULT, 10, crawl_interval, callback)}
             if len(result) + len(comment_list) > max_count:
                 comment_list = comment_list[:max_count - len(result)]
             if callback:  # If there is a callback function, execute it
                 await callback(video_id, comment_list)
+            if is_fetch_sub_comments:
+                for comment in comment_list:
+                    comment_id = comment['rpid']
+                    if comment.get("rcount", 0) > 0:
+                        await self.get_video_all_level_two_comments(video_id, comment_id, CommentOrderType.DEFAULT, 10, crawl_interval, callback)
             await asyncio.sleep(crawl_interval)
-            if not is_fetch_sub_comments:
-                result.extend(comment_list)
-                continue
+            result.extend(comment_list)
         return result
 
     async def get_video_all_level_two_comments(
diff --git a/tests/test_bilibili_client_comments.py b/tests/test_bilibili_client_comments.py
new file mode 100644
index 000000000..48992969e
--- /dev/null
+++ b/tests/test_bilibili_client_comments.py
@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from media_platform.bilibili.client import BilibiliClient
+
+
+@pytest.mark.asyncio
+async def test_video_comments_trim_first_level_before_fetching_sub_comments(monkeypatch):
+    client = object.__new__(BilibiliClient)
+    fetched_roots = []
+    saved_batches = []
+
+    async def fake_get_video_comments(video_id, order_mode, next):
+        return {
+            "cursor": {"is_end": True, "next": 0},
+            "replies": [
+                {"rpid": 1, "rcount": 1, "content": {"message": "a"}},
+                {"rpid": 2, "rcount": 1, "content": {"message": "b"}},
+            ],
+        }
+
+    async def fake_get_video_all_level_two_comments(
+        video_id,
+        level_one_comment_id,
+        order_mode,
+        ps,
+        crawl_interval,
+        callback,
+    ):
+        fetched_roots.append(level_one_comment_id)
+
+    async def fake_callback(video_id, comments):
+        saved_batches.append(comments)
+
+    async def fake_sleep(_):
+        return None
+
+    monkeypatch.setattr(client, "get_video_comments", fake_get_video_comments)
+    monkeypatch.setattr(client, "get_video_all_level_two_comments", fake_get_video_all_level_two_comments)
+    monkeypatch.setattr("media_platform.bilibili.client.asyncio.sleep", fake_sleep)
+
+    result = await client.get_video_all_comments(
+        video_id="123",
+        crawl_interval=0,
+        is_fetch_sub_comments=True,
+        callback=fake_callback,
+        max_count=1,
+    )
+
+    assert [comment["rpid"] for comment in saved_batches[0]] == [1]
+    assert fetched_roots == [1]
+    assert [comment["rpid"] for comment in result] == [1]

From ebebebe5246ee41780e23f5148a831d920f3266a Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 20:32:50 +0800
Subject: [PATCH 02/12] feat: parse bilibili article urls

---
 media_platform/bilibili/help.py            | 29 +++++++++++++++++++++-
 model/m_bilibili.py                        |  6 +++++
 tests/media_platform/bilibili/test_help.py | 25 +++++++++++++++++++
 3 files changed, 59 insertions(+), 1 deletion(-)
 create mode 100644 tests/media_platform/bilibili/test_help.py

diff --git a/media_platform/bilibili/help.py b/media_platform/bilibili/help.py
index 74fdfa24d..9830860d3 100644
--- a/media_platform/bilibili/help.py
+++ b/media_platform/bilibili/help.py
@@ -28,7 +28,7 @@
 from hashlib import md5
 from typing import Dict
 
-from model.m_bilibili import VideoUrlInfo, CreatorUrlInfo
+from model.m_bilibili import ArticleUrlInfo, CreatorUrlInfo, VideoUrlInfo
 from tools import utils
 
 
@@ -104,6 +104,33 @@ def parse_video_info_from_url(url: str) -> VideoUrlInfo:
     raise ValueError(f"Unable to parse video ID from URL: {url}")
 
 
+def parse_article_info_from_url(url: str) -> ArticleUrlInfo:
+    """
+    Parse article ID from Bilibili article URL.
+    Args:
+        url: Bilibili article link or cv id
+            - https://www.bilibili.com/read/cv123456
+            - cv123456
+            - 123456
+    Returns:
+        ArticleUrlInfo: Object containing article ID
+    """
+    if url.isdigit():
+        return ArticleUrlInfo(article_id=url)
+
+    cv_pattern = r"^cv(\d+)$"
+    match = re.search(cv_pattern, url)
+    if match:
+        return ArticleUrlInfo(article_id=match.group(1))
+
+    article_pattern = r"/read/cv(\d+)"
+    match = re.search(article_pattern, url)
+    if match:
+        return ArticleUrlInfo(article_id=match.group(1))
+
+    raise ValueError(f"Unable to parse article ID from URL: {url}")
+
+
 def parse_creator_info_from_url(url: str) -> CreatorUrlInfo:
     """
     Parse creator ID from Bilibili creator space URL
diff --git a/model/m_bilibili.py b/model/m_bilibili.py
index 2852144ee..cc7138fec 100644
--- a/model/m_bilibili.py
+++ b/model/m_bilibili.py
@@ -38,6 +38,12 @@ class VideoUrlInfo(BaseModel):
     video_type: str = Field(default="video", title="video type")
 
 
+class ArticleUrlInfo(BaseModel):
+    """Bilibili article URL information"""
+    article_id: str = Field(title="article id (cv id without cv prefix)")
+    article_type: str = Field(default="article", title="article type")
+
+
 class CreatorUrlInfo(BaseModel):
     """Bilibili creator URL information"""
     creator_id: str = Field(title="creator id (UID)")
diff --git a/tests/media_platform/bilibili/test_help.py b/tests/media_platform/bilibili/test_help.py
new file mode 100644
index 000000000..929ce6432
--- /dev/null
+++ b/tests/media_platform/bilibili/test_help.py
@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from media_platform.bilibili.help import parse_article_info_from_url
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("cv123456", "123456"),
+        ("123456", "123456"),
+        ("https://www.bilibili.com/read/cv123456", "123456"),
+        ("https://www.bilibili.com/read/cv123456?spm_id_from=333.999.0.0", "123456"),
+    ],
+)
+def test_parse_article_info_from_url(raw, expected):
+    article_info = parse_article_info_from_url(raw)
+    assert article_info.article_id == expected
+    assert article_info.article_type == "article"
+
+
+def test_parse_article_info_from_url_invalid():
+    with pytest.raises(ValueError):
+        parse_article_info_from_url("https://www.bilibili.com/video/BV1d54y1g7db")

From 56b858c45d3247f9fe5fcd4d983de3a613900e79 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 20:35:03 +0800
Subject: [PATCH 03/12] feat: add generic bilibili comment type support

---
 media_platform/bilibili/client.py             | 24 +++++++--
 media_platform/bilibili/field.py              |  5 ++
 .../bilibili/test_client_comments.py          | 50 +++++++++++++++++++
 3 files changed, 75 insertions(+), 4 deletions(-)
 create mode 100644 tests/media_platform/bilibili/test_client_comments.py

diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py
index f04c7f781..4394fa593 100644
--- a/media_platform/bilibili/client.py
+++ b/media_platform/bilibili/client.py
@@ -40,7 +40,7 @@
     from proxy.proxy_ip_pool import ProxyIpPool
 
 from .exception import DataFetchError
-from .field import CommentOrderType, SearchOrderType
+from .field import BilibiliCommentType, CommentOrderType, SearchOrderType
 from .help import BilibiliSign
 
 
@@ -241,6 +241,24 @@ async def get_video_media(self, url: str) -> Union[bytes, None]:
                 utils.logger.error(f"[BilibiliClient.get_video_media] {exc.__class__.__name__} for {exc.request.url} - {exc}")  # Keep original exception type name for developer debugging
                 return None
 
+    async def get_comments(
+        self,
+        oid: str,
+        comment_type: BilibiliCommentType,
+        order_mode: CommentOrderType = CommentOrderType.DEFAULT,
+        next: int = 0,
+    ) -> Dict:
+        """get comments by Bilibili comment object type
+        :param oid: Comment object ID
+        :param comment_type: Bilibili comment object type
+        :param order_mode: Sort order
+        :param next: Comment page selection
+        :return:
+        """
+        uri = "/x/v2/reply/wbi/main"
+        post_data = {"oid": oid, "mode": order_mode.value, "type": comment_type.value, "ps": 20, "next": next}
+        return await self.get(uri, post_data)
+
     async def get_video_comments(
         self,
         video_id: str,
@@ -253,9 +271,7 @@ async def get_video_comments(
         :param next: Comment page selection
         :return:
         """
-        uri = "/x/v2/reply/wbi/main"
-        post_data = {"oid": video_id, "mode": order_mode.value, "type": 1, "ps": 20, "next": next}
-        return await self.get(uri, post_data)
+        return await self.get_comments(video_id, BilibiliCommentType.VIDEO, order_mode, next)
 
     async def get_video_all_comments(
         self,
diff --git a/media_platform/bilibili/field.py b/media_platform/bilibili/field.py
index 4a4ec751b..b25773e68 100644
--- a/media_platform/bilibili/field.py
+++ b/media_platform/bilibili/field.py
@@ -52,3 +52,8 @@ class CommentOrderType(Enum):
 
     # By time
     TIME = 2
+
+
+class BilibiliCommentType(Enum):
+    VIDEO = 1
+    ARTICLE = 12
diff --git a/tests/media_platform/bilibili/test_client_comments.py b/tests/media_platform/bilibili/test_client_comments.py
new file mode 100644
index 000000000..f742748f0
--- /dev/null
+++ b/tests/media_platform/bilibili/test_client_comments.py
@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from media_platform.bilibili.client import BilibiliClient
+from media_platform.bilibili.field import BilibiliCommentType, CommentOrderType
+
+
+@pytest.fixture
+def bili_client():
+    return object.__new__(BilibiliClient)
+
+
+@pytest.mark.asyncio
+async def test_get_comments_uses_comment_type(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["uri"] = uri
+        captured["params"] = params
+        return {}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    await bili_client.get_comments(
+        oid="123456",
+        comment_type=BilibiliCommentType.ARTICLE,
+        order_mode=CommentOrderType.DEFAULT,
+        next=0,
+    )
+
+    assert captured["uri"] == "/x/v2/reply/wbi/main"
+    assert captured["params"]["oid"] == "123456"
+    assert captured["params"]["type"] == 12
+
+
+@pytest.mark.asyncio
+async def test_get_video_comments_keeps_type_1(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["params"] = params
+        return {}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    await bili_client.get_video_comments("998877")
+
+    assert captured["params"]["oid"] == "998877"
+    assert captured["params"]["type"] == 1

From 288a46a1cb9bb3e3a5979094a7bec6f0cea60a04 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 20:39:53 +0800
Subject: [PATCH 04/12] refactor: generalize bilibili comment pagination

---
 media_platform/bilibili/client.py             | 138 +++++++++++++++---
 .../bilibili/test_client_comments.py          |  91 ++++++++++++
 tests/test_bilibili_client_comments.py        |  14 +-
 3 files changed, 216 insertions(+), 27 deletions(-)

diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py
index 4394fa593..26cbc978e 100644
--- a/media_platform/bilibili/client.py
+++ b/media_platform/bilibili/client.py
@@ -273,17 +273,19 @@ async def get_video_comments(
         """
         return await self.get_comments(video_id, BilibiliCommentType.VIDEO, order_mode, next)
 
-    async def get_video_all_comments(
+    async def get_all_comments(
         self,
-        video_id: str,
+        oid: str,
+        comment_type: BilibiliCommentType,
         crawl_interval: float = 1.0,
         is_fetch_sub_comments=False,
         callback: Optional[Callable] = None,
         max_count: int = 10,
     ):
         """
-        get video all comments include sub comments
-        :param video_id:
+        get all comments include sub comments
+        :param oid:
+        :param comment_type:
         :param crawl_interval:
         :param is_fetch_sub_comments:
         :param callback:
@@ -299,15 +301,15 @@ async def get_video_all_comments(
             comments_res = None
             for attempt in range(max_retries):
                 try:
-                    comments_res = await self.get_video_comments(video_id, CommentOrderType.DEFAULT, next_page)
+                    comments_res = await self.get_comments(oid, comment_type, CommentOrderType.DEFAULT, next_page)
                     break  # Success
                 except DataFetchError as e:
                     if attempt < max_retries - 1:
                         delay = 5 * (2**attempt) + random.uniform(0, 1)
-                        utils.logger.warning(f"[BilibiliClient.get_video_all_comments] Retrying video_id {video_id} in {delay:.2f}s... (Attempt {attempt + 1}/{max_retries})")
+                        utils.logger.warning(f"[BilibiliClient.get_all_comments] Retrying oid {oid} in {delay:.2f}s... (Attempt {attempt + 1}/{max_retries})")
                         await asyncio.sleep(delay)
                     else:
-                        utils.logger.error(f"[BilibiliClient.get_video_all_comments] Max retries reached for video_id: {video_id}. Skipping comments. Error: {e}")
+                        utils.logger.error(f"[BilibiliClient.get_all_comments] Max retries reached for oid: {oid}. Skipping comments. Error: {e}")
                         is_end = True
                         break
             if not comments_res:
@@ -315,38 +317,74 @@ async def get_video_all_comments(
 
             cursor_info: Dict = comments_res.get("cursor")
             if not cursor_info:
-                utils.logger.warning(f"[BilibiliClient.get_video_all_comments] Could not find 'cursor' in response for video_id: {video_id}. Skipping.")
+                utils.logger.warning(f"[BilibiliClient.get_all_comments] Could not find 'cursor' in response for oid: {oid}. Skipping.")
                 break
 
             comment_list: List[Dict] = comments_res.get("replies", [])
 
             # Check if is_end and next exist
             if "is_end" not in cursor_info or "next" not in cursor_info:
-                utils.logger.warning(f"[BilibiliClient.get_video_all_comments] 'is_end' or 'next' not in cursor for video_id: {video_id}. Assuming end of comments.")
+                utils.logger.warning(f"[BilibiliClient.get_all_comments] 'is_end' or 'next' not in cursor for oid: {oid}. Assuming end of comments.")
                 is_end = True
             else:
                 is_end = cursor_info.get("is_end")
                 next_page = cursor_info.get("next")
 
             if not isinstance(is_end, bool):
-                utils.logger.warning(f"[BilibiliClient.get_video_all_comments] 'is_end' is not a boolean for video_id: {video_id}. Assuming end of comments.")
+                utils.logger.warning(f"[BilibiliClient.get_all_comments] 'is_end' is not a boolean for oid: {oid}. Assuming end of comments.")
                 is_end = True
             if len(result) + len(comment_list) > max_count:
                 comment_list = comment_list[:max_count - len(result)]
             if callback:  # If there is a callback function, execute it
-                await callback(video_id, comment_list)
+                await callback(oid, comment_list)
             if is_fetch_sub_comments:
                 for comment in comment_list:
                     comment_id = comment['rpid']
                     if comment.get("rcount", 0) > 0:
-                        await self.get_video_all_level_two_comments(video_id, comment_id, CommentOrderType.DEFAULT, 10, crawl_interval, callback)
+                        await self.get_all_level_two_comments(
+                            oid,
+                            comment_type,
+                            comment_id,
+                            CommentOrderType.DEFAULT,
+                            10,
+                            crawl_interval,
+                            callback,
+                        )
             await asyncio.sleep(crawl_interval)
             result.extend(comment_list)
         return result
 
-    async def get_video_all_level_two_comments(
+    async def get_video_all_comments(
         self,
         video_id: str,
+        crawl_interval: float = 1.0,
+        is_fetch_sub_comments=False,
+        callback: Optional[Callable] = None,
+        max_count: int = 10,
+    ):
+        """
+        get video all comments include sub comments
+        :param video_id:
+        :param crawl_interval:
+        :param is_fetch_sub_comments:
+        :param callback:
+        max_count: Maximum number of comments to crawl per note
+
+        :return:
+        """
+        return await self.get_all_comments(
+            video_id,
+            BilibiliCommentType.VIDEO,
+            crawl_interval,
+            is_fetch_sub_comments,
+            callback,
+            max_count,
+        )
+
+    async def get_all_level_two_comments(
+        self,
+        oid: str,
+        comment_type: BilibiliCommentType,
         level_one_comment_id: int,
         order_mode: CommentOrderType,
         ps: int = 10,
@@ -354,8 +392,9 @@ async def get_video_all_level_two_comments(
         callback: Optional[Callable] = None,
     ) -> Dict:
         """
-        get video all level two comments for a level one comment
-        :param video_id: Video ID
+        get all level two comments for a level one comment
+        :param oid: Comment object ID
+        :param comment_type: Bilibili comment object type
         :param level_one_comment_id: Level one comment ID
         :param order_mode:
         :param ps: Number of comments per page
@@ -366,26 +405,57 @@ async def get_video_all_level_two_comments(
 
         pn = 1
         while True:
-            result = await self.get_video_level_two_comments(video_id, level_one_comment_id, pn, ps, order_mode)
+            result = await self.get_level_two_comments(oid, comment_type, level_one_comment_id, pn, ps, order_mode)
             comment_list: List[Dict] = result.get("replies", [])
             if callback:  # If there is a callback function, execute it
-                await callback(video_id, comment_list)
+                await callback(oid, comment_list)
             await asyncio.sleep(crawl_interval)
             if (int(result["page"]["count"]) <= pn * ps):
                 break
 
             pn += 1
 
-    async def get_video_level_two_comments(
+    async def get_video_all_level_two_comments(
         self,
         video_id: str,
         level_one_comment_id: int,
+        order_mode: CommentOrderType,
+        ps: int = 10,
+        crawl_interval: float = 1.0,
+        callback: Optional[Callable] = None,
+    ) -> Dict:
+        """
+        get video all level two comments for a level one comment
+        :param video_id: Video ID
+        :param level_one_comment_id: Level one comment ID
+        :param order_mode:
+        :param ps: Number of comments per page
+        :param crawl_interval:
+        :param callback:
+        :return:
+        """
+        return await self.get_all_level_two_comments(
+            video_id,
+            BilibiliCommentType.VIDEO,
+            level_one_comment_id,
+            order_mode,
+            ps,
+            crawl_interval,
+            callback,
+        )
+
+    async def get_level_two_comments(
+        self,
+        oid: str,
+        comment_type: BilibiliCommentType,
+        level_one_comment_id: int,
         pn: int,
         ps: int,
         order_mode: CommentOrderType,
     ) -> Dict:
-        """get video level two comments
-        :param video_id: Video ID
+        """get level two comments
+        :param oid: Comment object ID
+        :param comment_type: Bilibili comment object type
         :param level_one_comment_id: Level one comment ID
         :param order_mode: Sort order
 
@@ -393,9 +463,9 @@ async def get_video_level_two_comments(
         """
         uri = "/x/v2/reply/reply"
         post_data = {
-            "oid": video_id,
+            "oid": oid,
             "mode": order_mode.value,
-            "type": 1,
+            "type": comment_type.value,
             "ps": ps,
             "pn": pn,
             "root": level_one_comment_id,
@@ -403,6 +473,30 @@ async def get_video_level_two_comments(
         result = await self.get(uri, post_data)
         return result
 
+    async def get_video_level_two_comments(
+        self,
+        video_id: str,
+        level_one_comment_id: int,
+        pn: int,
+        ps: int,
+        order_mode: CommentOrderType,
+    ) -> Dict:
+        """get video level two comments
+        :param video_id: Video ID
+        :param level_one_comment_id: Level one comment ID
+        :param order_mode: Sort order
+
+        :return:
+        """
+        return await self.get_level_two_comments(
+            video_id,
+            BilibiliCommentType.VIDEO,
+            level_one_comment_id,
+            pn,
+            ps,
+            order_mode,
+        )
+
     async def get_creator_videos(self, creator_id: str, pn: int, ps: int = 30, order_mode: SearchOrderType = SearchOrderType.LAST_PUBLISH) -> Dict:
         """get all videos for a creator
         :param creator_id: Creator ID
diff --git a/tests/media_platform/bilibili/test_client_comments.py b/tests/media_platform/bilibili/test_client_comments.py
index f742748f0..1ee085778 100644
--- a/tests/media_platform/bilibili/test_client_comments.py
+++ b/tests/media_platform/bilibili/test_client_comments.py
@@ -48,3 +48,94 @@ async def fake_get(uri, params=None, enable_params_sign=True):
 
     assert captured["params"]["oid"] == "998877"
     assert captured["params"]["type"] == 1
+
+
+@pytest.mark.asyncio
+async def test_get_all_comments_limits_first_level_before_fetching_sub_comments(monkeypatch, bili_client):
+    fetched_roots = []
+    saved_batches = []
+
+    async def fake_get_comments(oid, comment_type, order_mode, next):
+        return {
+            "cursor": {"is_end": True, "next": 0},
+            "replies": [
+                {"rpid": 1, "rcount": 1, "content": {"message": "a"}, "member": {"mid": 1, "uname": "u1"}},
+                {"rpid": 2, "rcount": 1, "content": {"message": "b"}, "member": {"mid": 2, "uname": "u2"}},
+            ],
+        }
+
+    async def fake_get_all_level_two_comments(
+        oid,
+        comment_type,
+        level_one_comment_id,
+        order_mode,
+        ps,
+        crawl_interval,
+        callback,
+    ):
+        fetched_roots.append(level_one_comment_id)
+
+    async def fake_callback(oid, comments):
+        saved_batches.append(comments)
+
+    async def fake_sleep(_):
+        return None
+
+    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
+    monkeypatch.setattr(bili_client, "get_all_level_two_comments", fake_get_all_level_two_comments)
+    monkeypatch.setattr("media_platform.bilibili.client.asyncio.sleep", fake_sleep)
+
+    result = await bili_client.get_all_comments(
+        oid="123",
+        comment_type=BilibiliCommentType.ARTICLE,
+        crawl_interval=0,
+        is_fetch_sub_comments=True,
+        callback=fake_callback,
+        max_count=1,
+    )
+
+    assert [comment["rpid"] for comment in saved_batches[0]] == [1]
+    assert fetched_roots == [1]
+    assert [comment["rpid"] for comment in result] == [1]
+
+
+@pytest.mark.asyncio
+async def test_get_video_all_comments_wrapper_preserves_trimmed_sub_comment_fetch(monkeypatch, bili_client):
+    fetched_roots = []
+    saved_batches = []
+
+    async def fake_get_all_comments(
+        oid,
+        comment_type,
+        crawl_interval,
+        is_fetch_sub_comments,
+        callback,
+        max_count,
+    ):
+        assert oid == "123"
+        assert comment_type is BilibiliCommentType.VIDEO
+        assert is_fetch_sub_comments is True
+        assert max_count == 1
+        await callback(
+            oid,
+            [{"rpid": 1, "rcount": 1, "content": {"message": "a"}, "member": {"mid": 1, "uname": "u1"}}],
+        )
+        fetched_roots.append(1)
+        return [{"rpid": 1, "rcount": 1, "content": {"message": "a"}, "member": {"mid": 1, "uname": "u1"}}]
+
+    async def fake_callback(oid, comments):
+        saved_batches.append(comments)
+
+    monkeypatch.setattr(bili_client, "get_all_comments", fake_get_all_comments)
+
+    result = await bili_client.get_video_all_comments(
+        video_id="123",
+        crawl_interval=0,
+        is_fetch_sub_comments=True,
+        callback=fake_callback,
+        max_count=1,
+    )
+
+    assert [comment["rpid"] for comment in saved_batches[0]] == [1]
+    assert fetched_roots == [1]
+    assert [comment["rpid"] for comment in result] == [1]
diff --git a/tests/test_bilibili_client_comments.py b/tests/test_bilibili_client_comments.py
index 48992969e..543740354 100644
--- a/tests/test_bilibili_client_comments.py
+++ b/tests/test_bilibili_client_comments.py
@@ -3,6 +3,7 @@
 import pytest
 
 from media_platform.bilibili.client import BilibiliClient
+from media_platform.bilibili.field import BilibiliCommentType
 
 
 @pytest.mark.asyncio
@@ -11,7 +12,8 @@ async def test_video_comments_trim_first_level_before_fetching_sub_comments(monk
     fetched_roots = []
     saved_batches = []
 
-    async def fake_get_video_comments(video_id, order_mode, next):
+    async def fake_get_comments(oid, comment_type, order_mode, next):
+        assert comment_type is BilibiliCommentType.VIDEO
         return {
             "cursor": {"is_end": True, "next": 0},
             "replies": [
@@ -20,14 +22,16 @@ async def fake_get_video_comments(video_id, order_mode, next):
             ],
         }
 
-    async def fake_get_video_all_level_two_comments(
-        video_id,
+    async def fake_get_all_level_two_comments(
+        oid,
+        comment_type,
         level_one_comment_id,
         order_mode,
         ps,
         crawl_interval,
         callback,
     ):
+        assert comment_type is BilibiliCommentType.VIDEO
         fetched_roots.append(level_one_comment_id)
 
     async def fake_callback(video_id, comments):
@@ -36,8 +40,8 @@ async def fake_callback(video_id, comments):
     async def fake_sleep(_):
         return None
 
-    monkeypatch.setattr(client, "get_video_comments", fake_get_video_comments)
-    monkeypatch.setattr(client, "get_video_all_level_two_comments", fake_get_video_all_level_two_comments)
+    monkeypatch.setattr(client, "get_comments", fake_get_comments)
+    monkeypatch.setattr(client, "get_all_level_two_comments", fake_get_all_level_two_comments)
     monkeypatch.setattr("media_platform.bilibili.client.asyncio.sleep", fake_sleep)
 
     result = await client.get_video_all_comments(

From 21abc7a5efd200a10fa2ced9afb637da336d03b7 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 23:00:24 +0800
Subject: [PATCH 05/12] feat: add bilibili article detail client

---
 media_platform/bilibili/client.py             | 13 ++++++++
 .../bilibili/test_client_article.py           | 30 +++++++++++++++++++
 2 files changed, 43 insertions(+)
 create mode 100644 tests/media_platform/bilibili/test_client_article.py

diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py
index 26cbc978e..4d9432f67 100644
--- a/media_platform/bilibili/client.py
+++ b/media_platform/bilibili/client.py
@@ -203,6 +203,19 @@ async def get_video_info(self, aid: Union[int, None] = None, bvid: Union[str, No
             params.update({"bvid": bvid})
         return await self.get(uri, params, enable_params_sign=False)
 
+    async def get_article_info(self, article_id: str) -> Dict:
+        """
+        Bilibili article detail api
+        :param article_id: Article cv id without cv prefix
+        :return:
+        """
+        if not article_id:
+            raise ValueError("Please provide article_id")
+
+        uri = "/x/article/view"
+        params = {"id": article_id}
+        return await self.get(uri, params, enable_params_sign=False)
+
     async def get_video_play_url(self, aid: int, cid: int) -> Dict:
         """
         Bilibli web video play url api
diff --git a/tests/media_platform/bilibili/test_client_article.py b/tests/media_platform/bilibili/test_client_article.py
new file mode 100644
index 000000000..7f77c91a2
--- /dev/null
+++ b/tests/media_platform/bilibili/test_client_article.py
@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from media_platform.bilibili.client import BilibiliClient
+
+
+@pytest.fixture
+def bili_client():
+    return object.__new__(BilibiliClient)
+
+
+@pytest.mark.asyncio
+async def test_get_article_info_calls_article_view_endpoint(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get(uri, params=None, enable_params_sign=True):
+        captured["uri"] = uri
+        captured["params"] = params
+        captured["enable_params_sign"] = enable_params_sign
+        return {"id": 123456, "title": "article title"}
+
+    monkeypatch.setattr(bili_client, "get", fake_get)
+
+    result = await bili_client.get_article_info("123456")
+
+    assert captured["uri"] == "/x/article/view"
+    assert captured["params"] == {"id": "123456"}
+    assert captured["enable_params_sign"] is False
+    assert result["title"] == "article title"

From 00f4b4354a5d6d2d7352d4d01cba4da4b80ffc5d Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 23:03:14 +0800
Subject: [PATCH 06/12] feat: add bilibili article database models

---
 database/models.py                            | 34 ++++++++++++
 .../database/test_bilibili_article_models.py  | 53 +++++++++++++++++++
 2 files changed, 87 insertions(+)
 create mode 100644 tests/database/test_bilibili_article_models.py

diff --git a/database/models.py b/database/models.py
index 011777d0a..44b7c55ea 100644
--- a/database/models.py
+++ b/database/models.py
@@ -69,6 +69,40 @@ class BilibiliVideoComment(Base):
     parent_comment_id = Column(String(255), comment='父评论ID')
     like_count = Column(Text, default='0', comment='点赞数')
 
+class BilibiliArticle(Base):
+    __tablename__ = 'bilibili_article'
+    id = Column(Integer, primary_key=True, comment='primary key')
+    article_id = Column(BigInteger, nullable=False, index=True, unique=True, comment='article id')
+    article_url = Column(Text, nullable=False, comment='article url')
+    title = Column(Text, comment='article title')
+    desc = Column(Text, comment='article description')
+    content = Column(Text, comment='article content')
+    creator_hash = Column(String(64), index=True, comment='anonymized creator id hash')
+    nickname = Column(Text, comment='masked nickname')
+    liked_count = Column(Text, comment='like count')
+    favorite_count = Column(Text, comment='favorite count')
+    share_count = Column(Text, comment='share count')
+    comment_count = Column(Text, comment='comment count')
+    create_time = Column(BigInteger, index=True, comment='create timestamp')
+    source_keyword = Column(Text, default='', comment='source keyword')
+    add_ts = Column(BigInteger, comment='add timestamp')
+    last_modify_ts = Column(BigInteger, comment='last modify timestamp')
+
+class BilibiliArticleComment(Base):
+    __tablename__ = 'bilibili_article_comment'
+    id = Column(Integer, primary_key=True, comment='primary key')
+    creator_hash = Column(String(64), index=True, comment='anonymized creator id hash')
+    nickname = Column(Text, comment='masked nickname')
+    add_ts = Column(BigInteger, comment='add timestamp')
+    last_modify_ts = Column(BigInteger, comment='last modify timestamp')
+    comment_id = Column(BigInteger, index=True, comment='comment id')
+    article_id = Column(BigInteger, index=True, comment='article id')
+    content = Column(Text, comment='comment content')
+    create_time = Column(BigInteger, comment='create timestamp')
+    sub_comment_count = Column(Text, comment='sub comment count')
+    parent_comment_id = Column(String(255), comment='parent comment id')
+    like_count = Column(Text, default='0', comment='like count')
+
 class BilibiliUpDynamic(Base):
     __tablename__ = 'bilibili_up_dynamic'
     id = Column(Integer, primary_key=True, comment='主键ID')
diff --git a/tests/database/test_bilibili_article_models.py b/tests/database/test_bilibili_article_models.py
new file mode 100644
index 000000000..be7224b03
--- /dev/null
+++ b/tests/database/test_bilibili_article_models.py
@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+
+from database.models import BilibiliArticle, BilibiliArticleComment
+
+
+def test_bilibili_article_table_name():
+    assert BilibiliArticle.__tablename__ == "bilibili_article"
+
+
+def test_bilibili_article_columns():
+    columns = set(BilibiliArticle.__table__.columns.keys())
+
+    assert {
+        "id",
+        "article_id",
+        "article_url",
+        "title",
+        "desc",
+        "content",
+        "creator_hash",
+        "nickname",
+        "liked_count",
+        "favorite_count",
+        "share_count",
+        "comment_count",
+        "create_time",
+        "source_keyword",
+        "add_ts",
+        "last_modify_ts",
+    }.issubset(columns)
+
+
+def test_bilibili_article_comment_table_name():
+    assert BilibiliArticleComment.__tablename__ == "bilibili_article_comment"
+
+
+def test_bilibili_article_comment_columns():
+    columns = set(BilibiliArticleComment.__table__.columns.keys())
+
+    assert {
+        "id",
+        "creator_hash",
+        "nickname",
+        "add_ts",
+        "last_modify_ts",
+        "comment_id",
+        "article_id",
+        "content",
+        "create_time",
+        "sub_comment_count",
+        "parent_comment_id",
+        "like_count",
+    }.issubset(columns)

From 71f45a4ef0b423679e930dfffa26728e8a9bafb4 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 23:10:26 +0800
Subject: [PATCH 07/12] feat: add bilibili article storage

---
 store/bilibili/__init__.py                 |  57 ++++++++++
 store/bilibili/_store_impl.py              | 116 ++++++++++++++++++++-
 tests/store/bilibili/test_article_store.py |  70 +++++++++++++
 3 files changed, 242 insertions(+), 1 deletion(-)
 create mode 100644 tests/store/bilibili/test_article_store.py

diff --git a/store/bilibili/__init__.py b/store/bilibili/__init__.py
index b5eb9215a..924cbc709 100644
--- a/store/bilibili/__init__.py
+++ b/store/bilibili/__init__.py
@@ -87,6 +87,63 @@ async def update_up_info(video_item: Dict):
     return
 
 
+async def update_bilibili_article(article_item: Dict):
+    author_info: Dict = article_item.get("author") or {}
+    article_stat: Dict = article_item.get("stats") or {}
+    article_id = str(article_item.get("id"))
+    save_content_item = {
+        "article_id": article_id,
+        "article_url": f"https://www.bilibili.com/read/cv{article_id}",
+        "title": article_item.get("title", "")[:500],
+        "desc": article_item.get("summary", "")[:500],
+        "content": article_item.get("content", ""),
+        "create_time": article_item.get("publish_time") or article_item.get("ctime"),
+        "creator_hash": anonymize_user_id(author_info.get("mid")),
+        "nickname": mask_nickname(author_info.get("name")),
+        "liked_count": str(article_stat.get("like", "")),
+        "favorite_count": str(article_stat.get("favorite", "")),
+        "share_count": str(article_stat.get("share", "")),
+        "comment_count": str(article_stat.get("reply", "")),
+        "last_modify_ts": utils.get_current_timestamp(),
+        "source_keyword": source_keyword_var.get(),
+    }
+    utils.logger.info(f"[store.bilibili.update_bilibili_article] bilibili article id:{article_id}, title:{save_content_item.get('title')}")
+    store = BiliStoreFactory.create_store()
+    if hasattr(store, "store_article"):
+        await store.store_article(article_item=save_content_item)
+    else:
+        await store.store_content(content_item=save_content_item)
+
+
+async def batch_update_bilibili_article_comments(article_id: str, comments: List[Dict]):
+    if not comments:
+        return
+    for comment_item in comments:
+        await update_bilibili_article_comment(article_id, comment_item)
+
+
+async def update_bilibili_article_comment(article_id: str, comment_item: Dict):
+    comment_id = str(comment_item.get("rpid"))
+    parent_comment_id = str(comment_item.get("parent", 0))
+    content: Dict = comment_item.get("content") or {}
+    user_info: Dict = comment_item.get("member") or {}
+    like_count: int = comment_item.get("like", 0)
+    save_comment_item = {
+        "comment_id": comment_id,
+        "parent_comment_id": parent_comment_id,
+        "create_time": comment_item.get("ctime"),
+        "article_id": str(article_id),
+        "content": content.get("message"),
+        "creator_hash": anonymize_user_id(user_info.get("mid")),
+        "nickname": mask_nickname(user_info.get("uname")),
+        "sub_comment_count": str(comment_item.get("rcount", 0)),
+        "like_count": like_count,
+        "last_modify_ts": utils.get_current_timestamp(),
+    }
+    utils.logger.info(f"[store.bilibili.update_bilibili_article_comment] Bilibili article comment: {comment_id}, content: {save_comment_item.get('content')}")
+    await BiliStoreFactory.create_store().store_article_comment(comment_item=save_comment_item)
+
+
 async def batch_update_bilibili_video_comments(video_id: str, comments: List[Dict]):
     if not comments:
         return
diff --git a/store/bilibili/_store_impl.py b/store/bilibili/_store_impl.py
index feaf280d3..56cbc4501 100644
--- a/store/bilibili/_store_impl.py
+++ b/store/bilibili/_store_impl.py
@@ -36,7 +36,13 @@
 import config
 from base.base_crawler import AbstractStore
 from database.db_session import get_session
-from database.models import BilibiliVideoComment, BilibiliVideo, BilibiliUpDynamic
+from database.models import (
+    BilibiliArticle,
+    BilibiliArticleComment,
+    BilibiliVideo,
+    BilibiliVideoComment,
+    BilibiliUpDynamic,
+)
 from tools.async_file_writer import AsyncFileWriter
 from tools import utils, words
 from var import crawler_type_var
@@ -64,6 +70,12 @@ async def store_content(self, content_item: Dict):
             item_type="videos"
         )
 
+    async def store_article(self, article_item: Dict):
+        await self.file_writer.write_to_csv(
+            item=article_item,
+            item_type="articles"
+        )
+
     async def store_comment(self, comment_item: Dict):
         """
         comment CSV storage implementation
@@ -78,6 +90,12 @@ async def store_comment(self, comment_item: Dict):
             item_type="comments"
         )
 
+    async def store_article_comment(self, comment_item: Dict):
+        await self.file_writer.write_to_csv(
+            item=comment_item,
+            item_type="article_comments"
+        )
+
     async def store_creator(self, creator: Dict):
         """
         creator CSV storage implementation
@@ -177,6 +195,54 @@ async def store_comment(self, comment_item: Dict):
                     setattr(comment_detail, key, value)
             await session.commit()
 
+    async def store_article(self, article_item: Dict):
+        article_id = int(article_item.get("article_id"))
+        article_item["article_id"] = article_id
+        article_item["create_time"] = int(article_item.get("create_time", 0) or 0)
+        article_item["liked_count"] = str(article_item.get("liked_count", ""))
+        article_item["favorite_count"] = str(article_item.get("favorite_count", ""))
+        article_item["share_count"] = str(article_item.get("share_count", ""))
+        article_item["comment_count"] = str(article_item.get("comment_count", ""))
+
+        async with get_session() as session:
+            result = await session.execute(select(BilibiliArticle).where(BilibiliArticle.article_id == article_id))
+            article_detail = result.scalar_one_or_none()
+
+            if not article_detail:
+                article_item["add_ts"] = utils.get_current_timestamp()
+                article_item["last_modify_ts"] = utils.get_current_timestamp()
+                new_article = BilibiliArticle(**article_item)
+                session.add(new_article)
+            else:
+                article_item["last_modify_ts"] = utils.get_current_timestamp()
+                for key, value in article_item.items():
+                    setattr(article_detail, key, value)
+            await session.commit()
+
+    async def store_article_comment(self, comment_item: Dict):
+        comment_id = int(comment_item.get("comment_id"))
+        comment_item["comment_id"] = comment_id
+        comment_item["article_id"] = int(comment_item.get("article_id", 0) or 0)
+        comment_item["create_time"] = int(comment_item.get("create_time", 0) or 0)
+        comment_item["like_count"] = str(comment_item.get("like_count", "0"))
+        comment_item["sub_comment_count"] = str(comment_item.get("sub_comment_count", "0"))
+        comment_item["parent_comment_id"] = str(comment_item.get("parent_comment_id", "0"))
+
+        async with get_session() as session:
+            result = await session.execute(select(BilibiliArticleComment).where(BilibiliArticleComment.comment_id == comment_id))
+            comment_detail = result.scalar_one_or_none()
+
+            if not comment_detail:
+                comment_item["add_ts"] = utils.get_current_timestamp()
+                comment_item["last_modify_ts"] = utils.get_current_timestamp()
+                new_comment = BilibiliArticleComment(**comment_item)
+                session.add(new_comment)
+            else:
+                comment_item["last_modify_ts"] = utils.get_current_timestamp()
+                for key, value in comment_item.items():
+                    setattr(comment_detail, key, value)
+            await session.commit()
+
     async def store_creator(self, creator: Dict):
         # 教学版：UP 主个人资料不再落库
         pass
@@ -231,6 +297,12 @@ async def store_content(self, content_item: Dict):
             item_type="contents"
         )
 
+    async def store_article(self, article_item: Dict):
+        await self.file_writer.write_single_item_to_json(
+            item=article_item,
+            item_type="articles"
+        )
+
     async def store_comment(self, comment_item: Dict):
         """
         comment JSON storage implementation
@@ -245,6 +317,12 @@ async def store_comment(self, comment_item: Dict):
             item_type="comments"
         )
 
+    async def store_article_comment(self, comment_item: Dict):
+        await self.file_writer.write_single_item_to_json(
+            item=comment_item,
+            item_type="article_comments"
+        )
+
     async def store_creator(self, creator: Dict):
         """
         creator JSON storage implementation
@@ -302,12 +380,24 @@ async def store_content(self, content_item: Dict):
             item_type="contents"
         )
 
+    async def store_article(self, article_item: Dict):
+        await self.file_writer.write_to_jsonl(
+            item=article_item,
+            item_type="articles"
+        )
+
     async def store_comment(self, comment_item: Dict):
         await self.file_writer.write_to_jsonl(
             item=comment_item,
             item_type="comments"
         )
 
+    async def store_article_comment(self, comment_item: Dict):
+        await self.file_writer.write_to_jsonl(
+            item=comment_item,
+            item_type="article_comments"
+        )
+
     async def store_creator(self, creator: Dict):
         await self.file_writer.write_to_jsonl(
             item=creator,
@@ -371,6 +461,30 @@ async def store_comment(self, comment_item: Dict):
         )
         utils.logger.info(f"[BiliMongoStoreImplement.store_comment] Saved comment {comment_id} to MongoDB")
 
+    async def store_article(self, article_item: Dict):
+        article_id = article_item.get("article_id")
+        if not article_id:
+            return
+
+        await self.mongo_store.save_or_update(
+            collection_suffix="articles",
+            query={"article_id": article_id},
+            data=article_item
+        )
+        utils.logger.info(f"[BiliMongoStoreImplement.store_article] Saved article {article_id} to MongoDB")
+
+    async def store_article_comment(self, comment_item: Dict):
+        comment_id = comment_item.get("comment_id")
+        if not comment_id:
+            return
+
+        await self.mongo_store.save_or_update(
+            collection_suffix="article_comments",
+            query={"comment_id": comment_id},
+            data=comment_item
+        )
+        utils.logger.info(f"[BiliMongoStoreImplement.store_article_comment] Saved article comment {comment_id} to MongoDB")
+
     async def store_creator(self, creator_item: Dict):
         # 教学版：UP 主个人资料不再落库
         pass
diff --git a/tests/store/bilibili/test_article_store.py b/tests/store/bilibili/test_article_store.py
new file mode 100644
index 000000000..be0a48a1e
--- /dev/null
+++ b/tests/store/bilibili/test_article_store.py
@@ -0,0 +1,70 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+
+from store import bilibili as bilibili_store
+
+
+@pytest.mark.asyncio
+async def test_update_bilibili_article_maps_article_fields(monkeypatch):
+    saved = {}
+
+    class FakeStore:
+        async def store_content(self, content_item):
+            saved.update(content_item)
+
+    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
+    monkeypatch.setattr(bilibili_store, "anonymize_user_id", lambda value: f"hash-{value}")
+    monkeypatch.setattr(bilibili_store, "mask_nickname", lambda value: f"masked-{value}")
+
+    await bilibili_store.update_bilibili_article({
+        "id": 123456,
+        "title": "title",
+        "summary": "summary",
+        "content": "content",
+        "publish_time": 1710000000,
+        "author": {"mid": 100, "name": "author"},
+        "stats": {"like": 1, "favorite": 2, "share": 3, "reply": 4},
+    })
+
+    assert saved["article_id"] == "123456"
+    assert saved["article_url"] == "https://www.bilibili.com/read/cv123456"
+    assert saved["title"] == "title"
+    assert saved["desc"] == "summary"
+    assert saved["creator_hash"] == "hash-100"
+    assert saved["nickname"] == "masked-author"
+    assert saved["liked_count"] == "1"
+    assert saved["favorite_count"] == "2"
+    assert saved["share_count"] == "3"
+    assert saved["comment_count"] == "4"
+
+
+@pytest.mark.asyncio
+async def test_update_bilibili_article_comment_maps_common_reply_fields(monkeypatch):
+    saved = {}
+
+    class FakeStore:
+        async def store_article_comment(self, comment_item):
+            saved.update(comment_item)
+
+    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
+    monkeypatch.setattr(bilibili_store, "anonymize_user_id", lambda value: f"hash-{value}")
+    monkeypatch.setattr(bilibili_store, "mask_nickname", lambda value: f"masked-{value}")
+
+    await bilibili_store.update_bilibili_article_comment("123456", {
+        "rpid": 9,
+        "parent": 0,
+        "ctime": 1710000000,
+        "content": {"message": "hello"},
+        "member": {"mid": 10, "uname": "user"},
+        "like": 5,
+        "rcount": 1,
+    })
+
+    assert saved["article_id"] == "123456"
+    assert saved["comment_id"] == "9"
+    assert saved["parent_comment_id"] == "0"
+    assert saved["content"] == "hello"
+    assert saved["creator_hash"] == "hash-10"
+    assert saved["nickname"] == "masked-user"
+    assert saved["like_count"] == 5

From c475539b91c154064075349c4baee7b4a0ffb604 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 23:14:19 +0800
Subject: [PATCH 08/12] feat: wire bilibili article detail crawling

---
 config/bilibili_config.py                     |   7 ++
 media_platform/bilibili/core.py               | 101 +++++++++++++++++-
 .../bilibili/test_core_article_flow.py        |  88 +++++++++++++++
 3 files changed, 194 insertions(+), 2 deletions(-)
 create mode 100644 tests/media_platform/bilibili/test_core_article_flow.py

diff --git a/config/bilibili_config.py b/config/bilibili_config.py
index 2c9f6ebd4..867f5cba5 100644
--- a/config/bilibili_config.py
+++ b/config/bilibili_config.py
@@ -32,6 +32,13 @@
     # ........................
 ]
 
+# Specify Bilibili article URL list for detail mode (supports full URL, cv ID, or numeric article ID)
+# Example:
+# - Full URL: "https://www.bilibili.com/read/cv123456"
+# - cv ID: "cv123456"
+# - Numeric article ID: "123456"
+BILI_SPECIFIED_ARTICLE_ID_LIST = []
+
 # Specify the URL list of Bilibili creators (supports full URL or UID)
 # Example:
 # - Full URL: "https://space.bilibili.com/434377496?spm_id_from=333.1007.0.0"
diff --git a/media_platform/bilibili/core.py b/media_platform/bilibili/core.py
index 14b509977..252eb564f 100644
--- a/media_platform/bilibili/core.py
+++ b/media_platform/bilibili/core.py
@@ -49,8 +49,8 @@
 
 from .client import BilibiliClient
 from .exception import DataFetchError
-from .field import SearchOrderType
-from .help import parse_video_info_from_url, parse_creator_info_from_url
+from .field import BilibiliCommentType, SearchOrderType
+from .help import parse_article_info_from_url, parse_creator_info_from_url, parse_video_info_from_url
 from .login import BilibiliLogin
 
 
@@ -117,6 +117,9 @@ async def start(self):
             elif config.CRAWLER_TYPE == "detail":
                 # Get the information and comments of the specified post
                 await self.get_specified_videos(config.BILI_SPECIFIED_ID_LIST)
+                article_id_list = getattr(config, "BILI_SPECIFIED_ARTICLE_ID_LIST", [])
+                if article_id_list:
+                    await self.get_specified_articles(article_id_list)
             elif config.CRAWLER_TYPE == "creator":
                 if config.CREATOR_MODE:
                     for creator_url in config.BILI_CREATOR_ID_LIST:
@@ -416,6 +419,100 @@ async def get_specified_videos(self, video_url_list: List[str]):
                 await self.get_bilibili_video(video_detail, semaphore)
         await self.batch_get_video_comments(video_aids_list)
 
+    async def get_specified_articles(self, article_url_list: List[str]):
+        """
+        get specified articles info from URLs or cv IDs
+        :param article_url_list: List of article URLs or cv IDs
+        :return:
+        """
+        utils.logger.info("[BilibiliCrawler.get_specified_articles] Parsing article URLs...")
+        article_id_list = []
+        for article_url in article_url_list:
+            try:
+                article_info = parse_article_info_from_url(article_url)
+                article_id_list.append(article_info.article_id)
+                utils.logger.info(f"[BilibiliCrawler.get_specified_articles] Parsed article ID: {article_info.article_id} from {article_url}")
+            except ValueError as e:
+                utils.logger.error(f"[BilibiliCrawler.get_specified_articles] Failed to parse article URL: {e}")
+                continue
+
+        semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
+        task_list = [self.get_article_info_task(article_id=article_id, semaphore=semaphore) for article_id in article_id_list]
+        article_details = await asyncio.gather(*task_list)
+        crawled_article_ids = []
+        for article_detail in article_details:
+            if article_detail is not None:
+                article_id = str(article_detail.get("id"))
+                if article_id:
+                    crawled_article_ids.append(article_id)
+                await bilibili_store.update_bilibili_article(article_detail)
+        await self.batch_get_article_comments(crawled_article_ids)
+
+    async def get_article_info_task(self, article_id: str, semaphore: asyncio.Semaphore) -> Optional[Dict]:
+        """
+        Get article detail task
+        :param article_id:
+        :param semaphore:
+        :return:
+        """
+        async with semaphore:
+            try:
+                result = await self.bili_client.get_article_info(article_id=article_id)
+                await asyncio.sleep(config.CRAWLER_MAX_SLEEP_SEC)
+                utils.logger.info(f"[BilibiliCrawler.get_article_info_task] Sleeping for {config.CRAWLER_MAX_SLEEP_SEC} seconds after fetching article details {article_id}")
+                return result
+            except DataFetchError as ex:
+                utils.logger.error(f"[BilibiliCrawler.get_article_info_task] Get article detail error: {ex}")
+                return None
+            except KeyError as ex:
+                utils.logger.error(f"[BilibiliCrawler.get_article_info_task] have not found article detail article_id:{article_id}, err: {ex}")
+                return None
+
+    async def batch_get_article_comments(self, article_id_list: List[str]):
+        """
+        batch get article comments
+        :param article_id_list:
+        :return:
+        """
+        if not config.ENABLE_GET_COMMENTS:
+            utils.logger.info("[BilibiliCrawler.batch_get_article_comments] Crawling comment mode is not enabled")
+            return
+
+        utils.logger.info(f"[BilibiliCrawler.batch_get_article_comments] article ids:{article_id_list}")
+        semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
+        task_list: List[Task] = []
+        for article_id in article_id_list:
+            task = asyncio.create_task(self.get_article_comments(article_id, semaphore), name=article_id)
+            task_list.append(task)
+        await asyncio.gather(*task_list)
+
+    async def get_article_comments(self, article_id: str, semaphore: asyncio.Semaphore):
+        """
+        get comment for article id
+        :param article_id:
+        :param semaphore:
+        :return:
+        """
+        async with semaphore:
+            try:
+                utils.logger.info(f"[BilibiliCrawler.get_article_comments] begin get article_id: {article_id} comments ...")
+                await asyncio.sleep(config.CRAWLER_MAX_SLEEP_SEC)
+                utils.logger.info(f"[BilibiliCrawler.get_article_comments] Sleeping for {config.CRAWLER_MAX_SLEEP_SEC} seconds after fetching comments for article {article_id}")
+                await self.bili_client.get_all_comments(
+                    oid=article_id,
+                    comment_type=BilibiliCommentType.ARTICLE,
+                    crawl_interval=config.CRAWLER_MAX_SLEEP_SEC,
+                    is_fetch_sub_comments=config.ENABLE_GET_SUB_COMMENTS,
+                    callback=bilibili_store.batch_update_bilibili_article_comments,
+                    max_count=config.CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES,
+                )
+
+            except DataFetchError as ex:
+                utils.logger.error(f"[BilibiliCrawler.get_article_comments] get article_id: {article_id} comment error: {ex}")
+            except Exception as e:
+                utils.logger.error(f"[BilibiliCrawler.get_article_comments] may be been blocked, err:{e}")
+                raise
+
     async def get_video_info_task(self, aid: int, bvid: str, semaphore: asyncio.Semaphore) -> Optional[Dict]:
         """
         Get video detail task
diff --git a/tests/media_platform/bilibili/test_core_article_flow.py b/tests/media_platform/bilibili/test_core_article_flow.py
new file mode 100644
index 000000000..6b18e1b05
--- /dev/null
+++ b/tests/media_platform/bilibili/test_core_article_flow.py
@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+
+import asyncio
+
+import pytest
+
+import config
+from media_platform.bilibili.core import BilibiliCrawler
+from media_platform.bilibili.field import BilibiliCommentType
+from store import bilibili as bilibili_store
+
+
+@pytest.fixture
+def crawler(monkeypatch):
+    crawler_obj = BilibiliCrawler()
+    monkeypatch.setattr(config, "MAX_CONCURRENCY_NUM", 2)
+    monkeypatch.setattr(config, "CRAWLER_MAX_SLEEP_SEC", 0)
+    monkeypatch.setattr(config, "ENABLE_GET_COMMENTS", True)
+    monkeypatch.setattr(config, "ENABLE_GET_SUB_COMMENTS", False)
+    monkeypatch.setattr(config, "CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES", 2)
+
+    async def fake_sleep(_):
+        return None
+
+    monkeypatch.setattr("media_platform.bilibili.core.asyncio.sleep", fake_sleep)
+    return crawler_obj
+
+
+@pytest.mark.asyncio
+async def test_get_specified_articles_fetches_detail_and_comments(monkeypatch, crawler):
+    stored_articles = []
+    comment_calls = []
+
+    class FakeClient:
+        async def get_article_info(self, article_id):
+            return {"id": int(article_id), "title": "article title"}
+
+        async def get_all_comments(
+            self,
+            oid,
+            comment_type,
+            crawl_interval,
+            is_fetch_sub_comments,
+            callback,
+            max_count,
+        ):
+            comment_calls.append({
+                "oid": oid,
+                "comment_type": comment_type,
+                "crawl_interval": crawl_interval,
+                "is_fetch_sub_comments": is_fetch_sub_comments,
+                "callback": callback,
+                "max_count": max_count,
+            })
+
+    async def fake_update_article(article_detail):
+        stored_articles.append(article_detail)
+
+    monkeypatch.setattr(bilibili_store, "update_bilibili_article", fake_update_article)
+    monkeypatch.setattr(crawler, "bili_client", FakeClient(), raising=False)
+
+    await crawler.get_specified_articles(["cv123456"])
+
+    assert stored_articles == [{"id": 123456, "title": "article title"}]
+    assert comment_calls[0]["oid"] == "123456"
+    assert comment_calls[0]["comment_type"] is BilibiliCommentType.ARTICLE
+    assert comment_calls[0]["callback"] is bilibili_store.batch_update_bilibili_article_comments
+    assert comment_calls[0]["max_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_batch_get_article_comments_respects_comment_flag(monkeypatch, crawler):
+    called = False
+
+    async def fake_get_article_comments(article_id, semaphore):
+        nonlocal called
+        called = True
+
+    monkeypatch.setattr(config, "ENABLE_GET_COMMENTS", False)
+    monkeypatch.setattr(crawler, "get_article_comments", fake_get_article_comments, raising=False)
+
+    await crawler.batch_get_article_comments(["123456"])
+
+    assert called is False
+
+
+def test_detail_start_config_has_separate_article_list():
+    assert hasattr(config, "BILI_SPECIFIED_ARTICLE_ID_LIST")

From 4f9aeac58019c8c4a5eba954ffdb2e629ac3a01e Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Wed, 1 Jul 2026 23:16:10 +0800
Subject: [PATCH 09/12] feat: add bilibili article comment wrappers

---
 media_platform/bilibili/client.py             | 41 +++++++++++++
 .../bilibili/test_client_comments.py          | 60 +++++++++++++++++++
 2 files changed, 101 insertions(+)

diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py
index 4d9432f67..9fffd3778 100644
--- a/media_platform/bilibili/client.py
+++ b/media_platform/bilibili/client.py
@@ -286,6 +286,20 @@ async def get_video_comments(
         """
         return await self.get_comments(video_id, BilibiliCommentType.VIDEO, order_mode, next)
 
+    async def get_article_comments(
+        self,
+        article_id: str,
+        order_mode: CommentOrderType = CommentOrderType.DEFAULT,
+        next: int = 0,
+    ) -> Dict:
+        """get article comments
+        :param article_id: Article ID
+        :param order_mode: Sort order
+        :param next: Comment page selection
+        :return:
+        """
+        return await self.get_comments(article_id, BilibiliCommentType.ARTICLE, order_mode, next)
+
     async def get_all_comments(
         self,
         oid: str,
@@ -394,6 +408,33 @@ async def get_video_all_comments(
             max_count,
         )
 
+    async def get_article_all_comments(
+        self,
+        article_id: str,
+        crawl_interval: float = 1.0,
+        is_fetch_sub_comments=False,
+        callback: Optional[Callable] = None,
+        max_count: int = 10,
+    ):
+        """
+        get article all comments include sub comments
+        :param article_id:
+        :param crawl_interval:
+        :param is_fetch_sub_comments:
+        :param callback:
+        max_count: Maximum number of comments to crawl per article
+
+        :return:
+        """
+        return await self.get_all_comments(
+            article_id,
+            BilibiliCommentType.ARTICLE,
+            crawl_interval,
+            is_fetch_sub_comments,
+            callback,
+            max_count,
+        )
+
     async def get_all_level_two_comments(
         self,
         oid: str,
diff --git a/tests/media_platform/bilibili/test_client_comments.py b/tests/media_platform/bilibili/test_client_comments.py
index 1ee085778..c0749f71e 100644
--- a/tests/media_platform/bilibili/test_client_comments.py
+++ b/tests/media_platform/bilibili/test_client_comments.py
@@ -50,6 +50,66 @@ async def fake_get(uri, params=None, enable_params_sign=True):
     assert captured["params"]["type"] == 1
 
 
+@pytest.mark.asyncio
+async def test_get_article_comments_uses_type_12(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get_comments(oid, comment_type, order_mode, next):
+        captured["oid"] = oid
+        captured["type"] = comment_type.value
+        return {}
+
+    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
+
+    await bili_client.get_article_comments("123456")
+
+    assert captured == {"oid": "123456", "type": 12}
+
+
+@pytest.mark.asyncio
+async def test_get_article_all_comments_delegates_to_generic(monkeypatch, bili_client):
+    captured = {}
+
+    async def fake_get_all_comments(
+        oid,
+        comment_type,
+        crawl_interval,
+        is_fetch_sub_comments,
+        callback,
+        max_count,
+    ):
+        captured["oid"] = oid
+        captured["type"] = comment_type.value
+        captured["crawl_interval"] = crawl_interval
+        captured["is_fetch_sub_comments"] = is_fetch_sub_comments
+        captured["callback"] = callback
+        captured["max_count"] = max_count
+        return [{"rpid": 1}]
+
+    async def fake_callback(oid, comments):
+        return None
+
+    monkeypatch.setattr(bili_client, "get_all_comments", fake_get_all_comments)
+
+    result = await bili_client.get_article_all_comments(
+        article_id="123456",
+        crawl_interval=0,
+        is_fetch_sub_comments=True,
+        callback=fake_callback,
+        max_count=3,
+    )
+
+    assert captured == {
+        "oid": "123456",
+        "type": 12,
+        "crawl_interval": 0,
+        "is_fetch_sub_comments": True,
+        "callback": fake_callback,
+        "max_count": 3,
+    }
+    assert result == [{"rpid": 1}]
+
+
 @pytest.mark.asyncio
 async def test_get_all_comments_limits_first_level_before_fetching_sub_comments(monkeypatch, bili_client):
     fetched_roots = []

From fccfc94aa00ef3d5f26f6e97068de45a17303253 Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Thu, 2 Jul 2026 11:33:24 +0800
Subject: [PATCH 10/12] chore: remove article support planning doc

---
 docs/plans/2026-07-01-bili-article-support.md | 931 ------------------
 1 file changed, 931 deletions(-)
 delete mode 100644 docs/plans/2026-07-01-bili-article-support.md

diff --git a/docs/plans/2026-07-01-bili-article-support.md b/docs/plans/2026-07-01-bili-article-support.md
deleted file mode 100644
index cd1261cfc..000000000
--- a/docs/plans/2026-07-01-bili-article-support.md
+++ /dev/null
@@ -1,931 +0,0 @@
-# Bilibili Article Support Implementation Plan
-
-> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
-
-**Goal:** Add complete Bilibili article support, including article URL parsing, article detail crawling, first-level comments, optional second-level comments, and dedicated storage.
-
-**Architecture:** Keep existing video public APIs compatible while extracting Bilibili comment crawling into a generic `type + oid` implementation. Article crawling is added as a first-class content flow with dedicated parsing, client methods, core orchestration, and storage models. The video comment pagination semantics have already been corrected: first-level comments are trimmed by `max_count` before saving and before second-level comment fetching, and retained first-level comments are always counted even when second-level comments are enabled.
-
-**Tech Stack:** Python async/await, Playwright-backed Bilibili API client, SQLAlchemy models, existing MediaCrawler store abstractions, pytest.
-
----
-
-## Context
-
-Current Bilibili code is video-centered:
-
-- `media_platform/bilibili/client.py` calls Bilibili comment APIs with `type=1`, which means video comments.
-- Bilibili article comments require the same comment API family but with `type=12` and `oid=<article cvid>`.
-- `media_platform/bilibili/help.py` only parses video BV URLs and creator URLs.
-- `store/bilibili` and `database/models.py` only have video and video comment storage models.
-
-Completed prerequisite: video comment crawling now trims first-level comments before fetching second-level comments and extends the local result in both modes. This behavior is covered by `tests/test_bilibili_client_comments.py`.
-
-The implementation should preserve existing video methods such as `get_video_comments()` as compatibility wrappers. The new generic implementation should become the only place that knows how to page comments and second-level comments, while preserving the corrected video semantics above.
-
-## External API Notes
-
-Use Bilibili's common comment model:
-
-- Video comments: `type=1`, `oid=<aid>`
-- Article comments: `type=12`, `oid=<cvid>`
-- First-level comments endpoint: `/x/v2/reply/wbi/main`
-- Second-level comments endpoint: `/x/v2/reply/reply`
-
-Before implementing article detail mapping, verify a real article detail response with one known `cv` article ID. Do not guess the final response shape for title/content/stat fields.
-
----
-
-### Task 1: Add Bilibili Article URL Model And Parser
-
-**Files:**
-- Modify: `model/m_bilibili.py`
-- Modify: `media_platform/bilibili/help.py`
-- Test: `tests/media_platform/bilibili/test_help.py`
-
-**Step 1: Write failing parser tests**
-
-Create or extend `tests/media_platform/bilibili/test_help.py`:
-
-```python
-import pytest
-
-from media_platform.bilibili.help import parse_article_info_from_url
-
-
-@pytest.mark.parametrize(
-    ("raw", "expected"),
-    [
-        ("cv123456", "123456"),
-        ("123456", "123456"),
-        ("https://www.bilibili.com/read/cv123456", "123456"),
-        ("https://www.bilibili.com/read/cv123456?spm_id_from=333.999.0.0", "123456"),
-    ],
-)
-def test_parse_article_info_from_url(raw, expected):
-    article_info = parse_article_info_from_url(raw)
-    assert article_info.article_id == expected
-    assert article_info.article_type == "article"
-
-
-def test_parse_article_info_from_url_invalid():
-    with pytest.raises(ValueError):
-        parse_article_info_from_url("https://www.bilibili.com/video/BV1d54y1g7db")
-```
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_help.py -v
-```
-
-Expected: FAIL because `parse_article_info_from_url` and `ArticleUrlInfo` do not exist.
-
-**Step 3: Add model and parser**
-
-In `model/m_bilibili.py`, add:
-
-```python
-class ArticleUrlInfo(BaseModel):
-    """Bilibili article URL information"""
-    article_id: str = Field(title="article id (cv id without cv prefix)")
-    article_type: str = Field(default="article", title="article type")
-```
-
-In `media_platform/bilibili/help.py`, import `ArticleUrlInfo` and add `parse_article_info_from_url(url: str) -> ArticleUrlInfo`.
-
-Parsing rules:
-
-- If input is digits, return it directly.
-- If input starts with `cv` followed by digits, strip `cv`.
-- If URL contains `/read/cv<digits>`, extract digits.
-- Otherwise raise `ValueError`.
-
-**Step 4: Run test to verify it passes**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_help.py -v
-```
-
-Expected: PASS.
-
-**Step 5: Commit**
-
-```bash
-git add model/m_bilibili.py media_platform/bilibili/help.py tests/media_platform/bilibili/test_help.py
-git commit -m "feat: parse bilibili article urls"
-```
-
----
-
-### Task 2: Add Generic Bilibili Comment Type Support
-
-**Files:**
-- Modify: `media_platform/bilibili/field.py`
-- Modify: `media_platform/bilibili/client.py`
-- Test: `tests/media_platform/bilibili/test_client_comments.py`
-
-**Step 1: Write failing client tests**
-
-Create `tests/media_platform/bilibili/test_client_comments.py` with a minimal client fixture that bypasses network by monkeypatching `get`.
-
-Test these cases:
-
-```python
-import pytest
-
-from media_platform.bilibili.field import BilibiliCommentType, CommentOrderType
-
-
-@pytest.mark.asyncio
-async def test_get_comments_uses_comment_type(monkeypatch, bili_client):
-    captured = {}
-
-    async def fake_get(uri, params=None, enable_params_sign=True):
-        captured["uri"] = uri
-        captured["params"] = params
-        return {}
-
-    monkeypatch.setattr(bili_client, "get", fake_get)
-
-    await bili_client.get_comments(
-        oid="123456",
-        comment_type=BilibiliCommentType.ARTICLE,
-        order_mode=CommentOrderType.DEFAULT,
-        next=0,
-    )
-
-    assert captured["uri"] == "/x/v2/reply/wbi/main"
-    assert captured["params"]["oid"] == "123456"
-    assert captured["params"]["type"] == 12
-```
-
-Also add a compatibility test:
-
-```python
-@pytest.mark.asyncio
-async def test_get_video_comments_keeps_type_1(monkeypatch, bili_client):
-    captured = {}
-
-    async def fake_get(uri, params=None, enable_params_sign=True):
-        captured["params"] = params
-        return {}
-
-    monkeypatch.setattr(bili_client, "get", fake_get)
-
-    await bili_client.get_video_comments("998877")
-
-    assert captured["params"]["oid"] == "998877"
-    assert captured["params"]["type"] == 1
-```
-
-If there is no reusable `bili_client` fixture, create a local fixture with dummy headers, page, and cookies. Monkeypatched tests should not touch network or Playwright.
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: FAIL because `BilibiliCommentType` and `get_comments()` do not exist.
-
-**Step 3: Implement comment type enum and generic first-level method**
-
-In `media_platform/bilibili/field.py`, add:
-
-```python
-class BilibiliCommentType(Enum):
-    VIDEO = 1
-    ARTICLE = 12
-```
-
-In `media_platform/bilibili/client.py`, add:
-
-```python
-async def get_comments(
-    self,
-    oid: str,
-    comment_type: BilibiliCommentType,
-    order_mode: CommentOrderType = CommentOrderType.DEFAULT,
-    next: int = 0,
-) -> Dict:
-    uri = "/x/v2/reply/wbi/main"
-    post_data = {
-        "oid": oid,
-        "mode": order_mode.value,
-        "type": comment_type.value,
-        "ps": 20,
-        "next": next,
-    }
-    return await self.get(uri, post_data)
-```
-
-Change `get_video_comments()` to call `get_comments(..., BilibiliCommentType.VIDEO, ...)`.
-
-**Step 4: Run test to verify it passes**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: PASS.
-
-**Step 5: Commit**
-
-```bash
-git add media_platform/bilibili/field.py media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
-git commit -m "feat: add generic bilibili comment type support"
-```
-
----
-
-### Task 3: Generalize Corrected Comment Pagination And Second-Level Comments
-
-**Files:**
-- Modify: `media_platform/bilibili/client.py`
-- Test: `tests/media_platform/bilibili/test_client_comments.py`
-- Existing regression: `tests/test_bilibili_client_comments.py`
-
-**Step 1: Write failing pagination tests**
-
-Add tests that monkeypatch first-level and second-level methods:
-
-```python
-@pytest.mark.asyncio
-async def test_get_all_comments_limits_first_level_before_fetching_sub_comments(monkeypatch, bili_client):
-    fetched_roots = []
-    saved_batches = []
-
-    async def fake_get_comments(oid, comment_type, order_mode, next):
-        return {
-            "cursor": {"is_end": True, "next": 0},
-            "replies": [
-                {"rpid": 1, "rcount": 1, "content": {"message": "a"}, "member": {"mid": 1, "uname": "u1"}},
-                {"rpid": 2, "rcount": 1, "content": {"message": "b"}, "member": {"mid": 2, "uname": "u2"}},
-            ],
-        }
-
-    async def fake_get_all_level_two_comments(oid, comment_type, level_one_comment_id, order_mode, ps, crawl_interval, callback):
-        fetched_roots.append(level_one_comment_id)
-
-    async def fake_callback(oid, comments):
-        saved_batches.append(comments)
-
-    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
-    monkeypatch.setattr(bili_client, "get_all_level_two_comments", fake_get_all_level_two_comments)
-
-    await bili_client.get_all_comments(
-        oid="123",
-        comment_type=BilibiliCommentType.ARTICLE,
-        crawl_interval=0,
-        is_fetch_sub_comments=True,
-        callback=fake_callback,
-        max_count=1,
-    )
-
-    assert len(saved_batches[0]) == 1
-    assert fetched_roots == [1]
-```
-
-Also add or preserve a video-wrapper regression equivalent to `tests/test_bilibili_client_comments.py`:
-
-- `get_video_all_comments(..., is_fetch_sub_comments=True, max_count=1)` saves and returns only one retained first-level comment.
-- Second-level comments are fetched only for that retained first-level comment.
-- This regression should continue to pass after video methods delegate to generic methods.
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: FAIL because `get_all_comments()` and generic second-level methods do not exist yet. The existing video-specific regression should already pass before this task and must continue to pass after the refactor.
-
-**Step 3: Implement generic pagination**
-
-In `media_platform/bilibili/client.py`, add:
-
-- `get_all_comments(oid, comment_type, crawl_interval, is_fetch_sub_comments, callback, max_count)`
-- `get_all_level_two_comments(oid, comment_type, level_one_comment_id, order_mode, ps, crawl_interval, callback)`
-- `get_level_two_comments(oid, comment_type, level_one_comment_id, pn, ps, order_mode)`
-
-Required behavior:
-
-1. Fetch a page of first-level comments.
-2. Validate `cursor.is_end` and `cursor.next`.
-3. Trim `comment_list` before saving and before fetching sub-comments.
-4. Save first-level comments with `callback(oid, comment_list)`.
-5. If `is_fetch_sub_comments` is true, fetch second-level comments only for retained first-level comments where `rcount > 0`.
-6. Always extend the local result with retained first-level comments so `max_count` works in both modes.
-
-Important: `max_count` limits retained first-level comments. It does not impose a global cap on second-level comments; second-level pagination remains controlled by the second-level API loop for each retained first-level comment.
-
-Update old video methods as wrappers:
-
-- `get_video_all_comments()` calls `get_all_comments(..., BilibiliCommentType.VIDEO, ...)`.
-- `get_video_all_level_two_comments()` calls generic second-level method with `VIDEO`.
-- `get_video_level_two_comments()` calls generic method with `VIDEO`.
-
-**Step 4: Run tests**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py tests/test_bilibili_client_comments.py -v
-```
-
-Expected: PASS.
-
-**Step 5: Commit**
-
-```bash
-git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
-git commit -m "refactor: generalize bilibili comment pagination"
-```
-
----
-
-### Task 4: Verify And Implement Article Detail Client
-
-**Files:**
-- Modify: `media_platform/bilibili/client.py`
-- Test: `tests/media_platform/bilibili/test_client_article.py`
-
-**Step 1: Manually verify article detail endpoint**
-
-Use one known public article ID and run a small script or interactive call through existing client authentication if needed. Candidate API endpoints to verify include Bilibili article/read APIs. Record the response fields used for:
-
-- article ID
-- title
-- summary or content
-- creator ID and name
-- publish timestamp
-- stats: likes, favorites, shares, comments
-
-Do not commit captured personal cookies or full response dumps.
-
-**Step 2: Write failing mapping test**
-
-Create `tests/media_platform/bilibili/test_client_article.py`:
-
-```python
-import pytest
-
-
-@pytest.mark.asyncio
-async def test_get_article_info_calls_expected_endpoint(monkeypatch, bili_client):
-    captured = {}
-
-    async def fake_get(uri, params=None, enable_params_sign=True):
-        captured["uri"] = uri
-        captured["params"] = params
-        return {"id": 123456, "title": "article title"}
-
-    monkeypatch.setattr(bili_client, "get", fake_get)
-
-    result = await bili_client.get_article_info("123456")
-
-    assert captured["params"]
-    assert result["title"] == "article title"
-```
-
-Adjust exact assertions after endpoint verification.
-
-**Step 3: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_article.py -v
-```
-
-Expected: FAIL because `get_article_info()` does not exist.
-
-**Step 4: Implement `get_article_info()`**
-
-Add a method to `BilibiliClient` that accepts article ID without the `cv` prefix and returns raw article detail data.
-
-Keep response normalization out of the client unless existing project style clearly normalizes in client. Current Bilibili video code stores raw-ish API detail and maps it in store layer, so follow that pattern.
-
-**Step 5: Run tests**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_article.py tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: PASS.
-
-**Step 6: Commit**
-
-```bash
-git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_article.py
-git commit -m "feat: add bilibili article detail client"
-```
-
----
-
-### Task 5: Add Article Database Models
-
-**Files:**
-- Modify: `database/models.py`
-- Test: `tests/database/test_bilibili_article_models.py`
-
-**Step 1: Write model metadata test**
-
-Create `tests/database/test_bilibili_article_models.py`:
-
-```python
-from database.models import BilibiliArticle, BilibiliArticleComment
-
-
-def test_bilibili_article_table_name():
-    assert BilibiliArticle.__tablename__ == "bilibili_article"
-
-
-def test_bilibili_article_comment_table_name():
-    assert BilibiliArticleComment.__tablename__ == "bilibili_article_comment"
-```
-
-Add assertions for key columns if existing database model tests follow that style.
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/database/test_bilibili_article_models.py -v
-```
-
-Expected: FAIL because models do not exist.
-
-**Step 3: Add models**
-
-In `database/models.py`, add:
-
-`BilibiliArticle` fields:
-
-- `id`
-- `article_id`
-- `article_url`
-- `title`
-- `desc`
-- `content`
-- `creator_hash`
-- `nickname`
-- `liked_count`
-- `favorite_count`
-- `share_count`
-- `comment_count`
-- `create_time`
-- `source_keyword`
-- `add_ts`
-- `last_modify_ts`
-
-`BilibiliArticleComment` fields:
-
-- `id`
-- `creator_hash`
-- `nickname`
-- `add_ts`
-- `last_modify_ts`
-- `comment_id`
-- `article_id`
-- `content`
-- `create_time`
-- `sub_comment_count`
-- `parent_comment_id`
-- `like_count`
-
-Follow existing type choices from `BilibiliVideo` and `BilibiliVideoComment`.
-
-**Step 4: Run test**
-
-Run:
-
-```bash
-pytest tests/database/test_bilibili_article_models.py -v
-```
-
-Expected: PASS.
-
-**Step 5: Commit**
-
-```bash
-git add database/models.py tests/database/test_bilibili_article_models.py
-git commit -m "feat: add bilibili article database models"
-```
-
----
-
-### Task 6: Add Article Store Methods
-
-**Files:**
-- Modify: `store/bilibili/__init__.py`
-- Modify: `store/bilibili/_store_impl.py`
-- Test: `tests/store/bilibili/test_article_store.py`
-
-**Step 1: Write failing store mapping tests**
-
-Create `tests/store/bilibili/test_article_store.py`.
-
-Use monkeypatching to avoid filesystem and database writes:
-
-```python
-import pytest
-
-from store import bilibili as bilibili_store
-
-
-@pytest.mark.asyncio
-async def test_update_bilibili_article_maps_article_fields(monkeypatch):
-    saved = {}
-
-    class FakeStore:
-        async def store_content(self, content_item):
-            saved.update(content_item)
-
-    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
-
-    await bilibili_store.update_bilibili_article({
-        "id": 123456,
-        "title": "title",
-        "summary": "summary",
-        "content": "content",
-        "publish_time": 1710000000,
-        "author": {"mid": 100, "name": "author"},
-        "stats": {"like": 1, "favorite": 2, "share": 3, "reply": 4},
-    })
-
-    assert saved["article_id"] == "123456"
-    assert saved["article_url"] == "https://www.bilibili.com/read/cv123456"
-    assert saved["title"] == "title"
-```
-
-Adjust raw field names after Task 4 endpoint verification.
-
-Add a comment mapping test:
-
-```python
-@pytest.mark.asyncio
-async def test_update_bilibili_article_comment_maps_common_reply_fields(monkeypatch):
-    saved = {}
-
-    class FakeStore:
-        async def store_article_comment(self, comment_item):
-            saved.update(comment_item)
-
-    monkeypatch.setattr(bilibili_store.BiliStoreFactory, "create_store", lambda: FakeStore())
-
-    await bilibili_store.update_bilibili_article_comment("123456", {
-        "rpid": 9,
-        "parent": 0,
-        "ctime": 1710000000,
-        "content": {"message": "hello"},
-        "member": {"mid": 10, "uname": "user"},
-        "like": 5,
-        "rcount": 1,
-    })
-
-    assert saved["article_id"] == "123456"
-    assert saved["comment_id"] == "9"
-    assert saved["content"] == "hello"
-```
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/store/bilibili/test_article_store.py -v
-```
-
-Expected: FAIL because article store methods do not exist.
-
-**Step 3: Add public store functions**
-
-In `store/bilibili/__init__.py`, add:
-
-- `update_bilibili_article(article_item: Dict)`
-- `batch_update_bilibili_article_comments(article_id: str, comments: List[Dict])`
-- `update_bilibili_article_comment(article_id: str, comment_item: Dict)`
-
-Use `anonymize_user_id()` and `mask_nickname()` consistently with video storage.
-
-**Step 4: Add store implementation methods**
-
-In `store/bilibili/_store_impl.py`:
-
-- Import `BilibiliArticle` and `BilibiliArticleComment`.
-- Add `store_article_comment()` to store classes that need a separate comment collection/table.
-- For file-based stores, write article content with `item_type="articles"` and article comments with `item_type="article_comments"`.
-- For DB store, upsert `BilibiliArticle` by `article_id` and `BilibiliArticleComment` by `comment_id`.
-- For Mongo store, use collection suffixes `articles` and `article_comments`.
-
-If changing the abstract store interface would affect every platform, avoid adding abstract methods. Use Bilibili store implementation methods directly where needed, or fall back to existing `store_comment()` only if the implementation can route by `item_type`.
-
-**Step 5: Run tests**
-
-Run:
-
-```bash
-pytest tests/store/bilibili/test_article_store.py tests/database/test_bilibili_article_models.py -v
-```
-
-Expected: PASS.
-
-**Step 6: Commit**
-
-```bash
-git add store/bilibili/__init__.py store/bilibili/_store_impl.py tests/store/bilibili/test_article_store.py
-git commit -m "feat: add bilibili article storage"
-```
-
----
-
-### Task 7: Wire Article Detail Flow Into Bilibili Core
-
-**Files:**
-- Modify: `media_platform/bilibili/core.py`
-- Modify: `config/bilibili_config.py`
-- Test: `tests/media_platform/bilibili/test_core_article_flow.py`
-
-**Step 1: Write failing orchestration tests**
-
-Create `tests/media_platform/bilibili/test_core_article_flow.py`.
-
-Test that article inputs are parsed and passed to article detail/comment methods. Use a `BilibiliCrawler` instance with monkeypatched `bili_client` and store functions.
-
-Important cases:
-
-- `get_specified_articles(["cv123456"])` fetches article detail for `123456`.
-- If comments are enabled, article comments are fetched.
-- Existing `get_specified_videos()` remains unchanged.
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_core_article_flow.py -v
-```
-
-Expected: FAIL because article core methods do not exist.
-
-**Step 3: Add config documentation**
-
-In `config/bilibili_config.py`, update `BILI_SPECIFIED_ID_LIST` comments to say it supports:
-
-- Bilibili video URL
-- BV number
-- Bilibili article URL
-- `cv` article ID
-- numeric article ID, when treated as article ID in article-specific config
-
-Prefer adding `BILI_SPECIFIED_ARTICLE_ID_LIST = []` if mixed numeric IDs would be ambiguous. A pure numeric input could be a creator ID or article ID depending on context, so article-specific config is safer.
-
-Recommended approach:
-
-- Keep `BILI_SPECIFIED_ID_LIST` for videos.
-- Add `BILI_SPECIFIED_ARTICLE_ID_LIST` for articles.
-- In `detail` mode, crawl both lists.
-
-**Step 4: Add core methods**
-
-In `media_platform/bilibili/core.py`, add:
-
-- `get_specified_articles(article_url_list: List[str])`
-- `batch_get_article_comments(article_id_list: List[str])`
-- `get_article_comments(article_id: str, semaphore: asyncio.Semaphore)`
-
-Behavior:
-
-1. Parse article IDs.
-2. Fetch article detail with `self.bili_client.get_article_info(article_id)`.
-3. Store with `bilibili_store.update_bilibili_article(article_detail)`.
-4. If `ENABLE_GET_COMMENTS`, fetch article comments with:
-   - `self.bili_client.get_all_comments(...)`
-   - `comment_type=BilibiliCommentType.ARTICLE`
-   - `callback=bilibili_store.batch_update_bilibili_article_comments`
-   - `max_count=config.CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES`
-
-In `start()`, under `CRAWLER_TYPE == "detail"`, call both:
-
-- `get_specified_videos(config.BILI_SPECIFIED_ID_LIST)`
-- `get_specified_articles(config.BILI_SPECIFIED_ARTICLE_ID_LIST)`
-
-Only call article flow if the list exists and is non-empty.
-
-**Step 5: Run tests**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_core_article_flow.py tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: PASS.
-
-**Step 6: Commit**
-
-```bash
-git add media_platform/bilibili/core.py config/bilibili_config.py tests/media_platform/bilibili/test_core_article_flow.py
-git commit -m "feat: wire bilibili article detail crawling"
-```
-
----
-
-### Task 8: Add Article Comment Wrapper Methods
-
-**Files:**
-- Modify: `media_platform/bilibili/client.py`
-- Test: `tests/media_platform/bilibili/test_client_comments.py`
-
-**Step 1: Write failing wrapper tests**
-
-Add tests:
-
-```python
-@pytest.mark.asyncio
-async def test_get_article_comments_uses_type_12(monkeypatch, bili_client):
-    captured = {}
-
-    async def fake_get_comments(oid, comment_type, order_mode, next):
-        captured["oid"] = oid
-        captured["type"] = comment_type.value
-        return {}
-
-    monkeypatch.setattr(bili_client, "get_comments", fake_get_comments)
-
-    await bili_client.get_article_comments("123456")
-
-    assert captured == {"oid": "123456", "type": 12}
-```
-
-**Step 2: Run test to verify it fails**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: FAIL because article wrappers do not exist.
-
-**Step 3: Add wrappers**
-
-Add:
-
-- `get_article_comments(article_id, order_mode=CommentOrderType.DEFAULT, next=0)`
-- `get_article_all_comments(article_id, crawl_interval=1.0, is_fetch_sub_comments=False, callback=None, max_count=10)`
-
-Both should delegate to generic comment methods with `BilibiliCommentType.ARTICLE`.
-
-**Step 4: Run tests**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili/test_client_comments.py -v
-```
-
-Expected: PASS.
-
-**Step 5: Commit**
-
-```bash
-git add media_platform/bilibili/client.py tests/media_platform/bilibili/test_client_comments.py
-git commit -m "feat: add bilibili article comment wrappers"
-```
-
----
-
-### Task 9: Documentation And User Configuration
-
-**Files:**
-- Modify: `README.md`
-- Modify: `docs/index.md`
-- Modify: `docs/项目架构文档.md`
-
-**Step 1: Update user-facing docs**
-
-Document Bilibili article support:
-
-- `BILI_SPECIFIED_ARTICLE_ID_LIST`
-- Supported formats:
-  - `https://www.bilibili.com/read/cv123456`
-  - `cv123456`
-  - `123456`
-- Comment flags:
-  - `ENABLE_GET_COMMENTS`
-  - `ENABLE_GET_SUB_COMMENTS`
-  - `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES`
-- Data outputs:
-  - `articles`
-  - `article_comments`
-
-**Step 2: Run docs grep**
-
-Run:
-
-```bash
-rg -n "BILI_SPECIFIED_ARTICLE_ID_LIST|article_comments|专栏" README.md docs config/bilibili_config.py
-```
-
-Expected: New config and docs are discoverable.
-
-**Step 3: Commit**
-
-```bash
-git add README.md docs/index.md docs/项目架构文档.md
-git commit -m "docs: document bilibili article crawling"
-```
-
----
-
-### Task 10: End-To-End Verification
-
-**Files:**
-- No code changes expected unless verification finds defects.
-
-**Step 1: Run focused tests**
-
-Run:
-
-```bash
-pytest tests/media_platform/bilibili tests/store/bilibili tests/database/test_bilibili_article_models.py -v
-```
-
-Expected: PASS.
-
-**Step 2: Run broader test suite**
-
-Run:
-
-```bash
-pytest tests -v
-```
-
-Expected: PASS, or document unrelated existing failures with exact failing tests.
-
-**Step 3: Run a dry manual detail crawl**
-
-Set a local config or command-line args for:
-
-- `PLATFORM=bili`
-- `CRAWLER_TYPE=detail`
-- one valid `BILI_SPECIFIED_ARTICLE_ID_LIST`
-- `ENABLE_GET_COMMENTS=True`
-- `ENABLE_GET_SUB_COMMENTS=False`
-- `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES=2`
-- `SAVE_DATA_OPTION=jsonl`
-
-Run the crawler and verify:
-
-- Article detail output exists.
-- Article comment output exists.
-- No video output is created unless video IDs are configured.
-
-**Step 4: Run optional second-level comment verification**
-
-Repeat with:
-
-- `ENABLE_GET_SUB_COMMENTS=True`
-- `CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES=1`
-
-Verify:
-
-- For both video and article flows, only retained first-level comments trigger second-level comment fetching.
-- Saved second-level comments have non-zero `parent_comment_id`.
-
-**Step 5: Final status**
-
-Run:
-
-```bash
-git status --short --branch
-```
-
-Expected: clean working tree after final commit.
-
----
-
-## Non-Goals For This Plan
-
-- Bilibili article search by keyword.
-- Crawling all articles from a creator homepage.
-- Downloading images embedded in article content.
-- Changing comment crawling behavior for non-Bilibili platforms.
-
-These can be added later once the specified-article flow is stable.

From 4761378b267ac5ea33a1a5f322d8b818c1d3503e Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Thu, 2 Jul 2026 12:20:52 +0800
Subject: [PATCH 11/12] feat: route bilibili article ids from cli

---
 cmd_arg/arg.py                 | 28 ++++++++++++++++++++++++++--
 tests/test_cmd_arg_bilibili.py | 25 +++++++++++++++++++++++++
 2 files changed, 51 insertions(+), 2 deletions(-)
 create mode 100644 tests/test_cmd_arg_bilibili.py

diff --git a/cmd_arg/arg.py b/cmd_arg/arg.py
index 20d1d3976..0573f97d6 100644
--- a/cmd_arg/arg.py
+++ b/cmd_arg/arg.py
@@ -151,6 +151,27 @@ def _normalize_tieba_creator_url(value: str) -> str:
     return f"https://tieba.baidu.com/home/main?id={value}"
 
 
+def _is_bilibili_article_id(value: str) -> bool:
+    """Return True for Bilibili article URLs, cv IDs, or numeric article IDs."""
+    value = value.strip()
+    return (
+        value.isdigit()
+        or re.fullmatch(r"cv\d+", value) is not None
+        or re.search(r"/read/cv\d+", value) is not None
+    )
+
+
+def _split_bilibili_specified_ids(values: list[str]) -> tuple[list[str], list[str]]:
+    video_ids: list[str] = []
+    article_ids: list[str] = []
+    for value in values:
+        if _is_bilibili_article_id(value):
+            article_ids.append(value)
+        else:
+            video_ids.append(value)
+    return video_ids, article_ids
+
+
 async def parse_cmd(argv: Optional[Sequence[str]] = None):
     """Parse command line arguments using Typer."""
 
@@ -255,7 +276,7 @@ def main(
             str,
             typer.Option(
                 "--specified_id",
-                help="Post/video ID list in detail mode, multiple IDs separated by commas (supports full URL or ID)",
+                help="Post/video/article ID list in detail mode, multiple IDs separated by commas (supports full URL or ID)",
                 rich_help_panel="Basic Configuration",
             ),
         ] = "",
@@ -371,7 +392,10 @@ def main(
             if platform == PlatformEnum.XHS:
                 config.XHS_SPECIFIED_NOTE_URL_LIST = specified_id_list
             elif platform == PlatformEnum.BILIBILI:
-                config.BILI_SPECIFIED_ID_LIST = specified_id_list
+                (
+                    config.BILI_SPECIFIED_ID_LIST,
+                    config.BILI_SPECIFIED_ARTICLE_ID_LIST,
+                ) = _split_bilibili_specified_ids(specified_id_list)
             elif platform == PlatformEnum.DOUYIN:
                 config.DY_SPECIFIED_ID_LIST = specified_id_list
             elif platform == PlatformEnum.WEIBO:
diff --git a/tests/test_cmd_arg_bilibili.py b/tests/test_cmd_arg_bilibili.py
new file mode 100644
index 000000000..a095caced
--- /dev/null
+++ b/tests/test_cmd_arg_bilibili.py
@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+
+import config
+import pytest
+from cmd_arg import parse_cmd
+
+
+@pytest.mark.asyncio
+async def test_bilibili_detail_cli_splits_video_and_article_ids():
+    await parse_cmd(
+        [
+            "--platform",
+            "bili",
+            "--type",
+            "detail",
+            "--specified_id",
+            "BV1Sz4y1U77N,https://www.bilibili.com/read/cv123456,cv654321",
+        ]
+    )
+
+    assert config.BILI_SPECIFIED_ID_LIST == ["BV1Sz4y1U77N"]
+    assert config.BILI_SPECIFIED_ARTICLE_ID_LIST == [
+        "https://www.bilibili.com/read/cv123456",
+        "cv654321",
+    ]

From 9e7e0fbca581ecba5010bf505ffb028e01402e6b Mon Sep 17 00:00:00 2001
From: chenjiahao <471929267@qq.com>
Date: Thu, 2 Jul 2026 14:09:28 +0800
Subject: [PATCH 12/12] Support mixed Bilibili specified IDs

---
 cmd_arg/arg.py                                | 28 +-----------
 config/bilibili_config.py                     | 14 +++---
 media_platform/bilibili/core.py               | 22 +++++++---
 media_platform/bilibili/help.py               | 26 +++++++++++
 .../bilibili/test_core_article_flow.py        | 44 ++++++++++++++++++-
 tests/test_cmd_arg_bilibili.py                |  6 +--
 6 files changed, 95 insertions(+), 45 deletions(-)

diff --git a/cmd_arg/arg.py b/cmd_arg/arg.py
index 0573f97d6..d1c7a6a7a 100644
--- a/cmd_arg/arg.py
+++ b/cmd_arg/arg.py
@@ -21,8 +21,8 @@
 from __future__ import annotations
 
 
-import sys
 import re
+import sys
 from enum import Enum
 from types import SimpleNamespace
 from typing import Iterable, Optional, Sequence, Type, TypeVar
@@ -151,27 +151,6 @@ def _normalize_tieba_creator_url(value: str) -> str:
     return f"https://tieba.baidu.com/home/main?id={value}"
 
 
-def _is_bilibili_article_id(value: str) -> bool:
-    """Return True for Bilibili article URLs, cv IDs, or numeric article IDs."""
-    value = value.strip()
-    return (
-        value.isdigit()
-        or re.fullmatch(r"cv\d+", value) is not None
-        or re.search(r"/read/cv\d+", value) is not None
-    )
-
-
-def _split_bilibili_specified_ids(values: list[str]) -> tuple[list[str], list[str]]:
-    video_ids: list[str] = []
-    article_ids: list[str] = []
-    for value in values:
-        if _is_bilibili_article_id(value):
-            article_ids.append(value)
-        else:
-            video_ids.append(value)
-    return video_ids, article_ids
-
-
 async def parse_cmd(argv: Optional[Sequence[str]] = None):
     """Parse command line arguments using Typer."""
 
@@ -392,10 +371,7 @@ def main(
             if platform == PlatformEnum.XHS:
                 config.XHS_SPECIFIED_NOTE_URL_LIST = specified_id_list
             elif platform == PlatformEnum.BILIBILI:
-                (
-                    config.BILI_SPECIFIED_ID_LIST,
-                    config.BILI_SPECIFIED_ARTICLE_ID_LIST,
-                ) = _split_bilibili_specified_ids(specified_id_list)
+                config.BILI_SPECIFIED_ID_LIST = specified_id_list
             elif platform == PlatformEnum.DOUYIN:
                 config.DY_SPECIFIED_ID_LIST = specified_id_list
             elif platform == PlatformEnum.WEIBO:
diff --git a/config/bilibili_config.py b/config/bilibili_config.py
index 867f5cba5..e8385e20b 100644
--- a/config/bilibili_config.py
+++ b/config/bilibili_config.py
@@ -21,10 +21,13 @@
 # Control the number of videos/posts crawled per day
 MAX_NOTES_PER_DAY = 1
 
-# Specify Bilibili video URL list (supports complete URL or BV number)
+# Specify Bilibili video/article URL list (supports complete URL, BV number, cv ID, or numeric article ID)
 # Example:
-# - Full URL: "https://www.bilibili.com/video/BV1dwuKzmE26/?spm_id_from=333.1387.homepage.video_card.click"
+# - Video URL: "https://www.bilibili.com/video/BV1dwuKzmE26/?spm_id_from=333.1387.homepage.video_card.click"
 # - BV number: "BV1d54y1g7db"
+# - Article URL: "https://www.bilibili.com/read/cv123456"
+# - cv ID: "cv123456"
+# - Numeric article ID: "123456"
 BILI_SPECIFIED_ID_LIST = [
     "https://www.bilibili.com/video/BV1dwuKzmE26/?spm_id_from=333.1387.homepage.video_card.click",
     "BV1Sz4y1U77N",
@@ -32,13 +35,6 @@
     # ........................
 ]
 
-# Specify Bilibili article URL list for detail mode (supports full URL, cv ID, or numeric article ID)
-# Example:
-# - Full URL: "https://www.bilibili.com/read/cv123456"
-# - cv ID: "cv123456"
-# - Numeric article ID: "123456"
-BILI_SPECIFIED_ARTICLE_ID_LIST = []
-
 # Specify the URL list of Bilibili creators (supports full URL or UID)
 # Example:
 # - Full URL: "https://space.bilibili.com/434377496?spm_id_from=333.1007.0.0"
diff --git a/media_platform/bilibili/core.py b/media_platform/bilibili/core.py
index 252eb564f..33538cdf6 100644
--- a/media_platform/bilibili/core.py
+++ b/media_platform/bilibili/core.py
@@ -50,7 +50,12 @@
 from .client import BilibiliClient
 from .exception import DataFetchError
 from .field import BilibiliCommentType, SearchOrderType
-from .help import parse_article_info_from_url, parse_creator_info_from_url, parse_video_info_from_url
+from .help import (
+    parse_article_info_from_url,
+    parse_creator_info_from_url,
+    parse_video_info_from_url,
+    split_bilibili_specified_ids,
+)
 from .login import BilibiliLogin
 
 
@@ -116,10 +121,7 @@ async def start(self):
                 await self.search()
             elif config.CRAWLER_TYPE == "detail":
                 # Get the information and comments of the specified post
-                await self.get_specified_videos(config.BILI_SPECIFIED_ID_LIST)
-                article_id_list = getattr(config, "BILI_SPECIFIED_ARTICLE_ID_LIST", [])
-                if article_id_list:
-                    await self.get_specified_articles(article_id_list)
+                await self.get_specified_ids(config.BILI_SPECIFIED_ID_LIST)
             elif config.CRAWLER_TYPE == "creator":
                 if config.CREATOR_MODE:
                     for creator_url in config.BILI_CREATOR_ID_LIST:
@@ -419,6 +421,16 @@ async def get_specified_videos(self, video_url_list: List[str]):
                 await self.get_bilibili_video(video_detail, semaphore)
         await self.batch_get_video_comments(video_aids_list)
 
+    async def get_specified_ids(self, specified_id_list: List[str]):
+        """
+        Dispatch mixed Bilibili specified IDs to video or article crawlers.
+        """
+        video_id_list, article_id_list = split_bilibili_specified_ids(specified_id_list)
+        if video_id_list:
+            await self.get_specified_videos(video_id_list)
+        if article_id_list:
+            await self.get_specified_articles(article_id_list)
+
     async def get_specified_articles(self, article_url_list: List[str]):
         """
         get specified articles info from URLs or cv IDs
diff --git a/media_platform/bilibili/help.py b/media_platform/bilibili/help.py
index 9830860d3..64ce6e443 100644
--- a/media_platform/bilibili/help.py
+++ b/media_platform/bilibili/help.py
@@ -131,6 +131,32 @@ def parse_article_info_from_url(url: str) -> ArticleUrlInfo:
     raise ValueError(f"Unable to parse article ID from URL: {url}")
 
 
+def is_bilibili_article_id(value: str) -> bool:
+    """
+    Return True for Bilibili article URLs, cv IDs, or numeric article IDs.
+    """
+    value = value.strip()
+    return (
+        value.isdigit()
+        or re.fullmatch(r"cv\d+", value) is not None
+        or re.search(r"/read/cv\d+", value) is not None
+    )
+
+
+def split_bilibili_specified_ids(values: list[str]) -> tuple[list[str], list[str]]:
+    """
+    Split mixed Bilibili specified IDs into video and article inputs.
+    """
+    video_ids: list[str] = []
+    article_ids: list[str] = []
+    for value in values:
+        if is_bilibili_article_id(value):
+            article_ids.append(value)
+        else:
+            video_ids.append(value)
+    return video_ids, article_ids
+
+
 def parse_creator_info_from_url(url: str) -> CreatorUrlInfo:
     """
     Parse creator ID from Bilibili creator space URL
diff --git a/tests/media_platform/bilibili/test_core_article_flow.py b/tests/media_platform/bilibili/test_core_article_flow.py
index 6b18e1b05..5d2e48e09 100644
--- a/tests/media_platform/bilibili/test_core_article_flow.py
+++ b/tests/media_platform/bilibili/test_core_article_flow.py
@@ -7,6 +7,7 @@
 import config
 from media_platform.bilibili.core import BilibiliCrawler
 from media_platform.bilibili.field import BilibiliCommentType
+from media_platform.bilibili.help import split_bilibili_specified_ids
 from store import bilibili as bilibili_store
 
 
@@ -84,5 +85,44 @@ async def fake_get_article_comments(article_id, semaphore):
     assert called is False
 
 
-def test_detail_start_config_has_separate_article_list():
-    assert hasattr(config, "BILI_SPECIFIED_ARTICLE_ID_LIST")
+def test_split_bilibili_specified_ids_separates_video_and_article_inputs():
+    video_ids, article_ids = split_bilibili_specified_ids([
+        "BV1Sz4y1U77N",
+        "https://www.bilibili.com/video/BV1d54y1g7db",
+        "https://www.bilibili.com/read/cv123456",
+        "cv654321",
+        "789012",
+    ])
+
+    assert video_ids == [
+        "BV1Sz4y1U77N",
+        "https://www.bilibili.com/video/BV1d54y1g7db",
+    ]
+    assert article_ids == [
+        "https://www.bilibili.com/read/cv123456",
+        "cv654321",
+        "789012",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_get_specified_ids_dispatches_videos_and_articles(monkeypatch, crawler):
+    video_calls = []
+    article_calls = []
+
+    async def fake_get_specified_videos(video_ids):
+        video_calls.append(video_ids)
+
+    async def fake_get_specified_articles(article_ids):
+        article_calls.append(article_ids)
+
+    monkeypatch.setattr(crawler, "get_specified_videos", fake_get_specified_videos)
+    monkeypatch.setattr(crawler, "get_specified_articles", fake_get_specified_articles)
+
+    await crawler.get_specified_ids([
+        "BV1Sz4y1U77N",
+        "cv123456",
+    ])
+
+    assert video_calls == [["BV1Sz4y1U77N"]]
+    assert article_calls == [["cv123456"]]
diff --git a/tests/test_cmd_arg_bilibili.py b/tests/test_cmd_arg_bilibili.py
index a095caced..bd7d9cf92 100644
--- a/tests/test_cmd_arg_bilibili.py
+++ b/tests/test_cmd_arg_bilibili.py
@@ -6,7 +6,7 @@
 
 
 @pytest.mark.asyncio
-async def test_bilibili_detail_cli_splits_video_and_article_ids():
+async def test_bilibili_detail_cli_keeps_video_and_article_ids_in_single_list():
     await parse_cmd(
         [
             "--platform",
@@ -18,8 +18,8 @@ async def test_bilibili_detail_cli_splits_video_and_article_ids():
         ]
     )
 
-    assert config.BILI_SPECIFIED_ID_LIST == ["BV1Sz4y1U77N"]
-    assert config.BILI_SPECIFIED_ARTICLE_ID_LIST == [
+    assert config.BILI_SPECIFIED_ID_LIST == [
+        "BV1Sz4y1U77N",
         "https://www.bilibili.com/read/cv123456",
         "cv654321",
     ]