"""无需模型下载的耐久性、取消和乐观编辑。"""
import asyncio
from contextlib import closing
from types import SimpleNamespace

import pytest
from fastapi.testclient import TestClient

from app.contracts import TranscriptEditRequest
from app.database.db import connect
from app.errors import ApiError
from app.services import transcription_service as jobs
from app.services.attachment_service import attachment_path


def text_attachment():
    path = attachment_path("lecture.txt")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("原始识别内容", encoding="utf-8")
    return path


def test_idempotency_edit_history_and_event_replay():
    text_attachment()

    async def scenario():
        first = await jobs.create_transcription("lecture.txt", idempotency_key="submit-1")
        repeated = await jobs.create_transcription("lecture.txt", idempotency_key="submit-1")
        assert first.job_id == repeated.job_id
        assert first.status == "completed"
        with pytest.raises(ApiError) as conflict:
            await jobs.create_transcription("lecture.txt", language="en", idempotency_key="submit-1")
        assert conflict.value.code == "IDEMPOTENCY_CONFLICT"
        revised = jobs.edit(first.job_id, TranscriptEditRequest(revision=1, text="校对内容"))
        assert revised.original_text == "原始识别内容"
        assert revised.revision == 2
        with pytest.raises(ApiError) as stale:
            jobs.edit(first.job_id, TranscriptEditRequest(revision=1, text="覆盖"))
        assert stale.value.code == "VERSION_CONFLICT"
        with closing(connect()) as conn:
            assert conn.execute("SELECT COUNT(*) FROM media_revisions").fetchone()[0] == 1
        events = jobs.events(first.job_id)
        assert [e["event"] for e in events] == ["Queued", "TranscriptionStarted", "Completed", "Revised"]
        assert jobs.events(first.job_id, events[-2]["sequence"]) == events[-1:]

    asyncio.run(scenario())


def test_cancel_before_start_retry_and_restart_recovery():
    text_attachment()

    async def scenario():
        job = await jobs.create_transcription("lecture.txt", wait=False)
        cancelled = await jobs.cancel(job.job_id)
        assert cancelled.status == "cancelled"
        next_job = await jobs.retry(job.job_id)
        assert next_job.previous_job_id == job.job_id
        assert next_job.job_id != job.job_id
        await jobs._tasks[jobs.task_key(next_job.job_id)]
        assert jobs.require_job(next_job.job_id).status == "completed"
        # 模拟已停止进程留下的持久作业。
        cancelled.status = "running"
        jobs.save(cancelled, "TranscriptionStarted")
        jobs.recover_interrupted()
        assert jobs.require_job(job.job_id).error_code == "TRANSCRIPTION_INTERRUPTED"

    asyncio.run(scenario())


def test_controlled_upload_and_async_http_flow():
    from app.main import app
    with TestClient(app) as client:
        assert client.post("/api/media/attachments?filename=a.wav", content=b"").status_code == 422
        uploaded = client.post("/api/media/attachments?filename=lecture.txt", content="真实转写文本".encode())
        assert uploaded.status_code == 201
        attachment_id = uploaded.json()["attachment_id"]
        assert client.get(f"/api/media/attachments/{attachment_id}").content == "真实转写文本".encode()
        response = client.post("/api/media/transcriptions", json={"attachment_id": attachment_id})
        assert response.status_code == 202 and response.json()["status"] == "queued"
        job_id = response.json()["job_id"]
        events = client.get(f"/api/media/transcriptions/{job_id}/events")
        assert "event: Completed" in events.text
        assert client.get("/api/media/transcriptions").json()["page"]["total"] == 1
        assert client.get(f"/api/media/transcriptions/{job_id}").json()["text"] == "真实转写文本"
        assert client.get(f"/api/media/transcriptions/{job_id}/events", headers={"Last-Event-ID": "bad"}).status_code == 422


