from __future__ import annotations

# ruff: noqa: E402
import io
import sqlite3
import sys
from pathlib import Path

import pytest
from reportlab.pdfgen import canvas

ROOT=Path(__file__).resolve().parents[1]
HEALTH=ROOT/"scripts"/"health"
sys.path.insert(0,str(HEALTH));sys.path.insert(0,str(ROOT/"tests"))

from dashboard_v5.document_reconciliation import changed_fragments,reconcile_documents
from dashboard_v5.document_review import quarantine_upload
from dashboard_v5.read_api import dispatch_api
from dashboard_v5.sprint6i_c_schema import assert_schema
from fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture
from migrate_sprint6i_c_schema import safe_migrate
import health_dashboard_action_worker as worker


def _metadata(kind="laboratory_report"):
    return {"document_date":"2026-07-18","document_type":kind,"institution":"Synthetic Clinic","personal_title":"Synthetic document","investigation_day":"2026-07-18","note":"synthetic only"}


def _pdf(text:str)->bytes:
    output=io.BytesIO();document=canvas.Canvas(output,pagesize=(400,300));document.drawString(24,250,text);document.save();return output.getvalue()


def _configure(monkeypatch,database:Path,root:Path)->None:
    monkeypatch.setattr(worker,"DASHBOARD_DB",database);monkeypatch.setattr(worker,"DOCUMENT_QUARANTINE",root/"quarantine");monkeypatch.setattr(worker,"DOCUMENT_STORAGE",root/"documents");monkeypatch.setattr(worker,"DASHBOARD_V5_FILE",None)


def _import(monkeypatch,database:Path,root:Path,text:str,kind="laboratory_report")->str:
    _configure(monkeypatch,database,root);metadata=_metadata(kind);uploaded=quarantine_upload(_pdf(text),metadata,root/"quarantine")
    return worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":uploaded["token"],"sha256":uploaded["sha256"],"metadata":metadata}))


def _candidate(database:Path,intake:str,kind:str="laboratory_value")->str:
    connection=sqlite3.connect(database);row=connection.execute("SELECT c.id FROM document_candidates c JOIN document_processing p ON p.document_id=c.document_id WHERE p.intake_id=? AND c.candidate_type=? AND c.status IN ('open','conflicting') ORDER BY c.id LIMIT 1",(intake,kind)).fetchone();connection.close();assert row;return str(row[0])


def _binding(database:Path,candidate:str)->dict[str,object]:
    projected=next(item for item in dispatch_api(database,"/api/v1/lab-review","status=all")["candidates"] if item["id"]==candidate)
    return {"expected_candidate_revision":projected["revision_number"],"preview_revision":projected["revision"]}


def _review(intake:str,candidate:str,decision:str,action:str,value="4.2",unit="mg/l"):
    payload={"version":1,"action":"document_review","operation":"candidate_decision","document_id":intake,"target_id":candidate,"value":value,"unit":unit,"decision":decision,"metadata":{}}
    connection=sqlite3.connect(worker.DASHBOARD_DB);kind=connection.execute("SELECT candidate_type FROM document_candidates WHERE id=?",(candidate,)).fetchone()[0];connection.close()
    if kind=="laboratory_value":payload.update(_binding(Path(worker.DASHBOARD_DB),candidate))
    worker.apply_document_review(worker.validate_document_action(payload),action)


def _review_source(intake:str)->None:
    for operation,target,action in (("original_review","","8"*32),("page_review","page_1","9"*32)):
        payload={"version":1,"action":"document_review","operation":operation,"document_id":intake,"target_id":target,"value":"","unit":"","decision":"","metadata":{}}
        worker.apply_document_review(worker.validate_document_action(payload),action)


def test_copy_first_schema_is_additive_idempotent_and_restore_proven(tmp_path):
    database=tmp_path/"health.db";backup=tmp_path/"backup"/"pre-6ic.db";build_dashboard_v5_fixture(database)
    assert safe_migrate(database,backup)=={"copy_migration":"ok","production_migration":"ok","idempotency":"ok","integrity":"ok","foreign_keys":"ok","restore_test":"ok"}
    connection=sqlite3.connect(database);assert_schema(connection);assert connection.execute("PRAGMA integrity_check").fetchone()[0]=="ok";connection.close();assert backup.stat().st_mode&0o777==0o600


