tasks.py

"""Doodba child project tasks.

This file is to be executed with https://www.pyinvoke.org/ in Python 3.8.1+.

Contains common helpers to develop using this child project.
"""

import json
import operator
import os
import platform
import re
import shutil
import stat
import subprocess
import tempfile
import time
import xml.etree.ElementTree as ET
from datetime import datetime
from glob import iglob
from itertools import chain
from logging import getLogger
from pathlib import Path
from shutil import which

from invoke import exceptions, task

try:
    import yaml
except ImportError:
    from invoke.util import yaml

yaml.SafeLoader.add_constructor(
    "!override",
    lambda loader, node: loader.construct_sequence(node),
)

PROJECT_ROOT = Path(__file__).parent.absolute()
SRC_PATH = PROJECT_ROOT / "odoo" / "custom" / "src"
UID_ENV = {
    "GID": os.environ.get("DOODBA_GID", str(os.getgid())),
    "UID": os.environ.get("DOODBA_UID", str(os.getuid())),
    "DOODBA_UMASK": os.environ.get("DOODBA_UMASK", "27"),
}
UID_ENV.update(
    {
        "DOODBA_GITAGGREGATE_GID": os.environ.get(
            "DOODBA_GITAGGREGATE_GID", UID_ENV["GID"]
        ),
        "DOODBA_GITAGGREGATE_UID": os.environ.get(
            "DOODBA_GITAGGREGATE_UID", UID_ENV["UID"]
        ),
    }
)
yaml.SafeLoader.add_constructor(
    "!override",
    lambda loader, node: loader.construct_sequence(node),
)
SERVICES_WAIT_TIME = int(os.environ.get("SERVICES_WAIT_TIME", 4))
ODOO_VERSION = float(
    yaml.safe_load((PROJECT_ROOT / "common.yaml").read_text())["services"]["odoo"][
        "build"
    ]["args"]["ODOO_VERSION"]
)
# Depending on the user's docker version either version of docker compose could not
# be available. We default to v2 and fallback to v1.

docker_compose_v2 = (
    subprocess.run([shutil.which("docker"), "compose"], capture_output=True).returncode
    == 0
)
DOCKER_COMPOSE_CMD = (
    f"{shutil.which('docker')} compose"
    if docker_compose_v2
    else shutil.which("docker-compose")
)

_logger = getLogger(__name__)


def _override_docker_command(service, command, file, orig_file=None):
    # Read config from main file
    if orig_file:
        with open(orig_file) as fd:
            orig_docker_config = yaml.safe_load(fd.read())
            docker_compose_file_version = orig_docker_config.get("version")
    else:
        docker_compose_file_version = "2.4"
    docker_config = {
        "services": {service: {"command": command}},
    }
    if not docker_compose_v2 and docker_compose_file_version:
        docker_config["version"] = docker_compose_file_version
    docker_config_yaml = yaml.dump(docker_config)
    file.write(docker_config_yaml)
    file.flush()


def _remove_auto_reload(file, orig_file):
    with open(orig_file) as fd:
        orig_docker_config = yaml.safe_load(fd.read())
    odoo_command = orig_docker_config["services"]["odoo"]["command"]
    new_odoo_command = []
    for flag in odoo_command:
        if flag.startswith("--dev"):
            flag = flag.replace("reload,", "")
        new_odoo_command.append(flag)
    _override_docker_command("odoo", new_odoo_command, file, orig_file=orig_file)


def _get_cwd_addon(file):
    cwd = Path(file).resolve()
    manifest_file = False
    while PROJECT_ROOT < cwd:
        manifest_file = (cwd / "__manifest__.py").exists() or (
            cwd / "__openerp__.py"
        ).exists()
        if manifest_file:
            return cwd.stem
        cwd = cwd.parent
        if cwd == PROJECT_ROOT:
            return None


def _scan_subrepos_and_add_path_mappings(
    cw_config,
    firefox_configuration,
    chrome_configuration,
):
    """Scan subrepos in SRC_PATH, configure folders & pathMappings."""
    for subrepo in SRC_PATH.glob("*"):
        if not subrepo.is_dir():
            continue
        if (subrepo / ".git").exists() and subrepo.name != "odoo":
            cw_config["folders"].append(
                {"path": str(subrepo.relative_to(PROJECT_ROOT))}
            )

        private_dir = subrepo / "odoo" / "custom" / "src" / "private"
        # Default scanning approach (1-level + addons/* + private/*)
        for addon in chain(
            subrepo.glob("*"),
            subrepo.glob("addons/*"),
            private_dir.glob("*"),
        ):
            if (addon / "__manifest__.py").is_file() or (
                addon / "__openerp__.py"
            ).is_file():
                url = f"http://localhost:{ODOO_VERSION:.0f}069/{addon.name}/static/"
                path = "${workspaceFolder:%s}/%s/static/" % (  # noqa: UP031
                    subrepo.name,
                    addon.relative_to(subrepo),
                )
                firefox_configuration["pathMappings"].append({"url": url, "path": path})
                chrome_configuration["pathMapping"][url] = path


def _modules_installed(c, modules_list, dbname="devel"):
    """Return set of module technical names installed in dbname."""
    if not modules_list:
        return set()
    # Quote module names safely for SQL IN (...)
    quoted = ",".join(repr(m) for m in modules_list if m)
    cmd = (
        f"{DOCKER_COMPOSE_CMD} exec -T db "
        f"psql -U odoo -d {dbname} -Atc "
        f'"select name from ir_module_module '
        f"where state='installed' and name in ({quoted});\""
    )
    res = c.run(cmd, hide=True, warn=True)
    return set(filter(None, res.stdout.splitlines()))


