import time
import uuid
from contextlib import contextmanager

import httpx
import pytest
import urllib3
from fastapi import HTTPException
from minio.error import S3Error
from sqlalchemy import func, inspect, select

from app.core.config import get_settings
from app.services.html_filter import filter_html

_PUBLIC_KEYS = {
    "siteName",
    "logoUrl",
    "bannerUrls",
    "icpText",
    "policeText",
    "friendLinks",
    "linkCirea",
    "linkHousing",
}


def test_object_key_keeps_safe_extension(monkeypatch):
    monkeypatch.setenv("MINIO_BASE_URL", "http://127.0.0.1:9000/gsfdc-public/")
    get_settings.cache_clear()
    try:
        from app.services.storage import make_object_key, public_object_url, safe_filename

        assert safe_filename("../../logo.PNG") == "logo.PNG"
        key = make_object_key("../../logo.PNG")
        assert ".." not in key
        assert key.endswith(".png")
        url = public_object_url(key)
        assert url == "http://127.0.0.1:9000/gsfdc-public/" + key
        assert "//" not in url.replace("://", "")
    finally:
        get_settings.cache_clear()


def test_html_whitelist_removes_active_content():
    html = (
        '<p onClick="alert(1)" class="keep">正文敏感词</p>'
        '<a href="javascript:alert(1)" title="t">链接</a>'
        "<script>alert(1)</script>"
        "<style>p{color:red}</style>"
        '<div>前<iframe src="https://evil.test">密文</iframe>后</div>'
        '<img src="https://evil.test/a.png" alt="x">'
        '<img src="http://127.0.0.1:9000/gsfdc-public/2026/10/19/abc.png" alt="ok">'
    )
    cleaned = filter_html(html, ["2026/10/19/abc.png"], "http://127.0.0.1:9000/gsfdc-public")
    lowered = cleaned.lower()
    assert "<script" not in lowered
    assert "onclick" not in lowered
    assert "javascript:" not in lowered
    assert "<style" not in lowered
    assert "<iframe" not in lowered
    assert "密文" not in cleaned
    assert "alert" not in lowered
    assert "evil.test" not in cleaned
    assert "正文敏感词" in cleaned
    assert 'class="keep"' in cleaned
    assert "链接" in cleaned
    assert 'title="t"' in cleaned
    assert 'alt="x"' in cleaned
    assert "2026/10/19/abc.png" in cleaned
    assert "前" in cleaned and "后" in cleaned


def _database_configured() -> bool:
    get_settings.cache_clear()
    return bool(get_settings().database_url.strip())


def _load_routers():
    try:
        from app.api.admin.content import router as admin_router
        from app.api.site.settings import router as site_router
    except ModuleNotFoundError:
        pytest.skip("app.db 尚未就绪")
    return admin_router, site_router


@pytest.fixture
def api_client(monkeypatch):
    monkeypatch.setenv("JWT_SECRET", "test-secret-must-be-32-characters!")
    monkeypatch.setenv("API_PAYLOAD_ENCRYPT", "false")
    get_settings.cache_clear()
    admin_router, site_router = _load_routers()
    from fastapi.testclient import TestClient

    from app.main import create_app

    app = create_app()
    paths = {getattr(route, "path", "") for route in app.routes}
    if "/api/admin/files" not in paths:
        app.include_router(admin_router, prefix="/api/admin")
    if "/api/site/settings" not in paths:
        app.include_router(site_router, prefix="/api/site")
    with TestClient(app) as client:
        yield client
    get_settings.cache_clear()


def _staff_headers() -> dict[str, str]:
    from app.core.security import create_access_token

    token = create_access_token("1", "staff")
    return {"Authorization": f"Bearer {token}"}


@contextmanager
def _session():
    from app.db import get_db

    resource = get_db()
    if hasattr(resource, "__next__"):
        db = next(resource)
        try:
            yield db
        finally:
            db.rollback()
            resource.close()
        return
    try:
        yield resource
    finally:
        resource.rollback()
        resource.close()


def _prepare_tables() -> None:
    from app.db import Base
    from app.models.content import AdminFile, SensitiveWord, SiteSetting, ensure_site_defaults

    with _session() as db:
        Base.metadata.create_all(
            bind=db.get_bind(),
            tables=[SiteSetting.__table__, AdminFile.__table__, SensitiveWord.__table__],
        )
        ensure_site_defaults(db)


def _table_names() -> set[str]:
    with _session() as db:
        return set(inspect(db.get_bind()).get_table_names())


