Create webdav_sync.py

webdav_sync.py

@@ -0,0 +1,195 @@
+# webdav_sync.py
+# -*- coding: utf-8 -*-
+"""
+WebDAV 同步器：启动时从远端拉取 history.json，随后每 30 分钟将本地变动回传到远端。
+- 仅依赖 requests
+- 支持 Basic Auth
+- 避免多进程/多实例重复同步：使用文件锁保证只有一个同步线程在运行
+环境变量（至少配置 WEBDAV_URL）：
+  WEBDAV_URL           : 远端文件的完整 URL（例如 https://dav.example.com/path/history.json）
+  WEBDAV_USERNAME      : （可选）用户名
+  WEBDAV_PASSWORD      : （可选）密码
+  WEBDAV_VERIFY_SSL    : （可选）默认为 true；设为 "false" 可跳过证书验证
+  WEBDAV_SYNC_INTERVAL : （可选）同步间隔秒，默认 1800（30 分钟）
+  WEBDAV_SYNC_ENABLED  : （可选）默认启用；设为 "false" 可关闭
+"""
+
+import os
+import time
+import json
+import hashlib
+import threading
+from contextlib import contextmanager
+from typing import Optional
+
+import requests
+
+DEFAULT_INTERVAL = 1800  # 30 minutes
+LOCKFILE_PATH = "/tmp/webdav_sync.lock"  # 进程间锁，确保只跑一个同步线程
+
+def _bool_env(name: str, default: bool) -> bool:
+    v = os.environ.get(name)
+    if v is None:
+        return default
+    return str(v).strip().lower() in ("1", "true", "yes", "on")
+
+def _int_env(name: str, default: int) -> int:
+    v = os.environ.get(name)
+    try:
+        return int(v) if v is not None else default
+    except ValueError:
+        return default
+
+def _sha1_of_file(path: str) -> Optional[str]:
+    if not os.path.isfile(path):
+        return None
+    h = hashlib.sha1()
+    with open(path, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+@contextmanager
+def _interprocess_lock(path: str):
+    """简易文件锁（独占创建）。拿不到锁就不抛异常，直接 yield False。"""
+    fd = None
+    acquired = False
+    try:
+        # O_EXCL + O_CREAT：若文件已存在会失败
+        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        os.write(fd, str(os.getpid()).encode("utf-8"))
+        acquired = True
+        yield True
+    except FileExistsError:
+        yield False
+    finally:
+        if fd is not None:
+            try:
+                os.close(fd)
+                os.unlink(path)
+            except Exception:
+                pass
+
+class WebDavSyncer:
+    def __init__(self, local_path: str):
+        self.local_path = local_path
+        self.url = os.environ.get("WEBDAV_URL", "").strip()
+        self.username = os.environ.get("WEBDAV_USERNAME", "")
+        self.password = os.environ.get("WEBDAV_PASSWORD", "")
+        self.verify_ssl = _bool_env("WEBDAV_VERIFY_SSL", True)
+        self.interval = _int_env("WEBDAV_SYNC_INTERVAL", DEFAULT_INTERVAL)
+        self.enabled = _bool_env("WEBDAV_SYNC_ENABLED", True) and bool(self.url)
+        self._lock = threading.Lock()
+        self._stop = threading.Event()
+        self._thread = None
+        self._last_pushed_hash = None
+
+    def _session(self) -> requests.Session:
+        s = requests.Session()
+        if self.username:
+            s.auth = (self.username, self.password)
+        s.verify = self.verify_ssl
+        s.headers.update({"User-Agent": "history-sync/1.0"})
+        return s
+
+    # ====== Pull（启动时执行一次）======
+    def initial_pull(self):
+        if not self.enabled:
+            print("[webdav-sync] Disabled or missing WEBDAV_URL; skip initial pull.")
+            return
+        try:
+            with self._session() as s:
+                # 先 HEAD 看看是否存在
+                h = s.head(self.url, timeout=10)
+                if h.status_code == 404:
+                    print(f"[webdav-sync] Remote not found (404): {self.url}. Skip pull.")
+                    return
+                if h.status_code >= 400:
+                    print(f"[webdav-sync] HEAD failed: {h.status_code} {h.text[:200]}")
+                    return
+                r = s.get(self.url, timeout=30)
+                if r.status_code == 404:
+                    print(f"[webdav-sync] Remote not found on GET: {self.url}.")
+                    return
+                r.raise_for_status()
+                content = r.content
+                # 简单校验是否是 JSON
+                try:
+                    json.loads(content.decode("utf-8"))
+                except Exception:
+                    print("[webdav-sync] Warning: remote content is not valid JSON; still writing as-is.")
+                # 写入本地
+                os.makedirs(os.path.dirname(self.local_path) or ".", exist_ok=True)
+                with open(self.local_path, "wb") as f:
+                    f.write(content)
+                self._last_pushed_hash = _sha1_of_file(self.local_path)
+                print(f"[webdav-sync] Pulled remote -> {self.local_path}")
+        except requests.RequestException as e:
+            print(f"[webdav-sync] Initial pull failed: {e}")
+
+    # ====== Push（周期性，把本地变更上��）======
+    def push_if_changed(self):
+        if not self.enabled:
+            return
+        current_hash = _sha1_of_file(self.local_path)
+        if not current_hash:
+            # 本地没有文件就不 push
+            return
+        if current_hash == self._last_pushed_hash:
+            return
+        try:
+            with self._session() as s, open(self.local_path, "rb") as f:
+                r = s.put(self.url, data=f, timeout=30)
+                if r.status_code >= 400:
+                    print(f"[webdav-sync] PUT failed: {r.status_code} {r.text[:200]}")
+                    return
+                self._last_pushed_hash = current_hash
+                print(f"[webdav-sync] Pushed local -> {self.url}")
+        except requests.RequestException as e:
+            print(f"[webdav-sync] Push failed: {e}")
+
+    def _loop(self):
+        # 周期循环
+        while not self._stop.is_set():
+            try:
+                with self._lock:
+                    self.push_if_changed()
+            except Exception as e:
+                print(f"[webdav-sync] Loop error: {e}")
+            self._stop.wait(self.interval)
+
+    def start(self):
+        if not self.enabled:
+            print("[webdav-sync] disabled; not starting background sync.")
+            return
+        # 进程间锁：只有一个进程会启动线程
+        with _interprocess_lock(LOCKFILE_PATH) as ok:
+            if not ok:
+                print("[webdav-sync] another process is already syncing; skip starting thread.")
+                return
+            # 记录当前 baseline
+            self._last_pushed_hash = _sha1_of_file(self.local_path)
+            self._thread = threading.Thread(target=self._loop, name="WebDavSyncer", daemon=True)
+            self._thread.start()
+            # 注意：离开 with 后锁文件会被释放——这是有意的；
+            # 目的只是避免“同时多个进程创建线程”。创建完成后就算释放，也只会有一个线程已在运行。
+
+    def stop(self):
+        self._stop.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=3)
+
+# ====== 供应用调用的便捷入口 ======
+
+def start_webdav_sync_if_configured(local_history_path: str):
+    """
+    在应用启动时调用：
+      1) 先尝试拉取远端文件覆盖本地
+      2) 启动后台线程每隔 interval 推送本地变更
+    """
+    syncer = WebDavSyncer(local_history_path)
+    # 先拉取一份，保证“新部署前从远端下载到运行目录”的要求
+    syncer.initial_pull()
+    # 再启动定时 push
+    syncer.start()
+    return syncer  # 如需在应用关闭时手动 stop，可保留引用