fileshare.py

"""FileShare - LAN file sharing with system tray UI.

Each PC runs this program. It hosts one or more share folders over HTTP on a
local port, and pulls every peer's share folders into per-peer subfolders under
`download_folder` — each shared folder appears as its own subdirectory beneath
the peer's name (downloads/<peer>/<folder>/...). No authentication. Peer list
and folders are in config.json; share_folder may be a single string or a list
of paths, and any relative path is resolved against the script/exe directory.
Ignore patterns (gitignore-like) are in .fileshareignore. Sync is one-way pull.

Run modes (auto-selected by OS, override with --tray / --headless):
  Windows  -> tray:     `pythonw fileshare.py` (or start.bat) shows a tray icon.
  Linux/*  -> headless:  `python3 fileshare.py` runs as a daemon (no UI). Stop
                         with SIGTERM/Ctrl-C; SIGUSR1 toggles pause, SIGUSR2
                         syncs now. Intended to run under systemd.
"""

import fnmatch
import gzip
import hashlib
import json
import os
import socket
import subprocess
import sys
import threading
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from logging import FileHandler, Formatter, INFO, getLogger
from pathlib import Path

# The tray UI (pystray + Pillow) is optional: it's only needed in tray mode,
# which is the Windows default. Headless mode (the Linux/server default) runs
# fine without these, so don't hard-fail at import time.
try:
    import pystray
    from PIL import Image, ImageDraw
    HAVE_TRAY = True
except ImportError:
    pystray = None
    Image = ImageDraw = None
    HAVE_TRAY = False


if getattr(sys, "frozen", False):
    SCRIPT_DIR = Path(sys.executable).parent.resolve()
else:
    SCRIPT_DIR = Path(__file__).parent.resolve()
CONFIG_PATH = SCRIPT_DIR / "config.json"
IGNORE_PATH = SCRIPT_DIR / ".fileshareignore"
LOG_DIR = SCRIPT_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)


def _make_logger(name: str, filename: str):
    logger = getLogger(name)
    logger.setLevel(INFO)
    if not logger.handlers:
        h = FileHandler(LOG_DIR / filename, encoding="utf-8")
        h.setFormatter(Formatter("%(asctime)s [%(levelname)s] %(message)s"))
        logger.addHandler(h)
    logger.propagate = False
    return logger


sync_log = _make_logger("fileshare.sync", "sync.log")
access_log = _make_logger("fileshare.access", "access.log")


DEFAULT_CONFIG = {
    # share_folder may be a single path string OR a list of paths. Paths can be
    # absolute or relative; relative paths are anchored to the script/exe
    # directory. Each entry's basename becomes its share name and must be unique.
    "share_folder": ["share"],
    "download_folder": "downloads",
    "port": 8765,
    "sync_interval_seconds": 10,
    "request_timeout_seconds": 5,
    "peers": [
        {"ip": "192.168.0.10", "name": "alice"},
        {"ip": "192.168.0.11", "name": "bob"}
    ]
}


def resolve_path(raw: str) -> Path:
    """Resolve a config path: relative paths are anchored to SCRIPT_DIR so the
    program behaves the same regardless of which cwd it was launched from.
    """
    p = Path(raw)
    if not p.is_absolute():
        p = SCRIPT_DIR / p
    return p.resolve()


class DuplicateShareFolderError(ValueError):
    """Two share-folder entries resolve to the same basename."""


def compute_share_folders(cfg: dict) -> list:
    """Return [(name, resolved_path), ...] for the configured share folders.

    Accepts ``share_folder`` as either a string (legacy single-folder form) or
    a list of strings. Raises DuplicateShareFolderError if two entries share a
    basename, since that name is the public identifier exposed to peers and
    used as the subfolder under each peer's downloads/<peer>/ directory.
    """
    raw = cfg.get("share_folder")
    if isinstance(raw, str):
        items = [raw]
    elif isinstance(raw, list):
        items = [x for x in raw if isinstance(x, str)]
    else:
        items = []
    result = []
    seen: dict = {}
    for item in items:
        item = item.strip()
        if not item:
            continue
        path = resolve_path(item)
        name = path.name
        if not name:
            continue
        if name in seen:
            raise DuplicateShareFolderError(
                f"Duplicate share folder name '{name}': "
                f"'{seen[name]}' and '{path}'. Each share folder must have a "
                f"unique basename — rename one of the folders or place them in "
                f"distinct parent directories with different names."
            )
        seen[name] = path
        result.append((name, path))
    return result


def show_fatal_error(msg: str):
    """Surface a fatal startup error in a way the user will actually see.

    Since the program normally runs via pythonw (no console), a stderr message
    is invisible. Use a Windows MessageBox as the primary channel.
    """
    sync_log.error("FATAL: %s", msg)
    if sys.platform == "win32":
        try:
            import ctypes
            ctypes.windll.user32.MessageBoxW(
                0, msg, "FileShare — configuration error", 0x10
            )
            return
        except Exception:
            pass
    sys.stderr.write(msg + "\n")

