"""Isolated database preparation and deterministic burst-dataset scheduling."""
from __future__ import annotations

import hashlib
import json
import math
from datetime import datetime, timedelta, timezone
from pathlib import Path
from uuid import UUID
from zoneinfo import ZoneInfo

from psycopg import connect, sql
from psycopg.conninfo import make_conninfo
from psycopg.rows import dict_row


def canonical_hash(value) -> str:
    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest()


def deterministic_standard_normal(*, seed: str, at: datetime, device_id: str, metric: str) -> float:
    """Return order-independent deterministic Gaussian noise for one observation."""
    material = f"{seed}\0{at.isoformat()}\0{device_id}\0{metric}".encode()
    digest = hashlib.sha256(material).digest()
    first = (int.from_bytes(digest[:8], "big") + 1) / (2**64 + 1)
    second = (int.from_bytes(digest[8:16], "big") + 1) / (2**64 + 1)
    return math.sqrt(-2.0 * math.log(first)) * math.cos(2.0 * math.pi * second)


def noisy_scada_observation(
    *,
    raw_value: float,
    device_type: str,
    at: datetime,
    device_id: str,
    seed: str,
    pressure_sigma_percent: float = 0.2,
    pressure_sigma_floor_m: float = 0.02,
    flow_sigma_percent: float = 0.5,
    clip_sigma: float = 3.0,
) -> float:
    """Convert model units and add reproducible zero-mean telemetry noise."""
    if not math.isfinite(raw_value):
        raise ValueError("model observation must be finite")
    if min(pressure_sigma_percent, pressure_sigma_floor_m, flow_sigma_percent, clip_sigma) < 0:
        raise ValueError("noise parameters must be non-negative")
    z = max(-clip_sigma, min(clip_sigma, deterministic_standard_normal(
        seed=seed, at=at, device_id=device_id, metric=device_type,
    )))
    if device_type == "pressure":
        converted = raw_value
        sigma = max(abs(converted) * pressure_sigma_percent / 100.0, pressure_sigma_floor_m)
        return max(0.0, converted + z * sigma)
    if device_type == "pipe_flow":
        converted = raw_value * 3.6  # L/s -> m3/h
        sigma = abs(converted) * flow_sigma_percent / 100.0
        return converted + z * sigma
    raise ValueError(f"unsupported SCADA device type: {device_type}")


