from __future__ import annotations

import subprocess
import sys
from dataclasses import dataclass
from types import SimpleNamespace

from services.crawler.application import MediaCrawlerRuntimeService
from services.crawler.domain import CrawlerProvider, CrawlerRuntimePaths, ProviderKind


@dataclass
class _FakePathState:
    exists: bool = True
    text: str = ""
    mtime: float = 1.0


class _FakePath:
    def __init__(self, value: str, storage: dict[str, _FakePathState]) -> None:
        self.value = value
        self._storage = storage

    def __truediv__(self, child: str) -> "_FakePath":
        return _FakePath(f"{self.value}/{child}", self._storage)

    def exists(self) -> bool:
        state = self._storage.get(self.value)
        return bool(state and state.exists)

    def stat(self):
        state = self._storage[self.value]
        return SimpleNamespace(st_mtime=state.mtime)

    def write_text(self, content: str, encoding: str = "utf-8") -> int:
        state = self._storage.setdefault(self.value, _FakePathState())
        state.exists = True
        state.text = content
        state.mtime += 1
        return len(content)

    def read_text(self, encoding: str = "utf-8") -> str:
        return self._storage[self.value].text

    def touch(self) -> None:
        state = self._storage.setdefault(self.value, _FakePathState())
        state.exists = True
        state.mtime += 1

    def mkdir(self, parents: bool = False, exist_ok: bool = False) -> None:
        state = self._storage.setdefault(self.value, _FakePathState())
        state.exists = True

    def __str__(self) -> str:
        return self.value


class _FakeMediaCrawler:
    def __init__(self, paths: CrawlerRuntimePaths) -> None:
        self.paths = paths

    def get_provider(self) -> CrawlerProvider:
        return CrawlerProvider(kind=ProviderKind.MEDIA_CRAWLER, paths=self.paths)

    def get_paths(self) -> CrawlerRuntimePaths:
        return self.paths

    def ensure_available(self) -> CrawlerRuntimePaths:
        return self.paths


def _build_paths() -> tuple[CrawlerRuntimePaths, dict[str, _FakePathState]]:
    storage: dict[str, _FakePathState] = {
        "/mediacrawler": _FakePathState(exists=True, mtime=1),
        "/mediacrawler/web_runtime.py": _FakePathState(exists=True, text="print('runtime')\n", mtime=1),
        "/mediacrawler/requirements.txt": _FakePathState(exists=True, text="playwright\n", mtime=5),
        "/mediacrawler/config/db_config.py": _FakePathState(exists=True, text="# placeholder\n", mtime=1),
        "/mediacrawler/config/base_config.py": _FakePathState(
            exists=True,
            text="\n".join(
                [
                    'PLATFORM = "xhs"',
                    "KEYWORDS = (",
                    '    "old"',
                    ")",
                    'CRAWLER_TYPE = "search"',
                    'SAVE_DATA_OPTION = "db"',
                    "CRAWLER_MAX_NOTES_COUNT = 10",
                    "ENABLE_GET_COMMENTS = False",
                    "CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES = 5",
                    "HEADLESS = False",
                    'USER_DATA_DIR = "%s_user_data_dir"',
                ]
            ),
            mtime=1,
        ),
        "/runtime/crawler/browser_data": _FakePathState(exists=True, mtime=1),
    }
    paths = CrawlerRuntimePaths(
        root=_FakePath("/mediacrawler", storage),
        runtime=_FakePath("/mediacrawler/web_runtime.py", storage),
        requirements=_FakePath("/mediacrawler/requirements.txt", storage),
        db_config=_FakePath("/mediacrawler/config/db_config.py", storage),
        base_config=_FakePath("/mediacrawler/config/base_config.py", storage),
        browser_data=_FakePath("/runtime/crawler/browser_data", storage),
    )
    return paths, storage


def test_build_runtime_env_uses_reloaded_settings_and_database_aliases():
    reload_calls: list[str] = []
    settings = SimpleNamespace(
        DB_HOST="db.internal",
        DB_PORT=15432,
        DB_USER="crawler_user",
        DB_PASSWORD="crawler_pwd",
        DB_NAME="crawler_db",
    )
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: reload_calls.append("reloaded"),
        settings_getter=lambda: settings,
        environ={"EXISTING": "1"},
    )

    env = service.build_runtime_env()

    assert reload_calls == ["reloaded"]
    assert env["EXISTING"] == "1"
    assert env["POSTGRES_DB_HOST"] == "db.internal"
    assert env["POSTGRES_DB_PORT"] == "15432"
    assert env["POSTGRES_DB_USER"] == "crawler_user"
    assert env["POSTGRES_DB_PWD"] == "crawler_pwd"
    assert env["POSTGRES_DB_NAME"] == "crawler_db"
    assert env["MYSQL_DB_HOST"] == "db.internal"
    assert env["MYSQL_DB_NAME"] == "crawler_db"
    assert env["BETTAFISH_CRAWLER_BROWSER_DATA_DIR"] == "/runtime/crawler/browser_data"
    assert env["MEDIACRAWLER_USER_DATA_DIR_PATTERN"] == "/runtime/crawler/browser_data/%s_user_data_dir"


