fix: harden fetch transactions and prepare v0.4.1 (#65)

Author: yuanrengu

.github/workflows/optional-dependencies.yml

@@ -0,0 +1,37 @@
+name: Optional dependencies
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "17 3 * * 1"
+
+jobs:
+  install:
+    runs-on: ubuntu-latest
+    timeout-minutes: 35
+    strategy:
+      fail-fast: false
+      matrix:
+        extra: [base, viz, ml]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+      - name: Install base
+        if: matrix.extra == 'base'
+        run: python -m pip install -e .
+      - name: Install optional extra
+        if: matrix.extra != 'base'
+        run: python -m pip install -e ".[${{ matrix.extra }}]"
+      - name: Smoke test base CLI
+        run: |
+          xcrawler --version
+          xcrawler demo --output /tmp/xcrawler-demo
+      - name: Smoke test viz imports
+        if: matrix.extra == 'viz'
+        run: python -c "import matplotlib; import visualize"
+      - name: Smoke test ML imports
+        if: matrix.extra == 'ml'
+        run: python -c "import sentence_transformers; import sklearn"

.github/workflows/publish.yml

@@ -0,0 +1,40 @@
+name: Publish to PyPI
+
+on:
+  release:
+    types: [published]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Build distributions
+        run: |
+          python -m pip install --upgrade build
+          python -m build
+      - uses: actions/upload-artifact@v4
+        with:
+          name: python-package-distributions
+          path: dist/
+
+  publish:
+    needs: build
+    runs-on: ubuntu-latest
+    environment:
+      name: pypi
+      url: https://pypi.org/project/xcrawler-ai/
+    permissions:
+      id-token: write
+    steps:
+      - uses: actions/download-artifact@v4
+        with:
+          name: python-package-distributions
+          path: dist/
+      - uses: pypa/gh-action-pypi-publish@release/v1

.github/workflows/test.yml

@@ -58,3 +58,21 @@ jobs:
           /tmp/xcrawler-wheel-test/bin/python -m pip install dist/*.whl
           /tmp/xcrawler-wheel-test/bin/xcrawler --version
           /tmp/xcrawler-wheel-test/bin/xcrawler --help
+
+  coverage:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install test dependencies
+        run: python -m pip install -e ".[test]"
+      - name: Enforce coverage floor
+        run: >-
+          python -m pytest
+          --cov=xcrawler
+          --cov=main
+          --cov=fetch_more_history
+          --cov-report=term-missing
+          --cov-fail-under=80

CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.4.1] - 2026-07-12
+
 ### Fixed
 - Full timeline fetching now retries transient failures and fails explicitly when any later page cannot be fetched, instead of returning a partial result as complete
 - `xcrawler fetch --replace` commits raw and translated snapshots together and preserves both previous files when translation is incomplete or either replacement fails
@@ -19,6 +21,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 - `fetch-more --pages` is now explicitly a shared HTTP request budget that includes retries
 - Full fetch behavior is documented as archive mode by default and snapshot mode with `--replace`
+- Full and incremental X API operations share one retry, rate-limit, and error-classification engine
+- `--replace --no-translate` filters retained translations to the new raw snapshot instead of leaving stale records
+- CI now enforces a coverage floor and performs scheduled base, visualization, and ML installation smoke tests
+- The PyPI distribution is named `xcrawler-ai` (the CLI and import package remain `xcrawler`) and uses Trusted Publishing
 
 ## [0.4.0] - 2026-07-11
 

README.md

@@ -22,7 +22,7 @@
 - 🧩 **模块化** — 统一 `xcrawler` CLI，支持可插拔存储和 LLM Provider
 
 ```bash
-python3 -m pip install -e ".[all]"          # 安装全功能依赖
+python3 -m pip install "xcrawler-ai[all]"   # 从 PyPI 安装全功能依赖
 xcrawler demo                               # 无需 API Key 的虚构数据示例
 xcrawler fetch --user MiracleHe             # 抓取 + 翻译 + 聚类
 xcrawler analyze interest --user MiracleHe  # 专业兴趣画像
@@ -64,15 +64,15 @@ xcrawler report --user MiracleHe            # 生成图表 + HTML 报告
 项目要求 Python 3.10 或更高版本。
 
 ```bash
-# 推荐：使用虚拟环境，安装全功能依赖
+# 推荐：使用虚拟环境，从 PyPI 安装全功能依赖
 python3 -m venv .venv
 source .venv/bin/activate
-python3 -m pip install -e ".[all]"
+python3 -m pip install "xcrawler-ai[all]"
 
-# 如果只需要基础 CLI / 抓取 / LLM 功能
-python3 -m pip install -e .
+# 如果只需要基础 CLI / 抓取 / LLM 功能（命令仍为 xcrawler）
+python3 -m pip install xcrawler-ai
 
-# 可选依赖：向量聚类/可视化
+# 从源码开发或按需安装向量聚类/可视化依赖
 python3 -m pip install -e ".[ml,viz]"
 ```
 
@@ -1006,7 +1006,13 @@ LLM 调用通过 `LLMProvider` 抽象保留 DeepSeek/OpenAI 兼容 Provider 入
 
 ## 🔄 更新日志
 
-### v0.4.0 - 可靠性、安全数据语义与无密钥 Demo 🆕
+### v0.4.1 - 抓取事务与增量状态机 🆕
+- ✅ **全量完整性**：任意分页失败都显式返回失败，禁止部分结果进入 snapshot
+- ✅ **增量状态机**：Forward/Backward 分阶段保存，记录请求、数据页、重试、stop reason 和 partial 状态
+- ✅ **数据契约**：raw schema 严格校验，译文携带原文与配置指纹
+- ✅ **统一请求引擎**：全量、增量、用户信息共用重试、限流和错误分类
+
+### v0.4.0 - 可靠性、安全数据语义与无密钥 Demo
 - ✅ **抓取可靠性**：增量抓取区分“无新数据”和网络失败，增加 429/5xx/超时重试
 - ✅ **数据安全**：全量抓取默认合并，`--replace` 显式覆盖，强制重翻改为全有或全无
 - ✅ **快照事务**：`--replace` 仅在抓取完整且翻译全部成功后同时提交 raw/translated，部分失败返回非零退出码

RELEASE_CHECKLIST.md

@@ -11,6 +11,7 @@
 - [ ] `xcrawler --help` 可正常显示。
 - [ ] `python3 -m build` 成功生成 wheel 和 sdist。
 - [ ] `python3 -m twine check dist/*` 通过。
+- [ ] PyPI Trusted Publisher 指向 `yuanrengu/xcrawler`、工作流 `publish.yml`、环境 `pypi`。
 - [ ] 在干净虚拟环境从 wheel 安装后，`xcrawler --version`、`xcrawler --help` 和 `xcrawler demo` 通过。
 - [ ] `README.md`、`QUICK_START.md`、`.env.example` 与当前 CLI 保持一致。
 - [ ] `CONTRIBUTING.md`、`SECURITY.md`、`LICENSE` 存在且链接有效。
@@ -22,13 +23,14 @@
 - [ ] 更新 `pyproject.toml` 中的版本号。
 - [ ] `xcrawler/__init__.py` 与 `pyproject.toml` 版本号一致。
 - [ ] `CHANGELOG.md` 将待发布内容从 `Unreleased` 移入当前版本和日期。
-- [ ] 创建并验证签名或 annotated Git tag，例如 `v0.4.0`。
+- [ ] 创建并验证签名或 annotated Git tag，例如 `v0.4.1`。
 - [ ] 在 README 更新对应版本的变更摘要。
-- [ ] 创建 GitHub Release，标题使用版本号，例如 `v0.4.0`。
+- [ ] 创建 GitHub Release，标题使用版本号，例如 `v0.4.1`。
 - [ ] Release notes 包含：新增能力、兼容性说明、迁移提示、已知限制。
 
 ## 发布后
 
 - [ ] 确认 Release 页面、源码包和文档链接可访问。
+- [ ] GitHub Release 发布后，`publish.yml` 成功将 `xcrawler-ai` 上传到 PyPI。
 - [ ] 创建下一阶段 roadmap 或 follow-up issues。
 - [ ] 如涉及安全或隐私变更，在 release notes 中单独说明。

fetch_more_history.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import argparse
-import json
 import os
 import time
 from dataclasses import asdict, dataclass
@@ -16,6 +15,7 @@
 import requests
 
 from xcrawler.clients import x_api
+from xcrawler.clients.retry import RequestAttemptsError, request_json_with_retries
 from xcrawler.clients.x_api import auth_headers
 from xcrawler.config import load_config
 from xcrawler.paths import ensure_dir
@@ -42,6 +42,12 @@
 class FetchError(RuntimeError):
     """抓取未能成功完成，不应将其视为“没有新数据”。"""
 
+    def __init__(self, message: str, *, requests_used: int = 0, retries: int = 0, stop_reason: str = "error"):
+        super().__init__(message)
+        self.requests_used = requests_used
+        self.retries = retries
+        self.stop_reason = stop_reason
+
 
 @dataclass
 class FetchBatchResult:
@@ -87,6 +93,9 @@ def save_fetch_status(
     forward: FetchBatchResult | None = None,
     backward: FetchBatchResult | None = None,
     error: str | None = None,
+    error_requests: int = 0,
+    error_retries: int = 0,
+    error_stop_reason: str | None = None,
 ) -> None:
     save_json(
         os.path.join(cache_dir, f"{username}_fetch_status.json"),
@@ -99,12 +108,13 @@ def save_fetch_status(
             "backward": backward.to_dict() if backward else None,
             "requests_used": (forward.requests_used if forward else 0) + (
                 backward.requests_used if backward else 0
-            ),
+            ) + error_requests,
             "data_pages": (forward.data_pages if forward else 0) + (
                 backward.data_pages if backward else 0
             ),
-            "retries": (forward.retries if forward else 0) + (backward.retries if backward else 0),
+            "retries": (forward.retries if forward else 0) + (backward.retries if backward else 0) + error_retries,
             "error": error,
+            "error_stop_reason": error_stop_reason,
         },
     )
 
@@ -160,63 +170,29 @@ def fetch_tweets_generic(
 
     page = 0
     while data_pages < max_pages_limit and requests_used < request_budget:
-        response = None
-        data = None
-        for attempt in range(1, max_retries + 1):
-            if requests_used >= request_budget:
-                raise FetchError(f"请求预算已用完（{requests_used}/{request_budget}）")
-            try:
-                requests_used += 1
-                response = requests.get(url, headers=headers, params=params, timeout=10)
-                if response.status_code == 429:
-                    reset_time = response.headers.get("x-rate-limit-reset")
-                    if reset_time is None:
-                        wait_seconds = min(2 ** (attempt - 1), 8)
-                    else:
-                        try:
-                            wait_seconds = max(0, int(float(reset_time)) - int(time.time()))
-                        except (TypeError, ValueError) as error:
-                            raise FetchError("API 限流重置时间无效") from error
-                    if wait_seconds > MAX_RATE_LIMIT_WAIT:
-                        raise FetchError(f"API 限流需等待 {wait_seconds} 秒，超过最大等待时间")
-                    if attempt == max_retries:
-                        raise FetchError(f"API 限流，重试 {max_retries} 次后仍未恢复")
-                    print(f"⏳ API 限流，{wait_seconds} 秒后重试 ({attempt}/{max_retries})...")
-                    retries += 1
-                    time.sleep(wait_seconds)
-                    continue
-
-                response.raise_for_status()
-                data = response.json()
-                break
-            except requests.exceptions.HTTPError as error:
-                status = error.response.status_code if error.response is not None else 0
-                if status in (401, 403):
-                    raise FetchError(f"认证失败（HTTP {status}），请检查 X_BEARER_TOKEN") from error
-                retryable = status >= 500 or status == 0
-                if not retryable or attempt == max_retries:
-                    raise FetchError(f"第 {page + 1} 页 HTTP 错误（{status or 'unknown'}）") from error
-                delay = min(2 ** (attempt - 1), 8)
-                print(f"⚠️ 第 {page + 1} 页 HTTP {status}，{delay} 秒后重试 ({attempt}/{max_retries})...")
-                retries += 1
-                time.sleep(delay)
-            except requests.exceptions.RequestException as error:
-                if attempt == max_retries:
-                    raise FetchError(f"第 {page + 1} 页网络错误，重试 {max_retries} 次后仍失败") from error
-                delay = min(2 ** (attempt - 1), 8)
-                print(f"⚠️ 第 {page + 1} 页网络错误，{delay} 秒后重试 ({attempt}/{max_retries})...")
-                retries += 1
-                time.sleep(delay)
-            except (json.JSONDecodeError, ValueError) as error:
-                if attempt == max_retries:
-                    raise FetchError(f"第 {page + 1} 页响应解析失败") from error
-                delay = min(2 ** (attempt - 1), 8)
-                print(f"⚠️ 第 {page + 1} 页响应解析失败，{delay} 秒后重试 ({attempt}/{max_retries})...")
-                retries += 1
-                time.sleep(delay)
-
-        if response is None or data is None:
-            raise FetchError(f"第 {page + 1} 页未能获取有效响应")
+        try:
+            attempt_result = request_json_with_retries(
+                url,
+                headers=headers,
+                params=params,
+                request_get=requests.get,
+                max_retries=max_retries,
+                request_budget=request_budget - requests_used,
+                page_number=page + 1,
+                max_rate_limit_wait=MAX_RATE_LIMIT_WAIT,
+                sleep=time.sleep,
+            )
+        except RequestAttemptsError as error:
+            raise FetchError(
+                str(error),
+                requests_used=requests_used + error.requests_used,
+                retries=retries + error.retries,
+                stop_reason=error.stop_reason,
+            ) from error
+        requests_used += attempt_result.requests_used
+        retries += attempt_result.retries
+        response = attempt_result.response
+        data = attempt_result.data
 
         try:
 
@@ -387,7 +363,15 @@ def main():
             )
         except FetchError as error:
             print(f"❌ 新推文抓取失败: {error}")
-            save_fetch_status(CACHE_DIR, TARGET_USERNAME, status="failed", error=str(error))
+            save_fetch_status(
+                CACHE_DIR,
+                TARGET_USERNAME,
+                status="failed",
+                error=str(error),
+                error_requests=error.requests_used,
+                error_retries=error.retries,
+                error_stop_reason=error.stop_reason,
+            )
             return 1
         remaining_pages_quota -= forward_result.requests_used
         tweets = forward_result.tweets
@@ -444,6 +428,9 @@ def main():
                 status="partial",
                 forward=forward_result,
                 error=str(error),
+                error_requests=error.requests_used,
+                error_retries=error.retries,
+                error_stop_reason=error.stop_reason,
             )
             return 2
         tweets = backward_result.tweets

main.py

@@ -14,7 +14,7 @@
 from xcrawler.paths import ensure_dir, translation_cache_path
 from xcrawler.services.embeddings import encode_texts_with_cache
 from xcrawler.services.llm_calls import LLMCallRecorder
-from xcrawler.services.records import make_translated_tweet
+from xcrawler.services.records import make_translated_tweet, normalize_translated_tweets
 from xcrawler.services.sampling import sample_evenly
 from xcrawler.services.translation import (
     translate_batch,
@@ -329,7 +329,22 @@ def main():
         # 3. 清洗 + 批量翻译
         if args.no_translate:
             if args.replace:
-                save_json(raw_file, raw_tweets)
+                translated_file = os.path.join(CACHE_DIR, f"{TARGET_USERNAME}_translated.json")
+                if os.path.exists(translated_file):
+                    existing_translated = normalize_translated_tweets(load_json(translated_file, default=[]))
+                    snapshot_ids = {str(tweet["id"]) for tweet in raw_tweets}
+                    retained_translations = [
+                        item
+                        for item in existing_translated
+                        if item.get("tweet_id") is not None and str(item["tweet_id"]) in snapshot_ids
+                    ]
+                    replace_json_files_atomically({
+                        raw_file: raw_tweets,
+                        translated_file: retained_translations,
+                    })
+                    print(f"🧹 snapshot 已移除 {len(existing_translated) - len(retained_translations)} 条过期译文")
+                else:
+                    save_json(raw_file, raw_tweets)
             print("⏭️  --no-translate 模式：跳过翻译和分析，仅保存原始推文")
             print(f"\n✅ 完成！原始推文已保存至: {raw_file}")
             return 0