def write_realtime_scada_day(
    *,
    business_dsn: str,
    timescale_dsn: str,
    start: datetime,
    interval_seconds: int = 900,
    duration_seconds: int = 86400,
    noise_seed: str,
    pressure_sigma_percent: float = 0.2,
    pressure_sigma_floor_m: float = 0.02,
    flow_sigma_percent: float = 0.5,
    clip_sigma: float = 3.0,
) -> dict:
    """Map one online-model day into noisy SCADA observations."""
    if start.tzinfo is None or duration_seconds <= 0 or interval_seconds <= 0:
        raise ValueError("an aware start and positive duration/interval are required")
    if duration_seconds % interval_seconds:
        raise ValueError("duration must be divisible by the sampling interval")
    expected_times = [
        start + timedelta(seconds=offset)
        for offset in range(0, duration_seconds, interval_seconds)
    ]
    end = start + timedelta(seconds=duration_seconds)
    with connect(business_dsn, row_factory=dict_row) as business:
        devices = business.execute(
            """SELECT device_id,device_type,node_id,link_id,measurement_unit
               FROM asset.scada_devices
               WHERE device_type IN ('pressure','pipe_flow')
               ORDER BY device_id"""
        ).fetchall()
    if not devices:
        raise ValueError("no pressure or pipe-flow SCADA bindings are available")
    records: list[tuple] = []
    raw_fingerprint_rows: list[tuple] = []
    with connect(timescale_dsn, row_factory=dict_row) as timescale:
        for device in devices:
            if device["device_type"] == "pressure":
                if not device["node_id"] or device["measurement_unit"] != "m":
                    raise ValueError("pressure devices must bind nodes and use metres")
                rows = timescale.execute(
                    """SELECT time,pressure AS value FROM realtime.node_results
                       WHERE node_id=%s AND time>=%s AND time<%s ORDER BY time""",
                    (device["node_id"], start, end),
                ).fetchall()
            else:
                if not device["link_id"] or device["measurement_unit"] != "m3/h":
                    raise ValueError("flow devices must bind links and use m3/h")
                rows = timescale.execute(
                    """SELECT time,flow AS value FROM realtime.link_results
                       WHERE link_id=%s AND time>=%s AND time<%s ORDER BY time""",
                    (device["link_id"], start, end),
                ).fetchall()
            actual_times = [row["time"] for row in rows]
            if actual_times != expected_times:
                raise ValueError(f"incomplete realtime samples for device {device['device_id']}")
            for row in rows:
                raw_value = float(row["value"])
                value = noisy_scada_observation(
                    raw_value=raw_value,
                    device_type=device["device_type"],
                    at=row["time"],
                    device_id=device["device_id"],
                    seed=noise_seed,
                    pressure_sigma_percent=pressure_sigma_percent,
                    pressure_sigma_floor_m=pressure_sigma_floor_m,
                    flow_sigma_percent=flow_sigma_percent,
                    clip_sigma=clip_sigma,
                )
                raw_fingerprint_rows.append((row["time"].isoformat(), device["device_id"], raw_value))
                records.append((row["time"], device["device_id"], value, None))
        collision = timescale.execute(
            "SELECT count(*) count FROM scada.measurements WHERE time>=%s AND time<%s",
            (start, end),
        ).fetchone()["count"]
        if collision:
            raise ValueError("SCADA target day is not empty")
        with timescale.cursor().copy(
            "COPY scada.measurements (time,device_id,monitored_value,cleaned_value) FROM STDIN"
        ) as copy:
            for record in records:
                copy.write_row(record)
        timescale.commit()
        persisted = timescale.execute(
            "SELECT count(*) count FROM scada.measurements WHERE time>=%s AND time<%s",
            (start, end),
        ).fetchone()["count"]
    if persisted != len(records):
        raise ValueError("SCADA normal-day read-back count does not match generated records")
    return {
        "start": start.isoformat(),
        "end_exclusive": end.isoformat(),
        "devices": len(devices),
        "samples_per_device": len(expected_times),
        "rows": persisted,
        "units": {"pressure": "m", "pipe_flow": "m3/h", "model_flow": "L/s"},
        "noise": {
            "seed": noise_seed,
            "distribution": "deterministic_clipped_gaussian",
            "pressure_sigma_percent": pressure_sigma_percent,
            "pressure_sigma_floor_m": pressure_sigma_floor_m,
            "flow_sigma_percent": flow_sigma_percent,
            "clip_sigma": clip_sigma,
        },
        "raw_sha256": canonical_hash(raw_fingerprint_rows),
        "value_sha256": canonical_hash(
            [(time.isoformat(), device_id, value) for time, device_id, value, _ in records]
        ),
    }


