"""联网自学配置与状态 — web_learning.yaml + web_learning_state.yaml。"""

from __future__ import annotations

import copy
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4

import yaml

_TOPIC_ID_RE = re.compile(r"^[a-z][a-z0-9_]{1,31}$")


class WebLearningConfigError(Exception):
    pass


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _config_path(config: dict[str, Any], home: Path | None = None) -> Path:
    soul = config.get("soul") or {}
    raw = soul.get("web_learning") or "config/soul/web_learning.yaml"
    path = Path(raw)
    if path.is_file():
        return path
    if home is not None:
        candidate = home / raw
        if candidate.is_file():
            return candidate
    return path


def _default_config() -> dict[str, Any]:
    return {
        "enabled": False,
        "auto_learning": False,
        "schedule": {
            "mode": "daily_window",
            "timezone": "Asia/Shanghai",
            "window_start": "00:00",
            "window_end": "01:00",
            "duration_minutes": 60,
            "minutes_per_batch": 15,
            "max_runs_per_day": 4,
        },
        "topics": [],
        "allowlist": [],
        "search": {"provider": "mock", "min_hits": 2, "max_results": 5},
        "ingest": {
            "trust_tier": "web",
            "reality_anchor": "multi_source",
            "subject_eligible": True,
            "memory_scope": "subject",
            "provenance_type": "external_web",
        },
        "synthesize": {
            "primary_target": "SL4",
            "sl5_touch": "forbidden",
            "values_touch": "forbidden",
            "label": "技能自学",
        },
    }


