ezshare_-_ezs_sync.py

#!/usr/bin/env python3

###############################################################################################
#ezs_sync.py: syncs files from a EZShare WiFi SDCard to the local directory
#
#Description:
#   syncs the remote filesystem in the EZShare this machine is connected to (via WiFi),to the
#   current local directory, using the device HTTP server; the general idea is to do what a
#   `wget -rN -cp` would do, if the URIs the EZShare exports did not mangle the file names and
#   therefore force us to take them from the anchor text, plus omit directory names from these
#   same URLs and so force us to also handle them, plus special treatment for timestamps in the
#   future and timezone conversion.
#
#Authors:
#   Initial code by Claude.AI: https://claude.ai/share/a98522d6-c5a6-41ac-8426-b4504a37cc25
#   Further debugging, fixing and enhancing: 2025/10/12 Durval Menezes
#   V2.2: performance optimizations (HTTP connection optimization and paralelization of
#         directory scannings and file downloads) plus bug fixes:
#           - Initial code by Claude.AI (Sonnet 4.6 extended):
#               https://claude.ai/share/1541fc98-d3f9-4c59-8310-39050fb76143
#           - Further debugging, fixing and enhancing: 2026/03/24 Durval Menezes
#   V2.3: adaptive concurrency — starts at INITIAL_WORKERS and backs off to MIN_WORKERS
#         on connection/timeout errors.
#   V2.5: fix attempt for ~4-minute hang at end of run (partially correct diagnosis).
#   V2.7: fix stall root cause: urllib3 silently absorbing timeouts via internal retries.
#   V2.8: fix crash on 503: remove status_forcelist; handle HTTPError separately without
#         touching the concurrency limiter.
#   V2.9: two fixes:
#         (a) A single file failure no longer aborts the entire run. _run_parallel()
#             collects per-future exceptions; run() reports all failures at the end and
#             exits with code 1 if any occurred. Every other file still downloads.
#         (b) HTTP errors (e.g. 503 overload) get their own retry budget MAX_HTTP_RETRIES,
#             set higher than MAX_RETRIES because 503s can persist for many seconds on the
#             EZShare while other concurrent downloads succeed.
#   V3.0: final polish:
#         - parse_size, parse_timestamp, set_file_timestamp, file_exists_with_same_attrs
#           converted to @staticmethod (don't use self; pylint fix).
#         - start_time moved from stats dict to self._start_time (stats dict should only
#           hold incrementable counters; start_time is not one).
#         - files_created/files_updated stats now recorded only after a successful
#           download, not before (previously a failed download still incremented them).
#         - size_downloaded initialised to 0 before the download loop (pylint fix for
#           possibly-undefined-variable; the variable is always set before use but the
#           analyser cannot see that through the while/break structure).
#         - _fetch_directory now retries on HTTPError from raise_for_status(), matching
#           download_file behaviour (a 503 on a directory listing no longer crashes
#           phase 1).
#         - IOError replaced with OSError (IOError is a deprecated alias in Python 3).
#         - Simplified UTC_OFFSET calculation (no need to name the intermediate).
#   V3.2: pylint3 compliance and final performance optimization:
#         - eliminated the easiest pylint3 warnings.
#         - replaced exponential backoff with linear backoff capped at MAX_WAIT,
#           reducing worst-case retry delay and variance across runs.
#   V3.3: consistency fix — _fetch_directory was still using the old exponential
#         backoff formula (2 ** attempts) after V3.2 changed download_file to linear
#         capped backoff. Both now use the same _retry_wait() helper, guaranteeing
#         identical behaviour and a single place to tune the formula.
#         Also: phase 1 executor max_workers capped at INITIAL_WORKERS (previously
#         it could spin up len(batch) threads even when batch < INITIAL_WORKERS,
#         creating more threads than the concurrency limiter would ever allow through).
#   V3.4: two bug fixes:
#         (a) DST/timezone bug: UTC_OFFSET was computed once at import time. When DST
#             changed (GMT-3 → GMT-4), the offset changed but all previously-downloaded
#             files on disk still had mtimes set with the old offset, causing every
#             file to show "timestamps differ by 3600.0 s" and be re-downloaded.
#             Fix: parse_timestamp now returns a UTC-aware datetime (tzinfo=timezone.utc)
#             instead of a naive local-time datetime. file_exists_with_same_attrs now
#             compares POSIX timestamps (stat.st_mtime vs timestamp.timestamp()) directly,
#             which are always UTC-based and DST-invariant. UTC_OFFSET is removed.
#         (b) KeyboardInterrupt not killing the process cleanly: Ctrl-C raised
#             KeyboardInterrupt in as_completed(), but Python's atexit handler registered
#             by ThreadPoolExecutor calls t.join() on all threads it ever created, so
#             the process waited for in-flight downloads to complete before exiting.
#             Fix: _run_parallel() catches KeyboardInterrupt and cancels all pending
#             (not-yet-started) futures before re-raising. main() catches the re-raised
#             KeyboardInterrupt and calls os._exit(130) to bypass atexit entirely,
#             letting daemon threads die with the process immediately.
#   V3.5: fix: directory scan failures were logged but not counted, not shown in the
#         summary, and did not affect the exit code — so a total failure to connect
#         to the card reported "DOWNLOAD COMPLETED" and exited 0. Added sdirs_failed
#         stat counter, updated summary header to distinguish scan-only failures from
#         mixed/download failures, and included the new counter in run()'s return value.
#   Link to Claude AI conversation up to this point: still the same URL as V2.2 above, ie:
#          https://claude.ai/share/1541fc98-d3f9-4c59-8310-39050fb76143
#License:
#   WTFPL: https://www.wtfpl.net/about/
###############################################################################################