def write_demo_scada_window(
    *,
    business_dsn: str,
    timescale_dsn: str,
    normal_source_start: datetime,
    target_start: datetime,
    day_count: int,
    event_start: datetime,
    event_run_id: str,
    noise_seed: str,
    interval_seconds: int = 900,
) -> dict:
    """Repeat one normal online day and replace one five-hour event window."""
    if any(value.tzinfo is None for value in (normal_source_start, target_start, event_start)):
        raise ValueError("normal, target, and event times must be timezone-aware")
    if day_count < 15 or interval_seconds <= 0 or 86400 % interval_seconds:
        raise ValueError("demo windows require at least 15 complete aligned days")
    target_end = target_start + timedelta(days=day_count)
    if not target_start <= event_start < target_end:
        raise ValueError("event must fall inside the demo window")
    run_uuid = UUID(str(event_run_id))
    with connect(business_dsn, row_factory=dict_row) as business:
        devices = business.execute(
            """SELECT device_id,device_type,node_id,link_id,measurement_unit
               FROM asset.scada_devices
               WHERE device_type IN ('pressure','pipe_flow') ORDER BY device_id"""
        ).fetchall()
    pressure_devices = [row for row in devices if row["device_type"] == "pressure"]
    flow_devices = [row for row in devices if row["device_type"] == "pipe_flow"]
    if any(not row["node_id"] or row["measurement_unit"] != "m" for row in pressure_devices):
        raise ValueError("pressure devices must bind nodes and use metres")
    if any(not row["link_id"] or row["measurement_unit"] != "m3/h" for row in flow_devices):
        raise ValueError("flow devices must bind links and use m3/h")
    normal_end = normal_source_start + timedelta(days=1)
    expected_normal_times = [
        normal_source_start + timedelta(seconds=offset)
        for offset in range(0, 86400, interval_seconds)
    ]
    with connect(timescale_dsn, row_factory=dict_row) as timescale:
        normal_nodes = timescale.execute(
            """SELECT time,node_id,pressure AS value FROM realtime.node_results
               WHERE time>=%s AND time<%s AND node_id=ANY(%s) ORDER BY time,node_id""",
            (normal_source_start, normal_end, [row["node_id"] for row in pressure_devices]),
        ).fetchall()
        normal_links = timescale.execute(
            """SELECT time,link_id,flow AS value FROM realtime.link_results
               WHERE time>=%s AND time<%s AND link_id=ANY(%s) ORDER BY time,link_id""",
            (normal_source_start, normal_end, [row["link_id"] for row in flow_devices]),
        ).fetchall()
        event_nodes = timescale.execute(
            """SELECT time,node_id,pressure AS value FROM analysis.node_results
               WHERE run_id=%s AND node_id=ANY(%s) ORDER BY time,node_id""",
            (run_uuid, [row["node_id"] for row in pressure_devices]),
        ).fetchall()
        event_links = timescale.execute(
            """SELECT time,link_id,flow AS value FROM analysis.link_results
               WHERE run_id=%s AND link_id=ANY(%s) ORDER BY time,link_id""",
            (run_uuid, [row["link_id"] for row in flow_devices]),
        ).fetchall()

        normal_node_values = {(row["time"], row["node_id"]): float(row["value"]) for row in normal_nodes}
        normal_link_values = {(row["time"], row["link_id"]): float(row["value"]) for row in normal_links}
        for device in devices:
            values = normal_node_values if device["device_type"] == "pressure" else normal_link_values
            element = device["node_id"] if device["device_type"] == "pressure" else device["link_id"]
            if any((at, element) not in values for at in expected_normal_times):
                raise ValueError(f"normal source day is incomplete for device {device['device_id']}")

        records: dict[tuple[datetime, str], tuple] = {}
        for day_offset in range(day_count):
            day_start = target_start + timedelta(days=day_offset)
            for sample_offset, source_at in enumerate(expected_normal_times):
                target_at = day_start + timedelta(seconds=sample_offset * interval_seconds)
                for device in pressure_devices:
                    raw = normal_node_values[(source_at, device["node_id"])]
                    value = noisy_scada_observation(
                        raw_value=raw, device_type="pressure", at=target_at,
                        device_id=device["device_id"], seed=noise_seed,
                    )
                    records[(target_at, device["device_id"])] = (
                        target_at, device["device_id"], value, None,
                    )
                for device in flow_devices:
                    raw = normal_link_values[(source_at, device["link_id"])]
                    value = noisy_scada_observation(
                        raw_value=raw, device_type="pipe_flow", at=target_at,
                        device_id=device["device_id"], seed=noise_seed,
                    )
                    records[(target_at, device["device_id"])] = (
                        target_at, device["device_id"], value, None,
                    )

        if not event_nodes or not event_links:
            raise ValueError("event analysis has no bound pressure or flow results")
        source_event_start = min(row["time"] for row in event_nodes + event_links)
        event_times = sorted({row["time"] for row in event_nodes + event_links})
        if len(event_times) != 21 or any(
            right - left != timedelta(seconds=interval_seconds)
            for left, right in zip(event_times, event_times[1:])
        ):
            raise ValueError("event analysis must contain 21 aligned fifteen-minute frames")
        event_node_values = {(row["time"], row["node_id"]): float(row["value"]) for row in event_nodes}
        event_link_values = {(row["time"], row["link_id"]): float(row["value"]) for row in event_links}
        for source_at in event_times:
            target_at = event_start + (source_at - source_event_start)
            if target_at >= target_end:
                raise ValueError("event results extend beyond the demo window")
            for device in pressure_devices:
                key = (source_at, device["node_id"])
                if key not in event_node_values:
                    raise ValueError(f"event pressure is incomplete for device {device['device_id']}")
                value = noisy_scada_observation(
                    raw_value=event_node_values[key], device_type="pressure", at=target_at,
                    device_id=device["device_id"], seed=noise_seed,
                )
                records[(target_at, device["device_id"])] = (
                    target_at, device["device_id"], value, None,
                )
            for device in flow_devices:
                key = (source_at, device["link_id"])
                if key not in event_link_values:
                    raise ValueError(f"event flow is incomplete for device {device['device_id']}")
                value = noisy_scada_observation(
                    raw_value=event_link_values[key], device_type="pipe_flow", at=target_at,
                    device_id=device["device_id"], seed=noise_seed,
                )
                records[(target_at, device["device_id"])] = (
                    target_at, device["device_id"], value, None,
                )

        ordered = [records[key] for key in sorted(records)]
        expected_count = day_count * len(expected_normal_times) * len(devices)
        if len(ordered) != expected_count:
            raise ValueError("demo SCADA window has an unexpected observation count")
        timescale.execute(
            "DELETE FROM scada.measurements WHERE time>=%s AND time<%s",
            (target_start, target_end),
        )
        with timescale.cursor().copy(
            "COPY scada.measurements (time,device_id,monitored_value,cleaned_value) FROM STDIN"
        ) as copy:
            for record in ordered:
                copy.write_row(record)
        timescale.commit()
        persisted = timescale.execute(
            """SELECT count(*) count,count(DISTINCT device_id) devices,
                      count(DISTINCT time) frames,count(cleaned_value) cleaned
               FROM scada.measurements WHERE time>=%s AND time<%s""",
            (target_start, target_end),
        ).fetchone()
    if persisted != {
        "count": expected_count,
        "devices": len(devices),
        "frames": day_count * len(expected_normal_times),
        "cleaned": 0,
    }:
        raise ValueError("demo SCADA read-back verification failed")
    return {
        "start": target_start.isoformat(),
        "end_exclusive": target_end.isoformat(),
        "day_count": day_count,
        "event_start": event_start.isoformat(),
        "event_end": (event_start + timedelta(seconds=20 * interval_seconds)).isoformat(),
        "event_run_id": str(run_uuid),
        "devices": len(devices),
        "frames": day_count * len(expected_normal_times),
        "rows": expected_count,
        "cleaned_rows": 0,
        "noise_seed": noise_seed,
        "value_sha256": canonical_hash(
            [(time.isoformat(), device_id, value) for time, device_id, value, _ in ordered]
        ),
    }