class WebLearningService:
    """读 yaml 配置 + 持久化主体侧联网自学开关与批次。"""

    def __init__(self, store, config: dict[str, Any], *, home: Path | None = None) -> None:
        self._store = store
        self._config = config
        self._home = home
        self._yaml_path = _config_path(config, home=home)
        self._state_path = store.subject_root / "web_learning_state.yaml"

    def load_yaml_config(self) -> dict[str, Any]:
        if not self._yaml_path.is_file():
            return _default_config()
        with self._yaml_path.open(encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        if not isinstance(data, dict):
            return _default_config()
        return data

    def load_state(self) -> dict[str, Any]:
        if not self._state_path.is_file():
            return self._empty_state()
        with self._state_path.open(encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        if not isinstance(data, dict):
            return self._empty_state()
        data.setdefault("enabled", False)
        data.setdefault("auto_learning", data.get("auto_web_search", False))
        data.setdefault("custom_queries", [])
        data.setdefault("active_topic_ids", [])
        data.setdefault("schedule", {})
        data.setdefault("topics", [])
        data.setdefault("batches", [])
        data.setdefault("runs_today", 0)
        data.setdefault("learning_minutes_today", 0)
        data.setdefault("runs_day", _today_key())
        return data

    def save_state(self, state: dict[str, Any]) -> None:
        self._state_path.parent.mkdir(parents=True, exist_ok=True)
        self._state_path.write_text(
            yaml.safe_dump(state, allow_unicode=True, sort_keys=False),
            encoding="utf-8",
        )

    @staticmethod
    def _empty_state() -> dict[str, Any]:
        return {
            "enabled": False,
            "auto_learning": False,
            "custom_queries": [],
            "active_topic_ids": [],
            "schedule": {},
            "topics": [],
            "batches": [],
            "runs_today": 0,
            "learning_minutes_today": 0,
            "runs_day": _today_key(),
            "last_batch_at": None,
        }

    def effective_schedule(self) -> dict[str, Any]:
        cfg = self.load_yaml_config()
        state = self.load_state()
        base = dict(cfg.get("schedule") or {})
        overlay = state.get("schedule") if isinstance(state.get("schedule"), dict) else {}
        return {**base, **overlay}

    def effective_topics(self) -> list[dict[str, Any]]:
        state = self.load_state()
        stored = state.get("topics")
        if isinstance(stored, list) and stored:
            return [t for t in stored if isinstance(t, dict) and t.get("id")]
        cfg = self.load_yaml_config()
        return [t for t in (cfg.get("topics") or []) if isinstance(t, dict) and t.get("id")]

    def _ensure_topics_materialized(self, state: dict[str, Any]) -> None:
        if state.get("topics"):
            return
        state["topics"] = copy.deepcopy(self.effective_topics())

    def active_topic_ids(self) -> list[str]:
        state = self.load_state()
        ids = [str(x) for x in (state.get("active_topic_ids") or []) if str(x).strip()]
        if ids:
            return ids
        return [str(t.get("id")) for t in self.effective_topics() if t.get("id")]

    @staticmethod
    def _validate_topic_id(topic_id: str) -> str:
        tid = (topic_id or "").strip().lower()
        if not _TOPIC_ID_RE.match(tid):
            raise WebLearningConfigError("topic id 须为小写字母开头，仅含 a-z、0-9、_")
        return tid

    def patch_settings(
        self,
        *,
        enabled: bool | None = None,
        auto_learning: bool | None = None,
        auto_web_search: bool | None = None,
        append_query: dict[str, str] | None = None,
        schedule: dict[str, Any] | None = None,
        active_topic_ids: list[str] | None = None,
    ) -> dict[str, Any]:
        state = self.load_state()
        if enabled is not None:
            state["enabled"] = bool(enabled)
        auto_val = auto_learning if auto_learning is not None else auto_web_search
        if auto_val is not None:
            state["auto_learning"] = bool(auto_val)
            state["auto_web_search"] = bool(auto_val)
        if schedule is not None:
            from niannian.soul.learning_schedule import validate_daily_window

            merged = {**self.effective_schedule(), **schedule}
            try:
                validate_daily_window(
                    str(merged.get("window_start") or "00:00"),
                    str(merged.get("window_end") or "01:00"),
                )
            except ValueError as exc:
                raise WebLearningConfigError(str(exc)) from exc
            dur = int(merged.get("duration_minutes") or 60)
            if dur < 15 or dur > 240:
                raise WebLearningConfigError("duration_minutes 须在 15–240")
            state["schedule"] = {
                k: merged[k]
                for k in ("window_start", "window_end", "duration_minutes", "timezone", "minutes_per_batch", "max_runs_per_day")
                if k in merged and merged[k] is not None
            }
        if active_topic_ids is not None:
            valid = {str(t.get("id")) for t in self.effective_topics()}
            picked = [str(x).strip() for x in active_topic_ids if str(x).strip() in valid]
            if active_topic_ids and not picked:
                raise WebLearningConfigError("active_topic_ids 无有效领域")
            state["active_topic_ids"] = picked
        if append_query:
            queries = list(state.get("custom_queries") or [])
            queries.append(
                {
                    "id": f"q_{uuid4().hex[:6]}",
                    "topic_id": str(append_query.get("topic_id") or "owner_interest"),
                    "query": str(append_query.get("query") or append_query.get("label") or ""),
                    "added_at": _now_iso(),
                }
            )
            state["custom_queries"] = queries[-30:]
        self.save_state(state)
        return self.to_api_dict()

    def upsert_topic(
        self,
        *,
        topic_id: str,
        label: str,
        seed_queries: list[str] | None = None,
    ) -> dict[str, Any]:
        tid = self._validate_topic_id(topic_id)
        name = (label or "").strip() or tid
        seeds = [s.strip() for s in (seed_queries or []) if isinstance(s, str) and s.strip()]
        state = self.load_state()
        self._ensure_topics_materialized(state)
        topics = list(state.get("topics") or [])
        entry = {"id": tid, "label": name, "seed_queries": seeds}
        replaced = False
        for i, t in enumerate(topics):
            if str(t.get("id")) == tid:
                topics[i] = {**t, **entry}
                replaced = True
                break
        if not replaced:
            topics.append(entry)
        state["topics"] = topics
        active = list(state.get("active_topic_ids") or [])
        if tid not in active and not active:
            state["active_topic_ids"] = [str(x.get("id")) for x in topics if x.get("id")]
        self.save_state(state)
        return self.to_api_dict()

    def delete_topic(self, topic_id: str) -> dict[str, Any]:
        tid = self._validate_topic_id(topic_id)
        state = self.load_state()
        self._ensure_topics_materialized(state)
        before = list(state.get("topics") or [])
        after = [t for t in before if str(t.get("id")) != tid]
        if len(after) == len(before):
            raise WebLearningConfigError(f"领域不存在: {tid}")
        state["topics"] = after
        state["active_topic_ids"] = [x for x in (state.get("active_topic_ids") or []) if x != tid]
        self.save_state(state)
        return self.to_api_dict()

    def can_run_batch(self, *, manual: bool = False) -> tuple[bool, str]:
        from niannian.soul.cultural_learning import CulturalLearningService
        from niannian.soul.learning_schedule import minutes_budget_left, schedule_gate

        state = self.load_state()
        if not state.get("enabled"):
            return False, "disabled"
        self._reset_runs_if_new_day(state)

        cultural = CulturalLearningService(self._store, self._config, home=self._home)
        sched = self.effective_schedule()
        ok_sched, reason = schedule_gate(
            sched,
            manual=manual,
            bootstrap_complete=cultural.is_bootstrap_complete(),
        )
        if not ok_sched:
            return False, reason

        max_runs = int(sched.get("max_runs_per_day", 4))
        if int(state.get("runs_today") or 0) >= max_runs:
            return False, "daily_limit"

        if not manual:
            budget = minutes_budget_left(
                sched,
                learning_minutes_today=int(state.get("learning_minutes_today") or 0),
            )
            per_batch = int(sched.get("minutes_per_batch") or 15)
            if budget < per_batch:
                return False, "duration_limit"
        return True, "ok"

    def record_batch(self, batch: dict[str, Any]) -> None:
        sched = self.effective_schedule()
        per_batch = int(sched.get("minutes_per_batch") or 15)
        state = self.load_state()
        self._reset_runs_if_new_day(state)
        batches = list(state.get("batches") or [])
        batches.append(batch)
        state["batches"] = batches[-30:]
        state["runs_today"] = int(state.get("runs_today") or 0) + 1
        state["learning_minutes_today"] = int(state.get("learning_minutes_today") or 0) + per_batch
        state["last_batch_at"] = batch.get("at") or _now_iso()
        self.save_state(state)

    def allowlist_patterns(self) -> list[str]:
        cfg = self.load_yaml_config()
        patterns = list(cfg.get("allowlist") or [])
        return [str(p) for p in patterns if str(p).strip()]

    def to_api_dict(self) -> dict[str, Any]:
        cfg = self.load_yaml_config()
        state = self.load_state()
        topics = self.effective_topics()
        active = self.active_topic_ids()
        batches = list(state.get("batches") or [])
        recent = list(reversed(batches[-10:]))
        search_cfg = cfg.get("search") or {}
        syntheses = [
            {
                "topic": b.get("topic"),
                "title": b.get("title"),
                "atom_ids": b.get("atom_ids") or [],
                "at": b.get("at"),
                "status": b.get("status"),
            }
            for b in recent
            if isinstance(b, dict)
        ]
        sched = self.effective_schedule()
        return {
            "enabled": bool(state.get("enabled")),
            "auto_learning": bool(state.get("auto_learning", state.get("auto_web_search"))),
            "auto_web_search": bool(state.get("auto_learning", state.get("auto_web_search"))),
            "schedule": sched,
            "schedule_label": self._schedule_label(sched),
            "personality_bootstrap_complete": self._bootstrap_complete(),
            "learning_minutes_today": int(state.get("learning_minutes_today") or 0),
            "duration_minutes": int(sched.get("duration_minutes") or 60),
            "active_topic_ids": active,
            "topics": [
                {
                    "id": t.get("id"),
                    "label": t.get("label"),
                    "seed_queries": list(t.get("seed_queries") or []),
                }
                for t in topics
            ],
            "search_provider": search_cfg.get("provider", "mock"),
            "allowlist_count": len(self.allowlist_patterns()),
            "last_batch_at": state.get("last_batch_at"),
            "recent_batches": recent,
            "recent_syntheses": syntheses,
            "runs_today": int(state.get("runs_today") or 0),
        }

    def _bootstrap_complete(self) -> bool:
        from niannian.soul.cultural_learning import CulturalLearningService

        return CulturalLearningService(self._store, self._config, home=self._home).is_bootstrap_complete()

    @staticmethod
    def _schedule_label(sched: dict[str, Any]) -> str:
        start = sched.get("window_start") or "00:00"
        end = sched.get("window_end") or "01:00"
        dur = sched.get("duration_minutes") or 60
        tz = sched.get("timezone") or "Asia/Shanghai"
        return f"每日 {start}–{end}（{tz}）· {dur} 分钟"

    def pick_search_task(
        self,
        *,
        topic_id: str | None = None,
        query: str | None = None,
    ) -> dict[str, Any]:
        if query and query.strip():
            return {
                "topic_id": topic_id or "adhoc",
                "topic_label": "即时查询",
                "query": query.strip(),
            }

        cfg = self.load_yaml_config()
        state = self.load_state()
        topics = self.effective_topics()
        active_ids = set(self.active_topic_ids())
        if topic_id:
            topics = [t for t in topics if t.get("id") == topic_id]
            if not topics:
                raise ValueError(f"unknown topic: {topic_id}")
        else:
            topics = [t for t in topics if str(t.get("id")) in active_ids]
            if not topics:
                raise ValueError("no active topics configured")

        batch_count = len(state.get("batches") or [])
        topic = topics[batch_count % len(topics)]
        tid = str(topic.get("id") or "general")

        custom = [
            q
            for q in (state.get("custom_queries") or [])
            if isinstance(q, dict) and str(q.get("query") or "").strip()
        ]
        topic_custom = [q for q in custom if q.get("topic_id") == tid]
        pool_custom = topic_custom or custom

        seeds = [s for s in (topic.get("seed_queries") or []) if isinstance(s, str) and s.strip()]
        if pool_custom:
            picked = pool_custom[batch_count % len(pool_custom)]
            return {
                "topic_id": tid,
                "topic_label": topic.get("label") or tid,
                "query": str(picked.get("query")),
            }
        if not seeds:
            raise ValueError("no seed queries for topic")
        return {
            "topic_id": tid,
            "topic_label": topic.get("label") or tid,
            "query": seeds[batch_count % len(seeds)],
        }

    @staticmethod
    def _reset_runs_if_new_day(state: dict[str, Any]) -> None:
        today = _today_key()
        if state.get("runs_day") != today:
            state["runs_day"] = today
            state["runs_today"] = 0
            state["learning_minutes_today"] = 0


def _today_key() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")
