Adds paths adding functionnality

Adds functions that allows adding new paths to the synchronisation.
When writing the new paths to the file if a parent directory is
synchronised all the childrens are removed.

README.md

@@ -1,4 +1,14 @@
 Unisync is a data synchronisation tool written in python and based on [unison](https://github.com/bcpierce00/unison).  
-The goal is to be able to keep data synchronised between multiple computers without needing to have all the data kept locally while at the same time being able to access everything.  
+I couldn't find a tool to fulfill the requirements I had for a synchronisation tool so I am creating my own as a wrapper around unison.  
 
-The development just started so the documentation will be written later.  
+# Goal
+
+Unisync purpose is to keep personal data synchronised between multiple machines without needing to have all the data present an all the machines at the same time. For example you might not need to have your movies on your laptop but still want them on your desktop at home or you might want to keep your old pictures only on a server.
+Unisync requires you to have a "server" (like a NAS at home) that will store all your data allowing you to only copy what you need when you need it.  
+The issue is that you need to know what data is stored on the server to avoid conflict if creating duplicate files or folders. To address this unisync places a symlink for every file you do not wish to keep locally and allows you to mount the remote filesystem (using sshfs) allowing you to access files that aren't synchronised.  
+
+# Developement
+
+Unisync was at first a simple bash script but as it grew more complex I started struggling to maintain it which is why I am porting it to python. It will make everything more robust, easier to maintain and to add functionalities.  
+I am in the early stages of the developement process, this should be usable in the upcoming weeks.  
+Help will be welcome in the future but is not desirable right now as I want to shape this the way I want to.

src/unisync/config.py

@@ -1,13 +1,17 @@
 # Copyright (C) 2025 Paul Retourné
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+from configparser import UNNAMED_SECTION
 from dataclasses import dataclass, field
+from pathlib import Path, PosixPath
 import ipaddress
 import configparser
-from configparser import UNNAMED_SECTION
 
 @dataclass
 class ServerConfig:
+    """
+    Dataclass keeping the config for connecting to the server
+    """
     user: str
     sshargs: list[str] | None = field(default_factory=list)
     hostname: str = ""
@@ -15,6 +19,9 @@ class ServerConfig:
     port: int = 22
 
     def __post_init__(self):
+        """
+        Make sure a remote is provided and the ip address is valid
+        """
         if self.ip == "" and self.hostname == "":
             raise ValueError("A remote must be provided (ip or hostname)")
 
@@ -26,13 +33,37 @@ class ServerConfig:
 
 @dataclass
 class RootsConfig:
+    """
+    Dataclass keeping the paths to the roots to synchronise
+    """
     local: str
     remote: str
 
+@dataclass
+class UnisonConfig:
+    """
+    Dataclass keeping unison specific configurations
+    """
+    bools: list = field(default_factory=list)
+    values: dict = field(default_factory=dict)
+
+@dataclass
+class OtherConfig:
+    """
+    Dataclass keeping miscellanous configuration options
+    """
+    cache_dir_path: PosixPath = Path("~/.unisync").expanduser()
+
 @dataclass
 class Config:
+    """
+    Main dataclass for the configurations
+    """
     server: ServerConfig
     roots: RootsConfig
+    unison: UnisonConfig
+    other: OtherConfig = field(default_factory=OtherConfig)
+
 
 def load_config(config_path:str) -> Config:
     """
@@ -42,7 +73,7 @@ def load_config(config_path:str) -> Config:
     Returns:
         Config: A populated Config object containing the loaded config.
     """
-    config = configparser.ConfigParser(allow_unnamed_section=True)
+    config = configparser.ConfigParser(allow_unnamed_section=True, allow_no_value=True)
     config.read(config_path)
 
     # Check if sections are provided
@@ -60,4 +91,17 @@ def load_config(config_path:str) -> Config:
             config.get(roots_section, "local"),
             config.get(roots_section, "remote")
         )
-    return Config(server_config, roots_config)
+
+    args_bool = list()
+    args_val = dict()
+    if "Unison" in config.sections():
+        for key, val in config.items("Unison"):
+            if key in config["DEFAULT"].keys():
+                continue
+            elif val == "" or val == None:
+                args_bool.append(key)
+            else:
+                args_val[key] = val
+    unison_config = UnisonConfig(args_bool, args_val)
+
+    return Config(server_config, roots_config, unison_config)

src/unisync/main.py

@@ -4,6 +4,7 @@
 import os
 from argparser import create_argparser
 from config import RootsConfig, ServerConfig, Config, load_config
+from synchroniser import Synchroniser
 
 def main():
     parser = create_argparser()
@@ -18,5 +19,26 @@ def main():
         # TODO make the command line arguments work and override the config options
         pass
 
+    synchroniser = Synchroniser(
+            config.roots.remote,
+            config.roots.local,
+            config.server.user,
+            config.server.ip if config.server.ip != "" else config.server.hostname,
+            config.server.port,
+            config.unison.bools,
+            config.unison.values
+    )
+
+    if synchroniser.create_ssh_master_connection() != 0:
+        print("Connection failed quitting")
+        return 1
+    print("Connected to the remote.")
+
+    synchroniser.sync_files(["salut"])
+
+    synchroniser.close_ssh_master_connection()
+
+
+
 if __name__ == "__main__":
     main()

src/unisync/paths.py