def create_empty_database(source_dsn: str, target: str, schema_template: str) -> None:
    if not target.startswith("tjw_bench_") or target.endswith("_template"):
        raise ValueError("benchmark targets must use the owned tjw_bench_ prefix")
    with connect(make_conninfo(source_dsn, dbname="postgres"), autocommit=True, row_factory=dict_row) as connection:
        if connection.execute("SELECT 1 FROM pg_database WHERE datname=%s", (target,)).fetchone():
            raise ValueError("target database already exists; explicit state is required for resumption")
        # Templates are intentionally connection-disabled; clone without altering that state.
        connection.execute(sql.SQL("CREATE DATABASE {} TEMPLATE {}").format(sql.Identifier(target), sql.Identifier(schema_template)))


def copy_project_inputs(source_dsn: str, target_dsn: str) -> dict:
    from scripts.check_burst_simulation_design import readonly
    from app.native.wndb.core.model_replace import _copy_order, _copy_table, _model_tables, _table_columns

    summaries = {}
    with readonly(source_dsn) as source, connect(target_dsn, row_factory=dict_row) as target:
        source.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY")
        tables = _model_tables(source)
        ordered = _copy_order(source, tables)
        for schema, table in reversed(ordered):
            target.execute(sql.SQL("DELETE FROM {}").format(sql.Identifier(schema, table)))
        for schema, table in ordered + [("asset", "scada_devices")]:
            columns = _table_columns(source, schema, table)
            if _table_columns(target, schema, table) != columns:
                raise ValueError(f"schema differs for {schema}.{table}")
            _copy_table(source, target, schema, table, columns)
            statement = sql.SQL("SELECT count(*) count FROM {}").format(sql.Identifier(schema, table))
            before, after = source.execute(statement).fetchone()["count"], target.execute(statement).fetchone()["count"]
            if before != after:
                raise ValueError("isolated model copy count mismatch")
            summaries[f"{schema}.{table}"] = before
        target.commit()
    return summaries