DEFAULT_IGNORE = """# FileShare ignore patterns (gitignore-style).
# Lines starting with # are comments. Trailing / marks a directory pattern.
# These apply to YOUR share folder before serving it to peers.

*.tmp
*.swp
*.part
~$*
Thumbs.db
.DS_Store
desktop.ini
__pycache__/
node_modules/
.git/
.venv/
"""


def load_config() -> dict:
    if not CONFIG_PATH.exists():
        CONFIG_PATH.write_text(
            json.dumps(DEFAULT_CONFIG, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )
    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


def load_ignore_patterns() -> list:
    if not IGNORE_PATH.exists():
        IGNORE_PATH.write_text(DEFAULT_IGNORE, encoding="utf-8")
    patterns = []
    with open(IGNORE_PATH, "r", encoding="utf-8") as f:
        for raw in f:
            line = raw.strip()
            if line and not line.startswith("#"):
                patterns.append(line)
    return patterns


def is_ignored(rel_path: str, patterns: list) -> bool:
    rel = rel_path.replace("\\", "/")
    parts = rel.split("/")
    name = parts[-1] if parts else rel
    for pat in patterns:
        if pat.endswith("/"):
            dir_pat = pat[:-1].lstrip("/")
            if any(fnmatch.fnmatch(p, dir_pat) for p in parts[:-1]):
                return True
        else:
            anchored = pat.startswith("/")
            p = pat.lstrip("/")
            if anchored:
                if fnmatch.fnmatch(rel, p):
                    return True
            else:
                if fnmatch.fnmatch(name, p) or fnmatch.fnmatch(rel, p):
                    return True
                if any(fnmatch.fnmatch(part, p) for part in parts):
                    return True
    return False


class HashCache:
    """Persistent SHA-256 cache keyed by absolute path.

    `(size, mtime)` is only a hint for skipping rehashing — the stored hash
    is always the source of truth. Wrong mtime causes a rehash, never a
    miscompare.
    """

    CHUNK = 1 << 20  # 1 MiB

    def __init__(self, cache_path: Path):
        self.cache_path = cache_path
        self.lock = threading.Lock()
        self.data = self._load()
        self._dirty = False

    def _load(self) -> dict:
        if not self.cache_path.exists():
            return {}
        try:
            return json.loads(self.cache_path.read_text(encoding="utf-8"))
        except Exception:
            return {}

    def save(self):
        with self.lock:
            if not self._dirty:
                return
            data_copy = dict(self.data)
            self._dirty = False
        try:
            tmp = self.cache_path.with_suffix(".tmp")
            tmp.write_text(json.dumps(data_copy, separators=(",", ":")), encoding="utf-8")
            tmp.replace(self.cache_path)
        except OSError as e:
            sync_log.error("Hash cache save failed: %s", e)
            with self.lock:
                self._dirty = True

    def hash_for(self, full_path: Path) -> str:
        key = str(full_path)
        try:
            st = full_path.stat()
        except OSError:
            return ""
        with self.lock:
            entry = self.data.get(key)
            if entry and entry[0] == st.st_size and abs(entry[1] - st.st_mtime) < 0.001:
                return entry[2]
        h = hashlib.sha256()
        try:
            with open(full_path, "rb") as f:
                while True:
                    chunk = f.read(self.CHUNK)
                    if not chunk:
                        break
                    h.update(chunk)
        except OSError:
            return ""
        digest = h.hexdigest()
        with self.lock:
            self.data[key] = [st.st_size, st.st_mtime, digest]
            self._dirty = True
        return digest

    def set(self, full_path: Path, file_hash: str):
        try:
            st = full_path.stat()
        except OSError:
            return
        with self.lock:
            self.data[str(full_path)] = [st.st_size, st.st_mtime, file_hash]
            self._dirty = True


class ListingCache:
    """Caches the /list response so it's rebuilt only when the share tree changes.

    Every request still does a cheap walk to build a signature over each served
    file's (path, size, mtime). The expensive work — hashing each file, JSON
    serialization, and gzip compression — happens only when that signature
    differs from the last build, so an unchanged tree costs just the walk.
    Pass ``force=True`` (manual "Sync now") to rebuild unconditionally.
    """

    def __init__(self):
        self.lock = threading.Lock()
        self._sig = None
        self._body = json.dumps({"files": []}).encode("utf-8")
        self._gzip = None  # gzip of _body, built lazily

    @staticmethod
    def _scan(share_folders, patterns):
        """Walk all share folders (applying ignore rules). Returns
        ``(entries, signature)`` where entries is a sorted list of
        ``(advertised_path, size, mtime, full_path)``."""
        entries = []
        for folder_name, share_root in share_folders:
            if not share_root.exists():
                continue
            for root, dirs, filenames in os.walk(share_root):
                rel_root = Path(root).relative_to(share_root)
                dirs[:] = [
                    d for d in dirs
                    if not is_ignored(
                        (rel_root / d).as_posix() if str(rel_root) != "." else d,
                        patterns,
                    )
                ]
                for fn in filenames:
                    rel = fn if str(rel_root) == "." else (rel_root / fn).as_posix()
                    if is_ignored(rel, patterns):
                        continue
                    full = Path(root) / fn
                    try:
                        st = full.stat()
                    except OSError:
                        continue
                    entries.append((f"{folder_name}/{rel}", st.st_size, st.st_mtime, full))
        entries.sort(key=lambda e: e[0])
        # Signature captures adds, deletes, renames, in-place edits, and
        # ignore-pattern changes (which alter the served entry set).
        hasher = hashlib.sha256()
        for path, size, mtime, _ in entries:
            hasher.update(f"{path}\x00{size}\x00{mtime!r}\n".encode("utf-8"))
        return entries, hasher.hexdigest()

    def _rebuild(self, entries, sig, hash_cache):
        # Hashing happens outside the lock (it can be slow on a cold cache).
        files = []
        for adv_path, size, mtime, full in entries:
            file_hash = hash_cache.hash_for(full) if hash_cache else ""
            if not file_hash:
                continue
            files.append({"path": adv_path, "size": size, "mtime": mtime, "hash": file_hash})
        body = json.dumps({"files": files}).encode("utf-8")
        with self.lock:
            self._sig = sig
            self._body = body
            self._gzip = None
        return len(files)

    def response(self, share_folders, patterns, hash_cache, want_gzip, force=False):
        """Return ``(data, is_gzip, file_count_or_None)`` for /list, rebuilding
        only when the tree changed or ``force`` is set."""
        entries, sig = self._scan(share_folders, patterns)
        count = None
        with self.lock:
            stale = force or sig != self._sig
        if stale:
            count = self._rebuild(entries, sig, hash_cache)
        with self.lock:
            if want_gzip:
                if self._gzip is None:
                    self._gzip = gzip.compress(self._body, compresslevel=6)
                return self._gzip, True, count
            return self._body, False, count


def fmt_bytes(n: int) -> str:
    if n is None:
        return "?"
    if n < 1024:
        return f"{n} B"
    v = float(n)
    for unit in ("KB", "MB", "GB", "TB"):
        v /= 1024
        if v < 1024:
            return f"{v:.1f} {unit}"
    return f"{v:.1f} PB"


class AppState:
    PEER_DEFAULT = {
        "ip": "", "online": False, "syncing": False, "paused": False,
        "last_sync": "-", "last_count": 0,
        "bytes_total": 0, "bytes_done": 0,
        "files_total": 0, "files_done": 0,
        "current_file": "",
    }

    def __init__(self):
        self.lock = threading.Lock()
        self.peer_status = {}
        self.config = {}
        self.ignore_patterns = []
        self.shutdown = threading.Event()
        self.local_ip = "?"
        self.hash_cache: "HashCache | None" = None
        self.listing_cache = ListingCache()
        self.shared_paused = False
        # List of (folder_name, resolved_path) tuples for the configured share
        # folders. Recomputed whenever config.json changes (rejecting reloads
        # that would introduce duplicate names).
        self.share_folders: list = []

    def toggle_share_paused(self) -> bool:
        with self.lock:
            self.shared_paused = not self.shared_paused
            return self.shared_paused

    def update_peer(self, name: str, **kwargs):
        with self.lock:
            if name not in self.peer_status:
                self.peer_status[name] = dict(self.PEER_DEFAULT)
            self.peer_status[name].update(kwargs)

    def add_peer_bytes(self, name: str, n: int):
        with self.lock:
            if name in self.peer_status:
                self.peer_status[name]["bytes_done"] += n

    def prune_peers(self, keep_names: set):
        with self.lock:
            for n in list(self.peer_status.keys()):
                if n not in keep_names:
                    del self.peer_status[n]

    def snapshot(self) -> dict:
        with self.lock:
            return {k: dict(v) for k, v in self.peer_status.items()}


state = AppState()


# ---------------------------------------------------------------------------
# HTTP server: serves /list, /file, /ping from share_folder
# ---------------------------------------------------------------------------
class ShareHandler(BaseHTTPRequestHandler):
    server_version = "FileShare/1.0"

    def log_message(self, format, *args):
        return  # silence stderr

    def do_GET(self):
        try:
            parsed = urllib.parse.urlparse(self.path)
            if parsed.path == "/ping":
                return self._send_text(200, "pong")
            if parsed.path == "/list":
                return self._handle_list(parsed)
            if parsed.path == "/file":
                return self._handle_file(parsed)
            self.send_error(404, "not found")
        except (ConnectionResetError, BrokenPipeError):
            pass
        except Exception as e:
            sync_log.exception("Server error: %s", e)
            try:
                self.send_error(500, str(e))
            except Exception:
                pass

    def _send_text(self, code: int, body: str):
        data = body.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _handle_list(self, parsed):
        if state.shared_paused:
            body = json.dumps({"files": [], "paused": True}).encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            access_log.info("LIST from %s (paused, 0 files)", self.client_address[0])
            return
        # Manual "Sync now" requests bypass the cache with ?fresh=1 so the
        # listing is rebuilt even when the tree looks unchanged.
        qs = urllib.parse.parse_qs(parsed.query)
        force = qs.get("fresh", ["0"])[0] in ("1", "true", "yes")
        want_gzip = "gzip" in self.headers.get("Accept-Encoding", "").lower()
        data, is_gzip, count = state.listing_cache.response(
            state.share_folders, state.ignore_patterns, state.hash_cache,
            want_gzip=want_gzip, force=force,
        )
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        if is_gzip:
            self.send_header("Content-Encoding", "gzip")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)
        access_log.info(
            "LIST from %s (%d bytes%s%s%s)",
            self.client_address[0], len(data),
            ", gzip" if is_gzip else "",
            ", fresh" if force else "",
            "" if count is None else f", rebuilt {count} files",
        )

    def _handle_file(self, parsed):
        if state.shared_paused:
            return self.send_error(503, "sharing paused")
        qs = urllib.parse.parse_qs(parsed.query)
        rel = qs.get("path", [""])[0]
        if not rel:
            return self.send_error(400, "missing path")
        # The first path component selects which configured share folder to
        # serve from — matches the prefix advertised by /list.
        rel_norm = rel.replace("\\", "/").lstrip("/")
        if "/" not in rel_norm:
            return self.send_error(400, "missing folder prefix")
        folder_name, inner = rel_norm.split("/", 1)
        share_root = None
        for n, p in state.share_folders:
            if n == folder_name:
                share_root = p
                break
        if share_root is None:
            return self.send_error(404, "share folder not found")
        target = (share_root / inner).resolve()
        try:
            target.relative_to(share_root)
        except ValueError:
            return self.send_error(403, "forbidden")
        if is_ignored(inner, state.ignore_patterns):
            return self.send_error(403, "ignored")
        if not target.is_file():
            return self.send_error(404, "not found")
        st = target.stat()
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Length", str(st.st_size))
        self.send_header("X-MTime", repr(st.st_mtime))
        self.end_headers()
        with open(target, "rb") as f:
            while True:
                chunk = f.read(65536)
                if not chunk:
                    break
                try:
                    self.wfile.write(chunk)
                except (ConnectionResetError, BrokenPipeError):
                    return
        access_log.info(
            "GET '%s' by %s (%d bytes)",
            rel, self.client_address[0], st.st_size,
        )


