WebDAV Concurrent Synchronization Script
📝 Python
User Safety: safe
Python
#!/usr/bin/env python3
# retoor <retoor@molodetz.nl>
"""as.py - asynchronous WebDAV synchronizer.
Walks a local source tree and mirrors it onto a remote WebDAV collection,
uploading many files concurrently. Re-running the script resumes naturally:
files already present remotely with a matching byte size are skipped.
Every individual file or directory failure is caught, logged and skipped so a
single unreadable or unreachable entry never aborts the whole run.
"""
from __future__ import annotations
import argparse
import asyncio
import base64
import logging
import os
import random
import sys
import time
from dataclasses import dataclass, field
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Optional
from urllib.parse import quote
import aiofiles
import aiohttp
from xml.etree import ElementTree as ET
DAV_NS = "DAV:"
PROG = "as.py"
VERSION = "1.0.0"
DEFAULT_HOST = "xtc-makes-you-free-sub1.your-storagebox.de"
DEFAULT_USER = "lololo-sub1"
DEFAULT_PASSWORD = "lalala*"
DEFAULT_CONCURRENCY = 16
DEFAULT_TIMEOUT_SECONDS = 60
DEFAULT_CHUNK_BYTES = 1024 * 1024
DEFAULT_RETRIES = 3
DEFAULT_BACKOFF_BASE = 1.0
DEFAULT_BACKOFF_MAX = 30.0
DEFAULT_BACKOFF_JITTER = 0.5
DEFAULT_SCAN_TIMEOUT = 10.0
DEFAULT_LOG_FILE = "as.log"
DEFAULT_LOG_BYTES = 10 * 1024 * 1024
DEFAULT_LOG_BACKUPS = 5
LOG_FORMAT = "%(asctime)s %(levelname)-8s [%(name)s] %(message)s"
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
logger = logging.getLogger("as")
def _basic_auth_header(user: str, password: str) -> str:
"""Build an HTTP Basic Authorization header value.
Encoded with base64 directly so it works on every aiohttp version.
"""
token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
return f"Basic {token}"
@dataclass
class Stats:
uploaded: int = 0
skipped: int = 0
created_dirs: int = 0
failed_files: int = 0
failed_dirs: int = 0
inaccessible_files: int = 0
bytes_uploaded: int = 0
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def bump(self, **kwargs: int) -> None:
async with self.lock:
for key, value in kwargs.items():
setattr(self, key, getattr(self, key) + value)
class WebDavClient:
"""Thin async wrapper around the WebDAV verbs we need."""
def __init__(self, host: str, user: str, password: str,
timeout_seconds: int, session: aiohttp.ClientSession,
retries: int = DEFAULT_RETRIES,
backoff_base: float = DEFAULT_BACKOFF_BASE,
backoff_max: float = DEFAULT_BACKOFF_MAX,
backoff_jitter: float = DEFAULT_BACKOFF_JITTER) -> None:
self.base = f"https://{host}"
self.auth_header = {
"Authorization": _basic_auth_header(user, password),
}
self.timeout = aiohttp.ClientTimeout(total=timeout_seconds)
self.session = session
self.retries = retries
self.backoff_base = backoff_base
self.backoff_max = backoff_max
self.backoff_jitter = backoff_jitter
def url_for(self, remote_path: str, directory: bool = False) -> str:
cleaned = remote_path if remote_path.startswith("/") else f"/{remote_path}"
encoded = "/".join(quote(part, safe="") for part in cleaned.split("/"))
if directory and not encoded.endswith("/"):
encoded += "/"
return f"{self.base}{encoded}"
@staticmethod
def _is_retryable_status(status: int) -> bool:
"""HTTP statuses that indicate a transient failure worth retrying."""
return status in (408, 425, 429, 500, 502, 503, 504)
@staticmethod
def _is_retryable_error(exc: BaseException) -> bool:
"""Network-level failures worth retrying on a flaky connection."""
if isinstance(exc, asyncio.TimeoutError):
return True
if isinstance(exc, aiohttp.ClientConnectionError):
return True
if isinstance(exc, aiohttp.ServerDisconnectedError):
return True
if isinstance(exc, aiohttp.ClientPayloadError):
return True
if isinstance(exc, OSError):
return True
return False
def _backoff_delay(self, attempt: int) -> float:
"""Exponential backoff with jitter, capped at a sane maximum."""
exponent = min(attempt - 1, 10)
base = min(self.backoff_base * (2 ** exponent), self.backoff_max)
if self.backoff_jitter <= 0:
return base
return random.uniform(base * (1 - self.backoff_jitter),
base * (1 + self.backoff_jitter))
async def _request_with_retry(self, method: str, url: str,
headers: dict, data=None) -> aiohttp.ClientResponse:
"""Issue a WebDAV request, retrying transient failures with backoff.
Returns the final response. Permanent HTTP errors (4xx other than the
retryable set) and permanent local errors are surfaced immediately.
"""
attempts = self.retries + 1
last_exc: Optional[BaseException] = None
last_status: Optional[int] = None
for attempt in range(1, attempts + 1):
try:
resp = await self.session.request(
method, url, timeout=self.timeout, headers=headers, data=data,
)
if not self._is_retryable_status(resp.status):
return resp
last_status = resp.status
await resp.release()
logger.warning(
"%s %s -> HTTP %s (attempt %d/%d, retrying)",
method, url, resp.status, attempt, attempts,
)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
if not self._is_retryable_error(exc):
raise
last_exc = exc
logger.warning(
"%s %s failed: %s (attempt %d/%d, retrying)",
method, url, exc, attempt, attempts,
)
if attempt < attempts:
await asyncio.sleep(self._backoff_delay(attempt))
if last_status is not None:
raise aiohttp.ClientResponseError(
None, (), status=last_status,
message=f"HTTP {last_status} after retries",
)
raise last_exc if last_exc is not None else RuntimeError("request failed")
async def propfind(self, remote_path: str) -> Optional[dict]:
"""Return {child_name: size_bytes} for a collection, or None if missing."""
url = self.url_for(remote_path, directory=True)
body = (
'<?xml version="1.0" encoding="utf-8"?>'
'<d:propfind xmlns:d="DAV:">'
"<d:prop><d:resourcetype/><d:getcontentlength/></d:prop>"
"</d:propfind>"
)
headers = {"Depth": "1", "Content-Type": "application/xml"}
headers.update(self.auth_header)
try:
resp = await self._request_with_retry(
"PROPFIND", url, headers=headers, data=body,
)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
logger.error("PROPFIND %s failed after retries: %s", remote_path, exc)
return None
try:
async with resp:
if resp.status in (404, 405):
return None
if resp.status != 207:
logger.warning("PROPFIND %s -> HTTP %s", remote_path, resp.status)
return None
text = await resp.text()
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
logger.error("PROPFIND %s body read failed: %s", remote_path, exc)
return None
return self._parse_multistatus(text)
@staticmethod
def _parse_multistatus(text: str) -> dict:
children: dict = {}
try:
root = ET.fromstring(text)
except ET.ParseError as exc:
logger.warning("Could not parse PROPFIND response: %s", exc)
return children
for response in root.iter(f"{{{DAV_NS}}}response"):
href_el = response.find(f"{{{DAV_NS}}}href")
if href_el is None or not href_el.text:
continue
href = href_el.text.strip()
name = href.rstrip("/").rsplit("/", 1)[-1]
if not name:
continue
is_collection = response.find(
f".//{{{DAV_NS}}}resourcetype/{{{DAV_NS}}}collection"
) is not None
if is_collection:
children[name] = -1
continue
length_el = response.find(f".//{{{DAV_NS}}}getcontentlength")
try:
children[name] = int(length_el.text) if length_el is not None else -1
except (TypeError, ValueError):
children[name] = -1
return children
async def mkcol(self, remote_path: str) -> bool:
url = self.url_for(remote_path, directory=True)
try:
resp = await self._request_with_retry(
"MKCOL", url, headers=self.auth_header,
)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
logger.error("MKCOL %s failed after retries: %s", remote_path, exc)
return False
async with resp:
if resp.status in (201, 405):
return True
logger.warning("MKCOL %s -> HTTP %s", remote_path, resp.status)
return False
async def put_file(self, remote_path: str, local_path: Path,
chunk_bytes: int) -> bool:
url = self.url_for(remote_path)
attempts = self.retries + 1
for attempt in range(1, attempts + 1):
try:
handle = await aiofiles.open(local_path, "rb")
except (OSError, PermissionError) as exc:
logger.error("Cannot open local file %s: %s", local_path, exc)
raise
async def stream_body():
try:
while True:
chunk = await handle.read(chunk_bytes)
if not chunk:
break
yield chunk
finally:
await handle.close()
retryable = False
try:
resp = await self.session.put(
url, timeout=self.timeout, data=stream_body(),
headers=self.auth_header,
)
if resp.status in (200, 201, 204):
await resp.release()
return True
retryable = self._is_retryable_status(resp.status)
logger.warning(
"PUT %s -> HTTP %s (attempt %d/%d)",
remote_path, resp.status, attempt, attempts,
)
await resp.release()
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
retryable = self._is_retryable_error(exc)
logger.warning(
"PUT %s failed: %s (attempt %d/%d)",
remote_path, exc, attempt, attempts,
)
finally:
if not handle.closed:
await handle.close()
if not retryable:
return False
if attempt < attempts:
await asyncio.sleep(self._backoff_delay(attempt))
return False
@dataclass
class DirTask:
rel: str
priority: int = 0
@dataclass
class FileTask:
rel: str
local: Path
size: Optional[int]
priority: int = 0
@dataclass
class DirInfo:
rel: str
local: Path
subdirs: list
files: list
size: int
UNKNOWN_SIZE = -1
UNKNOWN_PRIORITY = 1 << 62
_STALLED = object()
def _list_entries(path: Path):
"""Return a sorted list of child paths, or None if the dir is unreadable."""
try:
return sorted(path.iterdir(), key=lambda p: p.name)
except (OSError, PermissionError) as exc:
logger.error("Cannot read local dir %s: %s", path, exc)
return None
def _inspect_entry(path: Path):
"""Return (kind, size) for an entry, never raising on a broken FS.
kind is 'dir', 'file' or None; size is the byte count or None when it
cannot be determined.
"""
try:
if path.is_dir():
return ("dir", None)
except (OSError, PermissionError):
pass
try:
if path.is_file():
try:
return ("file", path.stat().st_size)
except (OSError, PermissionError, ValueError):
return ("file", None)
except (OSError, PermissionError):
pass
return (None, None)
async def _run_blocking(fn, timeout: float, *args):
"""Run a blocking filesystem call in a thread, time-boxed.
Returns the call's result, or the _STALLED sentinel if it does not finish
within `timeout` seconds. A stalled call never aborts the scan.
"""
try:
return await asyncio.wait_for(
asyncio.to_thread(fn, *args), timeout=timeout,
)
except asyncio.TimeoutError:
return _STALLED
async def _scan_dir(local_dir: Path, rel: str, manifest: dict,
op_timeout: float) -> int:
"""Recursively walk a directory, filling the manifest.
Returns the best-effort total size in bytes of this subtree. Every
filesystem interaction is individually time-boxed and guarded, so a single
unreadable or stalled directory is logged and skipped while the rest of the
tree continues to be scanned.
"""
result = await _run_blocking(_list_entries, op_timeout, local_dir)
if result is _STALLED:
logger.error("Timed out reading dir %s; skipping its subtree",
local_dir)
manifest[rel] = DirInfo(rel=rel, local=local_dir, subdirs=[], files=[],
size=0)
return 0
if result is None:
manifest[rel] = DirInfo(rel=rel, local=local_dir, subdirs=[], files=[],
size=0)
return 0
entries = result
subdirs: list = []
files: list = []
total = 0
for entry in entries:
name = entry.name
child_rel = f"{rel}/{name}" if rel else name
inspected = await _run_blocking(_inspect_entry, op_timeout, entry)
if inspected is _STALLED:
logger.error("Timed out inspecting %s; skipping it", entry)
continue
kind, size = inspected
if kind == "dir":
child_size = await _scan_dir(entry, child_rel, manifest, op_timeout)
subdirs.append((name, child_size))
total += child_size
elif kind == "file":
files.append((name, entry, size))
if size is not None:
total += size
else:
logger.warning("Skipping non-file/non-dir entry %s", entry)
manifest[rel] = DirInfo(rel=rel, local=local_dir, subdirs=subdirs,
files=files, size=total)
return total
class Synchronizer:
def __init__(self, source: Path, remote_root: str, client: WebDavClient,
concurrency: int, chunk_bytes: int, dry_run: bool,
stats: Stats, scan_timeout: float = 300.0) -> None:
self.source = source
self.remote_root = remote_root.rstrip("/")
self.client = client
self.concurrency = concurrency
self.chunk_bytes = chunk_bytes
self.dry_run = dry_run
self.stats = stats
self.scan_timeout = scan_timeout
self.queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self.manifest: dict = {}
self.seq = 0
self.pending = 0
self.pending_lock = asyncio.Lock()
self.stop_event = asyncio.Event()
def remote_path(self, rel: str) -> str:
return f"{self.remote_root}/{rel}" if rel else self.remote_root
def _next_seq(self) -> int:
self.seq += 1
return self.seq
async def _put(self, task) -> None:
"""Enqueue a task ordered by size (smallest first)."""
priority = task.priority if task.priority >= 0 else UNKNOWN_PRIORITY
await self.queue.put((priority, self._next_seq(), task))
async with self.pending_lock:
self.pending += 1
async def ensure_root(self) -> bool:
parts = [p for p in self.remote_root.split("/") if p]
current = ""
for part in parts:
current = f"{current}/{part}"
existing = await self.client.propfind(current)
if existing is None:
if self.dry_run:
logger.info("DRY-RUN would MKCOL %s", current)
else:
ok = await self.client.mkcol(current)
if ok:
await self.stats.bump(created_dirs=1)
logger.info("Created remote dir %s", current)
else:
logger.error("Failed to create remote dir %s", current)
return False
else:
logger.debug("Remote dir exists %s", current)
return True
async def scan_and_enqueue(self) -> bool:
"""Walk the local tree into a manifest, then enqueue the root dir.
Returns True if the root task was enqueued, False if the source root
itself could not be scanned. Individual unreadable or stalled
subdirectories are logged and skipped without aborting the run.
"""
logger.info("Scanning source tree for sizes...")
scanned: dict = {}
await _scan_dir(self.source, "", scanned, self.scan_timeout)
self.manifest = scanned
root = self.manifest.get("")
if root is None:
logger.error("Source could not be scanned: %s", self.source)
return False
logger.info("Scan complete: %d directories, %d bytes total",
len(self.manifest), root.size)
await self._put(DirTask(rel="", priority=root.size))
return True
async def process_dir(self, task: DirTask) -> None:
info = self.manifest.get(task.rel)
if info is None:
logger.error("No manifest entry for dir %s", task.rel)
await self.stats.bump(failed_dirs=1)
return
remote_dir = self.remote_path(task.rel)
remote_children: dict = {}
if not self.dry_run:
remote_children = await self.client.propfind(remote_dir) or {}
else:
logger.debug("DRY-RUN skip PROPFIND %s", remote_dir)
for name, child_size in info.subdirs:
rel = f"{task.rel}/{name}" if task.rel else name
if name not in remote_children:
if self.dry_run:
logger.info("DRY-RUN would MKCOL %s", self.remote_path(rel))
await self.stats.bump(created_dirs=1)
else:
ok = await self.client.mkcol(self.remote_path(rel))
if ok:
await self.stats.bump(created_dirs=1)
logger.info("Created remote dir %s", self.remote_path(rel))
else:
logger.error("Failed to create remote dir %s",
self.remote_path(rel))
await self.stats.bump(failed_dirs=1)
continue
await self._put(DirTask(rel=rel, priority=child_size))
for name, local, size in info.files:
rel = f"{task.rel}/{name}" if task.rel else name
remote_size = remote_children.get(name)
if size is not None and remote_size == size:
await self.stats.bump(skipped=1)
logger.debug("Skip (size match) %s", rel)
else:
priority = size if size is not None else UNKNOWN_SIZE
await self._put(FileTask(rel=rel, local=local, size=size,
priority=priority))
async def process_file(self, task: FileTask) -> None:
remote = self.remote_path(task.rel)
if self.dry_run:
label = task.size if task.size is not None else "unknown"
logger.info("DRY-RUN would upload %s (%s bytes)", task.rel, label)
await self.stats.bump(uploaded=1)
if task.size is not None:
await self.stats.bump(bytes_uploaded=task.size)
return
try:
ok = await self.client.put_file(remote, task.local, self.chunk_bytes)
except (OSError, PermissionError) as exc:
logger.error("Cannot read local file %s: %s", task.local, exc)
await self.stats.bump(inaccessible_files=1)
return
if ok:
await self.stats.bump(uploaded=1)
if task.size is not None:
await self.stats.bump(bytes_uploaded=task.size)
logger.info("Uploaded %s (%s bytes)", task.rel,
task.size if task.size is not None else "unknown")
else:
await self.stats.bump(failed_files=1)
logger.error("Failed to upload %s", task.rel)
async def worker(self, worker_id: int) -> None:
while True:
_, _, task = await self.queue.get()
try:
if isinstance(task, DirTask):
logger.debug("Worker %d processing dir %s", worker_id, task.rel)
await self.process_dir(task)
elif isinstance(task, FileTask):
logger.debug("Worker %d processing file %s", worker_id, task.rel)
await self.process_file(task)
elif task is None:
return
except Exception as exc: # noqa: BLE001 - never let a task kill a worker
logger.exception("Unexpected error processing %r: %s", task, exc)
finally:
self.queue.task_done()
async with self.pending_lock:
self.pending -= 1
if self.pending <= 0:
self.pending = 0
self.stop_event.set()
async def run(self) -> Stats:
logger.info("Source: %s", self.source)
logger.info("Remote root: %s", self.remote_path(""))
if not self.source.is_dir():
logger.error("Source is not a readable directory: %s", self.source)
return self.stats
if not self.dry_run:
if not await self.ensure_root():
logger.error("Could not ensure remote root exists; aborting.")
return self.stats
if not await self.scan_and_enqueue():
logger.error("Source scan failed; aborting.")
return self.stats
workers = [
asyncio.create_task(self.worker(i)) for i in range(self.concurrency)
]
await self.stop_event.wait()
for _ in workers:
await self.queue.put((0, self._next_seq(), None))
await asyncio.gather(*workers)
return self.stats
def positive_int(value: str) -> int:
"""Argparse type: require an integer strictly greater than zero."""
try:
parsed = int(value)
except ValueError:
raise argparse.ArgumentTypeError(f"'{value}' is not a valid integer")
if parsed < 1:
raise argparse.ArgumentTypeError(f"'{value}' must be >= 1")
return parsed
def non_negative_float(value: str) -> float:
"""Argparse type: require a non-negative number."""
try:
parsed = float(value)
except ValueError:
raise argparse.ArgumentTypeError(f"'{value}' is not a valid number")
if parsed < 0:
raise argparse.ArgumentTypeError(f"'{value}' must be >= 0")
return parsed
def jitter_fraction(value: str) -> float:
"""Argparse type: require a fraction between 0.0 and 1.0 inclusive."""
parsed = non_negative_float(value)
if parsed > 1.0:
raise argparse.ArgumentTypeError(f"'{value}' must be <= 1.0")
return parsed
def log_level(value: str) -> int:
"""Argparse type: accept a logging level name or numeric value."""
normalized = value.upper()
level = getattr(logging, normalized, None)
if isinstance(level, int):
return level
try:
return int(value)
except ValueError:
raise argparse.ArgumentTypeError(
f"'{value}' is not a valid log level "
"(use DEBUG, INFO, WARNING, ERROR, CRITICAL or a number)"
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog=PROG,
description=(
"Asynchronously mirror a local directory tree onto a remote WebDAV "
"collection, uploading many files concurrently.\n\n"
"The remote destination is created recursively when it does not "
"exist. Re-running the tool resumes naturally: files already "
"present remotely with a matching byte size are skipped, so an "
"interrupted run can simply be started again to finish the job.\n\n"
"Every individual file or directory failure is caught, logged and "
"skipped, so a single unreadable or unreachable entry never aborts "
"the whole run."
),
epilog=(
"examples:\n"
" %(prog)s /home/user/photos --remote /backup/photos\n"
" Mirror the local 'photos' directory into /backup/photos.\n\n"
" %(prog)s /data --remote /backup/data --concurrency 32 \\\n"
" --timeout 120 --verbose\n"
" Mirror /data with 32 parallel uploads, a 120s per-request\n"
" timeout and DEBUG-level logging.\n\n"
" %(prog)s /data --remote /backup/data --dry-run\n"
" Report what would be uploaded without touching the remote.\n\n"
"Credentials default to the configured Hetzner Storage Box and can "
"be overridden with --host/--user/--password or the AS_HOST, "
"AS_USER and AS_PASSWORD environment variables."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
add_help=True,
)
parser.add_argument(
"--version", action="version", version=f"%(prog)s {VERSION}",
help="Show the program version and exit.",
)
source_group = parser.add_argument_group(
"source and destination",
"What to copy and where to put it.",
)
source_group.add_argument(
"source", type=Path, metavar="SOURCE",
help="Local source directory to mirror onto the remote. Must be an "
"existing, readable directory.",
)
source_group.add_argument(
"--remote", required=True, metavar="PATH",
help="Remote destination path on the storage box, e.g. /backup/mydata. "
"Leading slash is optional. Any missing parent collections are "
"created recursively.",
)
connection_group = parser.add_argument_group(
"connection",
"WebDAV server credentials and endpoint. All default to the "
"configured Hetzner Storage Box.",
)
connection_group.add_argument(
"--host", default=DEFAULT_HOST, metavar="HOST",
help=f"Storage box hostname. (default: {DEFAULT_HOST})",
)
connection_group.add_argument(
"--user", default=DEFAULT_USER, metavar="USER",
help=f"WebDAV username. (default: {DEFAULT_USER})",
)
connection_group.add_argument(
"--password", default=DEFAULT_PASSWORD, metavar="PASSWORD",
help="WebDAV password. Prefer the AS_PASSWORD environment variable "
"over the command line to avoid exposing it in process listings.",
)
transfer_group = parser.add_argument_group(
"transfer",
"Tuning for throughput, reliability and concurrency.",
)
transfer_group.add_argument(
"--concurrency", type=positive_int, default=DEFAULT_CONCURRENCY,
metavar="N",
help=f"Maximum number of files uploaded in parallel. Higher values "
f"increase throughput but use more connections and memory. "
f"(default: {DEFAULT_CONCURRENCY})",
)
transfer_group.add_argument(
"--timeout", type=positive_int, default=DEFAULT_TIMEOUT_SECONDS,
metavar="SECONDS",
help=f"Per-request timeout in seconds. A request that exceeds this is "
f"aborted and the file is retried or skipped rather than hanging "
f"forever. (default: {DEFAULT_TIMEOUT_SECONDS})",
)
transfer_group.add_argument(
"--chunk", type=positive_int, default=DEFAULT_CHUNK_BYTES,
metavar="BYTES",
help=f"Size in bytes of each streaming read/write chunk while "
f"uploading. Files are streamed in chunks, never loaded whole "
f"into memory. (default: {DEFAULT_CHUNK_BYTES})",
)
transfer_group.add_argument(
"--retries", type=positive_int, default=DEFAULT_RETRIES,
metavar="N",
help=f"Number of times to retry a request that fails due to a transient "
f"network error or a 5xx/429 server response before giving up on "
f"that item. Applies to directory listings, directory creation and "
f"file uploads. Permanent errors (e.g. 401, 403, 404) are never "
f"retried. (default: {DEFAULT_RETRIES})",
)
transfer_group.add_argument(
"--backoff-base", type=non_negative_float, default=DEFAULT_BACKOFF_BASE,
metavar="SECONDS",
help=f"Base delay in seconds for the first retry. Each subsequent "
f"retry doubles the delay, up to --backoff-max. "
f"(default: {DEFAULT_BACKOFF_BASE})",
)
transfer_group.add_argument(
"--backoff-max", type=non_negative_float, default=DEFAULT_BACKOFF_MAX,
metavar="SECONDS",
help=f"Upper bound in seconds for the retry delay. "
f"(default: {DEFAULT_BACKOFF_MAX})",
)
transfer_group.add_argument(
"--backoff-jitter", type=jitter_fraction, default=DEFAULT_BACKOFF_JITTER,
metavar="FRACTION",
help=f"Random jitter applied to each retry delay, as a fraction of the "
f"delay, to avoid many concurrent workers retrying in lockstep. "
f"0 disables jitter, 1.0 allows up to double the delay. "
f"(default: {DEFAULT_BACKOFF_JITTER})",
)
transfer_group.add_argument(
"--scan-timeout", type=non_negative_float, default=DEFAULT_SCAN_TIMEOUT,
metavar="SECONDS",
help=f"Maximum seconds to wait on a single filesystem operation "
f"(listing a directory, classifying an entry, reading a size) "
f"during the source scan before giving up on that item. A "
f"stalled or unreadable directory is logged and skipped while "
f"the rest of the tree is still scanned. "
f"(default: {DEFAULT_SCAN_TIMEOUT})",
)
logging_group = parser.add_argument_group(
"logging",
"Control where and how much is logged.",
)
logging_group.add_argument(
"--log", type=Path, default=Path(DEFAULT_LOG_FILE), metavar="FILE",
help=f"Write a rotating log file to this path. "
f"(default: {DEFAULT_LOG_FILE})",
)
logging_group.add_argument(
"--log-level", type=log_level, default=None, metavar="LEVEL",
help="Console log verbosity: DEBUG, INFO, WARNING, ERROR or CRITICAL. "
"The log file always records DEBUG. (default: INFO)",
)
logging_group.add_argument(
"--verbose", action="store_true",
help="Shorthand for --log-level DEBUG. Enables per-file and per-worker "
"diagnostic logging.",
)
logging_group.add_argument(
"--quiet", action="store_true",
help="Suppress INFO logging on the console; only warnings and errors "
"are shown.",
)
behavior_group = parser.add_argument_group(
"behavior",
"Optional runtime behaviour.",
)
behavior_group.add_argument(
"--dry-run", action="store_true",
help="Do not contact the remote for writes. Walk the source tree and "
"report which directories would be created and which files would "
"be uploaded, without uploading anything.",
)
return parser
def configure_logging(log_path: Path, console_level: int) -> None:
root = logging.getLogger()
root.setLevel(logging.DEBUG)
formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
console = logging.StreamHandler(sys.stdout)
console.setLevel(console_level)
console.setFormatter(formatter)
root.addHandler(console)
try:
file_handler = RotatingFileHandler(
log_path, maxBytes=DEFAULT_LOG_BYTES, backupCount=DEFAULT_LOG_BACKUPS,
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
root.addHandler(file_handler)
except OSError as exc:
logger.warning("Could not open log file %s: %s", log_path, exc)
async def amain(args: argparse.Namespace) -> int:
if args.verbose:
console_level = logging.DEBUG
elif args.quiet:
console_level = logging.WARNING
else:
console_level = args.log_level if args.log_level is not None else logging.INFO
configure_logging(args.log, console_level)
stats = Stats()
timeout = aiohttp.ClientTimeout(total=args.timeout)
connector = aiohttp.TCPConnector(limit=args.concurrency, limit_per_host=args.concurrency)
async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
client = WebDavClient(
args.host, args.user, args.password, args.timeout, session,
retries=args.retries,
backoff_base=args.backoff_base,
backoff_max=args.backoff_max,
backoff_jitter=args.backoff_jitter,
)
sync = Synchronizer(args.source, args.remote, client,
args.concurrency, args.chunk, args.dry_run, stats,
scan_timeout=args.scan_timeout)
started = time.monotonic()
result = await sync.run()
elapsed = time.monotonic() - started
logger.info("=" * 60)
logger.info("Sync finished in %.2fs", elapsed)
logger.info("Directories created : %d", result.created_dirs)
logger.info("Files uploaded : %d", result.uploaded)
logger.info("Files skipped : %d", result.skipped)
logger.info("Bytes uploaded : %d", result.bytes_uploaded)
logger.info("Failed files : %d", result.failed_files)
logger.info("Failed dirs : %d", result.failed_dirs)
logger.info("Inaccessible files : %d", result.inaccessible_files)
logger.info("=" * 60)
return 0
def main() -> int:
parser = build_parser()
args = parser.parse_args()
if args.host == DEFAULT_HOST and os.environ.get("AS_HOST"):
args.host = os.environ["AS_HOST"]
if args.user == DEFAULT_USER and os.environ.get("AS_USER"):
args.user = os.environ["AS_USER"]
if args.password == DEFAULT_PASSWORD and os.environ.get("AS_PASSWORD"):
args.password = os.environ["AS_PASSWORD"]
try:
return asyncio.run(amain(args))
except KeyboardInterrupt:
print("\nInterrupted by user.", file=sys.stderr)
return 130
if __name__ == "__main__":
sys.exit(main())
Comments
Got already stabbed from behind. Fucker, I explicitly stated that the file system was corrupt and now i get this shit, does the thing even listen?
I just updated the script and hope it will do correctly now. Code you see above is the latest version. You can ignore error above.