def apply_project_runtime_migrations(target_dsn: str, repository_root: Path) -> list[str]:
    """Install the additive MCP and calculation queue tables in an owned target."""
    migrations = (
        "025_mcp_query_contexts.sql",
        "026_calculation_jobs.sql",
        "027_calculation_job_lifecycle.sql",
        "030_remove_calculation_queue_expiry.sql",
    )
    applied = []
    with connect(target_dsn) as connection:
        for filename in migrations:
            path = repository_root / "resources" / "sql" / filename
            connection.execute(path.read_text(encoding="utf-8"))
            applied.append(filename)
        connection.commit()
    return applied


async def register_benchmark_project(
    *,
    source_project_id: str,
    target: str,
    target_biz_dsn: str,
    target_iot_dsn: str,
):
    """Register owned databases and mirror source membership without role expansion."""
    from uuid import UUID

    from app.infra.db.metadb.database import SessionLocal
    from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository

    source_id = UUID(str(source_project_id))
    async with SessionLocal() as session:
        repository = MetadataRepository(session)
        source = await repository.get_project_by_id(source_id)
        if source is None:
            raise ValueError("source project metadata is unavailable")
        project = await repository.get_project_by_code(target)
        marker = {
            "benchmark_source_project_id": str(source_id),
            "benchmark_dataset": "burst-v1",
        }
        if project is None:
            project = await repository.create_project(
                name=f"{source.name} 爆管 Benchmark",
                code=target,
                description="隔离的爆管 benchmark 模拟与 SCADA 回放数据集",
                gs_workspace=target,
                map_extent=source.map_extent,
                status="active",
                map_config=source.map_config or {},
                settings={**(source.settings or {}), **marker},
            )
        elif any((project.settings or {}).get(key) != value for key, value in marker.items()):
            raise ValueError("existing target project is not owned by this benchmark dataset")
        await repository.upsert_project_database_config(
            project.id,
            db_role="biz_data",
            db_type="postgresql",
            dsn=target_biz_dsn,
            pool_min_size=1,
            pool_max_size=4,
        )
        await repository.upsert_project_database_config(
            project.id,
            db_role="iot_data",
            db_type="timescaledb",
            dsn=target_iot_dsn,
            pool_min_size=1,
            pool_max_size=4,
        )
        existing = {member.user_id for member in await repository.list_project_members(project.id)}
        for member in await repository.list_project_members(source_id):
            if member.user_id not in existing:
                await repository.add_project_member(
                    project.id, member.user_id, member.project_role
                )
        return project


