import numpy as np
import pandas as pd


def daily_features(df, active_periods):
    """各店舗の対象期間を日次にそろえ、前日と直前3暦日の特徴量を作る。"""
    data = df[["store", "date", "sales"]].copy()
    if data[["store", "date"]].isna().any().any() or data["store"].eq("").any():
        raise ValueError("店舗と日付は必須です")
    if not pd.api.types.is_datetime64_any_dtype(data["date"]):
        if not data["date"].astype("string").str.fullmatch(r"\d{4}-\d{2}-\d{2}").all():
            raise ValueError("日付文字列はYYYY-MM-DDで指定してください")
        data["date"] = pd.to_datetime(data["date"], format="%Y-%m-%d", errors="raise")
    if data["date"].dt.tz is not None:
        raise ValueError("タイムゾーンなしの日付を指定してください")
    if not data["date"].eq(data["date"].dt.normalize()).all():
        raise ValueError("日付は0時に統一してください")
    if data.duplicated(["store", "date"]).any():
        raise ValueError("1店舗1日1行に集約してください")
    if set(data["store"]) != set(active_periods):
        raise ValueError("全店舗の対象期間を指定してください")
    data["sales"] = pd.to_numeric(data["sales"], errors="raise").astype(float)
    if np.isinf(data["sales"]).any():
        raise ValueError("売上に無限大は使えません")

    frames = []
    for store, part in data.groupby("store", sort=True):
        bounds = active_periods[store]
        for bound in bounds:
            if isinstance(bound, str):
                if not pd.Series([bound]).str.fullmatch(r"\d{4}-\d{2}-\d{2}").all():
                    raise ValueError("対象期間の文字列はYYYY-MM-DDで指定してください")
        start, end = map(pd.Timestamp, bounds)
        if pd.isna(start) or pd.isna(end) or start > end:
            raise ValueError("対象期間が不正です")
        if any(d.tz is not None or d != d.normalize() for d in (start, end)):
            raise ValueError("対象期間もタイムゾーンなし・0時で指定してください")
        if not part["date"].between(start, end).all():
            raise ValueError("対象期間外のレコードがあります")
        days = pd.date_range(start, end, freq="D", name="date")
        part = part.set_index("date").sort_index()
        daily = part[["sales"]].reindex(days)
        daily["record_present"] = days.isin(part.index)
        daily["store"] = store
        # 欠測はNaNのまま残す。休業などで既知の0だけを入力に含める。
        past = daily["sales"].shift(1)
        daily["lag1"] = past
        daily["past3"] = past.rolling(3, min_periods=3).mean()
        frames.append(daily.reset_index())
    if not frames:
        raise ValueError("少なくとも1店舗のデータが必要です")
    return pd.concat(frames, ignore_index=True)[
        ["store", "date", "sales", "record_present", "lag1", "past3"]
    ]


def audit_daily_lags(df, active_periods):
    """元行ごとの参照日・履歴不足と、元行を分母にした診断集計を返す。"""
    daily = daily_features(df, active_periods)
    original = daily.loc[daily["record_present"]].copy()
    groups = original.groupby("store", sort=False)
    original["previous_observed_date"] = groups["date"].shift(1)
    original["expected_previous_date"] = original["date"] - pd.Timedelta(days=1)
    original["row_lag1"] = groups["sales"].shift(1)
    original["row_past3"] = groups["sales"].transform(
        lambda s: s.shift(1).rolling(3, min_periods=3).mean()
    )
    original = original.rename(columns={"lag1": "calendar_lag1", "past3": "calendar_past3"})
    original["no_previous_observation"] = original["previous_observed_date"].isna()
    original["date_mismatch"] = (
        ~original["no_previous_observation"]
        & original["previous_observed_date"].ne(original["expected_previous_date"])
    )

    # 参照日の分類は監査用。未来に分からない欠測有無を特徴量にしない。
    lookup = {(r.store, r.date): r.sales for r in original.itertuples()}
    missing, unknown, outside = [], [], []
    for row in original.itertuples():
        start, end = map(pd.Timestamp, active_periods[row.store])
        row_missing, row_unknown, row_outside = [], [], []
        for offset in (3, 2, 1):
            date = row.date - pd.Timedelta(days=offset)
            label = date.strftime("%Y-%m-%d")
            if not start <= date <= end:
                row_outside.append(label)
            elif (row.store, date) not in lookup:
                row_missing.append(label)
            elif pd.isna(lookup[(row.store, date)]):
                row_unknown.append(label)
        missing.append(row_missing)
        unknown.append(row_unknown)
        outside.append(row_outside)
    original["missing_dates"] = missing
    original["unknown_value_dates"] = unknown
    original["outside_period_dates"] = outside
    # record_presentはすべてTrueの元行に絞った後、レポートから除く。
    report = original.drop(columns="record_present").reset_index(drop=True)
    summaries = []
    for store, frame in report.groupby("store", sort=True):
        summary = {"store": store, "original_rows": len(frame),
                   "date_mismatch_rows": int(frame["date_mismatch"].sum()),
                   "no_previous_observation_rows": int(frame["no_previous_observation"].sum())}
        for name in ("calendar_lag1", "calendar_past3"):
            summary[f"{name}_nan_rows"] = int(frame[name].isna().sum())
            summary[f"{name}_nan_rate"] = float(frame[name].isna().mean())
        summaries.append(summary)
    return report, pd.DataFrame(summaries)