def test_terminology_export_and_privacy_cleanup():
    from app.main import app
    text_attachment()
    with TestClient(app) as client:
        created = client.post('/api/media/transcriptions', json={'attachment_id':'lecture.txt','terminology':{'识别':'校对'}}).json()
        job_id = created['job_id']
        client.get(f'/api/media/transcriptions/{job_id}/events')
        job = client.get(f'/api/media/transcriptions/{job_id}').json()
        assert job['text'] == '原始校对内容' and job['original_text'] == '原始识别内容'
        first = client.post(f'/api/media/transcriptions/{job_id}/notes', json={'title':'课程'}).json()
        again = client.post(f'/api/media/transcriptions/{job_id}/notes', json={'title':'课程'}).json()
        assert first['note_id'] == again['note_id']
        artifacts = client.post(f'/api/media/transcriptions/{job_id}/artifacts', json={
            'title': '课程', 'knowledge_title': '课程知识点',
            'provider_id': 'mock', 'model': 'mock-1',
        })
        assert artifacts.status_code == 201
        assert artifacts.json()['transcript']['note_id'] == first['note_id']
        assert artifacts.json()['knowledge_note']['note_id'] != first['note_id']
        repeated = client.post(f'/api/media/transcriptions/{job_id}/artifacts', json={
            'title': '课程', 'knowledge_title': '课程知识点',
            'provider_id': 'mock', 'model': 'mock-1',
        })
        assert repeated.json()['knowledge_note']['note_id'] == artifacts.json()['knowledge_note']['note_id']
        response = client.delete('/api/media/attachments/lecture.txt')
        assert first['note_id'] in response.json()['retained_note_ids']
        cleaned = client.get(f'/api/media/transcriptions/{job_id}').json()
        assert cleaned['text'] is None and cleaned['original_text'] is None and cleaned['corrections'] == []
        assert client.post(f'/api/media/transcriptions/{job_id}/retry').status_code == 409
        assert client.get('/api/media/attachments/lecture.txt').status_code == 404


def test_media_note_links_do_not_depend_on_rebuildable_note_projection():
    with closing(connect()) as conn:
        foreign_tables = {row[2] for row in conn.execute("PRAGMA foreign_key_list(media_notes)")}
    assert foreign_tables == {"media_jobs"}


def test_desktop_revision_conflict_recovers_marker_matched_note(monkeypatch):
    from app.services import note_service
    from app.services.media_notes import _create_note

    marker = "<!-- transcription:job:1:hash -->"
    recovered = SimpleNamespace(note_id="stable-note", title="Generated", markdown=f"{marker}\nbody")

    async def conflict(**_kwargs):
        raise ApiError(409, "REVISION_CONFLICT", "already written")

    monkeypatch.setattr(note_service, "create_note", conflict)
    monkeypatch.setattr(
        note_service, "list_notes",
        lambda **_kwargs: ([SimpleNamespace(note_id="stable-note", title="Generated")], 1),
    )

    async def get_note(note_id):
        return recovered if note_id == "stable-note" else None

    monkeypatch.setattr(note_service, "get_note", get_note)
    result = asyncio.run(_create_note(
        "Generated", recovered.markdown, SimpleNamespace(folder=""), marker
    ))
    assert result is recovered


def test_transcript_export_recovers_link_from_another_vault(monkeypatch):
    from app.contracts import TranscriptNoteRequest
    from app.services import note_service
    from app.services.media_notes import create_transcript_note

    text_attachment()

    async def scenario():
        job = await jobs.create_transcription("lecture.txt")
        options = TranscriptNoteRequest(title="跨库课程")
        original = await create_transcript_note(job.job_id, options)
        with closing(connect()) as conn:
            conn.execute(
                "UPDATE media_notes SET note_id=? WHERE job_id=?",
                ("note-from-another-vault", job.job_id),
            )
            conn.commit()

        real_get_note = note_service.get_note

        async def get_note(note_id):
            if note_id == "note-from-another-vault":
                return None
            return await real_get_note(note_id)

        monkeypatch.setattr(note_service, "get_note", get_note)
        recovered = await create_transcript_note(job.job_id, options)
        assert recovered.note_id == original.note_id
        with closing(connect()) as conn:
            linked = conn.execute(
                "SELECT note_id FROM media_notes WHERE job_id=?",
                (job.job_id,),
            ).fetchone()[0]
        assert linked == original.note_id

    asyncio.run(scenario())


