Validate and resolve file paths before API work

src/scrapingbee_cli/batch.py

@@ -217,6 +217,9 @@ def read_input_file(path: str, *, input_column: str | None = None) -> list[str]:
     If path ends with .csv, read as CSV using input_column (name or 0-based index)."""
     import sys as _sys
 
+    if path != "-":
+        path = str(Path(path).expanduser())
+
     if path == "-":
         lines = [line.strip() for line in _sys.stdin if line.strip()]
     elif path.lower().endswith(".csv"):

src/scrapingbee_cli/cli_utils.py

@@ -6,6 +6,7 @@
 import json
 import re
 import sys
+from pathlib import Path
 from typing import Any
 
 import click
@@ -243,7 +244,13 @@ def _batch_options(f: Any) -> Any:
         default=None,
         help="CSV input: column name or 0-based index.",
     )(f)
-    f = click.option("--output-dir", "output_dir", default=None, help="Batch output folder.")(f)
+    f = click.option(
+        "--output-dir",
+        "output_dir",
+        type=click.Path(),
+        default=None,
+        help="Batch output folder.",
+    )(f)
     f = click.option(
         "--output-format",
         "output_format",
@@ -313,21 +320,86 @@ def _batch_options(f: Any) -> Any:
     return f
 
 
+def resolve_output_path(path: str) -> str:
+    """Expand ``~`` in an output path."""
+    return str(Path(path).expanduser())
+
+
+def ensure_output_file_ready(
+    path: str,
+    *,
+    overwrite: bool = False,
+    skip_overwrite_check: bool = False,
+) -> str:
+    """Validate an output file path before any API work starts.
+
+    Expands ``~``, creates parent directories, and checks overwrite policy.
+    Returns the resolved path.
+    """
+    resolved = resolve_output_path(path)
+    parent = Path(resolved).parent
+    if str(parent) and not parent.exists():
+        try:
+            parent.mkdir(parents=True, exist_ok=True)
+        except OSError as e:
+            click.echo(f"Cannot create directory '{parent}': {e.strerror}", err=True)
+            raise SystemExit(1)
+    if not skip_overwrite_check:
+        confirm_overwrite(resolved, overwrite)
+    return resolved
+
+
+def ensure_output_dir_ready(path: str) -> str:
+    """Expand ``~`` and create an output directory before API work."""
+    resolved = resolve_output_path(path)
+    try:
+        Path(resolved).mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        click.echo(f"Cannot create directory '{resolved}': {e.strerror}", err=True)
+        raise SystemExit(1)
+    return resolved
+
+
+def ensure_input_file_ready(path: str) -> str:
+    """Validate an input file path before any API work starts.
+
+    Expands ``~`` and ensures the file exists and is readable. Stdin (``-``) is
+    passed through unchanged.
+    """
+    if path == "-":
+        return path
+    resolved = resolve_output_path(path)
+    p = Path(resolved)
+    if not p.is_file():
+        if p.exists():
+            click.echo(f"Not a file: '{resolved}'", err=True)
+        else:
+            click.echo(f"Input file not found: '{resolved}'", err=True)
+        raise SystemExit(1)
+    try:
+        with open(resolved, "rb"):
+            pass
+    except OSError as e:
+        click.echo(f"Cannot read from '{resolved}': {e.strerror}", err=True)
+        raise SystemExit(1)
+    return resolved
+
+
 def confirm_overwrite(path: str | None, overwrite: bool = False) -> None:
     """If path exists, prompt for confirmation unless --overwrite is set."""
     if not path:
         return
-    from pathlib import Path
+    resolved = resolve_output_path(path)
 
-    if Path(path).exists() and not overwrite:
+    if Path(resolved).exists() and not overwrite:
         # In REPL mode, prompt_toolkit owns the TTY (full-screen / alt-buffer),
         # so click.confirm reads from sys.stdin and blocks forever. Surface
         # the conflict as an error and tell the user to pass --overwrite.
         if is_repl_mode():
             raise click.UsageError(
-                f"'{path}' already exists. Re-run with --overwrite to replace it."
+                f"'{resolved}' already exists. Re-run with --overwrite to replace it."
             )
-        if not click.confirm(f"'{path}' already exists. Overwrite?"):
+        if not click.confirm(f"'{resolved}' already exists. Overwrite?"):
             click.echo("Cancelled.", err=True)
             raise SystemExit(0)
 
@@ -380,10 +452,6 @@ def store_common_options(obj: dict, **kwargs: Any) -> None:
     has_output_file = bool(obj.get("output_file"))
     has_output_dir = bool(obj.get("output_dir"))
 
-    # Check if output file already exists (skip for --update-csv which intentionally overwrites)
-    if has_output_file and not obj.get("update_csv"):
-        confirm_overwrite(obj["output_file"], obj.get("overwrite", False))
-
     # Mutual exclusion: --output-file and --output-dir
     if has_output_file and has_output_dir:
         click.echo(
@@ -520,6 +588,21 @@ def store_common_options(obj: dict, **kwargs: Any) -> None:
                 )
             raise SystemExit(1)
 
+    # Resolve paths before any API work (expand ~, mkdir outputs, validate inputs).
+    input_file_path = obj.get("input_file")
+    if isinstance(input_file_path, str):
+        obj["input_file"] = ensure_input_file_ready(input_file_path)
+    output_file_path = obj.get("output_file")
+    if isinstance(output_file_path, str):
+        obj["output_file"] = ensure_output_file_ready(
+            output_file_path,
+            overwrite=bool(obj.get("overwrite", False)),
+            skip_overwrite_check=bool(obj.get("update_csv")),
+        )
+    output_dir_path = obj.get("output_dir")
+    if isinstance(output_dir_path, str) and output_dir_path:
+        obj["output_dir"] = ensure_output_dir_ready(output_dir_path)
+
 
 def _parse_path(path: str) -> list[tuple[str, Any]]:
     """Parse a path expression into typed segments.
