def test_upload_rejects_unsupported_file_type(client):
    response = client.post(
        "/api/media/upload",
        files={"file": ("notes.txt", b"text", "text/plain")},
    )

    assert response.status_code == 400
    assert "Unsupported file type" in response.json()["detail"]


def test_upload_auto_creates_project(client, monkeypatch):
    uploaded = []
    monkeypatch.setattr("routers.media.upload_file", lambda object_name, data, content_type, length: uploaded.append(object_name))
    monkeypatch.setattr("routers.media.get_presigned_url", lambda object_name, expires=3600: f"http://storage/{object_name}")

    response = client.post(
        "/api/media/upload",
        files={"file": ("clip.mp4", b"video", "video/mp4")},
    )

    assert response.status_code == 201
    data = response.json()
    assert data["project_id"] is not None
    assert data["object_name"] == f"uploads/{data['project_id']}/clip.mp4"
    assert uploaded == ["uploads/general/clip.mp4", f"uploads/{data['project_id']}/clip.mp4"]


def test_upload_links_existing_project(client, monkeypatch):
    project = client.post("/api/projects", json={"name": "Existing"}).json()
    monkeypatch.setattr("routers.media.upload_file", lambda *args, **kwargs: None)
    monkeypatch.setattr("routers.media.get_presigned_url", lambda object_name, expires=3600: f"http://storage/{object_name}")

    response = client.post(
        "/api/media/upload",
        data={"project_id": str(project["id"])},
        files={"file": ("clip.mp4", b"video", "video/mp4")},
    )

    assert response.status_code == 201
    detail = client.get(f"/api/projects/{project['id']}").json()
    assert detail["video_path"] == f"uploads/{project['id']}/clip.mp4"


def test_upload_dicom_batch_filters_files_and_creates_project(client, monkeypatch):
    uploaded = []
    monkeypatch.setattr("routers.media.upload_file", lambda object_name, data, content_type, length: uploaded.append(object_name))

    response = client.post(
        "/api/media/upload/dicom",
        files=[
            ("files", ("10.dcm", b"dcm10", "application/dicom")),
            ("files", ("skip.txt", b"text", "text/plain")),
            ("files", ("2.dcm", b"dcm2", "application/dicom")),
            ("files", ("1.dcm", b"dcm1", "application/dicom")),
        ],
    )

    assert response.status_code == 201
    data = response.json()
    assert data["uploaded_count"] == 3
    assert uploaded == [
        f"uploads/{data['project_id']}/dicom/1.dcm",
        f"uploads/{data['project_id']}/dicom/2.dcm",
        f"uploads/{data['project_id']}/dicom/10.dcm",
    ]
    project_detail = client.get(f"/api/projects/{data['project_id']}").json()
    assert project_detail["name"] == "1.dcm"


def test_upload_dicom_batch_rejects_when_no_valid_dicom(client, monkeypatch):
    monkeypatch.setattr("routers.media.upload_file", lambda *args, **kwargs: None)

    response = client.post(
        "/api/media/upload/dicom",
        files=[
            ("files", ("notes.txt", b"text", "text/plain")),
        ],
    )

    assert response.status_code == 400
    assert response.json()["detail"] == "No valid DICOM files uploaded"


def test_parse_media_queues_background_task(client, monkeypatch):
    project = client.post("/api/projects", json={
        "name": "Parse Me",
        "video_path": "uploads/1/clip.mp4",
        "source_type": "video",
        "parse_fps": 5,
    }).json()

    class FakeAsyncResult:
        id = "celery-1"

    queued = []
    monkeypatch.setattr("routers.media.parse_project_media.delay", lambda task_id: queued.append(task_id) or FakeAsyncResult())
    published = []
    monkeypatch.setattr("routers.media.publish_task_progress_event", lambda task: published.append(task.id))

    response = client.post(f"/api/media/parse?project_id={project['id']}")

    assert response.status_code == 202
    data = response.json()
    assert data["task_type"] == "parse_video"
    assert data["status"] == "queued"
    assert data["progress"] == 0
    assert data["project_id"] == project["id"]
    assert data["celery_task_id"] == "celery-1"
    assert data["payload"] == {
        "source_type": "video",
        "parse_fps": 5.0,
        "max_frames": None,
        "target_width": 640,
    }
    assert queued == [data["id"]]
    assert published == [data["id"]]

    detail = client.get(f"/api/tasks/{data['id']}")
    assert detail.status_code == 200
    assert detail.json()["status"] == "queued"
    project_detail = client.get(f"/api/projects/{project['id']}").json()
    assert project_detail["status"] == "parsing"


