# tools/feed_tools.py — RSS/Atom feed monitoring.
# Stores feed list and seen-entry state under workspace/memory/ using atomic writes.
# Uses explicit SSL context and one retry for transient failures; fallback text when fetch fails.
from __future__ import annotations

import fcntl
import hashlib
import json
import os
import re
import ssl
import tempfile
import time
import uuid
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from urllib.parse import urlparse
from urllib.request import Request, urlopen
from urllib.error import URLError

from langchain_core.tools import tool

from tools._logging import _debug_log, _emit_input_status
from tools.job_stream_schema import (
    close_job_stream,
    error_job_stream,
    make_capabilities,
    metric_def,
    open_job_stream,
    update_job_stream,
)
from tools.plugin_stream_controls import clear_stream_state, get_action_count, get_control_value, set_control_value
from utils.llm_token_normalize import normalize_hex_id_token

_SSL_CTX = ssl.create_default_context()
try:
    import certifi
    _SSL_CTX.load_verify_locations(certifi.where())
except ImportError:
    pass
_FEED_FETCH_TIMEOUT = (5.0, 20.0)
_FEED_REQUEST_GAP = 2.0  # seconds between successive feed requests
_FEED_RETRY_DELAY = 2.5  # seconds before retry after first failure
_MAX_SEEN_GUIDS = 200     # cap stored GUIDs per feed to prevent unbounded growth
def _fallback_fetch_msg(url: str) -> str:
    return f" You can try read_webpage({url!r}) to see if the URL is reachable."
_FALLBACK_ADD_FEED = " If this is a normal webpage, try read_webpage(url) to confirm it."

_WHITESPACE_RE = re.compile(r"\s+")

# Namespaces for Atom
_ATOM_NS = "http://www.w3.org/2005/Atom"


def _get_workspace() -> str:
    from core import get_workspace_path
    return get_workspace_path()


def _feeds_path() -> tuple[str | None, str]:
    from utils.path_safety import safe_resolve_path
    return safe_resolve_path(_get_workspace(), "memory/feeds.json", allow_write=True, create_parents=True)


def _feed_state_path() -> tuple[str | None, str]:
    from utils.path_safety import safe_resolve_path
    return safe_resolve_path(_get_workspace(), "memory/feed_state.json", allow_write=True, create_parents=True)


def _normalize_feed_id(feed_id: str) -> str:
    return normalize_hex_id_token((feed_id or "").strip(), allowed_lengths=(8,))


def _normalize_state_keys(state: dict) -> tuple[dict, bool]:
    changed = False
    normalized: dict = {}
    for key, value in (state or {}).items():
        nk = _normalize_feed_id(str(key))
        if nk != str(key):
            changed = True
        normalized[nk] = value
    return normalized, changed


