from __future__ import annotations

import json
from pathlib import Path

from server.aps_domain.folder_pack import analyze_work_dir
from server.aps_domain.project_analyze import analyze_project_deep
from server.state.seed import empty_world


def _counter():
    values: dict[str, int] = {}

    def next_id(kind: str) -> int:
        values[kind] = values.get(kind, 0) + 1
        return values[kind]

    return next_id


def test_analyze_project_deep_readonly_does_not_mutate_world(tmp_path: Path, monkeypatch):
    def fail_write(*_args, **_kwargs):
        raise AssertionError("readonly analysis must not call write paths")

    def fail_ingest(*_args, **_kwargs):
        raise AssertionError("readonly analysis must not write knowledge")

    (tmp_path / "orders.csv").write_text(
        "orderNo,productCode,quantity,deliveryDate,customerName\n"
        "SO-1,PRODUCT-1,10,2026-08-20,CUSTOMER-1\n",
        encoding="utf-8-sig",
    )
    (tmp_path / "materials.csv").write_text(
        "code,name,type,unit,stock\n"
        "PRODUCT-1,Product 1,FINISHED_PRODUCT,pc,100\n",
        encoding="utf-8-sig",
    )
    (tmp_path / "routing.csv").write_text(
        "productCode,operationCode,operationName,seq,stdTimePerUnit\n"
        "PRODUCT-1,CUT,Cut,10,5\n",
        encoding="utf-8-sig",
    )
    (tmp_path / "equipment.csv").write_text(
        "code,name,capabilities\n"
        "EQ-1,Cutter,CUT\n",
        encoding="utf-8-sig",
    )

    read_world = empty_world()
    before = json.dumps(read_world, sort_keys=True, default=str)

    def fail_project_context(*_args, **_kwargs):
        raise AssertionError("readonly analysis must not call ProjectStore.snapshot")

    with monkeypatch.context() as patch:
        patch.setattr("server.aps_domain.importers.apply_import_commit", fail_write)
        patch.setattr("server.knowledge.ingest.ingest_project_analyze_report", fail_ingest)
        patch.setattr(
            "server.aps_domain.project_analyze._project_ctx",
            fail_project_context,
        )
        readonly = analyze_project_deep(
            read_world,
            input_path=str(tmp_path),
            next_id=_counter(),
            apply=False,
        )
    after = json.dumps(read_world, sort_keys=True, default=str)

    assert before == after
    assert readonly["readOnly"] is True
    assert readonly["applied"] == {}
    assert readonly["pendingAdoption"]["mode"] == "folder"
    assert readonly["pendingAdoption"]["batches"]
    assert all(batch["okRows"] for batch in readonly["pendingAdoption"]["batches"])
    assert readonly["pendingAdoption"]["filename"] == ""
    # 订单行引用了尚未采用的产品，预览阶段就无法通过关联校验：这份文件不会进入
    # 可提交批次，但必须原样出现在分析报告的 files 里，不能静默丢弃。
    assert set(readonly["pendingAdoption"]["sourcePaths"]) == {
        str(tmp_path / "materials.csv"),
        str(tmp_path / "routing.csv"),
        str(tmp_path / "equipment.csv"),
    }
    rejected = [
        item for item in (readonly["folder"] or {}).get("files") or []
        if item.get("name") == "orders.csv"
    ]
    assert rejected and rejected[0]["errorCount"] >= 1
    assert rejected[0]["errors"]
    assert not any(
        batch.get("kind") == "orders"
        for batch in readonly["pendingAdoption"]["batches"]
    )

    applied = analyze_project_deep(
        empty_world(),
        input_path=str(tmp_path),
        next_id=_counter(),
        apply=True,
    )

    for key in ("summary", "canSchedule", "orders", "materials", "routings", "equipment"):
        assert readonly[key] == applied[key]


