tasks.py

"""Tasks for maintaining the project.

Execute 'invoke --list' for guidance on using Invoke
"""

import os
import platform
import tempfile
import webbrowser
from pathlib import Path
from typing import Optional

from invoke import call, task
from invoke.context import Context
from invoke.runners import Result

ROOT_DIR = Path(__file__).parent
DOCS_DIR = ROOT_DIR.joinpath("docs")
DOCS_BUILD_DIR = DOCS_DIR.joinpath("_build")
DOCS_INDEX = DOCS_BUILD_DIR.joinpath("index.html")
COVERAGE_FILE = ROOT_DIR.joinpath(".coverage")
COVERAGE_DIR = ROOT_DIR.joinpath("htmlcov")
COVERAGE_REPORT = COVERAGE_DIR.joinpath("index.html")
SOURCE_DIR = ROOT_DIR.joinpath("src/ai_marketplace_monitor")
TEST_DIR = ROOT_DIR.joinpath("tests")
PYTHON_TARGETS = [
    SOURCE_DIR,
    TEST_DIR,
    DOCS_DIR.joinpath("conf.py"),
    ROOT_DIR.joinpath("noxfile.py"),
    Path(__file__),
]
PYTHON_TARGETS_STR = " ".join([str(p) for p in PYTHON_TARGETS])


def _run(c: Context, command: str) -> Optional[Result]:
    return c.run(command, pty=platform.system() != "Windows")


@task()
def clean_build(c: Context) -> None:
    """Clean up files from package building."""
    _run(c, "rm -fr build/")
    _run(c, "rm -fr dist/")
    _run(c, "rm -fr .eggs/")
    _run(c, "find . -name '*.egg-info' -exec rm -fr {} +")
    _run(c, "find . -name '*.egg' -exec rm -f {} +")


@task()
def clean_python(c: Context) -> None:
    """Clean up python file artifacts."""
    _run(c, "find . -name '*.pyc' -exec rm -f {} +")
    _run(c, "find . -name '*.pyo' -exec rm -f {} +")
    _run(c, "find . -name '*~' -exec rm -f {} +")
    _run(c, "find . -name '__pycache__' -exec rm -fr {} +")


@task()
def clean_tests(c: Context) -> None:
    """Clean up files from testing."""
    _run(c, f"rm -f {COVERAGE_FILE}")
    _run(c, f"rm -fr {COVERAGE_DIR}")
    _run(c, "rm -fr .pytest_cache")


@task()
def clean_docs(c: Context) -> None:
    """Clean up files from documentation builds."""
    _run(c, f"rm -fr {DOCS_BUILD_DIR}")


@task(pre=[clean_build, clean_python, clean_tests, clean_docs])
def clean(c: Context) -> None:
    """Run all clean sub-tasks."""


@task()
def install_hooks(c: Context) -> None:
    """Install pre-commit hooks."""
    _run(c, "uv run pre-commit install")


@task()
def hooks(c: Context) -> None:
    """Run pre-commit hooks."""
    _run(c, "uv run pre-commit run --all-files")


@task(name="format", help={"check": "Checks if source is formatted without applying changes"})
def format_(c: Context, check: bool = False) -> None:
    """Format code."""
    # Fix whitespace and end-of-file issues using the same tools as pre-commit
    if not check:
        _run(c, "uv run pre-commit run trailing-whitespace --all-files")
        _run(c, "uv run pre-commit run end-of-file-fixer --all-files")

    # Run isort and black
    isort_options = ["--check-only", "--diff"] if check else []
    _run(c, f"uv run isort {' '.join(isort_options)} {PYTHON_TARGETS_STR}")
    black_options = ["--diff", "--check"] if check else ["--quiet"]
    _run(c, f"uv run black {' '.join(black_options)} {PYTHON_TARGETS_STR}")


@task()
def ruff(c: Context) -> None:
    """Run ruff."""
    _run(c, f"uv run ruff check {PYTHON_TARGETS_STR}")


@task()
def security(c: Context) -> None:
    """Run security related checks."""
    with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
        temp_file = f.name
    try:
        _run(c, f"uv export --extra dev --format requirements-txt --no-hashes  > {temp_file}")
        _run(
            c,
            f"uv run pip-audit --requirement {temp_file} --format json",
        )
    finally:
        # Clean up
        if os.path.exists(temp_file):
            os.unlink(temp_file)
    return None


@task(pre=[ruff, call(format_, check=True)])
def lint(c: Context) -> None:
    """Run all linting."""


@task()
def mypy(c: Context) -> None:
    """Run mypy."""
    _run(c, f"uv run mypy {PYTHON_TARGETS_STR}")


@task()
def tests(c: Context) -> None:
    """Run tests."""
    pytest_options = ["--xdoctest", "--cov", "--cov-report=", "--cov-fail-under=0"]
    _run(c, f"uv run pytest {' '.join(pytest_options)} {TEST_DIR} {SOURCE_DIR}")