def run_server():
    port = state.config.get("port", 8765)
    try:
        srv = ThreadingHTTPServer(("0.0.0.0", port), ShareHandler)
    except OSError as e:
        sync_log.error("Server bind failed on port %d: %s", port, e)
        return
    sync_log.info("Server listening on 0.0.0.0:%d", port)
    try:
        srv.serve_forever(poll_interval=0.5)
    except Exception:
        sync_log.exception("Server crashed")


# ---------------------------------------------------------------------------
# Sync client: pulls each peer's share into downloads/<name>/
# ---------------------------------------------------------------------------
def sync_peer(peer: dict, download_root: Path, port: int, timeout: int,
              force_fresh: bool = False):
    ip = peer.get("ip", "").strip()
    if not ip:
        return
    name = (peer.get("name") or ip).strip()
    base = f"http://{ip}:{port}"
    peer_dir = download_root / name
    peer_dir.mkdir(parents=True, exist_ok=True)
    state.update_peer(
        name, ip=ip, syncing=True,
        bytes_total=0, bytes_done=0,
        files_total=0, files_done=0,
        current_file="",
    )

    try:
        # /list can be slow when the peer has a cold hash cache (each uncached
        # file is hashed inline before the response is sent), so give it a
        # generous floor — much longer than the default request_timeout used
        # for cheap calls like /ping. fresh=1 (manual Sync now) asks the peer to
        # rebuild its listing even if its tree looks unchanged. Accept-Encoding
        # opts into the peer's gzipped listing (older peers just ignore it).
        list_url = f"{base}/list" + ("?fresh=1" if force_fresh else "")
        req = urllib.request.Request(list_url, headers={"Accept-Encoding": "gzip"})
        with urllib.request.urlopen(req, timeout=max(timeout, 60)) as resp:
            raw = resp.read()
            if resp.headers.get("Content-Encoding", "").lower() == "gzip":
                raw = gzip.decompress(raw)
            listing = json.loads(raw.decode("utf-8"))
    except (urllib.error.URLError, socket.timeout, ConnectionError, OSError) as e:
        state.update_peer(name, online=False, syncing=False)
        sync_log.info("Peer %s (%s) offline: %s", name, ip, e)
        return
    except Exception as e:
        state.update_peer(name, online=False, syncing=False)
        sync_log.exception("List failed for %s (%s): %s", name, ip, e)
        return

    peer_paused = bool(listing.get("paused", False))
    state.update_peer(name, online=True, paused=peer_paused)
    files = listing.get("files", [])

    # First pass: decide what needs pulling (hash-based) so we can show total size.
    pending = []
    for f in files:
        rel = f.get("path", "")
        if not rel:
            continue
        remote_hash = f.get("hash", "")
        remote_size = int(f.get("size", 0))
        remote_mtime = float(f.get("mtime", 0))
        local = peer_dir / rel
        if not local.exists():
            pending.append((rel, remote_hash, remote_size, remote_mtime, local))
            continue
        if remote_hash:
            local_hash = state.hash_cache.hash_for(local) if state.hash_cache else ""
            if local_hash != remote_hash:
                pending.append((rel, remote_hash, remote_size, remote_mtime, local))
        else:
            # No hash advertised — fall back to size only.
            try:
                if local.stat().st_size != remote_size:
                    pending.append((rel, remote_hash, remote_size, remote_mtime, local))
            except OSError:
                pending.append((rel, remote_hash, remote_size, remote_mtime, local))

    total_bytes = sum(p[2] for p in pending)
    state.update_peer(
        name,
        bytes_total=total_bytes, bytes_done=0,
        files_total=len(pending), files_done=0,
    )

    downloaded = 0
    for idx, (rel, remote_hash, remote_size, remote_mtime, local) in enumerate(pending, 1):
        state.update_peer(name, current_file=rel)
        local.parent.mkdir(parents=True, exist_ok=True)
        url = f"{base}/file?path={urllib.parse.quote(rel)}"
        tmp = local.with_suffix(local.suffix + ".part")
        h = hashlib.sha256()
        progress_accum = 0
        PROGRESS_FLUSH = 1 << 20  # update tray every ~1 MiB
        try:
            with urllib.request.urlopen(url, timeout=max(timeout, 30)) as r:
                with open(tmp, "wb") as out:
                    while True:
                        chunk = r.read(65536)
                        if not chunk:
                            break
                        out.write(chunk)
                        h.update(chunk)
                        progress_accum += len(chunk)
                        if progress_accum >= PROGRESS_FLUSH:
                            state.add_peer_bytes(name, progress_accum)
                            progress_accum = 0
            if progress_accum:
                state.add_peer_bytes(name, progress_accum)
            actual_hash = h.hexdigest()
            if remote_hash and actual_hash != remote_hash:
                sync_log.error(
                    "Hash mismatch %s from %s (%s): expected %s got %s",
                    rel, name, ip, remote_hash[:12], actual_hash[:12],
                )
                tmp.unlink()
                continue
            os.utime(tmp, (remote_mtime, remote_mtime))
            if local.exists():
                local.unlink()
            tmp.rename(local)
            if state.hash_cache:
                state.hash_cache.set(local, actual_hash)
            downloaded += 1
            state.update_peer(name, files_done=idx)
            sync_log.info(
                "PULLED from %s (%s): %s (%s, sha256=%s)",
                name, ip, rel, fmt_bytes(remote_size), actual_hash[:12],
            )
        except Exception as e:
            sync_log.error("Pull failed %s from %s (%s): %s", rel, name, ip, e)
            if tmp.exists():
                try:
                    tmp.unlink()
                except OSError:
                    pass

    # Mirror deletes: anything under peer_dir that the peer no longer advertises
    # gets removed locally. Skipped while the peer is paused — their empty
    # listing in that state isn't authoritative, and treating it as truth would
    # wipe the entire mirror.
    deleted = 0
    if not peer_paused:
        remote_rels = {
            f.get("path", "").replace("\\", "/").lower()
            for f in files if f.get("path")
        }
        for root, _, filenames in os.walk(peer_dir):
            for fn in filenames:
                if fn.endswith(".part"):
                    continue  # in-flight download artifact, not a tracked file
                full = Path(root) / fn
                try:
                    rel_local = full.relative_to(peer_dir).as_posix()
                except ValueError:
                    continue
                if rel_local.lower() in remote_rels:
                    continue
                try:
                    full.unlink()
                    deleted += 1
                    sync_log.info("DELETED from %s (%s): %s", name, ip, rel_local)
                except OSError as e:
                    sync_log.error("Delete failed %s from %s: %s", rel_local, name, e)
        # Bottom-up prune of now-empty subdirectories so the mirror's directory
        # structure tracks the peer's, not just its files.
        for root, _, _ in os.walk(peer_dir, topdown=False):
            rp = Path(root)
            if rp == peer_dir:
                continue
            try:
                rp.rmdir()
            except OSError:
                pass

    state.update_peer(
        name,
        syncing=False,
        last_sync=datetime.now().strftime("%H:%M:%S"),
        last_count=downloaded,
        current_file="",
    )
    if downloaded or deleted:
        sync_log.info(
            "Sync %s (%s): %d file(s) updated, %d deleted",
            name, ip, downloaded, deleted,
        )