@task
def write_code_workspace_file(c, cw_path=None):
    """Generate code-workspace file definition.

    Some other tasks will call this one when needed, and since you cannot specify
    the file name there, if you want a specific one, you should call this task
    before.

    Most times you just can forget about this task and let it be run automatically
    whenever needed.

    If you don't define a workspace name, this task will reuse the 1st
    `doodba.*.code-workspace` file found inside the current directory.
    If none is found, it will default to `doodba.$(basename $PWD).code-workspace`.

    If you define it manually, remember to use the same prefix and suffix if you
    want it git-ignored by default.
    Example: `--cw-path doodba.my-custom-name.code-workspace`
    """
    root_name = f"doodba.{PROJECT_ROOT.name}"
    root_var = "${workspaceFolder:%s}" % root_name  # noqa: UP031
    if not cw_path:
        try:
            cw_path = next(PROJECT_ROOT.glob("doodba.*.code-workspace"))
        except StopIteration:
            cw_path = f"{root_name}.code-workspace"
    if not Path(cw_path).is_absolute():
        cw_path = PROJECT_ROOT / cw_path
    cw_config = {}
    try:
        with open(cw_path) as cw_fd:
            cw_config = json.load(cw_fd)
    except (FileNotFoundError, json.decoder.JSONDecodeError):
        pass  # Nevermind, we start with a new config
    # Static settings
    cw_config.setdefault("settings", {})
    cw_config["settings"].update(
        {
            "python.autoComplete.extraPaths": [f"{SRC_PATH}/odoo"],
            "python.analysis.extraPaths": [f"{SRC_PATH}/odoo"],
            "python.formatting.provider": "none",
            "python.linting.flake8Enabled": True,
            "python.linting.ignorePatterns": [f"{SRC_PATH}/odoo/**/*.py"],
            "python.linting.pylintArgs": [
                f"--init-hook=\"import sys;sys.path.append('{SRC_PATH}/odoo')\"",
                "--load-plugins=pylint_odoo",
            ],
            "python.linting.pylintEnabled": True,
            "python.defaultInterpreterPath": "python%s"
            % (2 if ODOO_VERSION < 11 else 3),
            "restructuredtext.confPath": "",
            "search.followSymlinks": False,
            "search.useIgnoreFiles": False,
            # Language-specific configurations
            "[python]": {"editor.defaultFormatter": "ms-python.black-formatter"},
            "[json]": {"editor.defaultFormatter": "esbenp.prettier-vscode"},
            "[jsonc]": {"editor.defaultFormatter": "esbenp.prettier-vscode"},
            "[markdown]": {"editor.defaultFormatter": "esbenp.prettier-vscode"},
            "[yaml]": {"editor.defaultFormatter": "esbenp.prettier-vscode"},
            "[xml]": {"editor.formatOnSave": False},
        }
    )
    # Launch configurations
    debugpy_configuration = {
        "name": "Attach Python debugger to running container",
        "type": "python",
        "request": "attach",
        "pathMappings": [
            {
                "localRoot": "${workspaceFolder:%s}/odoo" % root_name,  # noqa: UP031
                "remoteRoot": "/opt/odoo",
            }
        ],
        "port": int(ODOO_VERSION) * 1000 + 899,
        # HACK https://github.com/microsoft/vscode-python/issues/14820
        "host": "0.0.0.0",
    }
    firefox_configuration = {
        "type": "firefox",
        "request": "launch",
        "reAttach": True,
        "name": "Connect to firefox debugger",
        "url": f"http://localhost:{ODOO_VERSION:.0f}069/?debug=assets",
        "reloadOnChange": {
            "watch": f"{root_var}/odoo/custom/src/**/*.{'{js,css,scss,less}'}"
        },
        "skipFiles": ["**/lib/**"],
        "pathMappings": [],
    }
    chrome_executable = which("chrome") or which("chromium")
    chrome_configuration = {
        "type": "chrome",
        "request": "launch",
        "name": "Connect to chrome debugger",
        "url": f"http://localhost:{ODOO_VERSION:.0f}069/?debug=assets",
        "skipFiles": ["**/lib/**"],
        "trace": True,
        "pathMapping": {},
    }
    if chrome_executable:
        chrome_configuration["runtimeExecutable"] = chrome_executable

    cw_config["launch"] = {
        "compounds": [
            {
                "name": "Start Odoo and debug Python",
                "configurations": ["Attach Python debugger to running container"],
                "preLaunchTask": "Start Odoo in debug mode",
            },
            {
                "name": "Test and debug current module",
                "configurations": ["Attach Python debugger to running container"],
                "preLaunchTask": "Run Odoo Tests in debug mode for current module",
                "internalConsoleOptions": "openOnSessionStart",
            },
        ],
        "configurations": [
            debugpy_configuration,
            firefox_configuration,
            chrome_configuration,
        ],
    }
    # Configure workspace roots and path mappings
    cw_config["folders"] = []
    _scan_subrepos_and_add_path_mappings(
        cw_config,
        firefox_configuration,
        chrome_configuration,
    )

    cw_config["tasks"] = {
        "version": "2.0.0",
        "tasks": [
            {
                "label": "Start Odoo",
                "type": "process",
                "command": "invoke",
                "args": ["start", "--detach"],
                "presentation": {
                    "echo": True,
                    "reveal": "silent",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"label": "$(play-circle) Start Odoo"}},
            },
            {
                "label": "Install current module",
                "type": "process",
                "command": "invoke",
                "args": ["install", "--cur-file", "${file}", "restart"],
                "presentation": {
                    "echo": True,
                    "reveal": "always",
                    "focus": True,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {
                    "statusbar": {"label": "$(symbol-property) Install module"}
                },
            },
            {
                "label": "Run Odoo Tests for current module",
                "type": "process",
                "command": "invoke",
                "args": ["test", "--cur-file", "${file}"],
                "presentation": {
                    "echo": True,
                    "reveal": "always",
                    "focus": True,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"label": "$(beaker) Test module"}},
            },
            {
                "label": "Run Odoo Tests in debug mode for current module",
                "type": "process",
                "command": "invoke",
                "args": [
                    "test",
                    "--cur-file",
                    "${file}",
                    "--debugpy",
                ],
                "presentation": {
                    "echo": True,
                    "reveal": "silent",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"hide": True}},
            },
            {
                "label": "Start Odoo in debug mode",
                "type": "process",
                "command": "invoke",
                "args": ["start", "--detach", "--debugpy"],
                "presentation": {
                    "echo": True,
                    "reveal": "silent",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"hide": True}},
            },
            {
                "label": "Stop Odoo",
                "type": "process",
                "command": "invoke",
                "args": ["stop"],
                "presentation": {
                    "echo": True,
                    "reveal": "silent",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"label": "$(stop-circle) Stop Odoo"}},
            },
            {
                "label": "Restart Odoo",
                "type": "process",
                "command": "invoke",
                "args": ["restart"],
                "presentation": {
                    "echo": True,
                    "reveal": "silent",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {"statusbar": {"label": "$(history) Restart Odoo"}},
            },
            {
                "label": "See container logs",
                "type": "process",
                "command": "invoke",
                "args": ["logs"],
                "presentation": {
                    "echo": True,
                    "reveal": "always",
                    "focus": False,
                    "panel": "shared",
                    "showReuseMessage": True,
                    "clear": False,
                },
                "problemMatcher": [],
                "options": {
                    "statusbar": {"label": "$(list-selection) See container logs"}
                },
            },
        ],
    }
    # Sort project folders
    cw_config["folders"].sort(key=operator.itemgetter("path"))
    # Put Odoo folder just before private and top folder and map to debugpy
    odoo = SRC_PATH / "odoo"
    if odoo.is_dir():
        cw_config["folders"].append({"path": str(odoo.relative_to(PROJECT_ROOT))})
    # HACK https://github.com/microsoft/vscode/issues/95963 put private second to last
    private = SRC_PATH / "private"
    if private.is_dir():
        cw_config["folders"].append({"path": str(private.relative_to(PROJECT_ROOT))})
    # HACK https://github.com/microsoft/vscode/issues/37947 put top folder last
    cw_config["folders"].append({"path": ".", "name": root_name})
    with open(cw_path, "w") as cw_fd:
        json.dump(cw_config, cw_fd, indent=2)
        cw_fd.write("\n")