def _count_files() -> int:
    from app.models.content import AdminFile

    with _session() as db:
        return int(db.scalar(select(func.count()).select_from(AdminFile)) or 0)


@pytest.fixture
def ready_client(api_client):
    if not _database_configured():
        pytest.skip("DATABASE_URL 为空")
    _prepare_tables()
    return api_client


def test_file_list_matches_filename(ready_client):
    from app.db import SessionLocal
    from app.models.content import AdminFile

    token = uuid.uuid4().hex[:10]
    names = [f"检索甲-{token}.txt", f"检索乙-{token}.png", f"其他-{token}.txt"]
    db = SessionLocal()
    ids: list[int] = []
    try:
        for name in names:
            row = AdminFile(
                bucket="public",
                object_key=f"test/{token}/{name}",
                filename=name,
                content_type="text/plain",
                size=1,
            )
            db.add(row)
            db.flush()
            ids.append(row.id)
        db.commit()
    finally:
        db.close()

    headers = _staff_headers()
    try:
        found = ready_client.get(
            "/api/admin/files",
            headers=headers,
            params={"q": f"检索甲-{token}", "size": 20},
        )
        assert found.status_code == 200, found.text
        body = found.json()
        assert [item["filename"] for item in body["records"]] == [f"检索甲-{token}.txt"]
        assert body["total"] == 1
        wildcard = ready_client.get(
            "/api/admin/files",
            headers=headers,
            params={"q": "%", "size": 100},
        )
        assert wildcard.status_code == 200, wildcard.text
        assert f"检索甲-{token}.txt" not in [item["filename"] for item in wildcard.json()["records"]]
    finally:
        db = SessionLocal()
        try:
            for file_id in ids:
                row = db.get(AdminFile, file_id)
                if row is not None:
                    db.delete(row)
            db.commit()
        finally:
            db.close()


def test_admin_content_requires_login(api_client):
    checks = [
        ("get", "/api/admin/files", None),
        ("get", "/api/admin/files/1", None),
        ("get", "/api/admin/site-settings", None),
        ("get", "/api/admin/sensitive-words", None),
        ("put", "/api/admin/site-settings", {"site_name": "x"}),
        ("post", "/api/admin/sensitive-words", {"word": "x"}),
        ("post", "/api/admin/content/preview-filter", {"html": "<p>a</p>"}),
        ("delete", "/api/admin/sensitive-words/1", None),
        ("delete", "/api/admin/files/1", None),
    ]
    for method, path, payload in checks:
        kwargs = {"json": payload} if payload is not None else {}
        response = getattr(api_client, method)(path, **kwargs)
        assert response.status_code == 401, path
    uploaded = api_client.post(
        "/api/admin/files",
        data={"bucket": "public"},
        files={"file": ("a.txt", b"hi", "text/plain")},
    )
    assert uploaded.status_code == 401


def test_preview_filter_reports_words_without_new_tables(ready_client):
    headers = _staff_headers()
    word = "预览" + uuid.uuid4().hex[:12]
    created = ready_client.post("/api/admin/sensitive-words", json={"word": word}, headers=headers)
    assert created.status_code == 200, created.text
    word_id = created.json()["id"]
    try:
        before_tables = _table_names()
        before_count = len(ready_client.get("/api/admin/sensitive-words", headers=headers).json())
        html = f'<p onclick="alert(1)">正文{word}</p><script>alert(1){word}</script>'
        response = ready_client.post(
            "/api/admin/content/preview-filter",
            json={"html": html},
            headers=headers,
        )
        assert response.status_code == 200, response.text
        body = response.json()
        assert word in body["hits"]
        assert body["html"].count(word) == 1
        assert "<script" not in body["html"].lower()
        assert "onclick" not in body["html"].lower()
        assert _table_names() == before_tables
        after_count = len(ready_client.get("/api/admin/sensitive-words", headers=headers).json())
        assert after_count == before_count
    finally:
        ready_client.delete(f"/api/admin/sensitive-words/{word_id}", headers=headers)


def test_upload_over_limit_is_rejected(ready_client):
    headers = _staff_headers()
    current = ready_client.get("/api/admin/site-settings", headers=headers)
    assert current.status_code == 200, current.text
    original = current.json()["upload_max_bytes"]
    try:
        updated = ready_client.put(
            "/api/admin/site-settings",
            json={"upload_max_bytes": 4},
            headers=headers,
        )
        assert updated.status_code == 200, updated.text
        before = _count_files()
        rejected = ready_client.post(
            "/api/admin/files",
            headers=headers,
            data={"bucket": "public"},
            files={"file": ("note.txt", b"12345", "text/plain")},
        )
        assert rejected.status_code == 400
        assert rejected.json()["detail"] == "文件超过大小上限"
        assert _count_files() == before
    finally:
        ready_client.put(
            "/api/admin/site-settings",
            json={"upload_max_bytes": original},
            headers=headers,
        )