def test_parse_media_accepts_frame_sequence_options(client, monkeypatch):
    project = client.post("/api/projects", json={
        "name": "Parse Options",
        "video_path": "uploads/1/clip.mp4",
        "source_type": "video",
        "parse_fps": 30,
    }).json()

    class FakeAsyncResult:
        id = "celery-options"

    monkeypatch.setattr("routers.media.parse_project_media.delay", lambda task_id: FakeAsyncResult())
    monkeypatch.setattr("routers.media.publish_task_progress_event", lambda task: None)

    response = client.post(
        f"/api/media/parse?project_id={project['id']}&parse_fps=15&max_frames=120&target_width=960"
    )

    assert response.status_code == 202
    data = response.json()
    assert data["payload"] == {
        "source_type": "video",
        "parse_fps": 15.0,
        "max_frames": 120,
        "target_width": 960,
    }
    assert client.get(f"/api/projects/{project['id']}").json()["parse_fps"] == 15.0


def test_parse_task_runner_registers_frames(client, db_session, monkeypatch, tmp_path):
    from models import ProcessingTask
    from services.media_task_runner import run_parse_media_task

    project = client.post("/api/projects", json={
        "name": "Parse Me",
        "video_path": "uploads/1/clip.mp4",
        "source_type": "video",
        "parse_fps": 5,
    }).json()
    task = ProcessingTask(
        task_type="parse_video",
        status="queued",
        progress=0,
        project_id=project["id"],
        payload={"source_type": "video"},
    )
    db_session.add(task)
    db_session.commit()
    db_session.refresh(task)
    frame_file = tmp_path / "frame_000001.jpg"
    frame_file.write_bytes(b"fake image")

    monkeypatch.setattr("services.media_task_runner.download_file", lambda object_name: b"video")
    monkeypatch.setattr(
        "services.media_task_runner.parse_video",
        lambda local_path, output_dir, fps, max_frames=None, target_width=640: ([str(frame_file)], 25.0),
    )
    monkeypatch.setattr("services.media_task_runner.extract_thumbnail", lambda local_path, thumbnail_path: open(thumbnail_path, "wb").write(b"thumb"))
    monkeypatch.setattr("services.media_task_runner.upload_file", lambda *args, **kwargs: None)
    monkeypatch.setattr("services.media_task_runner.upload_frames_to_minio", lambda frame_files, project_id: [f"projects/{project_id}/frames/frame_000001.jpg"])
    monkeypatch.setattr("routers.projects.get_presigned_url", lambda object_name, expires=3600: f"http://storage/{object_name}")
    published = []
    monkeypatch.setattr(
        "services.media_task_runner.publish_task_progress_event",
        lambda event_task: published.append((event_task.status, event_task.progress, event_task.message)),
    )

    result = run_parse_media_task(db_session, task.id)

    assert result["frames_extracted"] == 1
    assert result["frame_sequence"] == {
        "original_fps": 25.0,
        "parse_fps": 5.0,
        "frame_count": 1,
        "duration_ms": 0.0,
        "target_width": 640,
        "frame_width": None,
        "frame_height": None,
        "max_frames": None,
        "object_prefix": f"projects/{project['id']}/frames",
    }
    db_session.refresh(task)
    assert task.status == "success"
    assert task.progress == 100
    assert ("running", 5, "后台解析已启动") in published
    assert ("success", 100, "解析完成") in published
    project_detail = client.get(f"/api/projects/{project['id']}").json()
    assert project_detail["status"] == "ready"
    frames = client.get(f"/api/projects/{project['id']}/frames").json()
    assert "frame_000001.jpg" in frames[0]["image_url"]
    assert frames[0]["timestamp_ms"] == 0.0
    assert frames[0]["source_frame_number"] == 0


def test_parse_task_runner_replaces_existing_frames_and_annotations(client, db_session, monkeypatch, tmp_path):
    from models import Annotation, Frame, Mask, ProcessingTask
    from services.media_task_runner import run_parse_media_task

    project = client.post("/api/projects", json={
        "name": "Reparse Me",
        "video_path": "uploads/1/clip.mp4",
        "source_type": "video",
        "status": "ready",
        "parse_fps": 30,
        "thumbnail_url": "projects/old/thumbnail.jpg",
    }).json()
    old_frame = Frame(project_id=project["id"], frame_index=0, image_url="projects/old/frame.jpg")
    db_session.add(old_frame)
    db_session.commit()
    db_session.refresh(old_frame)
    old_annotation = Annotation(project_id=project["id"], frame_id=old_frame.id, mask_data={"label": "old"})
    db_session.add(old_annotation)
    db_session.commit()
    db_session.refresh(old_annotation)
    db_session.add(Mask(annotation_id=old_annotation.id, mask_url="masks/old.png"))
    task = ProcessingTask(
        task_type="parse_video",
        status="queued",
        progress=0,
        project_id=project["id"],
        payload={"source_type": "video", "parse_fps": 10},
    )
    db_session.add(task)
    db_session.commit()
    db_session.refresh(task)

    frame_file = tmp_path / "frame_000000.jpg"
    frame_file.write_bytes(b"new frame")
    monkeypatch.setattr("services.media_task_runner.download_file", lambda object_name: b"video")
    monkeypatch.setattr(
        "services.media_task_runner.parse_video",
        lambda local_path, output_dir, fps, max_frames=None, target_width=640: ([str(frame_file)], 25.0),
    )
    monkeypatch.setattr("services.media_task_runner.extract_thumbnail", lambda local_path, thumbnail_path: open(thumbnail_path, "wb").write(b"thumb"))
    monkeypatch.setattr("services.media_task_runner.upload_file", lambda *args, **kwargs: None)
    monkeypatch.setattr("services.media_task_runner.upload_frames_to_minio", lambda frame_files, project_id: [f"projects/{project_id}/frames/frame_000000.jpg"])
    monkeypatch.setattr("services.media_task_runner.publish_task_progress_event", lambda task: None)

    result = run_parse_media_task(db_session, task.id)

    assert result["frames_extracted"] == 1
    frames = db_session.query(Frame).filter(Frame.project_id == project["id"]).all()
    assert len(frames) == 1
    assert frames[0].image_url == f"projects/{project['id']}/frames/frame_000000.jpg"
    assert db_session.query(Annotation).filter(Annotation.project_id == project["id"]).count() == 0
    assert db_session.query(Mask).count() == 0


