ObsiGate/backend/tools/schemas.py

"""Pydantic input/output models for the AI tool layer.

Input models double as JSON Schemas advertised to LLMs (via
``model_json_schema``) and as validation for arguments received over MCP.
"""

from __future__ import annotations

from typing import Any

from pydantic import BaseModel, Field


class ListVaultsInput(BaseModel):
    """No parameters — lists the vaults the caller can access."""


class ListDirectoryInput(BaseModel):
    """Browse a directory inside a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field("", description="Vault-relative directory path (empty = vault root)")


class ReadFileInput(BaseModel):
    """Read a text file's content from a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class SearchFulltextInput(BaseModel):
    """Full-text search across one or all accessible vaults."""

    q: str = Field(..., min_length=1, description="Search query")
    vault: str = Field("all", description="Vault name or 'all'")
    tag: str | None = Field(None, description="Optional comma-separated tag filter")
    limit: int = Field(50, ge=1, le=200, description="Maximum number of results")


class ListTagsInput(BaseModel):
    """List tags and their occurrence counts."""

    vault: str | None = Field(None, description="Vault name or 'all' (default: all accessible)")


class ListAllFilesInput(BaseModel):
    """List every file of a vault (optionally under a subdirectory)."""

    vault: str = Field(..., description="Vault name")
    dir: str = Field("", description="Vault-relative directory path (empty = vault root)")
    limit: int = Field(200, ge=1, le=2000, description="Maximum number of files")
    recursive: bool = Field(True, description="Recurse into subdirectories")


class ReadFileRawInput(BaseModel):
    """Read a text file's raw content (no HTML rendering)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class GetBacklinksInput(BaseModel):
    """List files linking to a target file via wikilinks."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the target file")


class ListBackupsInput(BaseModel):
    """List available backup versions of a file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class DiffBackupInput(BaseModel):
    """Unified diff between a backup version and another version or the current file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    version: int = Field(..., description="Backup timestamp used as the old/left side")
    compare_with: int | None = Field(
        None,
        description="Backup timestamp used as the new/right side (omit to compare with the current file)",
    )


class GetGraphInput(BaseModel):
    """Graph data (nodes and edges) for a vault or directory."""

    vault: str = Field(..., description="Vault name")
    path: str = Field("", description="Vault-relative directory path to focus on (empty = root)")
    depth: int = Field(1, ge=0, le=3, description="Expansion depth")
    scope: str = Field("directory", description="'directory' for a subtree, 'full' for the whole vault")
    tag: str = Field("", description="Only include files carrying this tag")


class SearchAdvancedInput(BaseModel):
    """Advanced full-text search (TF-IDF, facets, operators)."""

    q: str = Field("", description="Query (supports tag:, vault:, title:, path:, ext: operators)")
    vault: str = Field("all", description="Vault name or 'all'")
    tag: str | None = Field(None, description="Optional comma-separated tag filter")
    limit: int = Field(50, ge=1, le=200, description="Maximum number of results")
    offset: int = Field(0, ge=0, description="Pagination offset")
    sort: str = Field("relevance", description="Sort by 'relevance' or 'modified'")
    case_sensitive: bool = Field(False, description="Match case")
    whole_word: bool = Field(False, description="Match whole words only")
    regex: bool = Field(False, description="Treat query as a regular expression")
    include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
    exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
    created: str | None = Field(None, description="Created date filter (>date, <date, date..date)")
    modified: str | None = Field(None, description="Modified date filter (>date, <date, date..date, <Nd)")
    size: str | None = Field(None, description="Size filter (>size, <size, size..size, e.g. >1MB)")


class SearchPathsInput(BaseModel):
    """Search files and directories by path substring."""

    q: str = Field(..., min_length=1, description="Path substring to search for")
    vault: str = Field("all", description="Vault name or 'all'")


