feat: GUI 体验打磨 + 数据质量校验

GUI 体验打磨：
- 总览/同步/历史三页重构，新视觉规范（#f8f9fb 背景，border 替代 shadow）
- 实时产品列表、错误内联显示、阶段耗时拆分
- 重试失败产品、可点击数据目录（macOS）
- 后端扩展：四元组返回、progress_callback 扩展签名

数据质量校验：
- 10 项检查覆盖文件/内容/时间/覆盖/格式完整性
- repair_data_issues 支持 4 种自动修复动作
- CLI audit 子命令（--fix 模式）
- GUI 异步健康检查（确认→进度→结果三状态）
- 追加快捷路径改用原子写入（copy+replace）

407 测试通过

Author: Astro-Han

quantclass_sync_internal/cli.py

@@ -61,7 +61,7 @@
     split_products,
     utc_now_iso,
 )
-from .data_query import get_latest_run_summary, get_products_overview
+from .data_query import check_data_health, get_latest_run_summary, get_products_overview, repair_data_issues
 from .orchestrator import _build_headers, _execute_plans, load_catalog_or_raise, run_update_with_settings
 from .reporting import _append_result, _finalize_and_write_report, _new_report, resolve_report_path
 from .status_store import (
@@ -836,3 +836,66 @@ def cmd_gui(ctx: typer.Context) -> None:
     """
     from .gui import launch_gui
     launch_gui()
+
+@app.command("audit")
+@command_guard("audit")
+def cmd_audit(
+    ctx: typer.Context,
+    fix: bool = typer.Option(False, "--fix", help="自动修复可修复问题"),
+) -> None:
+    """数据质量全面检查"""
+    command_ctx = _init_command(ctx, "audit")
+    catalog = load_catalog_or_raise(command_ctx.catalog_file)
+    data_root = command_ctx.data_root
+
+    # CLI 进度回调：在同一行滚动显示当前扫描产品
+    def cli_progress(current, total, product, phase):
+        RICH_CONSOLE.print(f"  [{current + 1}/{total}] {product}", end="\r")
+
+    RICH_CONSOLE.print("[bold]数据质量检查[/bold]\n")
+    result = check_data_health(data_root, catalog, progress_callback=cli_progress)
+    RICH_CONSOLE.print()
+
+    issues = result["issues"]
+    summary = result["summary"]
+
+    if not issues:
+        RICH_CONSOLE.print("[green]✓ 数据健康，未发现问题[/green]")
+    else:
+        from collections import defaultdict
+        # 按问题类别分组显示
+        by_category = defaultdict(list)
+        for i in issues:
+            by_category[i["category"]].append(i)
+        category_names = {
+            "file_integrity": "文件完整性",
+            "content_integrity": "内容完整性",
+            "temporal_integrity": "时间完整性",
+            "coverage_integrity": "覆盖完整性",
+            "format_integrity": "格式完整性",
+        }
+        for cat, cat_issues in by_category.items():
+            RICH_CONSOLE.print(f"\n[bold]{category_names.get(cat, cat)}[/bold]")
+            for i in cat_issues:
+                sev_icon = "[red]●[/red]" if i["severity"] == "error" else "[yellow]○[/yellow]"
+                repair_tag = "[green]可修复[/green]" if i["repairable"] else "[dim]需处理[/dim]"
+                file_info = f" {i['file']}" if i["file"] else ""
+                RICH_CONSOLE.print(
+                    f"  {sev_icon} {i['product']}{file_info} - {i['detail']} {repair_tag}"
+                )
+
+    # 底部汇总行
+    RICH_CONSOLE.print(
+        f"\n{summary['scanned_products']} 产品 · {summary['scanned_files']} 文件"
+        f" · {summary['elapsed_seconds']}s"
+        f" · [red]{summary['by_severity'].get('error', 0)} 错误[/red]"
+        f" / [yellow]{summary['by_severity'].get('warning', 0)} 警告[/yellow]"
+    )
+
+    if fix and any(i["repairable"] for i in issues):
+        RICH_CONSOLE.print("\n[bold]修复可修复问题...[/bold]")
+        repair_result = repair_data_issues(data_root, issues)
+        RICH_CONSOLE.print(
+            f"  已修复 {len(repair_result['repaired'])} 个"
+            f"，失败 {len(repair_result['failed'])} 个"
+        )

quantclass_sync_internal/constants.py

@@ -200,6 +200,13 @@
     "default": {"max_attempts": 5, "timeout_seconds": 60, "backoff_cap_seconds": 8},
 }
 
+META_HEALTH_BASELINE_REL = Path(SYNC_META_DIRNAME) / "log" / "health_baseline.json"
+
+# 财务/公告类产品，日期连续性检查不适用
+FINANCIAL_PRODUCTS = {"stock-fin-data-xbx", "stock-fin-pre-fore-data-xbx"}
+NOTICE_PRODUCTS = {"stock-notices-title"}
+
+
 def normalize_product_name(product: str) -> str:
     """
     统一产品名写法。

quantclass_sync_internal/csv_engine.py

@@ -453,18 +453,30 @@ def _append_csv_rows(
     encoding: str,
 ) -> None:
     """
-    以 append 模式追加 CSV 数据行（不写表头/note）。
+    追加行到 CSV，使用 copy+append+replace 保证原子性。
+    调用方必须持有 _locked_target 文件锁。
 
     前置条件（由调用方保证）：
     - 文件末尾以 \\n 结尾
-    - 调用方持有文件锁
+    - 调用方持有文件锁（.lock 文件锁，非 inode 锁，os.replace 改 inode 安全）
 
     注意（跨层依赖）：中断恢复依赖 orchestrator 层在 sync
     完整成功后才更新 timestamp。
     """
-    with target.open("a", encoding=encoding, newline="") as f:
-        writer = csv.writer(f, delimiter=delimiter, lineterminator="\n")
-        writer.writerows(rows)
+    # 先 copy 到临时文件，追加后 replace，崩溃时临时文件留在磁盘但原文件不损坏
+    tmp_path = target.parent / f".tmp-append-{target.name}"
+    import shutil
+    try:
+        shutil.copy2(target, tmp_path)
+        with tmp_path.open("a", encoding=encoding, newline="") as f:
+            writer = csv.writer(f, delimiter=delimiter, lineterminator="\n")
+            writer.writerows(rows)
+        os.replace(tmp_path, target)
+    except BaseException:
+        # 写入失败时清理临时文件，保留原文件不动
+        if tmp_path.exists():
+            tmp_path.unlink()
+        raise
 
 
 def merge_payload(existing: Optional[CsvPayload], incoming: CsvPayload, rule: DatasetRule) -> Tuple[CsvPayload, int]:
@@ -661,6 +673,7 @@ def sync_payload_to_target(incoming: CsvPayload, target: Path, rule: DatasetRule
                                             existing_delimiter, rule.encoding,
                                         )
                                         audit.checked_files = 1
+                                        audit.append_fast = 1
                                         return "updated", len(incoming.rows), audit
 
         # --- 原有完整合并路径（不变）---