def sync_loop():
    while not state.shutdown.is_set():
        try:
            cfg = state.config
            peers = cfg.get("peers", []) or []
            port = cfg.get("port", 8765)
            timeout = cfg.get("request_timeout_seconds", 5)
            download_root = resolve_path(cfg.get("download_folder", "downloads"))
            download_root.mkdir(parents=True, exist_ok=True)

            names_in_cfg = {(p.get("name") or p.get("ip", "")).strip() for p in peers if p.get("ip")}
            state.prune_peers(names_in_cfg)

            threads = []
            for peer in peers:
                if not peer.get("ip"):
                    continue
                t = threading.Thread(
                    target=sync_peer,
                    args=(peer, download_root, port, timeout),
                    daemon=True,
                )
                t.start()
                threads.append(t)
            for t in threads:
                t.join(timeout=60)
        except Exception:
            sync_log.exception("Sync loop error")

        if state.hash_cache:
            state.hash_cache.save()
        interval = max(1, int(state.config.get("sync_interval_seconds", 10)))
        state.shutdown.wait(interval)


def config_watch_loop():
    last_cfg = 0.0
    last_ig = 0.0
    while not state.shutdown.is_set():
        try:
            if CONFIG_PATH.exists():
                m = CONFIG_PATH.stat().st_mtime
                if m != last_cfg:
                    last_cfg = m
                    try:
                        new_cfg = load_config()
                        # Compute (and validate) the new share folders before
                        # committing the config — a config that would collide
                        # on names is rejected so the live state stays usable.
                        new_folders = compute_share_folders(new_cfg)
                        state.config = new_cfg
                        state.share_folders = new_folders
                        for _, p in new_folders:
                            p.mkdir(parents=True, exist_ok=True)
                        sync_log.info(
                            "Config (re)loaded: %d peer(s), %d share folder(s)",
                            len(state.config.get("peers", [])),
                            len(new_folders),
                        )
                    except DuplicateShareFolderError as e:
                        sync_log.error("Config reload rejected: %s", e)
                    except Exception as e:
                        sync_log.error("Config reload failed: %s", e)
            if IGNORE_PATH.exists():
                m = IGNORE_PATH.stat().st_mtime
                if m != last_ig:
                    last_ig = m
                    state.ignore_patterns = load_ignore_patterns()
                    sync_log.info("Ignore patterns (re)loaded: %d pattern(s)", len(state.ignore_patterns))
        except Exception:
            sync_log.exception("Config watch error")
        state.shutdown.wait(2)


