import numpy as np
import pandas as pd
import re
from lightgbm import LGBMRegressor


def diagnose(model, X_valid, y_valid, y_train, *, X_train=None, training_log=None):
    """単出力LGBMRegressor用。X_validとy_validの行順一致は呼び出し側の契約。"""
    if not isinstance(model, LGBMRegressor):
        raise TypeError("modelは学習済みのLGBMRegressorにしてください")
    def vector(values, name):
        values = np.asarray(values, dtype=float)
        if values.ndim != 1 or not values.size or not np.isfinite(values).all():
            raise ValueError(f"{name}は空でない有限値の1次元配列にしてください")
        return values

    y_valid = vector(y_valid, "y_valid")
    y_train = vector(y_train, "y_train")
    if len(X_valid.shape) != 2 or X_valid.shape[0] != len(y_valid):
        raise ValueError("X_validは2次元で、y_validと同じ行数が必要です")
    if X_valid.shape[1] != model.n_features_in_:
        raise ValueError("X_validの列数が学習時と違います")
    if training_log is not None and not isinstance(training_log, str):
        raise TypeError("training_logは文字列またはNoneにしてください")
    for label, frame in [("X_valid", X_valid), ("X_train", X_train)]:
        if frame is None:
            continue
        if len(frame.shape) != 2 or frame.shape[1] != model.n_features_in_:
            raise ValueError(f"{label}の形状が学習時と違います")
        if isinstance(frame, pd.DataFrame):
            if list(frame.columns) != model.feature_name_ or not frame.columns.is_unique:
                raise ValueError(f"{label}の列名・列順が学習時と違います")
    if X_train is not None and X_train.shape[0] != len(y_train):
        raise ValueError("X_trainとy_trainの行数が違います")
    iteration = model.best_iteration_ if model.best_iteration_ > 0 else -1
    pred = vector(model.predict(
        X_valid, num_iteration=iteration, validate_features=True
    ), "prediction")
    if pred.shape != y_valid.shape:
        raise ValueError("予測とy_validの長さが違います")
    trees = model.booster_.dump_model(num_iteration=iteration)["tree_info"]
    used = set()

    def visit(node):
        if "split_feature" in node:
            used.add(node["split_feature"])
            visit(node["left_child"])
            visit(node["right_child"])

    for tree in trees:
        visit(tree["tree_structure"])
    mae = float(np.mean(np.abs(y_valid - pred)))
    baseline = float(np.mean(np.abs(y_valid - y_train.mean())))
    constant = bool(np.all(pred == pred[0]))
    if constant:
        action = "この検証入力では定数予測。入力の偏りと分割制約を点検（異常とは未確定）"
    elif mae >= baseline:
        action = "この検証組では平均予測を上回らず。有効性未確認、データと評価方法を点検"
    else:
        action = "この検証組では平均予測より良い。警告だけで設定を変えず検証継続"

    def columns(frame):
        if not isinstance(frame, pd.DataFrame):
            return None
        counts = frame.nunique(dropna=False)
        return dict(constant=counts[counts == 1].index.tolist(),
                    all_missing=frame.columns[frame.isna().all()].tolist())

    train_columns, valid_columns = columns(X_train), columns(X_valid)
    changed = None
    if train_columns is not None and valid_columns is not None:
        changed = [c for c in valid_columns["constant"] if c not in train_columns["constant"]]
    patterns = {
        "no_further_splits": r"No further splits with positive gain",
        "stopped_training": r"Stopped training because",
        "no_meaningful_features": r"There are no meaningful features",
        "zero_used_features": r"(?:number of )?used features:\s*0\b",
    }
    logs = {key: None if training_log is None else len(re.findall(pattern, training_log))
            for key, pattern in patterns.items()}
    checks = [action]
    if training_log is None:
        checks.append("学習ログ未提供。Stopped trainingとused features: 0の有無は未確認")
    elif not any(logs.values()):
        checks.append("対象ログの文字列一致なし。未発生の保証ではなく、ログ抑制・欠落・版差も確認")
    if len(y_valid) < 2:
        checks.append("検証が1行なので予測は必ず一定。予測変動の判断には複数行の検証が必要")
    if logs["no_meaningful_features"] or logs["zero_used_features"]:
        checks.append("特徴量が使えないログあり。全欠損・一定列とmin_child_samples/min_data_in_bin、Dataset事前除外を確認")
    if logs["stopped_training"]:
        checks.append("分割不能の停止ログあり。葉数とmin_child_samples/min_split_gain等を確認。検証指標によるearly stoppingとは別")
    if logs["no_further_splits"]:
        checks.append("木の追加分割が止まったログあり。回数だけで失敗とせず、葉数・検証MAEと併せて判断")
    if train_columns is None or valid_columns is None:
        checks.append("学習・検証双方のDataFrameがそろわず、列名による入力比較は未実施")
    if changed:
        checks.append("学習では変動し検証では一定の列あり。検証範囲や抽出条件を確認: " + ", ".join(changed))
    for label, info in [("学習", train_columns), ("検証", valid_columns)]:
        if info and info["all_missing"]:
            checks.append(label + "の全欠損列あり。読み込み・結合処理を確認: " + ", ".join(info["all_missing"]))
    return dict(
        prediction_trees=len(trees),
        valid_rows=len(y_valid),
        leaves_per_tree=[tree["num_leaves"] for tree in trees],
        split_features=len(used), prediction_std=float(np.std(pred)),
        constant_on_valid=constant, valid_mae=mae,
        train_mean_baseline_mae=baseline, next_check=action,
        log_observations=logs, train_columns=train_columns, valid_columns=valid_columns,
        varying_train_constant_valid=changed, checks=checks,
    )