def test_build_execution_context_reuses_runtime_env_and_paths():
    settings = SimpleNamespace(
        DB_HOST="db.internal",
        DB_PORT=15432,
        DB_USER="crawler_user",
        DB_PASSWORD="crawler_pwd",
        DB_NAME="crawler_db",
    )
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
        environ={"EXISTING": "1"},
    )

    context = service.build_execution_context()

    assert str(context.cwd) == "/mediacrawler"
    assert str(context.runtime_path) == "/mediacrawler/web_runtime.py"
    assert context.env["EXISTING"] == "1"
    assert context.env["POSTGRES_DB_HOST"] == "db.internal"


def test_get_browser_data_dir_returns_runtime_managed_writable_directory():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))

    assert str(service.get_browser_data_dir()) == "/runtime/crawler/browser_data"
    assert str(service.ensure_browser_data_dir()) == "/runtime/crawler/browser_data"
    assert service.get_browser_user_data_dir_pattern() == "/runtime/crawler/browser_data/%s_user_data_dir"


def test_build_completed_process_kwargs_reuses_execution_context_and_adds_text_run_defaults():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))
    execution_context = SimpleNamespace(cwd="/runtime-root", env={"ENV": "1"})

    kwargs = service.build_completed_process_kwargs(
        execution_context=execution_context,
        timeout=120,
    )

    assert kwargs == {
        "cwd": "/runtime-root",
        "env": {"ENV": "1"},
        "capture_output": True,
        "text": True,
        "encoding": "utf-8",
        "errors": "replace",
        "timeout": 120,
    }


def test_build_streaming_process_kwargs_reuses_execution_context_and_adds_text_popen_defaults():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))
    execution_context = SimpleNamespace(cwd="/runtime-root", env={"ENV": "1"})

    kwargs = service.build_streaming_process_kwargs(execution_context=execution_context)

    assert kwargs == {
        "cwd": "/runtime-root",
        "env": {"ENV": "1"},
        "stdout": subprocess.PIPE,
        "stderr": subprocess.STDOUT,
        "text": True,
        "encoding": "utf-8",
        "errors": "replace",
        "bufsize": 1,
    }


def test_get_default_save_data_option_follows_database_dialect():
    settings = SimpleNamespace(DB_DIALECT="postgresql")
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    assert service.get_default_save_data_option() == "postgres"


def test_resolve_save_data_option_prefers_explicit_value_and_falls_back_to_default():
    settings = SimpleNamespace(DB_DIALECT="postgresql")
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    assert service.resolve_save_data_option("json") == "json"
    assert service.resolve_save_data_option("") == "postgres"


def test_build_async_database_url_uses_normalized_runtime_settings():
    settings = SimpleNamespace(
        DB_DIALECT="mysql",
        DB_HOST="db.internal",
        DB_PORT=3307,
        DB_USER="crawler_user",
        DB_PASSWORD="secret value",
        DB_NAME="crawler_db",
        DB_CHARSET="utf8mb4",
    )
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    url = service.build_async_database_url()

    assert url == (
        "mysql+asyncmy://crawler_user:secret+value@db.internal:3307/"
        "crawler_db?charset=utf8mb4"
    )


def test_build_python_command_allows_reusing_python_entrypoint_builder():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))

    command = service.build_python_command("schema/init_database.py", "--seed", 1)

    assert command == [sys.executable, "schema/init_database.py", "--seed", "1"]


def test_build_login_status_command_spec_builds_canonical_status_runtime_command():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))

    command_spec = service.build_login_status_command_spec(
        platform="xhs",
        headless=True,
    )

    assert command_spec.command == [
        sys.executable,
        "/mediacrawler/web_runtime.py",
        "status",
        "--platform",
        "xhs",
        "--headless",
        "true",
    ]


def test_build_login_command_spec_builds_canonical_login_runtime_command():
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))

    command_spec = service.build_login_command_spec(
        platform="xhs",
        login_type="phone",
        headless=False,
        cookies="cookie=value",
        phone="13800138000",
    )

    assert command_spec.command == [
        sys.executable,
        "/mediacrawler/web_runtime.py",
        "login",
        "--platform",
        "xhs",
        "--login-type",
        "phone",
        "--headless",
        "false",
        "--cookies",
        "cookie=value",
        "--phone",
        "13800138000",
    ]