# ---------------------------------------------------------------------------
# Tray icon + menu
# ---------------------------------------------------------------------------
def make_icon_image(syncing: bool = False, paused: bool = False) -> Image.Image:
    img = Image.new("RGBA", (64, 64), (0, 0, 0, 0))
    d = ImageDraw.Draw(img)
    if paused:
        body = (130, 130, 130, 255)
        tab = (95, 95, 95, 255)
    elif syncing:
        body = (220, 140, 40, 255)
        tab = (180, 110, 30, 255)
    else:
        body = (60, 130, 200, 255)
        tab = (40, 100, 170, 255)
    d.polygon([(6, 20), (24, 10), (40, 10), (44, 20)], fill=tab)
    d.rounded_rectangle([6, 18, 58, 56], radius=4, fill=body, outline=(0, 0, 0, 255))
    d.line([(14, 32), (50, 32)], fill=(255, 255, 255, 255), width=2)
    d.line([(14, 40), (50, 40)], fill=(255, 255, 255, 255), width=2)
    d.line([(14, 48), (38, 48)], fill=(255, 255, 255, 255), width=2)
    return img


def open_path(p: Path):
    p = Path(p)
    try:
        os.startfile(str(p))
    except OSError:
        try:
            subprocess.Popen(["notepad.exe", str(p)])
        except Exception as e:
            sync_log.error("Open failed: %s (%s)", p, e)