def test_analyze_work_dir_sql_preview_without_apply(tmp_path: Path, monkeypatch):
    sql = tmp_path / "mini.sql"
    sql.write_text(
        """
CREATE TABLE `pl_order` (
  `id` int, `code` varchar(64), `business_code` varchar(64),
  `customer_business_code` varchar(64), `material_code` varchar(64),
  `material_name` varchar(64), `parent_code` varchar(64), `craftl_code` varchar(64),
  `machine_model` varchar(64), `dept` varchar(64), `people` varchar(64),
  `drawing_code` varchar(64), `quantity` decimal(10,2), `man_quantity` decimal(10,2),
  `all_quantity` decimal(10,2), `status` varchar(8), `stock_status` varchar(8),
  `distribution_status` varchar(8), `type` varchar(8), `description` varchar(64),
  `planned_end_time` datetime, `is_delete` varchar(8), `level` int
) ENGINE=InnoDB;
CREATE TABLE `r_production_craftl` (
  `id` int, `craftl_code` varchar(64), `craftl_name` varchar(64),
  `procedure_code` varchar(64), `procedure_name` varchar(64),
  `output_material_code` varchar(64), `output_material_name` varchar(64),
  `working_hours` decimal(10,2), `sort_no` int, `is_delete` varchar(8)
) ENGINE=InnoDB;
CREATE TABLE `md_equipment` (
  `id` int, `code` varchar(64), `name` varchar(64), `is_delete` varchar(8)
) ENGINE=InnoDB;
CREATE TABLE `md_material` (
  `id` int, `code` varchar(64), `name` varchar(64), `type` varchar(32),
  `unit` varchar(8), `is_delete` varchar(8)
) ENGINE=InnoDB;
INSERT INTO `pl_order` VALUES
(1,'ORD1',NULL,'C1','FG001','Product 1',NULL,'CRAFT1',NULL,NULL,NULL,NULL,10,NULL,NULL,'0','0','0','S',NULL,'2026-08-01 00:00:00','0',5);
INSERT INTO `r_production_craftl` VALUES
(1,'CRAFT1','Product 1','OP10','Cut','WIP-X','Semi',0.5,10,'0'),
(2,'CRAFT1','Product 1','OP20','Finish','FG001','Product 1',0.8,20,'0');
INSERT INTO `md_equipment` VALUES (1,'EQ1','Cutter','0');
INSERT INTO `md_material` VALUES (1,'FG001','Product 1','Product','pc','0');
""",
        encoding="utf-8",
    )

    class FakePS:
        def snapshot(self, include_messages=False):
            return {
                "projects": [{"id": "p1", "name": "Readonly", "workDir": str(tmp_path)}],
                "sessions": [{"id": "s1", "projectId": "p1"}],
                "files": [],
            }

    monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakePS())
    monkeypatch.setattr(
        "server.aps_domain.project_analyze._project_ctx_readonly",
        lambda _session_id: (
            {"id": "p1", "name": "Readonly", "workDir": str(tmp_path)},
            str(tmp_path),
        ),
    )
    world = empty_world()
    for key in ("flexOrders", "flexMaterials", "flexRoutings", "flexEquipment", "flexBom"):
        world[key] = []
    before = json.dumps(world, sort_keys=True, default=str)

    def fail_sql_write(*_args, **_kwargs):
        raise AssertionError("SQL preview must not be written")

    def fail_project_snapshot(*_args, **_kwargs):
        raise AssertionError("SQL preview must not call ProjectStore.snapshot")

    with monkeypatch.context() as patch:
        patch.setattr(
            "server.importers.sql_pack.apply_sql_pack_to_world",
            fail_sql_write,
        )
        patch.setattr(
            "server.aps_domain.folder_pack._project_work_dir",
            fail_project_snapshot,
        )
        report = analyze_work_dir(world, "s1", apply_sql=False)
    after = json.dumps(world, sort_keys=True, default=str)

    assert before == after
    assert report["sqlApplied"] is False
    assert report["sqlStats"]["orders"] >= 1
    assert report["sqlPreview"]["filename"] == "mini.sql"
    assert report["sqlPreview"]["fieldMapText"]
    assert report["canSchedule"] is True
    assert world["flexOrders"] == []
    assert world["flexRoutings"] == []
    assert world["flexEquipment"] == []

    applied = analyze_work_dir(empty_world(), "s1", apply_sql=True)
    assert applied["sqlApplied"] is True
    assert applied["sqlStats"] == report["sqlStats"]

    deep_world = empty_world()
    deep_before = json.dumps(deep_world, sort_keys=True, default=str)
    deep = analyze_project_deep(
        deep_world,
        input_path=str(sql),
        next_id=_counter(),
        apply=False,
        apply_sql=False,
    )
    deep_after = json.dumps(deep_world, sort_keys=True, default=str)

    assert deep_before == deep_after
    assert deep["readOnly"] is True
    assert deep["pendingAdoption"]["mode"] == "sql"
    assert deep["summary"]["orders"] >= 1
    assert deep["summary"]["routings"] >= 1
    assert deep["summary"]["equipment"] >= 1
    assert deep["canSchedule"] is True