def _pycharm_version_key(path):
    """Sort PyCharm config dirs by product priority and version."""
    name = path.name
    # Prefer modern PyCharm/PyCharmCE names over legacy PyCharmCE2020.x
    # style only by version
    product_priority = (
        1 if name.startswith("PyCharm") and not name.startswith("PyCharmCE") else 0
    )
    version_match = re.search(r"(\d{4})\.(\d+)", name)
    if version_match:
        version = tuple(map(int, version_match.groups()))
    else:
        version = (0, 0)
    return product_priority, version, name


def get_pycharm_config_dirs():
    """Return possible PyCharm config directories for Linux, macOS and Windows."""
    system = platform.system()
    if system == "Linux":
        base = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
        jetbrains = base / "JetBrains"
    elif system == "Darwin":
        jetbrains = Path.home() / "Library" / "Application Support" / "JetBrains"
    elif system == "Windows":
        appdata = os.environ.get("APPDATA")
        if not appdata:
            return []
        jetbrains = Path(appdata) / "JetBrains"
    else:
        return []
    candidates = list(jetbrains.glob("PyCharm*")) + list(jetbrains.glob("PyCharmCE*"))
    return sorted({p for p in candidates if p.is_dir()}, key=_pycharm_version_key)


def get_pycharm_tools_dir():
    """Return the latest existing PyCharm tools directory."""
    pycharm_dirs = get_pycharm_config_dirs()
    if not pycharm_dirs:
        return None
    return pycharm_dirs[-1] / "tools"


def write_pycharm_external_tools():
    """Ensure required PyCharm External Tools exist."""
    tools_dir = get_pycharm_tools_dir()
    if not tools_dir:
        # PyCharm not installed → silently skip
        return
    tools_dir.mkdir(parents=True, exist_ok=True)
    tools_path = tools_dir / "External Tools.xml"
    tools_to_ensure = {
        "Start Debug Test": "test --cur-file $FilePath$ --debugpy",
        "Start Debug": "start --detach --debugpy",
    }
    if tools_path.exists():
        try:
            root = ET.fromstring(tools_path.read_text(encoding="utf-8"))
        except ET.ParseError:
            root = ET.Element("toolSet", {"name": "External Tools"})
    else:
        root = ET.Element("toolSet", {"name": "External Tools"})
    existing = {tool.get("name") for tool in root.findall("tool")}
    for name, params in tools_to_ensure.items():
        if name in existing:
            continue
        tool = ET.SubElement(
            root,
            "tool",
            {
                "name": name,
                "showInMainMenu": "false",
                "showInEditor": "false",
                "showInProject": "false",
                "showInSearchPopup": "false",
                "disabled": "false",
                "useConsole": "true",
                "showConsoleOnStdOut": "false",
                "showConsoleOnStdErr": "false",
                "synchronizeAfterRun": "true",
            },
        )
        exec_node = ET.SubElement(tool, "exec")
        ET.SubElement(exec_node, "option", {"name": "COMMAND", "value": "invoke"})
        ET.SubElement(exec_node, "option", {"name": "PARAMETERS", "value": params})
        ET.SubElement(
            exec_node,
            "option",
            {"name": "WORKING_DIRECTORY", "value": "$ProjectFileDir$"},
        )
    ET.indent(root, space="  ")
    tools_path.write_text(
        ET.tostring(root, encoding="unicode") + "\n",
        encoding="utf-8",
    )