@@ -1819,10 +1902,18 @@ def write_output(
     elif fields:
         data = _filter_fields(data, fields)
     if output_path:
+        resolved = resolve_output_path(output_path)
+        parent = Path(resolved).parent
+        if str(parent) and not parent.exists():
+            try:
+                parent.mkdir(parents=True, exist_ok=True)
+            except OSError as e:
+                click.echo(f"Cannot create directory '{parent}': {e.strerror}", err=True)
+                raise SystemExit(1)
         try:
-            fh = open(output_path, "wb")
+            fh = open(resolved, "wb")
         except OSError as e:
-            click.echo(f"Cannot write to '{output_path}': {e.strerror}", err=True)
+            click.echo(f"Cannot write to '{resolved}': {e.strerror}", err=True)
             raise SystemExit(1)
         with fh:
             fh.write(data)
@@ -1832,7 +1923,7 @@ def write_output(
         # stdout/pipes clean for non-REPL use.
         from .theme import BEE_DIM, BEE_YELLOW, err_console
 
-        err_console.print(f"  [{BEE_DIM}]Saved to[/] [bold {BEE_YELLOW}]{output_path}[/]")
+        err_console.print(f"  [{BEE_DIM}]Saved to[/] [bold {BEE_YELLOW}]{resolved}[/]")
     else:
         # In REPL mode, truncate large text dumps to a tidy preview and surface
         # a path to the full output. Non-REPL invocations (`scrapingbee scrape ...`)

src/scrapingbee_cli/commands/crawl.py

@@ -14,6 +14,7 @@
     _validate_json_option,
     _validate_range,
     build_scrape_kwargs,
+    ensure_output_dir_ready,
     scrape_kwargs_to_api_params,
     store_common_options,
 )
@@ -427,7 +428,10 @@ def crawl_cmd(
     See https://www.scrapingbee.com/documentation/ for parameter details.
     """
     store_common_options(obj, **kwargs)
-    obj["output_dir"] = output_dir or ""
+    if output_dir:
+        obj["output_dir"] = ensure_output_dir_ready(output_dir)
+    else:
+        obj["output_dir"] = ""
     obj["concurrency"] = concurrency or 0
     obj["resume"] = resume
     obj["on_complete"] = on_complete
@@ -562,6 +566,7 @@ def crawl_cmd(
             custom_headers[k.strip()] = v.strip()
         out_dir = (obj.get("output_dir") or "").strip() or None
         out_dir = out_dir or default_crawl_output_dir()
+        out_dir = ensure_output_dir_ready(out_dir)
         allowed_list: list[str] | None = None
         if allowed_domains:
             allowed_list = [d.strip() for d in allowed_domains.split(",") if d.strip()]

src/scrapingbee_cli/commands/export.py

@@ -93,10 +93,12 @@ def export_cmd(
     input_path = Path(input_dir).resolve()
     output_file = obj.get("output_file")
 
-    # Check if output file already exists
-    from ..cli_utils import confirm_overwrite
+    # Check if output file already exists and prepare destination before reading inputs.
+    from ..cli_utils import ensure_output_file_ready
 
-    confirm_overwrite(output_file, overwrite)
+    if output_file:
+        output_file = ensure_output_file_ready(output_file, overwrite=overwrite)
+        obj["output_file"] = output_file
 
     # Load manifest for URL → relative-path mapping (optional)
     # Supports both old format (string values) and new format (dict values with "file" key).

tests/unit/test_repl_pty.py

@@ -314,3 +314,32 @@ def test_session_default_no_skip_warning_for_supported_command(tmp_path):
         )
     finally:
         child.close(force=True)
+
+
+@needs_cli
+def test_repl_tilde_output_file_validated_before_scrape(tmp_path):
+    """REPL expands ~/ output paths and rejects overwrites before any scrape work."""
+    home = tmp_path / "home"
+    home.mkdir()
+    out_dir = home / "Desktop" / "sb-cli-test" / "ss"
+    out_dir.mkdir(parents=True)
+    (out_dir / "screenshot.png").write_bytes(b"old")
+
+    child, screen, stream = _spawn(home)
+    try:
+        assert _pump_until(child, screen, stream, lambda s: "❯" in _text(s)), "no prompt"
+        child.send(
+            "scrape https://example.com --render-js false "
+            "--output-file ~/Desktop/sb-cli-test/ss/screenshot.png\r"
+        )
+        assert _pump_until(
+            child,
+            screen,
+            stream,
+            lambda s: "already exists" in _text(s) and "--overwrite" in _text(s),
+            timeout=10.0,
+        ), "overwrite error not shown before scrape"
+        t = _text(screen)
+        assert "Cannot write to '~/" not in t
+    finally:
+        child.close(force=True)