"""
HTTP Directory Downloader
Downloads files recursively from an HTTP directory listing, preserving timestamps
and checking file integrity, plus trying to be as robust as possible re: errors.

Performance architecture:
  Phase 1 - BFS directory tree in parallel batches, collecting download tasks
  Phase 2 - Execute all pending downloads in parallel; failures are collected and
            reported at the end — a single file failure never aborts the run.

Concurrency:
  Both phases share an AdaptiveConcurrencyLimiter that starts at INITIAL_WORKERS
  concurrent HTTP connections and decrements by 1 (down to MIN_WORKERS) each time
  the remote device returns a connection or timeout error.

Retry architecture (two layers — must not both be set high):
  urllib3 layer  — URLLIB3_RETRIES=0: raises immediately on any network error or
                   timeout. No internal retries. No status_forcelist (with total=0
                   that would cause MaxRetryError on 503 before our code sees it).
  Our layer      — download_file() has two separate retry budgets:
                   * MAX_RETRIES for connection/timeout (_BACKOFF_EXCEPTIONS):
                     also decrements the concurrency limiter (TCP-level overload).
                   * MAX_HTTP_RETRIES for HTTP errors (e.g. 503): retries without
                     touching the limiter; set higher because 503s can persist for
                     many seconds on the EZShare while other downloads succeed.
"""

import os
import re
import sys
import time
import shutil
import logging
import threading
from collections import deque
from contextlib import contextmanager
from datetime import datetime, timezone
from urllib.parse import urljoin
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
LOGGER = logging.getLogger(__name__)

# pylint: disable=bad-whitespace
MAX_RETRIES      = 5   # Retry budget for connection/timeout errors.
MAX_HTTP_RETRIES = 20  # Retry budget for HTTP errors (503 etc.). Higher because the
                       # EZShare can 503 for many seconds while other downloads proceed.
MAX_WAIT         = 2   # Maximum wait between retries
URLLIB3_RETRIES  = 0   # urllib3 internal retries. Must be 0: see module docstring.
INITIAL_WORKERS  = 6   # Starting concurrency.
MIN_WORKERS      = 2   # Floor for concurrency backoff.
REQUEST_TIMEOUT  = 30  # Seconds per SESSION.get() call hard ceiling.
CHUNK_SIZE       = 65536  # 64 KB read chunks.
# pylint: enable=bad-whitespace

