"""Canonicalise the Mendeley farm-digester workbook without altering raw files.

The four reactor sheets use mixed Excel/string date encodings.  Dates are
repaired monotonically in source-row order by considering both day-first and
month-first interpretations.  Every repaired value is flagged for audit.
"""

from __future__ import annotations

import hashlib
import json
from datetime import date, datetime
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd

REACTOR_SHEETS = {
    "RI-FLEX": ("R1", "flexible"),
    "R2-FLEX": ("R2", "flexible"),
    "R3-FIXED DOME": ("R3", "fixed_dome"),
    "R4-FIXED DOME": ("R4", "fixed_dome"),
}


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def _date_candidates(value: object) -> list[pd.Timestamp]:
    candidates: set[pd.Timestamp] = set()
    if pd.isna(value):
        return []
    if isinstance(value, (pd.Timestamp, datetime, date)):
        stamp = pd.Timestamp(value).normalize()
        candidates.add(stamp)
        if stamp.day <= 12 and stamp.month <= 12:
            candidates.add(pd.Timestamp(stamp.year, stamp.day, stamp.month))
    else:
        for dayfirst in (True, False):
            parsed = pd.to_datetime(value, dayfirst=dayfirst, errors="coerce")
            if not pd.isna(parsed):
                candidates.add(pd.Timestamp(parsed).normalize())
    return sorted(x for x in candidates if pd.Timestamp("2024-01-01") <= x <= pd.Timestamp("2025-02-01"))


def repair_monotonic_dates(values: Iterable[object]) -> tuple[list[pd.Timestamp | pd.NaT], list[bool]]:
    repaired: list[pd.Timestamp | pd.NaT] = []
    flags: list[bool] = []
    previous: pd.Timestamp | None = None
    for value in values:
        candidates = _date_candidates(value)
        if previous is not None:
            # Equality is retained as an auditable source duplicate. Choosing a
            # distant swapped interpretation to force strict increase would
            # silently corrupt all subsequent dates.
            forward = [x for x in candidates if x >= previous]
            chosen = min(forward, key=lambda x: (x - previous).days) if forward else pd.NaT
        else:
            chosen = min(candidates) if candidates else pd.NaT
        repaired.append(chosen)
        original = (
            pd.Timestamp(value).normalize()
            if isinstance(value, (pd.Timestamp, datetime, date))
            else None
        )
        flags.append(bool(pd.isna(chosen) or original is None or chosen != original))
        if not pd.isna(chosen):
            previous = chosen
    return repaired, flags


def _number(series: pd.Series) -> pd.Series:
    return pd.to_numeric(series.astype(str).str.extract(r"([-+]?\d*\.?\d+)")[0], errors="coerce")


def canonicalise(workbook: Path) -> pd.DataFrame:
    frames: list[pd.DataFrame] = []
    for sheet, (reactor, design) in REACTOR_SHEETS.items():
        source = pd.read_excel(workbook, sheet_name=sheet)
        dates, repaired = repair_monotonic_dates(source.iloc[:, 0].tolist())
        frame = pd.DataFrame(
            {
                "timestamp": dates,
                "reactor_id": reactor,
                "reactor_design": design,
                "source_sheet": sheet,
                "source_row": np.arange(2, len(source) + 2),
                "date_repaired": repaired,
                "manure_fed_kg": _number(source.iloc[:, 4]),
                "water_added_kg": _number(source.iloc[:, 5]),
                "cumulative_biogas_meter_reading": pd.to_numeric(source.iloc[:, 6], errors="coerce"),
                "air_temperature_c": pd.to_numeric(source.iloc[:, 7], errors="coerce"),
                "digester_temperature_c": pd.to_numeric(source.iloc[:, 8], errors="coerce"),
                "gas_temperature_c": pd.to_numeric(source.iloc[:, 18], errors="coerce"),
                "total_biogas_interval_ml": pd.to_numeric(source.iloc[:, 19], errors="coerce"),
                "total_biogas_stp_interval_ml": pd.to_numeric(source.iloc[:, 22], errors="coerce"),
                "methane_fraction_percent": pd.to_numeric(source.iloc[:, 15], errors="coerce"),
            }
        )
        frame["interval_days"] = frame["timestamp"].diff().dt.days
        frame["duplicate_timestamp"] = frame["timestamp"].duplicated(keep=False)
        frame["discontinuity"] = frame["interval_days"].notna() & frame["interval_days"].ne(1)
        # The source's methane-derived columns are not copied: methane fraction
        # is nearly absent, so their zero values are not measured methane.
        frames.append(frame)
    result = pd.concat(frames, ignore_index=True)
    result = result.sort_values(["timestamp", "reactor_id"], kind="stable").reset_index(drop=True)
    return result