def test_build_crawl_command_spec_resolves_default_save_option_and_optional_args():
    settings = SimpleNamespace(DB_DIALECT="postgresql")
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    command_spec = service.build_crawl_command_spec(
        platform="xhs",
        login_type="qrcode",
        crawler_type="search",
        keywords="museum",
        specified_ids="",
        creator_ids="author-1",
        start_page=2,
        max_notes=30,
        max_comments=15,
        enable_comments=True,
        enable_sub_comments=False,
        save_data_option="",
        headless=True,
        cookies="cookie=value",
        phone="13800138000",
    )

    assert command_spec.resolved_save_data_option == "postgres"
    assert command_spec.command == [
        sys.executable,
        "/mediacrawler/web_runtime.py",
        "crawl",
        "--platform",
        "xhs",
        "--login-type",
        "qrcode",
        "--crawler-type",
        "search",
        "--keywords",
        "museum",
        "--specified-ids",
        "",
        "--creator-ids",
        "author-1",
        "--start-page",
        "2",
        "--max-notes",
        "30",
        "--max-comments",
        "15",
        "--enable-comments",
        "true",
        "--enable-sub-comments",
        "false",
        "--save-option",
        "postgres",
        "--headless",
        "true",
        "--cookies",
        "cookie=value",
        "--phone",
        "13800138000",
    ]


def test_ensure_dependencies_installed_skips_when_marker_is_fresh():
    paths, storage = _build_paths()
    storage["/mediacrawler/.deps_installed"] = _FakePathState(exists=True, text="", mtime=10)
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))
    runner_calls: list[list[str]] = []

    def _runner(*args, **kwargs):
        runner_calls.append(list(args[0]))
        raise AssertionError("runner should not be called when dependencies are already fresh")

    assert service.ensure_dependencies_installed(runner=_runner) is True
    assert runner_calls == []


def test_ensure_dependencies_installed_falls_back_to_uv_and_creates_marker():
    paths, storage = _build_paths()
    service = MediaCrawlerRuntimeService(_FakeMediaCrawler(paths))
    commands: list[list[str]] = []

    def _runner(command, **kwargs):
        commands.append(list(command))
        if command[0] == "uv":
            return SimpleNamespace(returncode=0, stderr="")
        return SimpleNamespace(returncode=1, stderr="pip failed")

    assert service.ensure_dependencies_installed(runner=_runner) is True
    assert commands[0][:3] == [sys.executable, "-m", "pip"]
    assert commands[1][:3] == ["uv", "pip", "install"]
    assert storage["/mediacrawler/.deps_installed"].exists is True


def test_write_db_config_renders_project_database_settings():
    settings = SimpleNamespace(
        DB_DIALECT="postgresql",
        DB_HOST="db.internal",
        DB_PORT=25432,
        DB_USER="crawler_user",
        DB_PASSWORD="crawler_pwd",
        DB_NAME="crawler_db",
    )
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    db_config_path = service.write_db_config()
    content = db_config_path.read_text(encoding="utf-8")

    assert 'MYSQL_DB_HOST = "db.internal"' in content
    assert "MYSQL_DB_PORT = 25432" in content
    assert 'POSTGRES_DB_HOST = os.getenv("POSTGRES_DB_HOST", "db.internal")' in content
    assert 'POSTGRES_DB_NAME = os.getenv("POSTGRES_DB_NAME", "crawler_db")' in content


def test_write_base_config_updates_template_lines_and_flattens_keywords_block():
    settings = SimpleNamespace(DB_DIALECT="postgresql")
    paths, _storage = _build_paths()
    service = MediaCrawlerRuntimeService(
        _FakeMediaCrawler(paths),
        reload_settings_callable=lambda: None,
        settings_getter=lambda: settings,
    )

    base_config_path = service.write_base_config(
        platform="wb",
        keywords=["museum", "service"],
        crawler_type="detail",
        max_notes=88,
        max_comments=42,
        headless=False,
        enable_comments=False,
    )
    content = base_config_path.read_text(encoding="utf-8")

    assert 'PLATFORM = "wb"' in content
    assert 'KEYWORDS = "museum,service"' in content
    assert 'CRAWLER_TYPE = "detail"' in content
    assert 'SAVE_DATA_OPTION = "postgres"' in content
    assert "CRAWLER_MAX_NOTES_COUNT = 88" in content
    assert "CRAWLER_MAX_COMMENTS_COUNT_SINGLENOTES = 42" in content
    assert "ENABLE_GET_COMMENTS = False" in content
    assert "HEADLESS = False" in content
    assert 'USER_DATA_DIR = "/runtime/crawler/browser_data/%s_user_data_dir"' in content
    assert '    "old"' not in content