def write_pycharm_debug_run_configuration_file():
    """Create/update PyCharm run configuration to attach debugpy."""
    config_path = PROJECT_ROOT / ".idea" / "runConfigurations" / "Debug_odoo.xml"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    debugpy_port = int(ODOO_VERSION) * 1000 + 899
    xml = f"""<component name="ProjectRunConfigurationManager">
  <configuration default="false" name="Debug odoo" type="PythonDapAttachConfiguration">
    <pathMappings>
      <mapping local="$PROJECT_DIR$/odoo" remote="/opt/odoo" />
    </pathMappings>
    <option name="remoteAddress" value="localhost:{debugpy_port}" />
    <option name="remoteRoot" />
    <method v="2">
      <option
        name="ToolBeforeRunTask"
        enabled="true"
        actionId="Tool_External Tools_Start Debug"
      />
    </method>
  </configuration>
</component>
"""
    config_path.write_text(xml, encoding="utf-8")


def write_pycharm_attach_test_debug_run_configuration_file():
    """Create/update PyCharm DAP attach configuration for current module tests."""
    config_path = (
        PROJECT_ROOT
        / ".idea"
        / "runConfigurations"
        / "Test_and_debug_current_module.xml"
    )
    config_path.parent.mkdir(parents=True, exist_ok=True)
    debugpy_port = int(ODOO_VERSION) * 1000 + 899
    xml = f"""<component name="ProjectRunConfigurationManager">
  <configuration default="false" name="Test and debug current module"
    type="PythonDapAttachConfiguration">
    <pathMappings>
      <mapping local="$PROJECT_DIR$/odoo" remote="/opt/odoo" />
    </pathMappings>
    <option name="remoteAddress" value="localhost:{debugpy_port}" />
    <option name="remoteRoot" />
    <method v="2">
      <option
        name="ToolBeforeRunTask"
        enabled="true"
        actionId="Tool_External Tools_Start Debug Test"
      />
    </method>
  </configuration>
</component>
"""
    config_path.write_text(xml, encoding="utf-8")


def write_pycharm_debugger_configurations():
    write_pycharm_external_tools()
    write_pycharm_debug_run_configuration_file()
    write_pycharm_attach_test_debug_run_configuration_file()


@task
def develop(c):
    """Set up a basic development environment."""
    # Prepare environment
    auto = Path(PROJECT_ROOT, "odoo", "auto")
    addons = auto / "addons"
    addons.mkdir(parents=True, exist_ok=True)
    # Allow others writing, for podman support
    auto.chmod(0o777)
    addons.chmod(0o777)
    with c.cd(str(PROJECT_ROOT)):
        c.run("git init")
        c.run("ln -sf devel.yaml docker-compose.yml")
        write_code_workspace_file(c)
        write_pycharm_debugger_configurations()
        c.run("pre-commit install")


@task(develop)
def git_aggregate(c):
    """Download odoo & addons git code.

    Executes git-aggregator from within the doodba container.
    """
    with c.cd(str(PROJECT_ROOT)):
        c.run(
            DOCKER_COMPOSE_CMD + " run --rm -T devel-setup",
            env=UID_ENV,
        )
    write_code_workspace_file(c)
    for git_folder in SRC_PATH.glob("*/.git/.."):
        action = (
            "install"
            if (git_folder / ".pre-commit-config.yaml").is_file()
            else "uninstall"
        )
        with c.cd(str(git_folder)):
            c.run(f"pre-commit {action}")


@task(develop)
def closed_prs(c):
    """Test closed PRs from repos.yaml"""
    with c.cd(str(PROJECT_ROOT / "odoo/custom/src")):
        cmd = "gitaggregate -c {} show-closed-prs".format("repos.yaml")
        c.run(cmd, env=UID_ENV, pty=True)


@task()
def img_build(c, pull=True):
    """Build docker images."""
    cmd = DOCKER_COMPOSE_CMD + " build"
    if pull:
        cmd += " --pull"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd, env=UID_ENV, pty=True)


@task()
def img_pull(c, ignore_buildable=True):
    """Pull docker images."""
    cmd = DOCKER_COMPOSE_CMD + " pull"
    if ignore_buildable:
        cmd += " --ignore-buildable"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd, pty=True)


@task()
def lint(c, verbose=False):
    """Lint & format source code."""
    cmd = "pre-commit run --show-diff-on-failure --all-files --color=always"
    if verbose:
        cmd += " --verbose"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd)


@task()
def start(c, detach=True, debugpy=False, _reload=True, port_prefix=0):
    """Start environment."""
    cmd = DOCKER_COMPOSE_CMD + " up"
    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".yaml",
    ) as tmp_docker_compose_file:
        if debugpy or not _reload:
            # Remove auto-reload
            cmd = (
                DOCKER_COMPOSE_CMD + " -f docker-compose.yml "
                f"-f {tmp_docker_compose_file.name} up"
            )
            _remove_auto_reload(
                tmp_docker_compose_file,
                orig_file=PROJECT_ROOT / "docker-compose.yml",
            )
        if detach:
            cmd += " --detach"
        with c.cd(str(PROJECT_ROOT)):
            env = UID_ENV | {"DOODBA_DEBUGPY_ENABLE": str(int(debugpy))}
            if port_prefix:
                env["PORT_PREFIX"] = str(port_prefix)
            result = c.run(
                cmd,
                pty=True,
                env=env,
            )
            if not (
                "Recreating" in result.stdout
                or "Starting" in result.stdout
                or "Creating" in result.stdout
            ):
                restart(c)
        _logger.info("Waiting for services to spin up...")
        time.sleep(SERVICES_WAIT_TIME)