def test_byte_duplicate_has_no_second_user_decision(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);raw=_pdf("CRP 4.2 mg/l")
    _configure(monkeypatch,database,tmp_path)
    ids=[]
    for title in ("First","Second"):
        metadata=_metadata();metadata["personal_title"]=title;upload=quarantine_upload(raw,metadata,tmp_path/"quarantine");ids.append(worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":upload["token"],"sha256":upload["sha256"],"metadata":metadata})))
    connection=sqlite3.connect(database);rows=connection.execute("SELECT r.queue_bucket,r.open_decisions,p.duplicate_document_id FROM document_reconciliation r JOIN document_processing p ON p.document_id=r.document_id ORDER BY r.document_id DESC LIMIT 2").fetchall();connection.close()
    assert rows[0][0]=="no_action" and rows[0][1]==0 and rows[0][2] is not None


def test_identical_section_is_suppressed_and_near_match_only_returns_changes():
    assert changed_fragments("standard history remains unchanged","standard history remains unchanged")==[]
    changes=changed_fragments("standard history remains unchanged","standard history remains changed plus new statement")
    assert changes and any("changed" in item["current"] or "new" in item["current"] for item in changes)


def test_exact_database_and_workbook_lab_is_auto_linked(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"CRP 4.2 mg/l Referenz: 0-5 mg/l")
    connection=sqlite3.connect(database);connection.row_factory=sqlite3.Row;document_id=connection.execute("SELECT document_id FROM document_processing WHERE intake_id=?",(intake,)).fetchone()[0]
    connection.execute("INSERT INTO laborwerte(parameter_name,wert,einheit,abnahme_datum,verified_against_original,validierungsstatus) VALUES('CRP','4.2','mg/l','2026-07-18',1,'validiert')")
    result=reconcile_documents(connection,[{"parameter":"CRP","date":"2026-07-18","value":"=4.2","unit":"mg/l"}],now="2026-07-19T12:00:00+02:00")
    row=connection.execute("SELECT c.status,m.match_status FROM document_candidates c JOIN document_candidate_matches m ON m.candidate_id=c.id WHERE c.document_id=? AND c.candidate_type='laboratory_value'",(document_id,)).fetchone();connection.commit();connection.close()
    assert row["status"]=="already_present" and row["match_status"]=="exact_match" and result["exact_matches"]>=1


def test_lab_conflict_remains_blocked_with_observation_reference(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"CRP 4.2 mg/l Referenz: 0-5 mg/l")
    connection=sqlite3.connect(database);connection.row_factory=sqlite3.Row
    connection.execute("INSERT INTO laborwerte(parameter_name,wert,einheit,abnahme_datum,verified_against_original,validierungsstatus) VALUES('CRP','8.1','mg/l','2026-07-18',1,'validiert')")
    reconcile_documents(connection,[{"parameter":"CRP","date":"2026-07-18","value":"=9.1","unit":"mg/l"}],now="2026-07-19T12:01:00+02:00")
    row=connection.execute("SELECT c.status,m.match_status,m.reference_text FROM document_candidates c JOIN document_candidate_matches m ON m.candidate_id=c.id JOIN document_processing p ON p.document_id=c.document_id WHERE p.intake_id=? AND c.candidate_type='laboratory_value'",(intake,)).fetchone();connection.commit();connection.close()
    assert row["status"]=="conflicting" and row["match_status"]=="value_conflict" and row["reference_text"]


def test_prescription_is_not_classified_as_actual_administration(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"Medikament: SyntheticDrug 5 mg verordnet","doctor_report")
    candidate=_candidate(database,intake,"medication");_review_source(intake);_review(intake,candidate,"confirmed","1"*32,value="SyntheticDrug 5 mg verordnet",unit="")
    connection=sqlite3.connect(database);assert connection.execute("SELECT COUNT(*) FROM document_transfer_staging WHERE candidate_id=?",(candidate,)).fetchone()[0]==0;assert connection.execute("SELECT COUNT(*) FROM medication_administrations WHERE medication_name LIKE 'SyntheticDrug%'").fetchone()[0]==0;connection.close()


def test_review_then_preview_then_explicit_worker_transfer_is_idempotent(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"Befunddatum 2026-07-18\nCRP 4.2 mg/l Referenz: 0-5 mg/l");candidate=_candidate(database,intake)
    _review_source(intake);_review(intake,candidate,"confirmed","2"*32,value="CRP: 4.2",unit="mg/l")
    connection=sqlite3.connect(database);stage=connection.execute("SELECT status,old_value_json,new_value_json,source_page FROM document_transfer_staging WHERE candidate_id=?",(candidate,)).fetchone();before=connection.execute("SELECT COUNT(*) FROM laborwerte WHERE lower(parameter_name)='crp' AND befund_datum='2026-07-18'").fetchone()[0];connection.close();assert stage and stage[0]=="reviewed_pending_preview" and stage[1] and stage[2] and stage[3]==1
    transfer={"version":1,"action":"document_review","operation":"candidate_transfer","document_id":intake,"target_id":candidate,"value":"","unit":"","decision":"transfer","metadata":{},**_binding(database,candidate)}
    worker.apply_document_review(worker.validate_document_action(transfer),"3"*32);worker.apply_document_review(worker.validate_document_action(transfer),"4"*32)
    connection=sqlite3.connect(database);after=connection.execute("SELECT COUNT(*) FROM laborwerte WHERE lower(parameter_name)='crp' AND befund_datum='2026-07-18'").fetchone()[0];status=connection.execute("SELECT status FROM document_transfer_staging WHERE candidate_id=?",(candidate,)).fetchone()[0];connection.close();assert after==before+1 and status=="transferred"


def test_unreviewed_content_stays_out_of_verified_search_and_workspace_has_exact_page(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"UniqueSyntheticNeedle CRP 4.2 mg/l")
    assert dispatch_api(database,"/api/v1/search","q=UniqueSyntheticNeedle&include_machine=0")["groups"]["documents"]==[]
    listing=dispatch_api(database,"/api/v1/documents","limit=5");selected=next(item for item in listing["documents"] if item["review_id"]==intake);review=dispatch_api(database,f"/api/v1/documents/{selected['id']}/review","")
    assert review["pages"][0]["page"]==1 and "UniqueSyntheticNeedle" in review["pages"][0]["text"] and review["original_preview_url"]


def test_queue_is_one_compact_decision_count_not_overlapping_document_tasks(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);_import(monkeypatch,database,tmp_path,"CRP 4.2 mg/l")
    queue=dispatch_api(database,"/api/v1/document-review-queue","");assert [item["code"] for item in queue["groups"]].count("now_reviewable")==1;assert queue["total_decisions"]>=1


def test_stale_preview_is_blocked_after_manual_text_revision(tmp_path,monkeypatch):
    database=tmp_path/"health.db";build_dashboard_v5_fixture(database);intake=_import(monkeypatch,database,tmp_path,"Befunddatum 2026-07-18\nCRP 4.2 mg/l");candidate=_candidate(database,intake);_review_source(intake);_review(intake,candidate,"confirmed","5"*32,value="CRP: 4.2",unit="mg/l");binding=_binding(database,candidate)
    correction={"version":1,"action":"document_review","operation":"text_correction","document_id":intake,"target_id":"page_1","value":"CRP 4.3 mg/l","unit":"","decision":"","metadata":{}}
    worker.apply_document_review(worker.validate_document_action(correction),"6"*32)
    transfer={"version":1,"action":"document_review","operation":"candidate_transfer","document_id":intake,"target_id":candidate,"value":"","unit":"","decision":"transfer","metadata":{},**binding}
    with pytest.raises(RuntimeError,match="stale|revision conflict"):
        worker.apply_document_review(worker.validate_document_action(transfer),"7"*32)