def quality_summary(data: pd.DataFrame) -> dict:
    by_reactor = {}
    for reactor, part in data.groupby("reactor_id", sort=True):
        cadence = part["timestamp"].sort_values().diff().dt.days
        by_reactor[reactor] = {
            "rows": int(len(part)),
            "start": str(part["timestamp"].min().date()),
            "end": str(part["timestamp"].max().date()),
            "duplicate_timestamps": int(part["duplicate_timestamp"].sum()),
            "non_daily_intervals": int((cadence.notna() & cadence.ne(1)).sum()),
            "median_interval_days": float(cadence.dropna().median()),
            "target_missing": int(part["total_biogas_interval_ml"].isna().sum()),
            "temperature_missing": int(part["air_temperature_c"].isna().sum()),
            "digester_temperature_missing": int(part["digester_temperature_c"].isna().sum()),
            "methane_fraction_observations": int(part["methane_fraction_percent"].notna().sum()),
            "date_repairs": int(part["date_repaired"].sum()),
        }
    return {
        "rows": int(len(data)),
        "reactors": sorted(data["reactor_id"].unique().tolist()),
        "timestamp_start": str(data["timestamp"].min()),
        "timestamp_end": str(data["timestamp"].max()),
        "target_definition": "Total biogas meter increment between recorded observations, source-labelled mL",
        "target_is_methane": False,
        "by_reactor": by_reactor,
        "missing_fraction": {
            c: float(data[c].isna().mean())
            for c in [
                "total_biogas_interval_ml",
                "air_temperature_c",
                "digester_temperature_c",
                "manure_fed_kg",
                "methane_fraction_percent",
            ]
        },
    }


def _daily_weather(workbook: Path) -> pd.DataFrame:
    weather = pd.read_excel(workbook, sheet_name="Config 1", header=2, skiprows=[3])
    weather["timestamp"] = pd.to_datetime(weather.iloc[:, 0], errors="coerce")
    weather["date"] = weather["timestamp"].dt.normalize()
    air = pd.to_numeric(weather[" °C Air Temperature"], errors="coerce")
    rain = pd.to_numeric(weather[" mm Precipitation"], errors="coerce")
    solar = pd.to_numeric(weather[" W/m² Solar Radiation"], errors="coerce")
    weather = weather.assign(_air=air, _rain=rain, _solar=solar)
    return weather.groupby("date", as_index=False).agg(
        weather_air_temperature_mean_c=("_air", "mean"),
        weather_air_temperature_min_c=("_air", "min"),
        weather_air_temperature_max_c=("_air", "max"),
        weather_precipitation_sum_mm=("_rain", "sum"),
        weather_solar_radiation_mean_w_m2=("_solar", "mean"),
        weather_observation_count=("timestamp", "count"),
    ).rename(columns={"date": "timestamp"})


def process(
    workbook: Path,
    output_csv: Path,
    manifest_json: Path,
    weather_workbook: Path | None = None,
) -> dict:
    data = canonicalise(workbook)
    if weather_workbook is not None:
        data = data.merge(_daily_weather(weather_workbook), on="timestamp", how="left", validate="many_to_one")
    output_csv.parent.mkdir(parents=True, exist_ok=True)
    data.to_csv(output_csv, index=False, date_format="%Y-%m-%d")
    manifest = {
        "schema_version": "gfis-gate3-ds03-v1",
        "source_workbook": str(workbook),
        "source_workbook_sha256": sha256(workbook),
        "weather_workbook": str(weather_workbook) if weather_workbook else None,
        "weather_workbook_sha256": sha256(weather_workbook) if weather_workbook else None,
        "processed_file": str(output_csv),
        "processed_sha256": sha256(output_csv),
        "processing": [
            "Read four reactor sheets independently.",
            "Repair mixed Excel/string dates monotonically in source-row order; retain date_repaired flag.",
            "Parse numeric feed values without imputing raw observations.",
            "Preserve reactor identity and source row.",
            "Flag duplicate timestamps and non-daily intervals.",
            "Do not use zero-filled methane-derived source columns as measured methane.",
            "Aggregate 5-minute weather observations to daily summaries and left-join by calendar date.",
        ],
        "quality": quality_summary(data),
    }
    manifest_json.parent.mkdir(parents=True, exist_ok=True)
    manifest_json.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
    return manifest