class SuggestTagsInput(BaseModel):
    """Suggest tags matching a prefix."""

    q: str = Field(..., min_length=1, description="Tag prefix (with or without leading '#')")
    vault: str = Field("all", description="Vault name or 'all'")
    limit: int = Field(10, ge=1, le=50, description="Maximum number of suggestions")


class ListRecentInput(BaseModel):
    """List recently opened (or modified) files."""

    vault: str | None = Field(None, description="Optional single-vault filter")
    limit: int = Field(20, ge=1, le=200, description="Maximum number of files")
    mode: str = Field("opened", description="'opened' (history) or 'modified'")


# ── D. Mutations ───────────────────────────────────────────────────────────


class CreateFileInput(BaseModel):
    """Create a new text file in a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the new file")
    content: str = Field("", description="Initial file content")


class CreateDirectoryInput(BaseModel):
    """Create a new directory in a vault."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the new directory")


class EditFileInput(BaseModel):
    """Overwrite the full content of an existing file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    content: str = Field(..., description="New full content of the file")


class AppendToFileInput(BaseModel):
    """Append text to the end of an existing file."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    content: str = Field(..., description="Text to append")


class RenameFileInput(BaseModel):
    """Rename a file in place (same parent directory)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Current vault-relative file path")
    new_name: str = Field(..., description="New file name (no directory separator)")


class RenameDirectoryInput(BaseModel):
    """Rename a directory in place (same parent directory)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Current vault-relative directory path")
    new_name: str = Field(..., description="New directory name (no directory separator)")


class MovePathInput(BaseModel):
    """Move a file or directory to another directory within the same vault."""

    vault: str = Field(..., description="Vault name")
    source_path: str = Field(..., description="Current vault-relative path of the file/directory")
    destination_dir: str = Field("", description="Target directory path (empty = vault root)")


class ReplaceInFilesInput(BaseModel):
    """Find and replace text across vault files (destructive, dry-run by default)."""

    find: str = Field(..., min_length=1, description="Text or pattern to search for")
    replace: str = Field("", description="Replacement text")
    vault: str = Field("all", description="Vault name or 'all'")
    case_sensitive: bool = Field(False, description="Match case")
    whole_word: bool = Field(False, description="Match whole words only")
    regex: bool = Field(False, description="Treat 'find' as a regular expression")
    include_paths: str | None = Field(None, description="Comma-separated glob patterns to include")
    exclude_paths: str | None = Field(None, description="Comma-separated glob patterns to exclude")
    replace_all: bool = Field(False, description="Apply the replacement (otherwise only preview)")
    dry_run: bool | None = Field(
        None,
        description="Preview matches without writing. Defaults to the opposite of replace_all.",
    )


class DeleteFileInput(BaseModel):
    """Delete a file from a vault (destructive)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")


class DeleteDirectoryInput(BaseModel):
    """Delete a directory from a vault (destructive)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative directory path")
    recursive: bool = Field(True, description="Delete non-empty directories recursively")