def test_public_settings_hide_login_lock(ready_client):
    headers = _staff_headers()
    before = ready_client.get("/api/admin/site-settings", headers=headers)
    assert before.status_code == 200, before.text
    original = before.json()
    assert isinstance(original["upload_max_bytes"], int)
    assert isinstance(original["login_max_failures"], int)
    assert isinstance(original["login_lock_minutes"], int)
    assert isinstance(original["logoUrl"], str)
    assert isinstance(original["bannerUrls"], list)
    link = {"name": "示例链接", "url": "https://example.com/gsrea"}
    try:
        rejected = ready_client.put(
            "/api/admin/site-settings",
            json={"theme": "dark"},
            headers=headers,
        )
        assert rejected.status_code == 400
        saved = ready_client.put(
            "/api/admin/site-settings",
            json={
                "site_name": "甘肃价协测试站点",
                "icp_text": "陇ICP备测试号",
                "friend_links": [link],
            },
            headers=headers,
        )
        assert saved.status_code == 200, saved.text
        response = ready_client.get("/api/site/settings")
        assert response.status_code == 200, response.text
        body = response.json()
        assert set(body) == _PUBLIC_KEYS
        assert body["siteName"] == "甘肃价协测试站点"
        assert body["icpText"] == "陇ICP备测试号"
        assert body["friendLinks"] == [link]
        assert isinstance(body["logoUrl"], str)
        assert isinstance(body["bannerUrls"], list)
        text = response.text
        assert "login_max_failures" not in text
        assert "login_lock_minutes" not in text
        assert "loginMaxFailures" not in text
        assert "loginLockMinutes" not in text
        assert "upload_max_bytes" not in text
        assert "uploadMaxBytes" not in text
    finally:
        ready_client.put(
            "/api/admin/site-settings",
            json={
                "site_name": original["site_name"],
                "icp_text": original["icp_text"],
                "friend_links": original["friend_links"],
            },
            headers=headers,
        )


def _minio_ready() -> bool:
    from app.services.storage import ensure_buckets

    try:
        ensure_buckets()
    except (urllib3.exceptions.HTTPError, OSError, TimeoutError):
        return False
    except (HTTPException, S3Error):
        return False
    return True


def test_public_object_and_private_expiry(ready_client):
    if not _minio_ready():
        pytest.skip("MinIO 不可达")
    from app.models.content import AdminFile
    from app.services.storage import private_object_url

    headers = _staff_headers()
    payload = b"gsfdc-public-png"
    created_ids: list[int] = []
    try:
        uploaded = ready_client.post(
            "/api/admin/files",
            headers=headers,
            data={"bucket": "public"},
            files={"file": ("logo.png", payload, "image/png")},
        )
        assert uploaded.status_code == 200, uploaded.text
        public_body = uploaded.json()
        created_ids.append(public_body["id"])
        assert public_body["expiresAt"] is None
        base = get_settings().minio_base_url.strip().rstrip("/")
        assert public_body["url"].startswith(base + "/")
        assert "//" not in public_body["url"].replace("://", "")
        downloaded = httpx.get(public_body["url"], timeout=5.0, follow_redirects=True)
        assert downloaded.status_code == 200
        assert downloaded.content == payload

        private = ready_client.post(
            "/api/admin/files",
            headers=headers,
            data={"bucket": "private"},
            files={"file": ("secret.bin", b"gsfdc-private-bin", "application/octet-stream")},
        )
        assert private.status_code == 200, private.text
        private_body = private.json()
        created_ids.append(private_body["id"])
        assert private_body["expiresAt"]
        with _session() as db:
            row = db.get(AdminFile, private_body["id"])
            assert row is not None
            assert row.created_by == 1
            object_key = row.object_key
        fresh, _ = private_object_url(object_key, expires_seconds=60)
        assert httpx.get(fresh, timeout=5.0, follow_redirects=True).status_code == 200
        short, _ = private_object_url(object_key, expires_seconds=1)
        time.sleep(2)
        expired = httpx.get(short, timeout=5.0, follow_redirects=True)
        assert expired.status_code != 200
    finally:
        for file_id in created_ids:
            ready_client.delete(f"/api/admin/files/{file_id}", headers=headers)