from __future__ import annotations

from pathlib import Path
from types import SimpleNamespace

import pytest
from imagefind.config import Settings
from imagefind.resources import ResourceGovernor
from imagefind.scanner import Scanner
from imagefind.speech_quality import transcript_quality_state
from imagefind.uploads import UploadService
from imagefind.usage import StorageUsageService


def _upload(**values) -> dict:
    return {
        "id": "upload",
        "status": "completed",
        "phase": "ai_queued",
        "storage_backend": "openlist_native",
        "size_bytes": 100,
        "bytes_received": 100,
        "progress": 1,
        "external_progress": 1,
        "external_status": "provider-internal-state",
        "external_error": "provider-secret-error",
        "external_task_id": "provider-task-12345678",
        "deduplicated": 0,
        **values,
    }


def test_upload_public_contract_reports_ai_queue_without_leaking_provider_state():
    result = UploadService._public(_upload())

    assert result["stage"] == "ai_queued"
    assert result["stage_label"] == "已加入媒体库，AI 已排队"
    assert result["external_status"] == "completed"
    assert "external_state" not in result
    assert "external_error" not in result
    assert "provider-internal-state" not in str(result)
    assert "provider-secret-error" not in str(result)
    assert result["external_task_ref"] == "12345678"
    assert [stage["state"] for stage in result["stages"]] == ["completed"] * 4
    assert result["stages"][-1]["label"] == "加入媒体库并排队 AI"

    historical = UploadService._public(_upload(phase="cataloging"))
    assert historical["stage"] == "ai_queued"
    assert historical["stages"][-1]["state"] == "completed"


def test_native_restore_refresh_preserves_the_physical_object_mapping():
    restored = Scanner._native_refresh_record(
        None,
        {
            "physical_path": "cloud/library/opaque-name.bin",
            "physical_size_bytes": 356_793,
            "logical_size_bytes": 356_665,
        },
    )

    assert restored == {
        "storage_backend": "openlist_native",
        "external_target_path": "cloud/library/opaque-name.bin",
        "external_size_bytes": 356_793,
        "size_bytes": 356_665,
        "content_sha256": "",
    }
    assert Scanner._native_refresh_record(None, {"physical_path": "", "physical_size_bytes": 1}) is None


@pytest.mark.parametrize(
    ("score", "accepted", "rejected", "flags", "expected"),
    [
        (None, 0, 0, [], "empty"),
        (0.82, 8, 2, [], "ready"),
        (0.60, 5, 3, [], "filtered"),
        (0.90, 8, 0, ["language_script_conflict"], "low_quality"),
        (0.40, 2, 8, [], "low_quality"),
    ],
)
def test_transcript_quality_state_contract(score, accepted, rejected, flags, expected):
    assert transcript_quality_state(score, accepted, rejected, flags) == expected


def test_running_lane_clears_stale_io_pressure_reason(monkeypatch, tmp_path: Path):
    db = SimpleNamespace(setting=lambda _key, default=None: default)
    governor = ResourceGovernor(db, Settings(data_dir=tmp_path))
    governor._pressure_reasons["ai"] = "磁盘 I/O 持续繁忙，后台任务已让出资源"
    monkeypatch.setattr(
        governor,
        "sample",
        lambda: {
            "cpu_percent": 10,
            "memory_total_bytes": 8 * 1024**3,
            "memory_available_bytes": 6 * 1024**3,
            "disk_available_bytes": 100 * 1024**3,
            "io": {"supported": False},
        },
    )

    assert governor.pressure_reason(lane="ai", running=True) is None
    assert governor._pressure_reasons["ai"] is None


class _Sources:
    def __init__(self, source: dict, connector=None):
        self.source = source
        self._connector = connector

    def get(self, source_id: str):
        assert source_id == self.source["id"]
        return self.source

    def connector(self, source_id: str):
        assert source_id == self.source["id"]
        return self._connector