@task(
    help={
        "modules": "Comma-separated list of modules to install.",
        "core": "Install all core addons. Default: False",
        "extra": "Install all extra addons. Default: False",
        "private": "Install all private addons. Default: False",
        "enterprise": "Install all enterprise addons. Default: False",
        "cur-file": "Path to the current file."
        " Addon name will be obtained from there to install.",
    },
)
def install(
    c,
    modules=None,
    cur_file=None,
    core=False,
    extra=False,
    private=False,
    enterprise=False,
):
    """Install Odoo addons

    By default, installs addon from directory being worked on,
    unless other options are specified.
    """
    if not (modules or core or extra or private or enterprise):
        cur_module = _get_cwd_addon(cur_file or Path.cwd())
        if not cur_module:
            raise exceptions.ParseError(
                msg="Odoo addon to install not found. "
                "You must provide at least one option for modules"
                " or be in a subdirectory of one."
                " See --help for details."
            )
        modules = cur_module
    cmd = DOCKER_COMPOSE_CMD + " run --rm odoo addons init"
    if core:
        cmd += " --core"
    if extra:
        cmd += " --extra"
    if private:
        cmd += " --private"
    if enterprise:
        cmd += " --enterprise"
    if modules:
        cmd += f" -w {modules}"
    with c.cd(str(PROJECT_ROOT)):
        c.run(DOCKER_COMPOSE_CMD + " stop odoo")
        c.run(
            cmd,
            env=UID_ENV,
            pty=True,
        )


@task(
    help={
        "module": "Specific Odoo module to update.",
        "all": "Update all modules. Takes a lot of time. [default: False]",
        "repo": "Update all modules from a specific repository.",
        "msgmerge": "Merge .pot changes into all .po files. [default: True]",
        "fuzzy_matching": "Use fuzzy matching when merging. [default: False]",
        "purge_old_translations": "Remove lines with old translations. [default: True]",
        "remove_dates": "Remove dates from .po files. [default: True]",
    }
)
def updatepot(
    c,
    module=None,
    _all=False,
    repo=None,
    msgmerge=True,
    fuzzy_matching=False,
    purge_old_translations=True,
    remove_dates=True,
):
    """Updates POT of a given module"""
    if not module and not _all and not repo:
        cur_module = _get_cwd_addon(Path.cwd())
        if not cur_module:
            raise exceptions.ParseError(
                msg="Odoo addon to update translation not found "
                "You must provide at least one of: -m {module}, "
                "be in the subdirectory of a module, --all or -r {repo} "
                "See --help for details."
            )
        module = cur_module

    cmd = (
        DOCKER_COMPOSE_CMD
        + f" run --rm  -v {PROJECT_ROOT}/odoo/custom:/tmp/odoo/custom:rw,z "
        f"-v {PROJECT_ROOT}/odoo/auto:/tmp/odoo/auto:rw,z odoo "
        "click-odoo-makepot --addons-dir "
        f"{'/tmp/odoo/auto/addons' if not repo else '/tmp/odoo/custom/src/' + repo}/"
    )

    cmd += " --msgmerge" if msgmerge else " --no-msgmerge"
    cmd += " --no-fuzzy-matching" if not fuzzy_matching else " --fuzzy-matching"
    cmd += (
        " --purge-old-translations"
        if purge_old_translations
        else " --no-purge-old-translations"
    )
    if not _all and not repo:
        cmd += f" -m {module}"

    with c.cd(str(PROJECT_ROOT)):
        c.run(DOCKER_COMPOSE_CMD + " stop odoo")
        c.run(
            cmd,
            env=UID_ENV,
            pty=True,
        )
    glob = (
        f"{PROJECT_ROOT}/odoo/custom/src/{'*' if not repo else repo}"
        f"/{'*' if _all or repo else module}/i18n/"
    )
    new_files = iglob(f"{glob}/*.po*")
    for new_file in new_files:
        file_name = os.path.basename(new_file)
        if file_name.endswith("~"):
            Path(new_file).unlink()
            continue
        content = Path(new_file).read_text()
        new_lines = []
        for line in content.splitlines():
            if remove_dates and (
                line.startswith('"POT-Creation-Date')
                or line.startswith('"PO-Revision-Date')
            ):
                continue
            new_lines.append(line)
        content = "\n".join(new_lines)
        Path(new_file).write_text(content.strip() + "\n")
    _logger.info(".po[t] files updated")
    precommit_cmd = (
        f"pre-commit run --files {' '.join(iglob(f'{glob}/*.po*'))} --color=always"
    )
    if not repo and module:
        for folder in iglob(f"{PROJECT_ROOT}/odoo/custom/src/*/*"):
            if os.path.isdir(folder) and os.path.basename(folder) == module:
                repo = os.path.basename(os.path.dirname(folder))
                break
    precommit_folder = (
        str(PROJECT_ROOT) + f"/odoo/custom/src/{repo}" if repo != "private" else ""
    )
    with c.cd(str(precommit_folder)):
        c.run(precommit_cmd)