@@ -0,0 +1,112 @@
+# Copyright (C) 2025 Paul Retourné
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+
+import os.path
+import subprocess
+
+from pathlib import Path, PosixPath
+
+class PathsManager:
+
+    def __init__(self, local_dir:PosixPath, cache_dir:PosixPath):
+        """
+        Creates a PathsManager with the necessary data
+        Args:
+            local_dir: Path to the top directory of the synchronisation
+            cache_dir: Path to the cache directory that contains the paths file
+        """
+        if not local_dir.is_dir():
+            raise ValueError("Invalid local directory")
+        self.local_dir = local_dir
+
+        if not cache_dir.is_dir():
+            raise ValueError("Invalid cache directory")
+        self.cache_dir = cache_dir
+
+        self.paths_file:PosixPath = self.cache_dir / "paths"
+        if not self.paths_file.is_file():
+            raise ValueError("The paths file does not exist")
+
+
+    def user_select_files(self, choice_timeout:int=120) -> list[str]:
+        """
+        Make the user select files in the top directory.
+        Currently uses nnn for the selection.
+        The goal is to replace it in order to avoid using external programs.
+        Args:
+            choice_timeout: Time given to make choices in nnn
+        Returns:
+            list[str]: The list of paths that was selected relative to the top directory
+        Raise:
+            TimeoutExpired: User took too long to choose
+            CalledProcessError: An unknown error occured during the selection
+        """
+        command = [
+                "/usr/bin/nnn",
+                "-H",
+                "-p", "-",
+                self.local_dir
+                ]
+        nnn_process = subprocess.Popen(command, stdout=subprocess.PIPE)
+        try:
+            ret_code = nnn_process.wait(timeout=choice_timeout)
+        except subprocess.TimeoutExpired as e:
+            print("Choice timeout expired", file=sys.stderr)
+            raise e
+
+        if ret_code != 0:
+            print("File selection failed", file=sys.stderr)
+            raise subprocess.CalledProcessError("File selection failed")
+
+        paths_list:list[str] = []
+        while (next_path := nnn_process.stdout.readline()) != b'':
+            next_path = next_path.decode().strip()
+            # Make the path relative to the top directory
+            next_path = next_path[len(self.local_dir):].lstrip("/")
+            paths_list.append(next_path)
+        return paths_list
+
+    def add_files_to_sync(self):
+        while true:
+            try:
+                paths = self.user_select_files()
+                break
+            except subprocess.TimeoutExpired:
+                if input("Timeout expired do you want to retry (y/n): ") != "y":
+                    raise
+        self.write_new_paths(paths)
+
+    def get_paths_to_sync(self) -> list[str]:
+        """
+        Return the paths to synchronise as list.
+        """
+        paths:list[str] = self.paths_file.read_text().split("\n")
+        if paths[-1] == "":
+            paths.pop()
+        return paths
+
+    def write_new_paths(self, paths:list[str]):
+        """
+        Writes a list of new paths to the file
+        """
+        current_paths = self.get_paths_to_sync()
+        paths_to_add = list()
+        # Check if one of the parent is already being synchronised
+        # If so there is no need to add the child path
+        for new_path in paths:
+            is_contained = False
+            for existing in current_paths:
+                common = os.path.commonpath([new_path, existing])
+                if common == existing:
+                    is_contained = True
+                    break
+
+            if not is_contained and new_path not in paths_to_add:
+                paths_to_add.append(new_path)
+
+        with self.paths_file.open("w") as f:
+            for p in paths_to_add:
+                f.write(p + "\n")
+
+

src/unisync/synchroniser.py

@@ -7,6 +7,8 @@ import sys
 import time
 import logging
 
+from pathlib import Path
+
 logger = logging.getLogger(__name__)
 
 class Synchroniser:
@@ -131,8 +133,58 @@ class Synchroniser:
             command.append(remote_root)
             command.append("-batch")
 
-        print(command)
         proc = subprocess.Popen(command)
         ret_code = proc.wait()
         return ret_code
 
+    def update_links(self, background:bool=True):
+        """
+        Update the links on the remote.
+        First calls cleanlinks to remove deadlinks and empty directories.
+        Then calls lndir to create the new links.
+        Args:
+        - background: controls if the update is done in the background or waited for
+        """
+
+        link_update_script = (f"cd {self.remote_dir}/links && "
+        "cleanlinks && "
+        "lndir -withrevinfo -ignorelinks -silent ../.data .;")
+
+        if background:
+            link_background_wrapper = f"nohup bash -c \"{link_update_script}\" > /dev/null 2>&1 < /dev/null &"
+        else:
+            link_background_wrapper = link_update_script
+
+        command = [
+                "/usr/bin/ssh",
+                "-S", self.control_path,
+                f"{self.remote_user}@{self.remote_ip}",
+                "-p", str(self.remote_port),
+                link_background_wrapper
+                ]
+
+        link_update_process = subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+        if not background:
+            print("Starting links update.")
+            link_update_process.wait()
+            print("Done")
+
+    def mount_remote_dir(self):
+        """
+        Mount the remote directory to make the local links work.
+        This is achieved using sshfs.
+        Raise:
+        - subprocess.CalledProcessError: An error occured with sshfs
+        """
+        command = [
+                "/usr/bin/sshfs",
+                "-o", "ControlPath={self.control_path}",
+                "-o", "ServerAliveInterval=15",
+                "-p", str(self.remote_port),
+                f"{self.remote_user}@{self.remote_ip}:{self.remote_dir}/.data",
+                # Get the absolute path to the correct .data directory resolving symlinks
+                str(Path(f"{self.local}/../.data").resolve())
+                ]
+        completed_process = subprocess.run(command)
+        completed_process.check_returncode()