def test_artifact_host_writes_use_distinct_child_operations(monkeypatch):
    from app import host_bridge
    from app.contracts import TranscriptArtifactsRequest
    from app.services import media_notes

    operations = []
    text_attachment()
    job = asyncio.run(jobs.create_transcription("lecture.txt"))

    async def transcript(_job_id, _options):
        operations.append(host_bridge.operation_id.get())
        return SimpleNamespace(note_id="transcript-note")

    async def knowledge(*_args):
        return "<!-- knowledge-note:test -->\n# Knowledge"

    async def create(*_args, **_kwargs):
        operations.append(host_bridge.operation_id.get())
        return SimpleNamespace(note_id="knowledge-note")

    monkeypatch.setattr(media_notes, "create_transcript_note", transcript)
    monkeypatch.setattr(media_notes, "_knowledge_markdown", knowledge)
    monkeypatch.setattr(media_notes, "_create_note", create)

    token = host_bridge.operation_id.set("11111111-1111-4111-8111-111111111111")
    try:
        result = asyncio.run(media_notes.create_transcript_artifacts(
            job.job_id,
            TranscriptArtifactsRequest(
                title="Transcript", knowledge_title="Knowledge",
                provider_id="mock", model="mock-1",
            ),
        ))
    finally:
        host_bridge.operation_id.reset(token)

    assert result["transcript"].note_id == "transcript-note"
    assert result["knowledge_note"].note_id == "knowledge-note"
    assert len(operations) == 2
    assert operations[0] != operations[1]
    assert all(operation and operation != "11111111-1111-4111-8111-111111111111" for operation in operations)


def test_course_note_blocks_are_recomposed_with_markdown_and_plot_tools(monkeypatch):
    from app.container import container
    from app.services.media_notes import _compose_course_blocks

    names = []
    original = container.tools.execute

    async def execute(call, context):
        names.append(call.name)
        return await original(call, context)

    monkeypatch.setattr(container.tools, "execute", execute)
    markdown = """## 算法
```python
left += 1
```
```mermaid
flowchart LR
  A --> B
```
```function_plot
domain: -4, 4
range: -1, 8
y = x^2
```"""
    rendered = asyncio.run(_compose_course_blocks(markdown, "media-test"))
    assert names == ["markdown.compose", "markdown.compose", "function_plot.compose"]
    assert "```python\nleft += 1\n```" in rendered
    assert "```mermaid\nflowchart LR" in rendered
    assert "```function-plot\ndomain: -4, 4" in rendered


def test_course_note_rejects_invalid_function_plot():
    from app.services.media_notes import _compose_course_blocks

    with pytest.raises(ApiError) as invalid:
        asyncio.run(_compose_course_blocks(
            "```function-plot\ndomain: -4, 4\ny = __import__('os')\n```",
            "media-test",
        ))
    assert invalid.value.code == "KNOWLEDGE_NOTE_VISUAL_INVALID"


def test_local_only_export_and_rebuild_keep_local_embedding_policy(monkeypatch):
    from types import SimpleNamespace
    from app.contracts import TranscriptNoteRequest, IndexRebuildRequest
    from app.local_models.runtime import LocalEmbedding
    from app.retrieval import routed_vectors
    from app.services import note_service, index_service
    from app.services.media_notes import create_transcript_note
    calls = []
    class Routing:
        async def embed(self, texts, *, local_only=False):
            calls.append(local_only)
            assert local_only
            return SimpleNamespace(source='local', model_id='local-test', dimensions=2,
                                   vectors=[[1.0, 0.0] for _ in texts], fallback_reason=None)
    monkeypatch.setattr(routed_vectors, 'get_model_routing', lambda: Routing())
    monkeypatch.setattr(note_service, 'embedding', LocalEmbedding())
    text_attachment()
    async def scenario():
        job = await jobs.create_transcription('lecture.txt', local_only=True)
        note = await create_transcript_note(job.job_id, TranscriptNoteRequest(title='Private'))
        assert note.markdown.startswith('---\nembedding_local_only: true\n---')
        await note_service.update_note(note.note_id, markdown=note.markdown.replace(
            'embedding_local_only: true', 'embedding_local_only: true # keep local'))
        await index_service.rebuild(IndexRebuildRequest())
        assert len(calls) >= 3 and all(calls)
    asyncio.run(scenario())