class RestoreBackupInput(BaseModel):
    """Restore a file from one of its backup versions."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative file path")
    version: int = Field(..., description="Backup timestamp to restore")


class WebSearchInput(BaseModel):
    """Search the public web through the configured meta-search engine."""

    query: str = Field(..., description="Search terms")
    max_results: int = Field(5, ge=1, le=10, description="Number of results to return")
    category: str = Field("", description="Optional engine category: general, news, it, science")
    language: str = Field("", description="Optional language code, e.g. 'fr'")
    page: int = Field(1, ge=1, le=10, description="Result page number")


class FetchUrlInput(BaseModel):
    """Fetch one public web page and return its readable text."""

    url: str = Field(..., description="Absolute http(s) URL of a public page")
    render: bool = Field(
        False,
        description="Render JavaScript with the optional Playwright worker (dynamic SPA pages)",
    )


class CrawlSiteInput(BaseModel):
    """Crawl a small public site (same-host only) and save a digest into a vault."""

    url: str = Field(..., description="Absolute http(s) URL where the crawl starts")
    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the digest file to write (.md)")
    max_pages: int = Field(5, ge=1, le=20, description="Maximum number of pages to crawl")


class GitProviderInput(BaseModel):
    """Base fields for connected-source tools (Gitea / GitHub)."""

    provider: str = Field(..., description="'gitea' (OBSIGATE_GITEA_URL) or 'github'")
    repo: str = Field("", description="Optional 'owner/name' repository filter")
    limit: int = Field(20, ge=1, le=50, description="Maximum number of entries")


class GitSearchIssuesInput(BaseModel):
    """Search issues/pull requests on a connected Gitea or GitHub instance."""

    provider: str = Field(..., description="'gitea' or 'github'")
    query: str = Field(..., min_length=1, description="Search keywords")
    repo: str = Field("", description="Optional 'owner/name' scope (empty = instance-wide)")
    state: str = Field("open", description="'open' or 'closed'")
    limit: int = Field(10, ge=1, le=20, description="Maximum number of issues")


class GitGetFileInput(BaseModel):
    """Read a file from a connected Gitea or GitHub repository."""

    provider: str = Field(..., description="'gitea' or 'github'")
    repo: str = Field(..., description="'owner/name' repository")
    path: str = Field(..., description="Repository-relative file path")
    ref: str = Field("", description="Optional branch/tag/commit (empty = default branch)")


class SpreadsheetInput(BaseModel):
    """Create an .xlsx spreadsheet in a vault from rows of cells."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the file to write (.xlsx)")
    rows: list[list[str | int | float | bool | None]] = Field(
        ..., description="Rows of cell values (first row = header)"
    )
    sheet_name: str = Field("Feuille1", description="Worksheet name")
    overwrite: bool = Field(True, description="Replace an existing file (with backup)")