@task(
    help={
        "modules": "Comma-separated list of modules to uninstall.",
    },
)
def uninstall(
    c,
    modules=None,
    cur_file=None,
):
    """Uninstall Odoo addons

    By default, uninstalls addon from directory being worked on,
    unless other options are specified.
    """
    if not modules:
        cur_module = _get_cwd_addon(cur_file or Path.cwd())
        if not cur_module:
            raise exceptions.ParseError(
                msg="Odoo addon to uninstall not found. "
                "You must provide at least one option for modules"
                " or be in a subdirectory of one."
                " See --help for details."
            )
        modules = cur_module
    cmd = (
        DOCKER_COMPOSE_CMD
        + f" run --rm odoo click-odoo-uninstall -m {modules or cur_module}"
    )
    with c.cd(str(PROJECT_ROOT)):
        c.run(
            cmd,
            env=UID_ENV,
            pty=True,
        )


def _get_module_dependencies(
    c, modules=None, core=False, extra=False, private=False, enterprise=False
):
    """Returns a list of the addons' dependencies

    By default, refers to the addon from directory being worked on,
    unless other options are specified.
    """
    # Get list of dependencies for addon
    cmd = DOCKER_COMPOSE_CMD + " run --rm odoo addons list --dependencies"
    if core:
        cmd += " --core"
    if extra:
        cmd += " --extra"
    if private:
        cmd += " --private"
    if enterprise:
        cmd += " --enterprise"
    if modules:
        cmd += f" -w {modules}"
    with c.cd(str(PROJECT_ROOT)):
        dependencies = c.run(
            cmd,
            env=UID_ENV,
            hide="stdout",
        ).stdout.splitlines()[-1]
    return dependencies


def _test_in_debug_mode(c, odoo_command):
    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yaml"
    ) as tmp_docker_compose_file:
        cmd = (
            DOCKER_COMPOSE_CMD + " -f docker-compose.yml "
            f"-f {tmp_docker_compose_file.name} up -d"
        )
        _override_docker_command(
            "odoo",
            odoo_command,
            file=tmp_docker_compose_file,
            orig_file=Path(str(PROJECT_ROOT), "docker-compose.yml"),
        )
        with c.cd(str(PROJECT_ROOT)):
            c.run(
                cmd,
                env=UID_ENV | {"DOODBA_DEBUGPY_ENABLE": "1"},
                pty=True,
            )
        _logger.info("Waiting for services to spin up...")
        time.sleep(SERVICES_WAIT_TIME)


def _get_module_list(
    c,
    modules=None,
    core=False,
    extra=False,
    private=False,
    enterprise=False,
    only_installable=True,
):
    """Returns a list of addons according to the passed parameters.

    By default, refers to the addon from directory being worked on,
    unless other options are specified.
    """
    # Get list of dependencies for addon
    cmd = DOCKER_COMPOSE_CMD + " run --rm odoo addons list"
    if core:
        cmd += " --core"
    if extra:
        cmd += " --extra"
    if private:
        cmd += " --private"
    if enterprise:
        cmd += " --enterprise"
    if modules:
        cmd += f" -w {modules}"
    if only_installable:
        cmd += " --installable"
    with c.cd(str(PROJECT_ROOT)):
        module_list = c.run(
            cmd,
            env=UID_ENV,
            pty=True,
            hide="stdout",
        ).stdout.splitlines()[-1]
    return module_list


@task(
    help={
        "modules": "Comma-separated list of modules to test.",
        "core": "Test all core addons. Default: False",
        "extra": "Test all extra addons. Default: False",
        "private": "Test all private addons. Default: False",
        "enterprise": "Test all enterprise addons. Default: False",
        "skip": "Comma-separated list of modules to skip. Default: ''",
        "debugpy": "Whether or not to run tests in a VSCode debugging session. "
        "Default: False",
        "cur-file": "Path to the current file."
        " Addon name will be obtained from there to run tests",
        "mode": "Mode in which tests run. Options: ['init'(default), 'update']",
        "db_filter": "DB_FILTER regex to pass to the test container Set to ''"
        " to disable. Default: '^devel$'",
        "tags": "Comma-separated list of tags to test. Default: ',/'.join(modules)",
    },
)
def test(
    c,
    modules=None,
    core=False,
    extra=False,
    private=False,
    enterprise=False,
    skip="",
    debugpy=False,
    cur_file=None,
    mode="init",
    db_filter="^devel$",
    tags=None,
):
    """Run Odoo tests

    By default, tests addon from directory being worked on,
    unless other options are specified.

    NOTE: Odoo must be restarted manually after this to go back to normal mode
    """
    if not (modules or core or extra or private or enterprise):
        cur_module = _get_cwd_addon(cur_file or Path.cwd())
        if not cur_module:
            raise exceptions.ParseError(
                msg="Odoo addon to install not found. "
                "You must provide at least one option for modules"
                " or be in a subdirectory of one."
                " See --help for details."
            )
        modules = cur_module
    else:
        modules = _get_module_list(c, modules, core, extra, private, enterprise)
    odoo_command = ["odoo", "--test-enable", "--stop-after-init", "--workers=0"]
    if mode == "init":
        if ODOO_VERSION >= 19:
            mods = [m for m in modules.split(",") if m]
            installed = _modules_installed(c, mods)
            to_install = [m for m in mods if m not in installed]
            to_update = sorted(installed)

            if to_install:
                odoo_command.extend(["-i", ",".join(to_install)])
            if to_update:
                odoo_command.extend(["-u", ",".join(to_update)])
        else:
            odoo_command.append("-i")
    elif mode == "update":
        odoo_command.append("-u")
    else:
        raise exceptions.ParseError(
            msg="Available modes are 'init' or 'update'. See --help for details."
        )
    # Skip test in some modules
    modules_list = modules.split(",")
    for m_to_skip in skip.split(","):
        if not m_to_skip:
            continue
        if m_to_skip not in modules_list:
            _logger.warning(
                "%s not found in the list of addons to test: %s", m_to_skip, modules
            )
            continue
        modules_list.remove(m_to_skip)
    modules = ",".join(modules_list)
    if not (mode == "init" and ODOO_VERSION >= 19):
        odoo_command.append(modules)
    if ODOO_VERSION >= 12:
        # Limit tests to explicit list
        # Filter spec format (comma-separated)
        # [-][tag][/module][:class][.method]
        test_tags = f"/{',/'.join(modules_list)}"
        if tags:
            test_tags = tags
        odoo_command.extend(["--test-tags", test_tags])
    if debugpy:
        _test_in_debug_mode(c, odoo_command)
    else:
        cmd = [DOCKER_COMPOSE_CMD, "run", "--rm"]
        if db_filter:
            cmd.extend(["-e", f"DB_FILTER='{db_filter}'"])
        cmd.append("odoo")
        cmd.extend(odoo_command)
        with c.cd(str(PROJECT_ROOT)):
            c.run(
                " ".join(cmd),
                env=UID_ENV,
                pty=True,
            )