def test_parse_dicom_reads_files_in_natural_filename_order(monkeypatch, tmp_path):
    from pathlib import Path

    import numpy as np

    from services.frame_parser import parse_dicom

    dcm_dir = tmp_path / "dcm"
    output_dir = tmp_path / "frames"
    dcm_dir.mkdir()
    for name in ["10.dcm", "2.dcm", "1.dcm"]:
        (dcm_dir / name).write_bytes(b"dcm")

    read_order = []

    class FakeDicom:
        pixel_array = np.ones((2, 2), dtype=np.uint8)

    def fake_dcmread(path):
        read_order.append(Path(path).name)
        return FakeDicom()

    def fake_imwrite(path, image, params=None):
        Path(path).write_bytes(image.tobytes())
        return True

    monkeypatch.setattr("services.frame_parser.dcmread", fake_dcmread)
    monkeypatch.setattr("services.frame_parser.cv2.imwrite", fake_imwrite)

    frame_files = parse_dicom(str(dcm_dir), str(output_dir))

    assert read_order == ["1.dcm", "2.dcm", "10.dcm"]
    assert [Path(path).name for path in frame_files] == ["frame_000000.jpg", "frame_000001.jpg", "frame_000002.jpg"]


def test_parse_task_runner_downloads_dicom_objects_in_natural_filename_order(client, db_session, monkeypatch, tmp_path):
    from types import SimpleNamespace

    from models import ProcessingTask
    from services.media_task_runner import run_parse_media_task

    project = client.post("/api/projects", json={
        "name": "DICOM",
        "video_path": "uploads/1/dicom",
        "source_type": "dicom",
        "parse_fps": 30,
    }).json()
    task = ProcessingTask(
        task_type="parse_dicom",
        status="queued",
        progress=0,
        project_id=project["id"],
        payload={"source_type": "dicom"},
    )
    db_session.add(task)
    db_session.commit()
    db_session.refresh(task)

    class FakeClient:
        def list_objects(self, bucket, prefix, recursive=True):
            return [
                SimpleNamespace(object_name=f"{prefix}/10.dcm"),
                SimpleNamespace(object_name=f"{prefix}/2.dcm"),
                SimpleNamespace(object_name=f"{prefix}/1.dcm"),
            ]

    downloaded = []
    frame_files = []
    for idx in range(3):
        frame_file = tmp_path / f"frame_{idx:06d}.jpg"
        frame_file.write_bytes(b"fake image")
        frame_files.append(str(frame_file))

    monkeypatch.setattr("services.media_task_runner.get_minio_client", lambda: FakeClient())
    monkeypatch.setattr(
        "services.media_task_runner.download_file",
        lambda object_name: downloaded.append(object_name) or b"dcm",
    )
    monkeypatch.setattr("services.media_task_runner.parse_dicom", lambda *args, **kwargs: frame_files)
    monkeypatch.setattr(
        "services.media_task_runner.upload_frames_to_minio",
        lambda frames, project_id: [f"projects/{project_id}/frames/{idx}.jpg" for idx, _ in enumerate(frames)],
    )
    monkeypatch.setattr("services.media_task_runner.publish_task_progress_event", lambda task: None)

    result = run_parse_media_task(db_session, task.id)

    assert result["frames_extracted"] == 3
    assert downloaded == [
        "uploads/1/dicom/1.dcm",
        "uploads/1/dicom/2.dcm",
        "uploads/1/dicom/10.dcm",
    ]


def test_parse_task_runner_skips_already_cancelled_task(db_session):
    from models import ProcessingTask
    from services.media_task_runner import run_parse_media_task

    task = ProcessingTask(
        task_type="parse_video",
        status="cancelled",
        progress=100,
        message="任务已取消",
        project_id=1,
        payload={"source_type": "video"},
    )
    db_session.add(task)
    db_session.commit()
    db_session.refresh(task)

    result = run_parse_media_task(db_session, task.id)

    assert result["status"] == "cancelled"
    assert result["message"] == "任务已取消"