class DocxInput(BaseModel):
    """Create a .docx Word document in a vault from paragraphs."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the file to write (.docx)")
    title: str = Field("", description="Optional document title (heading 1)")
    paragraphs: list[str] = Field(..., description="Paragraph texts, in order")
    overwrite: bool = Field(True, description="Replace an existing file (with backup)")


class ListXlsxSheetsInput(BaseModel):
    """List the sheets of an existing .xlsx workbook (#153 A6)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the .xlsx file")


class XlsxToMarkdownInput(BaseModel):
    """Read one sheet of an existing .xlsx workbook as markdown (#153 A6)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the .xlsx file")
    sheet: str = Field(
        "", description="Sheet name (empty = the first/active sheet)"
    )


class SearchWorkbookInput(BaseModel):
    """Find a text across the sheets of a spreadsheet (#156 A14)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(
        ..., description="Vault-relative path of the file (.xlsx, .xlsm or .csv)"
    )
    query: str = Field(..., description="Text to look for")
    sheet: str = Field("", description="Restrict to one sheet (empty = all sheets)")
    case_sensitive: bool = Field(False, description="Match case")


class AnalyzeRangeInput(BaseModel):
    """Aggregate the values of an A1 range (#156 A14)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(
        ..., description="Vault-relative path of the file (.xlsx, .xlsm or .csv)"
    )
    sheet: str = Field(
        "", description="Sheet name (empty = the first/active sheet)"
    )
    range: str = Field(
        "",
        description="A1 range to analyse (e.g. 'B2:B50'); empty = the whole sheet",
    )


class UpdateXlsxCellsInput(BaseModel):
    """Batch-edit cells of an existing .xlsx workbook (#153 A6)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the .xlsx file")
    sheet: str = Field(
        "", description="Worksheet title to edit (ignored for a .csv)"
    )
    cells: dict[str, str | int | float | bool | None] = Field(
        ..., description="A1 reference -> new value (max 500 per call)"
    )
    allow_formula: bool = Field(
        False,
        description="Store '='/'@' values as real formulas (off by default, DDE guard)",
    )
    force: bool = Field(
        False,
        description="Write even when features openpyxl cannot rewrite would be dropped",
    )


class AppendXlsxRowsInput(BaseModel):
    """Append rows at the end of a sheet of an existing .xlsx (#153 A6)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the .xlsx file")
    sheet: str = Field(..., description="Worksheet title to extend")
    rows: list[list[str | int | float | bool | None]] = Field(
        ..., description="Rows of cell values, appended below the last used row (max 500)"
    )
    allow_formula: bool = Field(
        False,
        description="Store '='/'@' values as real formulas (off by default, DDE guard)",
    )
    force: bool = Field(
        False,
        description="Write even when features openpyxl cannot rewrite would be dropped",
    )


class EditXlsxStructureInput(BaseModel):
    """Structural CRUD on an existing workbook (#156 A14)."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the .xlsx/.xlsm file")
    actions: list[dict[str, Any]] = Field(
        ...,
        description=(
            "Ordered structural actions (1-50): sheet_add/sheet_rename/"
            "sheet_duplicate/sheet_delete, row_insert/row_delete/"
            "col_insert/col_delete"
        ),
    )
    force: bool = Field(
        False,
        description="Write even when features openpyxl cannot rewrite would be dropped",
    )


class CsvInput(BaseModel):
    """Create a .csv file in a vault from rows of cells."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the file to write (.csv)")
    rows: list[list[str | int | float | bool | None]] = Field(
        ..., description="Rows of cell values (first row = header)"
    )
    delimiter: str = Field(",", description="Field separator (',' ';' '\\t')")
    overwrite: bool = Field(True, description="Replace an existing file (with backup)")


class PdfInput(BaseModel):
    """Create a .pdf document in a vault from markdown-ish content."""

    vault: str = Field(..., description="Vault name")
    path: str = Field(..., description="Vault-relative path of the file to write (.pdf)")
    title: str = Field("Document", description="Document title")
    content: str = Field(..., description="Content (headings with #/##, then paragraphs)")
    overwrite: bool = Field(True, description="Replace an existing file (with backup)")


class FindDuplicatesInput(BaseModel):
    """Find candidate duplicate notes in a vault (#166)."""

    vault: str = Field(..., description="Vault name")
    threshold: float = Field(0.75, ge=0.3, le=1.0, description="Minimum similarity score")
    limit: int = Field(20, ge=1, le=200, description="Maximum number of pairs")
    subdir: str = Field("", description="Vault-relative directory scope (empty = whole vault)")


class MergeDuplicatesInput(BaseModel):
    """Merge one note into another, then delete the source (#166, destructive)."""

    vault: str = Field(..., description="Vault name")
    source_path: str = Field(..., description="Vault-relative path of the note to absorb")
    target_path: str = Field(..., description="Vault-relative path of the surviving note")
    strategy: str = Field("append", description="'append', 'prefer_target' or 'prefer_source'")


class NotifyExternalInput(BaseModel):
    """Send a notification through external channels (#168)."""

    title: str = Field(..., min_length=1, description="Notification title")
    message: str = Field(..., min_length=1, description="Notification body")
    trigger: str = Field("manual", description="Trigger scope: manual, schedule_failure, schedule_success")
    channel_id: str = Field("", description="Single channel id (empty = broadcast to trigger)")


class CreateScheduledTaskInput(BaseModel):
    """Create an automatic task executed by the scheduler (#170)."""

    name: str = Field(..., min_length=1, description="Task display name")
    action: dict[str, Any] = Field(..., description="{kind, params} (create_file, append_to_file, notify)")
    schedule: dict[str, Any] = Field(..., description="{kind, ...} (interval_hours, daily_time, once_at)")


class DeleteScheduledTaskInput(BaseModel):
    """Delete a scheduled task by id (#170)."""

    task_id: str = Field(..., min_length=1, description="Task id")


class RunScheduledTaskInput(BaseModel):
    """Execute a scheduled task immediately (#170)."""

    task_id: str = Field(..., min_length=1, description="Task id")


class ToolResult(BaseModel):
    """Uniform result returned by :func:`backend.tools.registry.call_tool`."""

    ok: bool = True
    data: Any = None
    error: str | None = None