@task(
    help={"purge": "Remove all related containers, networks images and volumes"},
)
def stop(c, purge=False):
    """Stop and (optionally) purge environment."""
    cmd = f"{DOCKER_COMPOSE_CMD} down --remove-orphans"
    if purge:
        cmd += " --rmi local --volumes"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd, pty=True)


def _db_run_cmd(odoo_command):
    """Return the compose command running a DB management command in odoo."""
    cmd = f"{DOCKER_COMPOSE_CMD} run --rm -l traefik.enable=false"
    if ODOO_VERSION >= 20:
        # `odoo db` rejects the `--load-language` the entrypoint adds for a missing DB
        cmd += " -e INITIAL_LANG="
    return f"{cmd} odoo {odoo_command}"


def _dropdb_cmd(dbname):
    """Return the command to drop a DB and its filestore."""
    if ODOO_VERSION >= 20:
        return _db_run_cmd(f"odoo db drop {dbname}")
    return _db_run_cmd(f"click-odoo-dropdb {dbname}")


def _copydb_cmd(source_db, destination_db):
    """Return the command to copy a DB and its filestore."""
    if ODOO_VERSION >= 20:
        return _db_run_cmd(f"odoo db duplicate {source_db} {destination_db}")
    return _db_run_cmd(f"click-odoo-copydb {source_db} {destination_db}")


@task(
    help={
        "dbname": "The DB that will be DESTROYED and recreated. Default: 'devel'.",
        "modules": "Comma-separated list of modules to install. Default: 'base'.",
        "core": "Install all core addons. Default: False",
        "extra": "Install all extra addons. Default: False",
        "private": "Install all private addons. Default: False",
        "enterprise": "Install all enterprise addons. Default: False",
        "populate": "Run preparedb task right after (only available for v11+)."
        " Default: True",
        "dependencies": "Install only the dependencies of the specified addons."
        "Default: False",
        "demo": "Create the DB with demo data. Default: True.",
    },
)
def resetdb(
    c,
    modules=None,
    core=False,
    extra=False,
    private=False,
    enterprise=False,
    dbname="devel",
    populate=True,
    dependencies=False,
    demo=True,
):
    """Reset the specified database with the specified modules.

    Uses click-odoo-initdb behind the scenes, which has a caching system that
    makes DB resets quicker. See its docs for more info.
    """
    if dependencies:
        modules = _get_module_dependencies(c, modules, core, extra, private, enterprise)
    elif core or extra or private or enterprise:
        modules = _get_module_list(c, modules, core, extra, private, enterprise)
    else:
        modules = modules or "base"
    with c.cd(str(PROJECT_ROOT)):
        c.run(f"{DOCKER_COMPOSE_CMD} stop odoo", pty=True)
        _run = f"{DOCKER_COMPOSE_CMD} run --rm -l traefik.enable=false odoo"
        c.run(
            _dropdb_cmd(dbname),
            env=UID_ENV,
            warn=True,
            pty=True,
        )
        lang = os.getenv("INITIAL_LANG")
        lang_opt = f" --lang {lang}" if lang else ""
        demo_opt = " --demo" if demo else " --no-demo"
        c.run(
            f"{_run} click-odoo-initdb -n {dbname} -m {modules}{lang_opt}{demo_opt}",
            env=UID_ENV,
            pty=True,
        )
    if populate and ODOO_VERSION < 11:
        _logger.warn(
            f"Skipping populate task as it is not available in v{ODOO_VERSION}"
        )
        populate = False
    if populate:
        preparedb(c)


@task()
def preparedb(c):
    """Run the `preparedb` script inside the container

    Populates the DB with some helpful config
    """
    if ODOO_VERSION < 11:
        raise exceptions.PlatformError(
            "The preparedb script is not available for Doodba environments bellow v11."
        )
    with c.cd(str(PROJECT_ROOT)):
        c.run(
            f"{DOCKER_COMPOSE_CMD} run --rm -l traefik.enable=false odoo preparedb",
            env=UID_ENV,
            pty=True,
        )


@task()
def restart(c, quick=True):
    """Restart odoo container(s)."""
    cmd = f"{DOCKER_COMPOSE_CMD} restart"
    if quick:
        cmd = f"{cmd} -t0"
    cmd = f"{cmd} odoo odoo_proxy"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd, env=UID_ENV, pty=True)


@task(
    help={
        "container": "Names of the containers from which logs will be obtained."
        " You can specify a single one, or several comma-separated names."
        " Default: None (show logs for all containers)"
    },
)
def logs(c, tail=10, follow=True, container=None):
    """Obtain last logs of current environment."""
    cmd = f"{DOCKER_COMPOSE_CMD} logs"
    if follow:
        cmd += " -f"
    if tail:
        cmd += f" --tail {tail}"
    if container:
        cmd += f" {container.replace(',', ' ')}"
    with c.cd(str(PROJECT_ROOT)):
        c.run(cmd, pty=True)