def _read_json(path: str, default):
    try:
        if not os.path.isfile(path):
            return default
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def _atomic_write_json(path: str, data) -> None:
    parent = os.path.dirname(path)
    os.makedirs(parent, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=parent, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            json.dump(data, f, indent=2, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
            fcntl.flock(f, fcntl.LOCK_UN)
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _fetch_feed(url: str) -> tuple[bool, bytes, str]:
    """Fetch URL; return (success, raw_bytes, error_message). Uses SSL context and one retry."""
    def _do_fetch() -> tuple[bool, bytes, str]:
        try:
            req = Request(url, headers={"User-Agent": "Mugi-Feed-Reader/1.0"})
            with urlopen(req, timeout=_FEED_FETCH_TIMEOUT[1], context=_SSL_CTX) as resp:
                return True, resp.read(), ""
        except URLError as e:
            return False, b"", str(e)
        except Exception as e:
            return False, b"", str(e)

    ok, raw, err = _do_fetch()
    if ok:
        return ok, raw, err
    time.sleep(_FEED_RETRY_DELAY)
    return _do_fetch()


def _looks_like_feed(raw: bytes) -> bool:
    """Quick check that the content looks like RSS or Atom XML."""
    head = raw[:512].lstrip()
    return b"<rss" in head or b"<feed" in head or b"<rdf:RDF" in head


def _parse_feed_entries(raw: bytes) -> list[dict]:
    """Parse RSS or Atom XML and return list of {guid, title, link, published}."""
    try:
        root = ET.fromstring(raw)
    except ET.ParseError:
        return []

    entries: list[dict] = []
    tag = root.tag

    # RSS 2.0 / RSS 1.0 (RDF)
    if "rss" in tag or "RDF" in tag:
        items = root.findall(".//item")
        for item in items:
            guid_el = item.find("guid")
            link_el = item.find("link")
            title_el = item.find("title")
            pub_el = item.find("pubDate") or item.find("dc:date", {"dc": "http://purl.org/dc/elements/1.1/"})
            guid = (guid_el.text if guid_el is not None else None) or (link_el.text if link_el is not None else None) or ""
            entries.append({
                "guid": guid.strip(),
                "title": (title_el.text or "").strip() if title_el is not None else "",
                "link": (link_el.text or "").strip() if link_el is not None else "",
                "published": (pub_el.text or "").strip() if pub_el is not None else "",
            })

    # Atom
    elif tag == f"{{{_ATOM_NS}}}feed" or tag == "feed":
        ns = {"a": _ATOM_NS} if "{" in tag else {}
        prefix = f"{{{_ATOM_NS}}}" if "{" in tag else ""
        for entry in root.findall(f"{prefix}entry"):
            id_el = entry.find(f"{prefix}id")
            title_el = entry.find(f"{prefix}title")
            link_el = entry.find(f"{prefix}link")
            pub_el = entry.find(f"{prefix}published") or entry.find(f"{prefix}updated")
            guid = (id_el.text or "").strip() if id_el is not None else ""
            link = ""
            if link_el is not None:
                link = link_el.get("href", "") or (link_el.text or "")
            entries.append({
                "guid": guid,
                "title": (title_el.text or "").strip() if title_el is not None else "",
                "link": link.strip(),
                "published": (pub_el.text or "").strip() if pub_el is not None else "",
            })

    return entries


@tool
def add_feed(url: str, label: str = "") -> str:
    """Add an RSS or Atom feed URL to the feed list for periodic new-item monitoring.

    Args:
        url: The HTTP/HTTPS URL of an RSS or Atom feed.
        label: Optional human-readable label (e.g. "Hacker News").

    Validates that the URL returns a parseable feed, then stores it.
    Returns a confirmation with the assigned feed ID, or an error.
    """
    url = re.sub(r"\s+", "", url or "")
    _emit_input_status(f"add_feed: {url[:80]}")
    try:
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https") or not parsed.netloc:
            return f"ERROR: URL must use http or https with a hostname. Got: {url!r}"
    except Exception as e:
        return f"ERROR: Invalid URL: {e}"

    feeds_path, err = _feeds_path()
    if err:
        return f"ERROR: Cannot resolve feeds path: {err}"
    state_path, err2 = _feed_state_path()
    if err2:
        return f"ERROR: Cannot resolve feed state path: {err2}"

    feeds: list[dict] = _read_json(feeds_path, [])
    if not isinstance(feeds, list):
        feeds = []

    # Deduplicate
    for entry in feeds:
        if entry.get("url") == url.strip():
            return f"Already subscribed: {url}\n  ID: {entry['id']}\n  Label: {entry.get('label') or '(none)'}"

    # Quick validation fetch
    ok, raw, fetch_err = _fetch_feed(url.strip())
    if not ok:
        return f"ERROR: Could not fetch feed: {fetch_err}{_FALLBACK_ADD_FEED}"
    if not _looks_like_feed(raw):
        return f"ERROR: URL does not appear to be an RSS or Atom feed. Check the URL and try again."

    # Parse to get initial GUIDs (so first check_feeds only reports items *after* subscription)
    initial_entries = _parse_feed_entries(raw)
    initial_guids = [e["guid"] for e in initial_entries if e.get("guid")][:_MAX_SEEN_GUIDS]

    feed_id = str(uuid.uuid4())[:8]
    entry = {
        "id": feed_id,
        "url": url.strip(),
        "label": (label or "").strip(),
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    feeds.append(entry)
    _atomic_write_json(feeds_path, feeds)

    state: dict = _read_json(state_path, {})
    if not isinstance(state, dict):
        state = {}
    state[feed_id] = {"last_seen_guids": initial_guids, "last_checked_ts": datetime.now(timezone.utc).isoformat()}
    _atomic_write_json(state_path, state)

    _debug_log(f"[feeds] Added: {feed_id} → {url}")
    return (
        f"Feed added.\n"
        f"  ID: {feed_id}\n"
        f"  URL: {url}\n"
        f"  Label: {entry['label'] or '(none)'}\n"
        f"  Initial entries: {len(initial_entries)} (baseline set, only *new* items will be reported)"
    )


@tool
def list_feeds() -> str:
    """List all RSS/Atom feeds currently subscribed for monitoring.

    Returns a formatted list with IDs, labels, and last-checked timestamps,
    or a message if no feeds are subscribed.
    """
    _emit_input_status("list_feeds")
    feeds_path, err = _feeds_path()
    if err:
        return f"ERROR: {err}"

    feeds: list[dict] = _read_json(feeds_path, [])
    if not isinstance(feeds, list) or not feeds:
        return "No feeds subscribed. Use add_feed(url) to start monitoring an RSS or Atom feed."

    state_path, _ = _feed_state_path()
    state: dict = _read_json(state_path, {}) if state_path else {}
    if not isinstance(state, dict):
        state = {}

    lines = [f"Feed subscriptions ({len(feeds)}):"]
    for f in feeds:
        fid = f.get("id", "?")
        url = f.get("url", "?")
        label = f.get("label") or ""
        added = f.get("created_at", "?")
        last = (state.get(fid) or {}).get("last_checked_ts") or "never"
        line = f"  [{fid}] {url}"
        if label:
            line += f" — {label}"
        line += f"\n       Added: {added} | Last checked: {last}"
        lines.append(line)
    return "\n".join(lines)


@tool
def remove_feed(feed_id: str) -> str:
    """Remove an RSS/Atom feed from the subscription list by its ID.

    Args:
        feed_id: The feed ID (from list_feeds).

    Returns a confirmation or error if not found.
    """
    feed_id = _normalize_feed_id(feed_id)
    _emit_input_status(f"remove_feed: {feed_id}")
    feeds_path, err = _feeds_path()
    if err:
        return f"ERROR: {err}"
    state_path, err2 = _feed_state_path()
    if err2:
        return f"ERROR: {err2}"

    feeds: list[dict] = _read_json(feeds_path, [])
    if not isinstance(feeds, list):
        feeds = []

    removed = None
    new_list = []
    for f in feeds:
        if _normalize_feed_id(str(f.get("id", ""))) == feed_id:
            removed = f
        else:
            new_list.append(f)

    if removed is None:
        ids = [f.get("id", "?") for f in feeds]
        return f"ID not found: {feed_id!r}. Current IDs: {ids}"

    _atomic_write_json(feeds_path, new_list)

    state: dict = _read_json(state_path, {})
    if isinstance(state, dict):
        state, _ = _normalize_state_keys(state)
    if isinstance(state, dict) and feed_id in state:
        del state[feed_id]
        _atomic_write_json(state_path, state)

    url = removed.get("url", "?")
    label = removed.get("label") or ""
    return f"Removed feed: [{feed_id}] {url}" + (f" ({label})" if label else "")


@tool
def check_feeds(max_new_per_feed: int = 5) -> str:
    """Check all subscribed RSS/Atom feeds for new items since the last check.

    Fetches each feed, compares entries to the stored baseline, and returns
    a summary of new items. Updates the baseline. Call this during the feeds
    heartbeat or when the user asks 'any new posts?' or 'check my feeds'.

    Args:
        max_new_per_feed: Maximum new items to report per feed (default 5).
    """
    _emit_input_status("Checking feeds…")
    stream_id = f"collector_feed:{int(time.time() * 1000)}"
    open_job_stream(
        plugin_id="collector_feed",
        plugin_name="Feed Collector",
        title="Check Feeds",
        job_type="check_feeds",
        stream_id=stream_id,
        controls=[
            {"id": "max_new_per_feed", "type": "slider", "label": "Max New / Feed", "min": 1, "max": 20, "step": 1, "default": int(max_new_per_feed)},
            {"id": "cancel", "type": "button", "label": "Cancel", "action_id": "cancel_job", "style": "destructive"},
        ],
        capabilities=make_capabilities(cancel=True, progress=True, metrics=True, phases=True, batch_items=True),
        metric_schema=[
            metric_def("feeds_total", "Feeds Total", format="number"),
            metric_def("feeds_checked", "Feeds Checked", format="number"),
            metric_def("new_items_total", "New Items", format="number"),
            metric_def("errors_total", "Errors", format="number"),
        ],
    )
    set_control_value(stream_id, "max_new_per_feed", int(max_new_per_feed))
    try:
        feeds_path, err = _feeds_path()
        if err:
            error_job_stream(plugin_id="collector_feed", plugin_name="Feed Collector", stream_id=stream_id, phase="setup", message=f"ERROR: {err}")
            return f"ERROR: {err}"
        state_path, err2 = _feed_state_path()
        if err2:
            error_job_stream(plugin_id="collector_feed", plugin_name="Feed Collector", stream_id=stream_id, phase="setup", message=f"ERROR: {err2}")
            return f"ERROR: {err2}"

        feeds: list[dict] = _read_json(feeds_path, [])
        if not isinstance(feeds, list) or not feeds:
            close_job_stream(plugin_id="collector_feed", plugin_name="Feed Collector", stream_id=stream_id, status="completed", message="No feeds subscribed.")
            return "No feeds subscribed. Use add_feed(url, label) to subscribe."

        state: dict = _read_json(state_path, {})
        if not isinstance(state, dict):
            state = {}
        else:
            state, migrated = _normalize_state_keys(state)
            if migrated:
                _atomic_write_json(state_path, state)

        results: list[str] = []
        errors_total = 0
        new_items_total = 0
        feeds_total = len(feeds)
        for i, feed in enumerate(feeds):
            if get_action_count(stream_id, "cancel_job") > 0:
                close_job_stream(
                    plugin_id="collector_feed",
                    plugin_name="Feed Collector",
                    stream_id=stream_id,
                    status="stopped",
                    message="Feed check cancelled by user.",
                    metrics={
                        "feeds_total": float(feeds_total),
                        "feeds_checked": float(i),
                        "new_items_total": float(new_items_total),
                        "errors_total": float(errors_total),
                    },
                )
                return "Feed check cancelled by user."
            if i > 0:
                time.sleep(_FEED_REQUEST_GAP)
            live_max = get_control_value(stream_id, "max_new_per_feed", max_new_per_feed)
            try:
                max_new_per_feed = max(1, min(20, int(float(live_max))))
            except Exception:
                pass
            fid = feed.get("id", "")
            url = feed.get("url", "")
            label = feed.get("label") or url
            update_job_stream(
                plugin_id="collector_feed",
                plugin_name="Feed Collector",
                stream_id=stream_id,
                phase="check_feed",
                status="running",
                progress_pct=float((i / max(feeds_total, 1)) * 100.0),
                message=f"Checking {label}",
                metrics={
                    "feeds_total": float(feeds_total),
                    "feeds_checked": float(i),
                    "new_items_total": float(new_items_total),
                    "errors_total": float(errors_total),
                },
                extra={"item_label": label},
            )
            if not url:
                continue
            ok, raw, fetch_err = _fetch_feed(url)
            if not ok:
                msg = f"[{label}] Fetch failed: {fetch_err}."
                if url:
                    msg += _fallback_fetch_msg(url)
                results.append(msg)
                errors_total += 1
                continue
            entries = _parse_feed_entries(raw)
            if not entries:
                results.append(f"[{label}] Could not parse feed entries.")
                errors_total += 1
                continue
            feed_state = state.get(fid, {})
            if not isinstance(feed_state, dict):
                feed_state = {}
            seen_guids: set[str] = set(feed_state.get("last_seen_guids", []))
            new_entries = [e for e in entries if e.get("guid") and e["guid"] not in seen_guids]
            new_entries = new_entries[:max_new_per_feed]
            all_current_guids = [e["guid"] for e in entries if e.get("guid")]
            updated_seen = list(dict.fromkeys(all_current_guids + list(seen_guids)))[:_MAX_SEEN_GUIDS]
            state[fid] = {
                "last_seen_guids": updated_seen,
                "last_checked_ts": datetime.now(timezone.utc).isoformat(),
            }
            if new_entries:
                lines = [f"New from [{label}] ({len(new_entries)} item{'s' if len(new_entries) != 1 else ''}):"]
                for e in new_entries:
                    title = e.get("title") or "(no title)"
                    link = e.get("link") or ""
                    pub = e.get("published") or ""
                    line = f"  - {title}"
                    if pub:
                        line += f" ({pub})"
                    if link:
                        line += f"\n    {link}"
                    lines.append(line)
                results.append("\n".join(lines))
                new_items_total += len(new_entries)
            _debug_log(f"[feeds] {label}: {len(new_entries)} new items")

        _atomic_write_json(state_path, state)
        close_job_stream(
            plugin_id="collector_feed",
            plugin_name="Feed Collector",
            stream_id=stream_id,
            status="completed",
            message="Feed check complete.",
            metrics={
                "feeds_total": float(feeds_total),
                "feeds_checked": float(feeds_total),
                "new_items_total": float(new_items_total),
                "errors_total": float(errors_total),
            },
        )
        if not results:
            return "No new feed items."
        return "\n\n".join(results)
    except Exception as e:
        error_job_stream(
            plugin_id="collector_feed",
            plugin_name="Feed Collector",
            stream_id=stream_id,
            phase="check_feeds",
            message=f"Feed check failed: {e}",
        )
        return f"Feed check failed: {e}"
    finally:
        clear_stream_state(stream_id)