_FILE_PATTERN = re.compile(
    r'(\d{4}-\s*\d{1,2}-\s*\d{1,2}\s+\d{1,2}:\s*\d{1,2}:\s*\d{1,2})'
    r'\s+(\d+(?:KB|MB|GB|B)?)\s*<a href="([^"]+)">\s*([^<]+)</a>'
)
_DIR_PATTERN = re.compile(
    r'(\d{4}-\s*\d{1,2}-\s*\d{1,2}\s+\d{1,2}:\s*\d{1,2}:\s*\d{1,2})'
    r'\s+\&lt;DIR\&gt;\s*<a href="([^"]+)">\s*([^<]+)</a>'
)
_TIMESTAMP_PATTERN = re.compile(
    r'(\d{4})-\s*(\d{1,2})-\s*(\d{1,2})\s+(\d{1,2}):\s*(\d{1,2}):\s*(\d{1,2})'
)

# Errors that indicate TCP-level overload: retry AND decrement the concurrency limiter.
# requests.exceptions.ConnectionError normally wraps the underlying socket exceptions,
# but with URLLIB3_RETRIES=0 some may surface unwrapped, so we list both layers.
_BACKOFF_EXCEPTIONS = (
    requests.exceptions.ConnectionError,
    requests.exceptions.Timeout,
    ConnectionRefusedError,
    ConnectionResetError,
    TimeoutError,
)