def _ensure_dir(path_str: str) -> Path:
    p = Path(path_str)
    p.mkdir(parents=True, exist_ok=True)
    return p


def _menu_action_factory(fn):
    def _action(icon, item):
        try:
            fn()
        except Exception:
            sync_log.exception("Menu action failed")
    return _action


def build_menu_items():
    items = []
    snap = state.snapshot()

    share_state = "PAUSED" if state.shared_paused else "ON"
    folder_count = len(state.share_folders)
    if folder_count != 1:
        share_state += f" ({folder_count} folders)"
    header = (
        f"This PC: {state.local_ip}  (port {state.config.get('port', 8765)})"
        f"  ·  Sharing: {share_state}"
    )
    items.append(pystray.MenuItem(header, None, enabled=False))
    items.append(pystray.Menu.SEPARATOR)

    if snap:
        items.append(pystray.MenuItem("Peers:", None, enabled=False))
        for name, st in sorted(snap.items()):
            if st["syncing"]:
                if st["bytes_total"] > 0:
                    pct = int(min(100, st["bytes_done"] * 100 / st["bytes_total"]))
                    progress = (
                        f"{fmt_bytes(st['bytes_done'])} / "
                        f"{fmt_bytes(st['bytes_total'])} · {pct}%"
                    )
                    file_part = ""
                    if st["current_file"]:
                        cf = st["current_file"]
                        if len(cf) > 32:
                            cf = "..." + cf[-29:]
                        file_part = f" · {cf}"
                    tag = (
                        f"  ⟳ {progress} "
                        f"({st['files_done']}/{st['files_total']}){file_part}"
                    )
                else:
                    tag = "  ⟳ checking..."
            elif not st["online"]:
                tag = "  ○ offline"
            elif st.get("paused"):
                tag = "  ⏸ paused (their share)"
            else:
                tag = f"  ● online  (last: {st['last_sync']}, +{st['last_count']})"
            label = f"  {name} [{st['ip']}]{tag}"
            items.append(pystray.MenuItem(label, None, enabled=False))
    else:
        items.append(pystray.MenuItem("No peers configured", None, enabled=False))

    items.append(pystray.Menu.SEPARATOR)
    items.append(pystray.MenuItem(
        "Open config.json",
        _menu_action_factory(lambda: open_path(CONFIG_PATH)),
    ))
    items.append(pystray.MenuItem(
        "Open .fileshareignore",
        _menu_action_factory(lambda: open_path(IGNORE_PATH)),
    ))
    share_folders = list(state.share_folders)
    if not share_folders:
        items.append(pystray.MenuItem("Open share folder", None, enabled=False))
    elif len(share_folders) == 1:
        _, only_path = share_folders[0]
        items.append(pystray.MenuItem(
            "Open share folder",
            _menu_action_factory(
                lambda p=only_path: open_path(_ensure_dir(str(p)))
            ),
        ))
    else:
        # Default-arg binding pins each lambda to its own path; without it,
        # every menu entry would open the last folder in the list.
        sub_items = [
            pystray.MenuItem(
                fname,
                _menu_action_factory(
                    lambda p=fpath: open_path(_ensure_dir(str(p)))
                ),
            )
            for fname, fpath in share_folders
        ]
        items.append(pystray.MenuItem(
            f"Open share folder ({len(share_folders)})",
            pystray.Menu(*sub_items),
        ))
    items.append(pystray.MenuItem(
        "Open downloads folder",
        _menu_action_factory(
            lambda: open_path(_ensure_dir(
                str(resolve_path(state.config.get("download_folder", "downloads")))
            ))
        ),
    ))
    items.append(pystray.MenuItem(
        "Open logs folder",
        _menu_action_factory(lambda: open_path(LOG_DIR)),
    ))
    items.append(pystray.Menu.SEPARATOR)
    pause_label = "Resume my sharing" if state.shared_paused else "Pause my sharing"
    items.append(pystray.MenuItem(pause_label, _menu_action_factory(toggle_share)))
    items.append(pystray.MenuItem("Sync now", _menu_action_factory(trigger_sync_now)))
    items.append(pystray.MenuItem("Quit", _quit_action))
    return items


