#!/usr/bin/env python3 """Cross-platform unattended sender. Credentials must be provided explicitly.""" import fnmatch import hashlib import json import logging import os from pathlib import Path import time import urllib.parse import urllib.request SERVER = os.environ.get("TP_URL", "https://s.aisai.cc").rstrip("/") TOKEN = os.environ.get("TP_TOKEN", "") SYNC_DIR = Path(os.environ.get("TP_DIR", str(Path.home() / "TransferSync") if os.name == "nt" else "/usr/filesync")).resolve() PATTERNS = [pattern.strip() for pattern in os.environ.get("TP_INCLUDE", "*").split(",") if pattern.strip()] AUTO_NAS = os.environ.get("TP_AUTO_NAS", "1") REMOTE_DIR = os.environ.get("TP_REMOTE_DIR", "mdsync").strip().strip("/") INTERVAL = max(15, int(os.environ.get("TP_INTERVAL", "1800"))) POLL_INTERVAL = max(5, int(os.environ.get("TP_POLL_INTERVAL", "15"))) JOB_ID = hashlib.sha256(str(SYNC_DIR).encode()).hexdigest()[:12] _state_dir = os.environ.get("STATE_DIRECTORY") or os.environ.get("XDG_STATE_HOME") _default_state = Path(_state_dir) / f".tp_filesync_state_{JOB_ID}.json" if _state_dir else Path.home() / f".tp_filesync_state_{JOB_ID}.json" STATE_FILE = Path(os.environ.get("TP_STATE_FILE", str(_default_state))) class SyncError(Exception): def __init__(self, uploaded, failed): self.uploaded = uploaded super().__init__("%d file(s) failed; %d uploaded" % (failed, uploaded)) def api(path, data=None): headers = {"Authorization": "Bearer " + TOKEN, "User-Agent": "TransferPro-Client/3.0"} if data is not None: headers["Content-Type"] = "application/json" data = json.dumps(data).encode() request = urllib.request.Request(SERVER + path, headers=headers, data=data) with urllib.request.urlopen(request, timeout=30) as response: return json.load(response) def content_hash(path): digest = hashlib.sha256() with path.open("rb") as payload: for chunk in iter(lambda: payload.read(65536), b""): digest.update(chunk) return digest.hexdigest() def remote_relative_path(path): local_rel = path.relative_to(SYNC_DIR).as_posix() if REMOTE_DIR: return f"{REMOTE_DIR}/{local_rel}" return local_rel def upload_file(path, auto_nas, sha256): relative = remote_relative_path(path) existing = api("/api/files/check", {"relativePath": relative, "size": path.stat().st_size, "sha256": sha256, "autoNas": str(auto_nas) == "1"}) if existing.get("duplicate"): logging.info("Already synchronized %s", relative) return False headers = {"Authorization": "Bearer " + TOKEN, "User-Agent": "TransferPro-Client/3.0", "Content-Type": "application/octet-stream", "Content-Length": str(path.stat().st_size), "X-File-Name": urllib.parse.quote(path.name), "X-Relative-Path": urllib.parse.quote(relative), "X-Auto-Nas": str(auto_nas), "X-Content-SHA256": sha256} with path.open("rb") as payload: request = urllib.request.Request(SERVER + "/api/cli/drop", data=payload, headers=headers, method="POST") with urllib.request.urlopen(request, timeout=180) as response: response.read() uploaded = response.headers.get("X-Upload-Deduplicated") != "true" logging.info("Uploaded %s", relative) return uploaded def sync_directory(auto_nas=AUTO_NAS): try: state = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {} if not isinstance(state, dict): state = {} except (ValueError, OSError): state = {} count = 0 failed = 0 for root, directories, names in os.walk(SYNC_DIR, followlinks=False): directories[:] = [name for name in directories if not name.startswith(".") and not (Path(root) / name).is_symlink()] for name in names: path = Path(root) / name relative = path.relative_to(SYNC_DIR).as_posix() if path.is_symlink() or name.startswith(".") or name.startswith("~$") or path.suffix in (".tmp", ".part", ".crdownload"): continue if not any(fnmatch.fnmatch(name, pattern) or fnmatch.fnmatch(relative, pattern) for pattern in PATTERNS): continue try: info = path.stat() if not info.st_size or time.time() - info.st_mtime < 3: continue version = "%s:%s" % (info.st_size, info.st_mtime_ns) previous = state.get(str(path)) if previous == version or isinstance(previous, dict) and previous.get("version") == version: continue sha256 = content_hash(path) if not isinstance(previous, dict) or previous.get("sha256") != sha256: if upload_file(path, auto_nas, sha256): count += 1 state[str(path)] = {"version": version, "sha256": sha256} except Exception as error: failed += 1 logging.error("Upload failed for %s: %s", relative, error) temporary = STATE_FILE.with_suffix(".tmp") STATE_FILE.parent.mkdir(parents=True, exist_ok=True) temporary.write_text(json.dumps(state), encoding="utf-8") os.replace(temporary, STATE_FILE) if failed: raise SyncError(count, failed) return count def report(count, message, status="success"): try: api("/api/vps/trigger", {"action": "report", "status": status, "count": count, "message": message}) except Exception as error: logging.warning("Status report failed: %s", error) def sync_and_report(message, auto_nas=AUTO_NAS, report_unchanged=True): try: count = sync_directory(auto_nas) except Exception as error: logging.error("Sync incomplete: %s", error) report(getattr(error, "uploaded", 0), str(error), "error") return if count or report_unchanged: report(count, message) def main(): if not TOKEN or not SERVER.startswith("https://"): raise SystemExit("Configure HTTPS TP_URL and an authorized TP_TOKEN") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") SYNC_DIR.mkdir(parents=True, exist_ok=True) sync_and_report("Daemon started") last_sync = time.monotonic() consecutive_poll_errors = 0 while True: try: signal = api("/api/vps/trigger") consecutive_poll_errors = 0 if signal.get("trigger"): command = signal.get("command", {}) use_nas = "1" if command.get("autoNas", AUTO_NAS == "1") else "0" sync_and_report("Manual trigger completed", use_nas) last_sync = time.monotonic() except Exception as error: consecutive_poll_errors += 1 if consecutive_poll_errors == 3 or consecutive_poll_errors % 40 == 0: logging.warning("Polling notice (%d consecutive): %s", consecutive_poll_errors, error) if time.monotonic() - last_sync >= INTERVAL: sync_and_report("Scheduled sync completed", report_unchanged=False) last_sync = time.monotonic() time.sleep(POLL_INTERVAL) if __name__ == "__main__": main()