服务端同步改为按行合并去重

serve/backend/app/api/data_sync.py

@@ -2,11 +2,13 @@
 from __future__ import annotations
 
 import logging
+import os
 import tarfile
 import tempfile
 from io import BytesIO
 from pathlib import Path
 
+import polars as pl
 from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, Form
 
 from app.config import settings
@@ -29,6 +31,89 @@ SYNCABLE_PARTS = {
 }
 
 
+def _resolve_unique_key(columns: list[str]) -> list[str] | None:
+    """根据列名推断去重键。
+
+    策略:
+      - 含 symbol + datetime: 按 [symbol, datetime] 去重(分钟 K)
+      - 含 symbol + date:    按 [symbol, date]    去重(日 K / 复权因子)
+      - 含 symbol:           按 [symbol]          去重(标的维表)
+      - 其他: 无法推断，回退到覆盖
+    """
+    cols = set(columns)
+    if "symbol" not in cols:
+        return None
+    if "datetime" in cols:
+        return ["symbol", "datetime"]
+    if "date" in cols:
+        return ["symbol", "date"]
+    return ["symbol"]
+
+
+def _merge_parquet(target: Path, new_bytes: bytes) -> tuple[bool, str]:
+    """把 new_bytes 代表的 Parquet 与 target 已有文件合并去重。
+
+    返回 (是否成功落盘, 操作描述):
+      - created:   目标不存在，直接新建
+      - merged:    与已有文件按主键合并去重(保留后写入的记录)
+      - empty_new: 上传文件为空，保留旧文件
+      - overwritten_no_key:   无法推断去重键，回退覆盖
+      - overwritten_corrupt_existing: 已有文件损坏，回退覆盖
+      - invalid_new: 上传文件不是合法 Parquet，未落盘
+    """
+    if not target.exists():
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_bytes(new_bytes)
+        return True, "created"
+
+    try:
+        new_df = pl.read_parquet(BytesIO(new_bytes))
+    except Exception as exc:
+        logger.warning("failed to parse uploaded parquet %s: %s", target.name, exc)
+        return False, "invalid_new"
+
+    if new_df.is_empty():
+        return True, "empty_new"
+
+    key = _resolve_unique_key(new_df.columns)
+    if key is None:
+        logger.warning(
+            "no unique key for %s (columns=%s), overwriting",
+            target.name,
+            new_df.columns,
+        )
+        target.write_bytes(new_bytes)
+        return True, "overwritten_no_key"
+
+    try:
+        existing_df = pl.read_parquet(target)
+    except Exception as exc:
+        logger.warning("existing parquet %s corrupt, overwriting: %s", target, exc)
+        target.write_bytes(new_bytes)
+        return True, "overwritten_corrupt_existing"
+
+    if existing_df.is_empty():
+        merged = new_df
+        action = "created"
+    else:
+        merged = pl.concat([existing_df, new_df], how="diagonal_relaxed").unique(
+            subset=key, keep="last"
+        )
+        action = "merged"
+
+    # 原子写入，防止写入过程中崩溃导致文件损坏
+    tmp = target.with_suffix(f".tmp-{os.getpid()}")
+    try:
+        merged.write_parquet(tmp)
+        os.replace(tmp, target)
+    except Exception:
+        if tmp.exists():
+            tmp.unlink(missing_ok=True)
+        raise
+
+    return True, action
+
+
 def _verify_sync_key(request: Request) -> None:
     """简单的鉴权 — 校验 X-Sync-Key 头。"""
     key = request.headers.get("X-Sync-Key", "")
@@ -105,6 +190,7 @@ async def upload_sync(
         raise HTTPException(status_code=400, detail="上传文件为空")
 
     extracted = 0
+    action_counter: dict[str, int] = {}
     try:
         with tarfile.open(fileobj=BytesIO(raw), mode="r:gz") as tar:
             for member in tar.getmembers():
@@ -123,13 +209,15 @@ async def upload_sync(
                     with tar.extractfile(member) as src:
                         if src is None:
                             continue
-                        target.write_bytes(src.read())
+                        ok, action = _merge_parquet(target, src.read())
+                        if ok:
                             extracted += 1
+                            action_counter[action] = action_counter.get(action, 0) + 1
     except tarfile.TarError as exc:
         raise HTTPException(status_code=400, detail=f"压缩包解析失败: {exc}") from exc
 
     # 刷新缓存
     _refresh_views(request)
 
-    logger.info("sync uploaded: parts=%s files=%d", parts, extracted)
-    return {"ok": True, "parts": sorted(requested), "file_count": extracted}
+    logger.info("sync uploaded: parts=%s files=%d actions=%s", parts, extracted, action_counter)
+    return {"ok": True, "parts": sorted(requested), "file_count": extracted, "actions": action_counter}

serve/backend/uv.lock

@@ -2491,7 +2491,7 @@ all = [
 
 [[package]]
 name = "tickflow-stock-panel-backend"
-version = "0.1.66"
+version = "0.1.70"
 source = { editable = "." }
 dependencies = [
     { name = "apscheduler" },