def test_readonly_knowledge_preview_never_seeds_or_applies_templates(monkeypatch):
    from server.aps_domain.project_analyze import (
        _fill_from_knowledge,
        _recommend_templates_readonly,
    )
    from server.knowledge import routing_templates

    def fail_seed():
        raise AssertionError("readonly preview must not seed routing templates")

    def fail_apply(*_args, **_kwargs):
        raise AssertionError("readonly preview must not apply routing templates")

    monkeypatch.setattr(routing_templates, "seed_templates", fail_seed)
    monkeypatch.setattr(routing_templates, "apply_template_to_product", fail_apply)

    recommendations = _recommend_templates_readonly("箱体 支架 铣削", top_k=1)
    assert recommendations
    assert recommendations[0]["code"] == "TPL-MILLING"

    world = empty_world()
    world["flexOrders"] = [{
        "orderNo": "ORDER-1",
        "productCode": "PRODUCT-1",
        "productName": "箱体支架",
        "status": "NEW",
    }]
    world["flexMaterials"] = [{"code": "PRODUCT-1", "name": "箱体支架"}]
    world["flexRoutings"] = []

    actions = _fill_from_knowledge(world, ["PRODUCT-1"], apply=False)

    assert any(action.get("type") == "template_applied" for action in actions)
    assert world["flexRoutings"]


def test_readonly_project_context_uses_ro_query_without_writes(tmp_path: Path, monkeypatch):
    import sqlite3

    from server.aps_domain.project_analyze import _project_ctx_readonly

    db_path = tmp_path / "master.db"
    connection = sqlite3.connect(db_path)
    connection.executescript(
        """
CREATE TABLE aps_chat_sessions (
  id TEXT, project_id TEXT, scope TEXT, tenant_uuid TEXT, deleted INTEGER
);
CREATE TABLE aps_workspace_projects (
  id TEXT, name TEXT, work_dir TEXT, tenant_uuid TEXT, deleted INTEGER
);
CREATE TABLE aps_project_members (
  id INTEGER, project_id TEXT, tenant_uuid TEXT, user_id INTEGER,
  status TEXT, deleted INTEGER
);
INSERT INTO aps_chat_sessions VALUES ('s1', 'p1', 'project', 'platform', 0);
INSERT INTO aps_workspace_projects VALUES ('p1', 'Readonly', 'C:/data', 'platform', 0);
INSERT INTO aps_project_members VALUES (1, 'p1', 'platform', 0, 'active', 0);
""",
    )
    connection.commit()
    connection.close()
    before = db_path.read_bytes()

    monkeypatch.setenv("APS_DB_PATH", str(db_path))
    monkeypatch.delenv("APS_PROJECTS_PATH", raising=False)

    project, work = _project_ctx_readonly("s1")

    assert project == {"id": "p1", "name": "Readonly", "workDir": "C:/data"}
    assert work == "C:/data"
    assert db_path.read_bytes() == before