def write_analysis_scada(
    *,
    business_dsn: str,
    timescale_dsn: str,
    run_id: str,
) -> dict:
    """Materialize one completed hydraulic run as bound SCADA observations."""
    run_uuid = UUID(str(run_id))
    with connect(business_dsn, row_factory=dict_row) as business:
        devices = business.execute(
            """SELECT device_id,device_type,node_id,link_id,measurement_unit
               FROM asset.scada_devices
               WHERE device_type IN ('pressure','pipe_flow')
               ORDER BY device_id"""
        ).fetchall()
    pressure = [row for row in devices if row["device_type"] == "pressure"]
    flows = [row for row in devices if row["device_type"] == "pipe_flow"]
    if any(not row["node_id"] or row["measurement_unit"] != "m" for row in pressure):
        raise ValueError("pressure devices must bind nodes and use metres")
    if any(not row["link_id"] or row["measurement_unit"] != "m3/h" for row in flows):
        raise ValueError("flow devices must bind links and use m3/h")

    pressure_by_node: dict[str, list[str]] = {}
    flow_by_link: dict[str, list[str]] = {}
    for row in pressure:
        pressure_by_node.setdefault(row["node_id"], []).append(row["device_id"])
    for row in flows:
        flow_by_link.setdefault(row["link_id"], []).append(row["device_id"])

    records: list[tuple] = []
    with connect(timescale_dsn, row_factory=dict_row) as timescale:
        node_rows = timescale.execute(
            """SELECT time,node_id,pressure FROM analysis.node_results
               WHERE run_id=%s AND node_id=ANY(%s) ORDER BY time,node_id""",
            (run_uuid, list(pressure_by_node)),
        ).fetchall()
        link_rows = timescale.execute(
            """SELECT time,link_id,flow FROM analysis.link_results
               WHERE run_id=%s AND link_id=ANY(%s) ORDER BY time,link_id""",
            (run_uuid, list(flow_by_link)),
        ).fetchall()
        for row in node_rows:
            records.extend(
                (row["time"], device_id, float(row["pressure"]), None)
                for device_id in pressure_by_node[row["node_id"]]
            )
        for row in link_rows:
            records.extend(
                (row["time"], device_id, float(row["flow"]) * 3.6, None)
                for device_id in flow_by_link[row["link_id"]]
            )
        if not records:
            raise ValueError("analysis run has no bound SCADA results")
        start = min(row[0] for row in records)
        end = max(row[0] for row in records)
        collision = timescale.execute(
            "SELECT count(*) count FROM scada.measurements WHERE time BETWEEN %s AND %s",
            (start, end),
        ).fetchone()["count"]
        if collision:
            raise ValueError("SCADA target interval is not empty")
        with timescale.cursor().copy(
            "COPY scada.measurements (time,device_id,monitored_value,cleaned_value) FROM STDIN"
        ) as copy:
            for record in sorted(records):
                copy.write_row(record)
        timescale.commit()
        persisted = timescale.execute(
            "SELECT count(*) count FROM scada.measurements WHERE time BETWEEN %s AND %s",
            (start, end),
        ).fetchone()["count"]
    if persisted != len(records):
        raise ValueError("SCADA read-back count does not match generated records")
    samples_per_device: dict[str, int] = {}
    for _, device_id, _, _ in records:
        samples_per_device[device_id] = samples_per_device.get(device_id, 0) + 1
    if len(set(samples_per_device.values())) != 1:
        raise ValueError("bound SCADA devices have inconsistent sample counts")
    return {
        "run_id": str(run_uuid),
        "start": start.isoformat(),
        "end": end.isoformat(),
        "devices": len(samples_per_device),
        "samples_per_device": next(iter(samples_per_device.values())),
        "rows": persisted,
        "value_sha256": canonical_hash(
            [(time.isoformat(), device_id, value) for time, device_id, value, _ in sorted(records)]
        ),
    }


def make_schedule(design) -> list[dict]:
    """Seventeen-day slots keep fifteen clean training days plus a recovery buffer."""
    local = ZoneInfo("Asia/Shanghai")
    first_date = datetime(2024, 1, 16, tzinfo=local)
    rows = []
    pipes = {pipe.pipe_id: pipe for pipe in design.pipes}
    for case in design.cases:
        for level in ("light", "medium", "severe"):
            source = case.occurred_at.astimezone(local)
            at = first_date + timedelta(days=17 * len(rows), hours=source.hour, minutes=source.minute)
            rows.append({"case_key": f"{case.case_id}-{level}", "case_id": case.case_id,
                         "severity": level, "pipe_id": case.pipe_id,
                         "area_cm2": pipes[case.pipe_id].candidate_orifice_areas_cm2[level],
                         "event_at": at.isoformat(), "history_start": (at - timedelta(days=15)).isoformat(),
                         "event_end": (at + timedelta(seconds=design.duration_seconds)).isoformat(),
                         "reserved_end": (at + timedelta(seconds=design.duration_seconds, days=1)).isoformat()})
    previous = None
    for row in rows:
        start = datetime.fromisoformat(row["history_start"])
        if previous is not None and start <= previous:
            raise ValueError("dataset date reservations overlap")
        previous = datetime.fromisoformat(row["reserved_end"])
    if previous is not None and previous >= datetime.now(timezone.utc):
        raise ValueError("benchmark slots must remain historical")
    return rows