def toggle_share():
    paused = state.toggle_share_paused()
    if paused:
        sync_log.info("Local sharing PAUSED by user")
    else:
        sync_log.info("Local sharing RESUMED by user")


def _quit_action(icon, item):
    state.shutdown.set()
    icon.stop()


_sync_trigger = threading.Event()


def trigger_sync_now():
    _sync_trigger.set()
    # Wake the sync loop by clearing+setting shutdown? No — we use a separate event.
    # Simpler: do an immediate one-off sync in a worker thread.
    threading.Thread(target=_oneshot_sync, daemon=True).start()


def _oneshot_sync():
    try:
        cfg = state.config
        peers = cfg.get("peers", []) or []
        port = cfg.get("port", 8765)
        timeout = cfg.get("request_timeout_seconds", 5)
        download_root = resolve_path(cfg.get("download_folder", "downloads"))
        download_root.mkdir(parents=True, exist_ok=True)
        threads = []
        for peer in peers:
            if not peer.get("ip"):
                continue
            t = threading.Thread(
                target=sync_peer,
                args=(peer, download_root, port, timeout),
                kwargs={"force_fresh": True},
                daemon=True,
            )
            t.start()
            threads.append(t)
        for t in threads:
            t.join(timeout=60)
    except Exception:
        sync_log.exception("Manual sync failed")