def _usage(source: dict, *, rclone=None, connector=None) -> StorageUsageService:
    return StorageUsageService(
        db=None,
        settings=SimpleNamespace(),
        rclone=rclone,
        sources=_Sources(source, connector),
        cache_seconds=30,
    )


def test_local_capacity_uses_source_filesystem(monkeypatch, tmp_path: Path):
    source = {"id": "local", "kind": "local", "config": {"path": str(tmp_path)}}
    monkeypatch.setattr(
        "imagefind.usage.shutil.disk_usage",
        lambda path: SimpleNamespace(total=10_000, used=4_000, free=6_000)
        if path == tmp_path
        else None,
    )

    result = _usage(source).capacity("local")
    assert (result["scope"], result["provider"], result["status"]) == (
        "local",
        "local",
        "available",
    )
    assert result["available_bytes"] == 6_000


def test_openlist_capacity_uses_remote_quota_and_cache():
    source = {
        "id": "cloud",
        "kind": "webdav",
        "config": {"driver": "alist", "storage_backend": "openlist_native"},
    }
    calls = []
    rclone = SimpleNamespace(
        about=lambda value: calls.append(value["id"])
        or {"total_bytes": 5_000, "used_bytes": 1_000, "available_bytes": 4_000}
    )
    usage = _usage(source, rclone=rclone)

    first = usage.capacity("cloud")
    second = usage.capacity("cloud")
    assert first["scope"] == "remote"
    assert first["provider"] == "rclone"
    assert first["available_bytes"] == 4_000
    assert second["available_bytes"] == 4_000
    assert calls == ["cloud"]


def test_remote_capacity_manual_override_wins_without_querying_the_nas_or_provider():
    total = 6 * 1024**4
    available = 4 * 1024**4
    source = {
        "id": "cloud",
        "kind": "webdav",
        "config": {
            "driver": "alist",
            "storage_backend": "openlist_native",
            "capacity_override_total_bytes": total,
            "capacity_override_available_bytes": available,
        },
    }
    rclone = SimpleNamespace(about=lambda _source: (_ for _ in ()).throw(AssertionError("must not query")))

    result = _usage(source, rclone=rclone).capacity("cloud")

    assert result["provider"] == "manual"
    assert result["scope"] == "remote"
    assert result["total_bytes"] == total
    assert result["used_bytes"] == total - available
    assert result["available_bytes"] == available


def test_webdav_capacity_supports_quota_and_distinguishes_unsupported_from_unreachable():
    xml = b"""<?xml version='1.0'?>
    <d:multistatus xmlns:d='DAV:'><d:response><d:propstat><d:prop>
    <d:quota-used-bytes>300</d:quota-used-bytes>
    <d:quota-available-bytes>700</d:quota-available-bytes>
    </d:prop></d:propstat></d:response></d:multistatus>"""

    class Connector:
        base_url = "https://dav.example/root/"

        def __init__(self, response):
            self.client = SimpleNamespace(request=lambda *_args, **_kwargs: response)
            self.closed = False

        def close(self):
            self.closed = True

    source = {"id": "dav", "kind": "webdav", "config": {"driver": "webdav"}}
    response = SimpleNamespace(status_code=207, content=xml)
    connector = Connector(response)
    available = _usage(source, connector=connector).capacity("dav")
    assert available["provider"] == "webdav"
    assert available["total_bytes"] == 1_000
    assert connector.closed is True

    no_quota = Connector(SimpleNamespace(status_code=207, content=b"<d:multistatus xmlns:d='DAV:'/>"))
    unsupported = _usage(source, connector=no_quota).capacity("dav")
    assert unsupported["status"] == "unsupported"

    failed = Connector(SimpleNamespace(status_code=503, content=b""))
    unreachable = _usage(source, connector=failed).capacity("dav")
    assert unreachable["status"] == "unreachable"
    assert unreachable["total_bytes"] == 0