@task
def after_update(c):
    """Execute some actions after a copier update or init"""
    # Make custom build scripts executable
    if ODOO_VERSION < 11:
        files = (
            Path(PROJECT_ROOT, "odoo", "custom", "build.d", "20-update-pg-repos"),
            Path(PROJECT_ROOT, "odoo", "custom", "build.d", "10-fix-certs"),
        )
        for script_file in files:
            # Ignore if, for some reason, the file didn't end up in the generated
            # project despite of the correct version (e.g. Copier exclusions)
            if not script_file.exists():
                continue
            cur_stat = script_file.stat()
            # Like chmod ug+x
            script_file.chmod(cur_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP)
    else:
        # Remove version-specific build scripts if the copier update didn't
        # HACK: https://github.com/copier-org/copier/issues/461
        files = (
            Path(PROJECT_ROOT, "odoo", "custom", "build.d", "20-update-pg-repos"),
            Path(PROJECT_ROOT, "odoo", "custom", "build.d", "10-fix-certs"),
        )
        for script_file in files:
            # missing_ok argument would take care of this, but it was only added for
            # Python 3.8
            if script_file.exists():
                script_file.unlink()


@task(
    help={
        "source_db": "The source DB name. Default: 'devel'.",
        "destination_db": (
            "The destination DB name. Default: '[SOURCE_DB_NAME]-[CURRENT_DATE]'"
        ),
    },
)
def snapshot(
    c,
    source_db="devel",
    destination_db=None,
):
    """Snapshot current database and filestore.

    Uses the DB management commands of the odoo image behind the scenes.
    """
    if not destination_db:
        destination_db = f"{source_db}-{datetime.now().strftime('%Y_%m_%d-%H_%M')}"
    with c.cd(str(PROJECT_ROOT)):
        cur_state = c.run(f"{DOCKER_COMPOSE_CMD} stop odoo db", pty=True).stdout
        _logger.info("Snapshoting current %s DB to %s", (source_db, destination_db))
        c.run(
            _copydb_cmd(source_db, destination_db),
            env=UID_ENV,
            pty=True,
        )
        if "Stopping" in cur_state:
            # Restart services if they were previously active
            c.run(f"{DOCKER_COMPOSE_CMD} start odoo db", pty=True)


@task(
    help={
        "snapshot_name": "The snapshot name. If not provided,"
        "the script will try to find the last snapshot"
        " that starts with the destination_db name",
        "destination_db": "The destination DB name. Default: 'devel'",
    },
)
def restore_snapshot(
    c,
    snapshot_name=None,
    destination_db="devel",
):
    """Restore database and filestore snapshot.

    Uses the DB management commands of the odoo image behind the scenes.
    """
    with c.cd(str(PROJECT_ROOT)):
        cur_state = c.run(f"{DOCKER_COMPOSE_CMD} stop odoo db", pty=True).stdout
        if not snapshot_name:
            # List DBs
            res = c.run(
                f"{DOCKER_COMPOSE_CMD} run --rm -e LOG_LEVEL=WARNING odoo psql -tc"
                " 'SELECT datname FROM pg_database;'",
                env=UID_ENV,
                hide="stdout",
            )
            db_list = []
            for db in res.stdout.splitlines():
                # Parse and filter DB List
                if not db.lstrip().startswith(destination_db):
                    continue
                db_name = db.lstrip()
                try:
                    db_date = datetime.strptime(
                        db_name.lstrip(f"{destination_db}-"), "%Y_%m_%d-%H_%M"
                    )
                    db_list.append((db_name, db_date))
                except ValueError:
                    continue
            snapshot_name = max(db_list, key=operator.itemgetter(1))[0]
            if not snapshot_name:
                raise exceptions.PlatformError(
                    f"No snapshot found for destination_db {destination_db}"
                )
        _logger.info("Restoring snapshot %s to %s", (snapshot_name, destination_db))
        c.run(
            _dropdb_cmd(destination_db),
            env=UID_ENV,
            warn=True,
            pty=True,
        )
        c.run(
            _copydb_cmd(snapshot_name, destination_db),
            env=UID_ENV,
            pty=True,
        )
        if "Stopping" in cur_state:
            c.run(f"{DOCKER_COMPOSE_CMD} start odoo db", pty=True)


@task(
    help={
        "module_name": "Name of the module to scaffold.",
        "path": "Path where to create the module. Default: current directory.",
    },
)
def scaffold(
    c,
    module_name,
    path=None,
):
    """Scaffold a new Odoo module.

    Creates a new Odoo module with the basic structure using odoo scaffold command.
    """
    if not module_name:
        raise exceptions.ParseError(
            msg="Module name is required. See --help for details."
        )

    # Use current directory if no path specified, otherwise use the specified path
    target_path = path or str(Path.cwd())

    # Convert the target path to be relative to PROJECT_ROOT for the container
    target_path_abs = Path(target_path).resolve()
    if not target_path_abs.is_relative_to(PROJECT_ROOT):
        raise exceptions.ParseError(
            msg=f"Path '{target_path}' must be within the project directory."
        )

    # Convert to container path
    container_path = str(target_path_abs.relative_to(PROJECT_ROOT))
    if container_path == ".":
        container_path = ""

    cmd = (
        f"{DOCKER_COMPOSE_CMD} run --rm -v "
        f'"{PROJECT_ROOT}:/tmp/project:rw" '
        f"odoo odoo scaffold {module_name} /tmp/project/{container_path}"
    )

    with c.cd(str(PROJECT_ROOT)):
        c.run(
            cmd,
            env=UID_ENV,
            pty=True,
        )