@task(
    help={
        "fmt": "Build a local report: report, html, json, annotate, html, xml.",
        "open_browser": "Open the coverage report in the web browser (requires --fmt html)",
    }
)
def coverage(c: Context, fmt: str = "report", open_browser: bool = False) -> None:
    """Create coverage report."""
    if any(Path().glob(".coverage.*")):
        _run(c, "uv run coverage combine")
    _run(c, f"uv run coverage {fmt} -i")
    if fmt == "html" and open_browser:
        webbrowser.open(COVERAGE_REPORT.as_uri())


@task(
    help={
        "serve": "Build the docs watching for changes",
        "open_browser": "Open the docs in the web browser",
    }
)
def docs(c: Context, serve: bool = False, open_browser: bool = False) -> None:
    """Build documentation."""
    _run(c, f"uv run sphinx-apidoc -f -o {DOCS_DIR} {SOURCE_DIR}")
    build_docs = f"uv run sphinx-build -b html {DOCS_DIR} {DOCS_BUILD_DIR}"
    _run(c, build_docs)
    if open_browser:
        webbrowser.open(DOCS_INDEX.absolute().as_uri())
    if serve:
        _run(c, f"uv run watchmedo shell-command -p '*.rst;*.md' -c '{build_docs}' -R -D .")


@task(
    help={
        "part": "Part of the version to be bumped.",
        "dry_run": "Don't write any files, just pretend. (default: False)",
    }
)
def version(c: Context, part: str, dry_run: bool = False) -> None:
    """Bump version."""
    bump_options = ["--dry-run"] if dry_run else []
    _run(c, f"uv run bump2version {' '.join(bump_options)} {part}")


@task(
    help={
        "version": "Version to release (e.g., 1.0.0)",
    }
)
def release(c: Context, version: str) -> None:
    """Release a new version by updating version files and building/uploading to PyPI."""
    if not version or version.startswith("-"):
        print("Error: Please provide a valid version number (e.g., inv release --version=1.0.0)")
        return

    print(f"Releasing version {version}...")

    # Update version in all required files
    files_to_update = [
        ("tests/test_aimm.py", 'assert version == "', '"'),
        ("src/ai_marketplace_monitor/__init__.py", '__version__ = "', '"'),
        ("pyproject.toml", 'version = "', '"'),
    ]

    for file_path, prefix, suffix in files_to_update:
        print(f"Updating version in {file_path}")
        with open(file_path, "r") as f:
            content = f.read()

        # Find and replace the version line
        lines = content.split("\n")
        for i, line in enumerate(lines):
            if prefix in line:
                start = line.find(prefix) + len(prefix)
                end = line.find(suffix, start)
                if end != -1:
                    lines[i] = line[:start] + version + line[end:]
                    break

        with open(file_path, "w") as f:
            f.write("\n".join(lines))

    # Update CHANGELOG.md
    print("Updating CHANGELOG.md...")
    changelog_path = "CHANGELOG.md"
    with open(changelog_path, "r") as f:
        changelog_content = f.read()

    # Find the [Unreleased] section
    lines = changelog_content.split("\n")
    updated_lines = []
    i = 0
    while i < len(lines):
        if lines[i].strip() == "## [Unreleased]":
            # Add the [Unreleased] header
            updated_lines.append(lines[i])

            # Collect all content under [Unreleased] until the next section
            i += 1
            unreleased_content = []
            while i < len(lines) and not lines[i].startswith("## ["):
                if lines[i].strip():  # Non-empty content line
                    unreleased_content.append(lines[i])
                i += 1

            # If there's content to move to the new version
            if unreleased_content:
                # Keep [Unreleased] section empty
                updated_lines.append("")

                # Add new version section with the content
                updated_lines.append(f"## [{version}]")
                updated_lines.append("")
                updated_lines.extend(unreleased_content)
            else:
                # Just add empty line after [Unreleased]
                updated_lines.append("")

            # Continue with the rest of the file (next version sections)
            if i < len(lines) and lines[i].startswith("## ["):
                updated_lines.append("")  # Add blank line before next section
                updated_lines.append(lines[i])
                i += 1
        else:
            updated_lines.append(lines[i])
            i += 1

    # Write back the updated changelog
    with open(changelog_path, "w") as f:
        f.write("\n".join(updated_lines))

    # Clean previous builds
    print("Cleaning previous builds...")
    _run(c, "rm -fr build/ dist/ *.egg-info")

    # Build the package
    print("Building package...")
    _run(c, "uv run python -m build")

    # Upload to PyPI
    print("Uploading to PyPI...")
    _run(c, "uv run python -m twine upload dist/*")

    print(f"Successfully released version {version}!")