def tray_refresh_loop(icon: "pystray.Icon"):
    last_visual = None
    while not state.shutdown.is_set():
        try:
            snap = state.snapshot()
            any_syncing = any(s["syncing"] for s in snap.values())
            visual = (any_syncing, state.shared_paused)
            if visual != last_visual:
                icon.icon = make_icon_image(syncing=any_syncing, paused=state.shared_paused)
                last_visual = visual
            icon.menu = pystray.Menu(*build_menu_items())
            try:
                icon.update_menu()
            except Exception:
                pass
        except Exception:
            sync_log.exception("Tray refresh error")
        state.shutdown.wait(1.5)


def detect_local_ip() -> str:
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except OSError:
        try:
            return socket.gethostbyname(socket.gethostname())
        except OSError:
            return "?"


def _start_workers() -> bool:
    """Load config and launch the worker threads. Returns False on fatal error."""
    state.config = load_config()
    state.ignore_patterns = load_ignore_patterns()
    try:
        state.share_folders = compute_share_folders(state.config)
    except DuplicateShareFolderError as e:
        # Duplicate folder names are unrecoverable: the name is the public
        # identifier sent to peers and used as their on-disk subdir, so we
        # refuse to start until the user fixes config.json.
        show_fatal_error(str(e))
        return False
    for _, path in state.share_folders:
        path.mkdir(parents=True, exist_ok=True)
    download_root = resolve_path(state.config.get("download_folder", "downloads"))
    download_root.mkdir(parents=True, exist_ok=True)
    state.local_ip = detect_local_ip()
    state.hash_cache = HashCache(SCRIPT_DIR / ".fileshare_cache.json")

    sync_log.info(
        "=== FileShare starting (local IP %s, port %d, %d share folder(s)) ===",
        state.local_ip,
        state.config.get("port", 8765),
        len(state.share_folders),
    )

    threading.Thread(target=run_server, daemon=True).start()
    threading.Thread(target=sync_loop, daemon=True).start()
    threading.Thread(target=config_watch_loop, daemon=True).start()
    return True


def run_tray():
    """Run with the system-tray UI (Windows default; needs pystray + Pillow)."""
    if not HAVE_TRAY:
        show_fatal_error(
            "Tray mode requires pystray and Pillow. Install them with "
            "`pip install -r requirements.txt`, or run headless with --headless."
        )
        return
    icon = pystray.Icon(
        "FileShare",
        make_icon_image(),
        f"FileShare ({state.local_ip})",
        menu=pystray.Menu(*build_menu_items()),
    )
    threading.Thread(target=tray_refresh_loop, args=(icon,), daemon=True).start()
    icon.run()


def run_headless():
    """Run without a tray icon: block until a termination signal arrives.

    This is the Linux/server default. The worker threads are already running;
    here we just keep the process alive and translate signals into actions so
    the program behaves well under systemd:
      SIGTERM / SIGINT -> graceful shutdown (systemctl stop / Ctrl-C)
      SIGUSR1          -> toggle local-share pause   (Unix only)
      SIGUSR2          -> trigger an immediate sync   (Unix only)
    """
    import signal

    def _stop(signum, _frame):
        sync_log.info("Received signal %s, shutting down", signum)
        state.shutdown.set()

    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)
    if hasattr(signal, "SIGUSR1"):
        signal.signal(signal.SIGUSR1, lambda *_: toggle_share())
    if hasattr(signal, "SIGUSR2"):
        signal.signal(signal.SIGUSR2, lambda *_: trigger_sync_now())

    port = state.config.get("port", 8765)
    sync_log.info("Running headless (PID %d). Send SIGTERM to stop.", os.getpid())
    print(
        f"FileShare running headless on {state.local_ip}:{port} "
        f"(PID {os.getpid()}). Press Ctrl-C to stop.",
        flush=True,
    )
    try:
        while not state.shutdown.is_set():
            state.shutdown.wait(1)
    except KeyboardInterrupt:
        state.shutdown.set()


def _select_mode(argv) -> str:
    """Decide tray vs headless. Explicit flags win; otherwise default by OS:
    Windows -> tray, everything else -> headless."""
    if "--headless" in argv:
        return "headless"
    if "--tray" in argv:
        return "tray"
    return "tray" if sys.platform == "win32" else "headless"


def main():
    if not _start_workers():
        return

    mode = _select_mode(sys.argv[1:])
    try:
        if mode == "tray":
            run_tray()
        else:
            run_headless()
    finally:
        state.shutdown.set()
        if state.hash_cache:
            state.hash_cache.save()
        sync_log.info("=== FileShare stopped ===")


if __name__ == "__main__":
    main()