def requests_retry_session(psession=None):
    """Return a session with urllib3 retry disabled and keep-alive off.

    URLLIB3_RETRIES=0: any error surfaces immediately to our retry loop.
    No status_forcelist: with total=0, that causes MaxRetryError on 503 before
    our code sees the response; we handle HTTP errors ourselves.
    Connection:close: prevents pooled sockets from hanging on teardown against
    the EZShare's unresponsive TCP stack.
    """
    session = psession or requests.Session()
    session.headers.update({'Connection': 'close'})
    retry = Retry(
        total=URLLIB3_RETRIES,
        connect=URLLIB3_RETRIES,
        read=URLLIB3_RETRIES,
        backoff_factor=0,
        method_whitelist=frozenset(['GET', 'POST']),  # Ensure POST is included
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session


SESSION = requests_retry_session()


def _retry_wait(attempt):
    """Return the number of seconds to sleep before retry attempt N (1-based).

    Linear growth capped at MAX_WAIT: 2, 2, 2, ... seconds.
    Keeping the cap low ensures the total wait across MAX_HTTP_RETRIES retries
    stays bounded; the device typically recovers within one or two retries anyway.
    """
    return min(attempt + 1, MAX_WAIT)


def _run_parallel(tasks, work_fn, max_workers, label=""):
    """Submit tasks, collect results and failures, shut down without waiting.

    Exceptions from individual futures are collected rather than re-raised,
    so one failing task never prevents the others from completing.

    Returns (results, failures) where:
      results  = list of (task, return_value) for successful futures
      failures = list of (task, exception)    for failed futures
    """
    executor = ThreadPoolExecutor(max_workers=max_workers,
                                  thread_name_prefix=f"ezs-{label}")
    try:
        future_map = {executor.submit(work_fn, task): task for task in tasks}
        results  = []
        failures = []
        for future in as_completed(future_map):
            task = future_map[future]
            exc  = future.exception()
            if exc is not None:
                failures.append((task, exc))
            else:
                results.append((task, future.result()))
        return results, failures
    except KeyboardInterrupt:
        # Cancel futures that haven't started yet so no new work begins.
        # Already-running futures cannot be interrupted mid-flight, but they
        # will be abandoned when os._exit() is called in main().
        for future in future_map:
            future.cancel()
        raise
    finally:
        # All futures are done at this point (as_completed exhausted).
        # wait=False avoids joining threads that may be blocked in socket
        # cleanup against the EZShare's unresponsive TCP stack.
        executor.shutdown(wait=False)


class AdaptiveConcurrencyLimiter:
    """Limits concurrent HTTP connections with runtime backoff on errors.

    Worker threads wrap their SESSION.get() calls with the connection()
    context manager.  On TCP-level failure they call on_error(), which
    decrements the active limit by 1 (floor: minimum).
    """

    def __init__(self, initial, minimum):
        # pylint: disable=bad-whitespace
        self._limit    = initial
        self._minimum  = minimum
        self._active   = 0
        self._backoffs = 0
        self._cond     = threading.Condition(threading.Lock())
        # pylint: enable=bad-whitespace

    @contextmanager
    def connection(self):
        """Acquire one connection slot, yield, then release and wake waiters."""
        with self._cond:
            while self._active >= self._limit:
                self._cond.wait()
            self._active += 1
        try:
            yield
        finally:
            with self._cond:
                self._active -= 1
                self._cond.notify_all()

    def on_error(self):
        """Decrement the limit by 1 on a connection/timeout error (floor: minimum)."""
        with self._cond:
            if self._limit > self._minimum:
                self._limit -= 1
                self._backoffs += 1
                LOGGER.warning("Concurrency backoff #%d: reducing limit to %d (min=%d)",
                               self._backoffs, self._limit, self._minimum)
            else:
                LOGGER.debug("Connection error at minimum concurrency (%d); not reducing",
                             self._minimum)

    @property
    def current_limit(self):
        """Current active concurrency limit."""
        with self._cond:
            return self._limit

    @property
    def backoff_count(self):
        """Number of times the limit has been decremented."""
        with self._cond:
            return self._backoffs


class EZShareDirectoryDownloader:
    """EZShare recursive directory downloader."""

    def __init__(self, base_url, base_dir='.'):
        # pylint: disable=bad-whitespace
        self.base_url    = base_url
        self.base_dir    = os.path.abspath(base_dir)
        self._start_time = time.time()
        self._stats_lock = threading.Lock()
        self._limiter    = AdaptiveConcurrencyLimiter(initial=INITIAL_WORKERS,
                                                      minimum=MIN_WORKERS)
        # pylint: enable=bad-whitespace

        self.stats = {
            'sdirs_checked':    0,
            'sdirs_created':    0,
            'sdirs_failed':     0,
            'files_checked':    0,
            'files_created':    0,
            'files_updated':    0,
            'bytes_checked':    0,
            'files_downloaded': 0,
            'bytes_downloaded': 0,
            'files_failed':     0,
        }

    def _update_stats(self, **kwargs):
        """Atomically increment one or more stat counters. Thread-safe."""
        with self._stats_lock:
            for key, value in kwargs.items():
                self.stats[key] += value

    # ------------------------------------------------------------------
    # Parsing helpers
    # ------------------------------------------------------------------

    @staticmethod
    def parse_size(size_str):
        """Convert size string (e.g. '7750KB') to (bytes, unit_granularity) tuple."""
        size_str = size_str.strip().upper()
        if size_str.endswith('KB'):
            return int(size_str[:-2]) * 1024, 1024
        if size_str.endswith('MB'):
            return int(float(size_str[:-2]) * 1024 * 1024), 1024 * 1024
        if size_str.endswith('GB'):
            return int(float(size_str[:-2]) * 1024 * 1024 * 1024), 1024 * 1024 * 1024
        if size_str.endswith('B'):
            return int(size_str[:-1]), 1
        return int(size_str), 1

    @staticmethod
    def parse_timestamp(timestamp_str):
        """Parse an EZShare timestamp string to a UTC-aware datetime.

        Handles the padded-space format: '2021-11- 6   16:17: 0'
        The EZShare serves timestamps in UTC. Returning a timezone-aware datetime
        (tzinfo=timezone.utc) means comparisons and POSIX conversions are correct
        regardless of the local machine's DST state at the time of the call.
        Note: a small number of files (e.g. Journal.dat) are served with non-UTC
        timestamps; those will be re-downloaded on each run, which is acceptable.
        """
        match = _TIMESTAMP_PATTERN.search(timestamp_str.strip())
        if not match:
            raise ValueError(f"Cannot parse timestamp: {timestamp_str!r}")
        year, month, day, hour, minute, second = (int(x) for x in match.groups())
        return datetime(year, month, day, hour, minute, second, tzinfo=timezone.utc)

    # ------------------------------------------------------------------
    # File attribute checks
    # ------------------------------------------------------------------

    @staticmethod
    def set_file_timestamp(filepath, timestamp):
        """Set the modification time of a local file to match the remote."""
        mod_time = timestamp.timestamp()
        os.utime(filepath, (mod_time, mod_time))

    @staticmethod
    def file_exists_with_same_attrs(filepath, size_bytes, size_tolerance, timestamp):
        """Return True if the local file exists and matches the remote size and timestamp.

        Timestamps are compared as POSIX values (seconds since epoch) so DST changes
        on the local machine are irrelevant — both sides are always UTC-based.
        A file whose remote timestamp is in the future is always treated as changed
        (handles the Resmed AirSense Journal.dat edge case where timestamp=2030 but
        content changes).
        """
        LOGGER.debug("checking file %s", filepath)
        if not os.path.exists(filepath):
            LOGGER.debug("local file does not exist")
            return False
        stat = os.stat(filepath)
        if abs(stat.st_size - size_bytes) > size_tolerance:
            LOGGER.debug("size differs by more than %d byte tolerance", size_tolerance)
            return False
        if timestamp > datetime.now(timezone.utc):
            LOGGER.debug("remote timestamp is in the future: treating as changed")
            return False
        # Compare as POSIX timestamps: stat.st_mtime is seconds-since-epoch (UTC-based),
        # and timestamp.timestamp() on a UTC-aware datetime is also seconds-since-epoch.
        time_diff = abs(stat.st_mtime - timestamp.timestamp())
        if time_diff > 1:
            LOGGER.debug("timestamps differ by %.1f s (> 1 s tolerance)", time_diff)
            return False
        LOGGER.debug("local and remote files match; skipping")
        return True

    # ------------------------------------------------------------------
    # Downloading
    # ------------------------------------------------------------------

    def download_file(self, url, filepath, size_expected, size_tolerance):
        """Download url to filepath and verify its size.

        Two separate retry budgets:
          _BACKOFF_EXCEPTIONS (connection/timeout): up to MAX_RETRIES attempts;
            also decrements the concurrency limiter (TCP-level overload).
          HTTPError (e.g. 503): up to MAX_HTTP_RETRIES attempts; does NOT touch
            the limiter — server overload is not a concurrency problem.
        """
        LOGGER.info("Downloading: %s -> %s", url, filepath)

        # pylint: disable=bad-whitespace
        conn_attempts  = 0
        http_attempts  = 0
        size_downloaded = 0  # initialised here so the post-loop check always sees it
        # pylint: enable=bad-whitespace

        while True:
            try:
                with self._limiter.connection():
                    with SESSION.get(url, stream=True,
                                     timeout=REQUEST_TIMEOUT) as response:
                        response.raise_for_status()
                        size_downloaded = 0
                        with open(filepath, 'wb') as file_handle:
                            for chunk in response.iter_content(chunk_size=CHUNK_SIZE):
                                if chunk:
                                    file_handle.write(chunk)
                                    size_downloaded += len(chunk)
                break  # success

            except _BACKOFF_EXCEPTIONS as exc:
                conn_attempts += 1
                self._limiter.on_error()
                LOGGER.warning("Connection error (attempt %d/%d): %s",
                               conn_attempts, MAX_RETRIES, exc)
                if conn_attempts >= MAX_RETRIES:
                    LOGGER.error("No connection retries left, aborting: %s", filepath)
                    raise
                wait = _retry_wait(conn_attempts)
                LOGGER.info("Retrying in %d s", wait)
                time.sleep(wait)

            except requests.exceptions.HTTPError as exc:
                http_attempts += 1
                status = exc.response.status_code if exc.response is not None else '?'
                LOGGER.warning("HTTP %s error (attempt %d/%d): %s",
                               status, http_attempts, MAX_HTTP_RETRIES, exc)
                if http_attempts >= MAX_HTTP_RETRIES:
                    LOGGER.error("No HTTP retries left, aborting: %s", filepath)
                    raise
                wait = _retry_wait(http_attempts)
                LOGGER.info("Retrying in %d s", wait)
                time.sleep(wait)

        if abs(size_downloaded - size_expected) > size_tolerance:
            raise OSError(f"Size mismatch for {filepath}: "
                          f"expected {size_expected}, got {size_downloaded}")

        self._update_stats(files_downloaded=1, bytes_downloaded=size_downloaded)
        return size_downloaded

    def _execute_download_task(self, task):
        """Execute a single pending download task. Thread-safe.
        task = (timestamp, size_bytes, size_tolerance, full_url, filepath)

        files_created/files_updated are recorded only after a successful download
        so that failed tasks do not inflate those counters.
        Exceptions propagate to _run_parallel, which collects them without
        aborting other concurrent tasks.
        """
        timestamp, size_bytes, size_tolerance, full_url, filepath = task
        if os.path.isdir(filepath):
            LOGGER.info("path %s exists as a directory, removing", filepath)
            shutil.rmtree(filepath)
        file_existed = os.path.exists(filepath)
        self.download_file(full_url, filepath, size_bytes, size_tolerance)
        self.set_file_timestamp(filepath, timestamp)
        # Count only after success so failures don't inflate these stats.
        if file_existed:
            self._update_stats(files_updated=1)
        else:
            self._update_stats(files_created=1)

    # ------------------------------------------------------------------
    # Directory crawling
    # ------------------------------------------------------------------

    def _fetch_directory(self, task):
        """Fetch one directory listing. task = (url, local_dir).

        Returns (pending_downloads, subdir_tasks). Acquires one concurrency slot
        for the HTTP fetch. Retries on both connection errors (with limiter backoff)
        and HTTP errors (without limiter backoff), matching download_file behaviour.
        Thread-safe.
        """
        url, local_dir = task
        local_dir = os.path.abspath(local_dir)

        if os.path.isfile(local_dir):
            LOGGER.info("path %s exists as a file, removing", local_dir)
            os.remove(local_dir)

        dir_already_existed = os.path.exists(local_dir)
        os.makedirs(local_dir, exist_ok=True)
        self._update_stats(sdirs_checked=1,
                           sdirs_created=0 if dir_already_existed else 1)

        LOGGER.debug("fetching directory listing: %s", url)

        conn_attempts = 0
        http_attempts = 0
        html = None

        while html is None:
            try:
                with self._limiter.connection():
                    response = SESSION.get(url, timeout=REQUEST_TIMEOUT)
                    response.raise_for_status()
                    html = response.text

            except _BACKOFF_EXCEPTIONS as exc:
                conn_attempts += 1
                self._limiter.on_error()
                LOGGER.warning("Connection error fetching %s (attempt %d/%d): %s",
                               url, conn_attempts, MAX_RETRIES, exc)
                if conn_attempts >= MAX_RETRIES:
                    raise OSError(
                        f"Failed to fetch directory listing {url} "
                        f"after {MAX_RETRIES} attempts: {exc}"
                    ) from exc
                time.sleep(_retry_wait(conn_attempts))

            except requests.exceptions.HTTPError as exc:
                http_attempts += 1
                status = exc.response.status_code if exc.response is not None else '?'
                LOGGER.warning("HTTP %s fetching %s (attempt %d/%d): %s",
                               status, url, http_attempts, MAX_HTTP_RETRIES, exc)
                if http_attempts >= MAX_HTTP_RETRIES:
                    raise OSError(
                        f"Failed to fetch directory listing {url} "
                        f"after {MAX_HTTP_RETRIES} HTTP attempts: {exc}"
                    ) from exc
                time.sleep(_retry_wait(http_attempts))

        pending_downloads = []
        for match in _FILE_PATTERN.finditer(html):
            timestamp_str, size_str, file_url, filename = match.groups()
            try:
                # pylint: disable=bad-whitespace
                timestamp              = self.parse_timestamp(timestamp_str)
                size_bytes, size_units = self.parse_size(size_str)
                size_tolerance         = size_units
                filepath               = os.path.join(local_dir, filename)
                # pylint: enable=bad-whitespace
                self._update_stats(files_checked=1, bytes_checked=size_bytes)
                if self.file_exists_with_same_attrs(filepath, size_bytes,
                                                    size_tolerance, timestamp):
                    LOGGER.debug("unchanged, skipping: %s", filename)
                    continue
                full_url = urljoin(url, file_url)
                pending_downloads.append(
                    (timestamp, size_bytes, size_tolerance, full_url, filepath))
            except Exception as exc:  # pylint: disable=broad-except
                LOGGER.error("Error parsing file entry '%s': %s", filename, exc)
                raise

        subdir_tasks = []
        for match in _DIR_PATTERN.finditer(html):
            _, dir_uri, sdirname = match.groups()
            if sdirname in ('.', '..'):
                LOGGER.debug("skipping pseudo-directory '%s'", sdirname)
                continue
            subdir_tasks.append((urljoin(url, dir_uri),
                                 os.path.join(local_dir, sdirname)))

        return pending_downloads, subdir_tasks

    # ------------------------------------------------------------------
    # Main execution
    # ------------------------------------------------------------------

    def run(self):
        """Two-phase execution:
          Phase 1 — BFS the remote directory tree in parallel batches.
          Phase 2 — Download all pending files in parallel; per-file failures are
                    collected and reported at the end rather than aborting the run.

        Returns True if all files succeeded, False if any failed.
        """
        LOGGER.info("Starting download from: %s (initial concurrency: %d, min: %d)",
                    self.base_url, INITIAL_WORKERS, MIN_WORKERS)

        # Phase 1: parallel BFS directory crawl
        all_download_tasks = []
        pending_dirs = deque([(self.base_url, self.base_dir)])
        while pending_dirs:
            batch = []
            while pending_dirs and len(batch) < INITIAL_WORKERS:
                batch.append(pending_dirs.popleft())
            results, failures = _run_parallel(
                batch, self._fetch_directory,
                max_workers=min(len(batch), INITIAL_WORKERS), label="phase1")
            for _, (download_tasks, subdir_tasks) in results:
                all_download_tasks.extend(download_tasks)
                pending_dirs.extend(subdir_tasks)
            for task, exc in failures:
                LOGGER.error("Failed to scan directory %s: %s", task[0], exc)
            self._update_stats(sdirs_failed=len(failures))

        LOGGER.info("Directory scan complete: %d directories checked, %d failed, %d files to download.",
                    self.stats['sdirs_checked'], self.stats['sdirs_failed'], len(all_download_tasks))

        # Phase 2: parallel file downloads — collect failures, never abort early
        download_failures = []
        if all_download_tasks:
            _, download_failures = _run_parallel(
                all_download_tasks, self._execute_download_task,
                max_workers=INITIAL_WORKERS, label="phase2")
            self._update_stats(files_failed=len(download_failures))
            for task, exc in download_failures:
                _, _, _, url, filepath = task
                LOGGER.error("FAILED: %s -> %s: %s", url, filepath, exc)

        elapsed = time.time() - self._start_time
        any_failed = download_failures or self.stats['sdirs_failed']
        print("\n" + "=" * 60)
        if not any_failed:
            print("DOWNLOAD COMPLETED")
        elif self.stats['sdirs_failed'] and not download_failures:
            print(f"SCAN FAILED: {self.stats['sdirs_failed']} director(ies) could not be listed")
        else:
            total = len(download_failures) + self.stats['sdirs_failed']
            print(f"DOWNLOAD COMPLETED WITH {total} FAILURE(S)")
        print("=" * 60)
        print(f"Time elapsed:          {elapsed:.2f} seconds")
        print(f"Concurrency (start):   {INITIAL_WORKERS}")
        print(f"Concurrency (final):   {self._limiter.current_limit}")
        print(f"Concurrency backoffs:  {self._limiter.backoff_count}")
        print(f"Directories checked:   {self.stats['sdirs_checked']}")
        print(f"Directories created:   {self.stats['sdirs_created']}")
        print(f"Directories failed:    {self.stats['sdirs_failed']}")
        print(f"Files checked:         {self.stats['files_checked']}")
        print(f"Files created:         {self.stats['files_created']}")
        print(f"Files updated:         {self.stats['files_updated']}")
        print(f"Files failed:          {self.stats['files_failed']}")
        print(f"Bytes checked:         {self.stats['bytes_checked']:,} bytes")
        print(f"Files downloaded:      {self.stats['files_downloaded']}")
        print(f"Bytes downloaded:      {self.stats['bytes_downloaded']:,} bytes")
        print("=" * 60)

        return not any_failed


def main():
    """Main entry point. Exits with code 1 if any file failed to download."""
    url = 'http://192.168.4.1/dir?dir=A:'
    downloader = EZShareDirectoryDownloader(url)
    try:
        success = downloader.run()
        sys.exit(0 if success else 1)
    except KeyboardInterrupt:
        LOGGER.warning("Interrupted.")
        # os._exit bypasses Python's atexit handlers, including the one registered
        # by ThreadPoolExecutor that calls t.join() on every thread it created.
        # Without this, Ctrl-C would still wait for all in-flight downloads to
        # finish before the process exited. 130 = 128 + SIGINT, the conventional
        # exit code for a process killed by Ctrl-C.
        os._exit(130)  # pylint: disable=protected-access


if __name__ == "__main__":
    main()

# Eof ezs_sync.py