"""Privacy-safe mail traffic statistics.

Stores only: direction (inbound/outbound), local mailbox, size in bytes, timestamp.
Never stores subject, body, remote recipients, or message content.
"""

from __future__ import annotations

import hashlib
import logging
import os
import re
import subprocess
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import func
from sqlalchemy.orm import Session

from . import models

logger = logging.getLogger(__name__)

QID_FROM_RE = re.compile(
    r"(?P<qid>[A-F0-9]+):\s+from=<(?P<from>[^>]*)>,\s+size=(?P<size>\d+)",
    re.IGNORECASE,
)
QID_TO_RE = re.compile(
    r"(?P<qid>[A-F0-9]+):\s+to=<(?P<to>[^>]*)>.*?status=(?P<status>\w+)",
    re.IGNORECASE,
)
JOURNAL_TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2}T[0-9:.+-]+)")


def _utc_now() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)


def _parse_journal_ts(line: str) -> datetime | None:
    match = JOURNAL_TS_RE.match(line.strip())
    if not match:
        return None
    raw = match.group(1)
    try:
        # journalctl -o short-iso-precise
        if raw.endswith("Z"):
            dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
        else:
            dt = datetime.fromisoformat(raw)
        if dt.tzinfo is not None:
            dt = dt.astimezone(UTC).replace(tzinfo=None)
        return dt
    except ValueError:
        return None


def local_mailboxes(db: Session) -> set[str]:
    rows = (
        db.query(models.Account.local_part, models.Domain.name)
        .join(models.Domain, models.Account.domain_id == models.Domain.id)
        .filter(models.Account.is_active.is_(True), models.Domain.is_active.is_(True))
        .all()
    )
    return {f"{local}@{domain}".lower() for local, domain in rows}


def local_domains(db: Session) -> set[str]:
    return {name.lower() for (name,) in db.query(models.Domain.name).filter(models.Domain.is_active.is_(True)).all()}


def _event_key(direction: str, mailbox: str, qid: str, size: int) -> str:
    material = f"{direction}|{mailbox.lower()}|{qid.upper()}|{size}"
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:64]


def _format_size(num: int) -> str:
    value = float(max(0, int(num)))
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if value < 1024 or unit == "TB":
            if unit == "B":
                return f"{int(value)} {unit}"
            return f"{value:.1f} {unit}"
        value /= 1024
    return f"{int(num)} B"


def collect_from_journal(db: Session, *, since: str | None = None, max_lines: int = 20000) -> dict[str, Any]:
    """Parse Postfix journal lines and insert privacy-safe traffic events."""
    mailboxes = local_mailboxes(db)
    domains = local_domains(db)
    if not mailboxes and not domains:
        return {"inserted": 0, "parsed": 0, "detail": "no local mailboxes"}

    state = db.query(models.MailStatsState).filter(models.MailStatsState.id == 1).first()
    if not state:
        state = models.MailStatsState(id=1)
        db.add(state)
        db.flush()

    cmd = [
        "journalctl",
        "-u",
        "postfix",
        "-u",
        "postfix@-",
        "--no-pager",
        "-o",
        "short-iso-precise",
        "-n",
        str(max_lines),
    ]
    if since:
        cmd.extend(["--since", since])
    elif state.last_run_at:
        # Small overlap to avoid missing lines at boundary.
        cmd.extend(["--since", (state.last_run_at - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S")])
    else:
        cmd.extend(["--since", "7 days ago"])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, check=False)
        lines = (result.stdout or "").splitlines()
    except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
        state.last_run_at = _utc_now()
        state.last_status = f"error: {exc}"
        db.commit()
        return {"inserted": 0, "parsed": 0, "error": str(exc)}

    sizes: dict[str, int] = {}
    senders: dict[str, str] = {}
    events: list[tuple[str, str, int, datetime, str]] = []  # direction, mailbox, size, when, key

    for line in lines:
        event_at = _parse_journal_ts(line) or _utc_now()
        from_match = QID_FROM_RE.search(line)
        if from_match:
            qid = from_match.group("qid").upper()
            sizes[qid] = int(from_match.group("size"))
            sender = (from_match.group("from") or "").strip().lower()
            if sender:
                senders[qid] = sender
            continue
        to_match = QID_TO_RE.search(line)
        if not to_match:
            continue
        status = (to_match.group("status") or "").lower()
        if status not in {"sent", "delivered"}:
            continue
        qid = to_match.group("qid").upper()
        rcpt = (to_match.group("to") or "").strip().lower()
        size = sizes.get(qid, 0)
        sender = senders.get(qid, "")

        # Inbound: delivered to a local mailbox.
        if rcpt in mailboxes or (rcpt and rcpt.split("@")[-1] in domains and rcpt in mailboxes):
            if rcpt in mailboxes:
                key = _event_key("inbound", rcpt, qid, size)
                events.append(("inbound", rcpt, size, event_at, key))

        # Outbound: local mailbox is the envelope sender (submission/relay).
        if sender in mailboxes:
            # Count once per queue id + sender (not per recipient) for privacy and size accuracy.
            key = _event_key("outbound", sender, qid, size)
            events.append(("outbound", sender, size, event_at, key))

    # Deduplicate within this batch.
    unique: dict[str, tuple[str, str, int, datetime]] = {}
    for direction, mailbox, size, event_at, key in events:
        unique[key] = (direction, mailbox, size, event_at)

    inserted = 0
    for key, (direction, mailbox, size, event_at) in unique.items():
        exists = db.query(models.MailTrafficEvent.id).filter(models.MailTrafficEvent.event_key == key).first()
        if exists:
            continue
        db.add(
            models.MailTrafficEvent(
                direction=direction,
                mailbox=mailbox[:320],
                size_bytes=max(0, int(size)),
                event_at=event_at,
                event_key=key,
                source="postfix",
            )
        )
        inserted += 1

    # Retention purge
    retention_days = int(os.getenv("LIMRISTEM_MAIL_STATS_RETENTION_DAYS", "400") or 400)
    if retention_days > 0:
        cutoff = _utc_now() - timedelta(days=retention_days)
        db.query(models.MailTrafficEvent).filter(models.MailTrafficEvent.event_at < cutoff).delete(synchronize_session=False)

    state.last_run_at = _utc_now()
    state.last_status = f"ok inserted={inserted} candidates={len(unique)} lines={len(lines)}"
    db.commit()
    return {"inserted": inserted, "candidates": len(unique), "lines": len(lines), "status": state.last_status}


def list_traffic(
    db: Session,
    *,
    direction: str | None = None,
    mailbox: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[dict[str, Any]]:
    query = db.query(models.MailTrafficEvent)
    if direction in {"inbound", "outbound"}:
        query = query.filter(models.MailTrafficEvent.direction == direction)
    if mailbox:
        query = query.filter(models.MailTrafficEvent.mailbox == mailbox.strip().lower())
    rows = (
        query.order_by(models.MailTrafficEvent.event_at.desc(), models.MailTrafficEvent.id.desc())
        .offset(max(0, offset))
        .limit(min(1000, max(1, limit)))
        .all()
    )
    return [
        {
            "id": row.id,
            "direction": row.direction,
            "mailbox": row.mailbox,
            "size_bytes": int(row.size_bytes or 0),
            "size_human": _format_size(int(row.size_bytes or 0)),
            "event_at": row.event_at,
        }
        for row in rows
    ]


def _day_bucket(column):
    # Portable-ish: MySQL DATE(), SQLite date()
    bind = column.table.bind if hasattr(column, "table") else None
    return func.date(column)


def aggregate_by_day(db: Session, *, days: int = 30, mailbox: str | None = None) -> list[dict[str, Any]]:
    days = max(1, min(3660, int(days)))
    start = _utc_now() - timedelta(days=days - 1)
    start = start.replace(hour=0, minute=0, second=0, microsecond=0)
    day_expr = func.date(models.MailTrafficEvent.event_at).label("day")
    query = db.query(
        day_expr,
        models.MailTrafficEvent.direction,
        func.count(models.MailTrafficEvent.id),
        func.coalesce(func.sum(models.MailTrafficEvent.size_bytes), 0),
    ).filter(models.MailTrafficEvent.event_at >= start)
    if mailbox:
        query = query.filter(models.MailTrafficEvent.mailbox == mailbox.strip().lower())
    rows = query.group_by(day_expr, models.MailTrafficEvent.direction).all()

    by_day: dict[str, dict[str, Any]] = {}
    for i in range(days):
        d = (start + timedelta(days=i)).date().isoformat()
        by_day[d] = {
            "day": d,
            "inbound_count": 0,
            "outbound_count": 0,
            "inbound_bytes": 0,
            "outbound_bytes": 0,
        }
    for day, direction, count, total_bytes in rows:
        key = str(day)
        if key not in by_day:
            by_day[key] = {
                "day": key,
                "inbound_count": 0,
                "outbound_count": 0,
                "inbound_bytes": 0,
                "outbound_bytes": 0,
            }
        if direction == "inbound":
            by_day[key]["inbound_count"] = int(count)
            by_day[key]["inbound_bytes"] = int(total_bytes)
        else:
            by_day[key]["outbound_count"] = int(count)
            by_day[key]["outbound_bytes"] = int(total_bytes)
    return [by_day[k] for k in sorted(by_day.keys())]


def aggregate_by_mailbox(db: Session, *, days: int = 30, limit: int = 50) -> list[dict[str, Any]]:
    days = max(1, min(3660, int(days)))
    start = _utc_now() - timedelta(days=days)
    rows = (
        db.query(
            models.MailTrafficEvent.mailbox,
            models.MailTrafficEvent.direction,
            func.count(models.MailTrafficEvent.id),
            func.coalesce(func.sum(models.MailTrafficEvent.size_bytes), 0),
        )
        .filter(models.MailTrafficEvent.event_at >= start)
        .group_by(models.MailTrafficEvent.mailbox, models.MailTrafficEvent.direction)
        .all()
    )
    by_box: dict[str, dict[str, Any]] = {}
    for mailbox, direction, count, total_bytes in rows:
        item = by_box.setdefault(
            mailbox,
            {
                "mailbox": mailbox,
                "inbound_count": 0,
                "outbound_count": 0,
                "inbound_bytes": 0,
                "outbound_bytes": 0,
            },
        )
        if direction == "inbound":
            item["inbound_count"] = int(count)
            item["inbound_bytes"] = int(total_bytes)
        else:
            item["outbound_count"] = int(count)
            item["outbound_bytes"] = int(total_bytes)
    ranked = sorted(
        by_box.values(),
        key=lambda item: item["inbound_count"] + item["outbound_count"],
        reverse=True,
    )
    return ranked[: max(1, min(500, limit))]


def summary_stats(db: Session) -> dict[str, Any]:
    now = _utc_now()

    def window_counts(delta: timedelta) -> dict[str, int]:
        start = now - delta
        rows = (
            db.query(models.MailTrafficEvent.direction, func.count(models.MailTrafficEvent.id))
            .filter(models.MailTrafficEvent.event_at >= start)
            .group_by(models.MailTrafficEvent.direction)
            .all()
        )
        out = {"inbound": 0, "outbound": 0}
        for direction, count in rows:
            out[str(direction)] = int(count)
        return out

    def window_bytes(delta: timedelta) -> dict[str, int]:
        start = now - delta
        rows = (
            db.query(models.MailTrafficEvent.direction, func.coalesce(func.sum(models.MailTrafficEvent.size_bytes), 0))
            .filter(models.MailTrafficEvent.event_at >= start)
            .group_by(models.MailTrafficEvent.direction)
            .all()
        )
        out = {"inbound": 0, "outbound": 0}
        for direction, total in rows:
            out[str(direction)] = int(total)
        return out

    day = window_counts(timedelta(days=1))
    week = window_counts(timedelta(days=7))
    month = window_counts(timedelta(days=30))
    year = window_counts(timedelta(days=365))
    day_b = window_bytes(timedelta(days=1))
    week_b = window_bytes(timedelta(days=7))
    month_b = window_bytes(timedelta(days=30))
    year_b = window_bytes(timedelta(days=365))

    return {
        "today": {"inbound": day["inbound"], "outbound": day["outbound"], "inbound_bytes": day_b["inbound"], "outbound_bytes": day_b["outbound"]},
        "last_7_days": {
            "inbound": week["inbound"],
            "outbound": week["outbound"],
            "inbound_bytes": week_b["inbound"],
            "outbound_bytes": week_b["outbound"],
            "avg_inbound_per_day": round(week["inbound"] / 7, 2),
            "avg_outbound_per_day": round(week["outbound"] / 7, 2),
        },
        "last_30_days": {
            "inbound": month["inbound"],
            "outbound": month["outbound"],
            "inbound_bytes": month_b["inbound"],
            "outbound_bytes": month_b["outbound"],
            "avg_inbound_per_day": round(month["inbound"] / 30, 2),
            "avg_outbound_per_day": round(month["outbound"] / 30, 2),
        },
        "last_365_days": {
            "inbound": year["inbound"],
            "outbound": year["outbound"],
            "inbound_bytes": year_b["inbound"],
            "outbound_bytes": year_b["outbound"],
            "avg_inbound_per_day": round(year["inbound"] / 365, 2),
            "avg_outbound_per_day": round(year["outbound"] / 365, 2),
            "avg_inbound_per_month": round(year["inbound"] / 12, 2),
            "avg_outbound_per_month": round(year["outbound"] / 12, 2),
        },
        "retention_days": int(os.getenv("LIMRISTEM_MAIL_STATS_RETENTION_DAYS", "400") or 400),
        "privacy_note": "Only mailbox, direction, size and timestamp are stored. No subject, body or remote addresses.",
    }


def chart_svg(daily: list[dict[str, Any]], *, width: int = 720, height: int = 220) -> str:
    """Simple dual-series SVG bar chart without external JS (CSP-friendly)."""
    if not daily:
        return '<svg xmlns="http://www.w3.org/2000/svg" width="720" height="80"><text x="12" y="40" fill="#617084">No traffic data yet</text></svg>'
    pad_l, pad_r, pad_t, pad_b = 36, 12, 16, 36
    inner_w = width - pad_l - pad_r
    inner_h = height - pad_t - pad_b
    max_v = max(1, max(int(d["inbound_count"] + d["outbound_count"]) for d in daily))
    n = len(daily)
    slot = inner_w / n
    bar_w = max(2, slot * 0.35)
    parts = [
        f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {width} {height}" role="img" aria-label="Mail traffic chart">',
        f'<rect width="{width}" height="{height}" fill="transparent"/>',
    ]
    # grid
    for i in range(0, 5):
        y = pad_t + inner_h * i / 4
        parts.append(f'<line x1="{pad_l}" y1="{y:.1f}" x2="{width - pad_r}" y2="{y:.1f}" stroke="#d4dae4" stroke-width="1"/>')
        val = int(max_v * (1 - i / 4))
        parts.append(f'<text x="4" y="{y + 4:.1f}" font-size="10" fill="#617084">{val}</text>')
    for idx, day in enumerate(daily):
        x0 = pad_l + idx * slot
        in_h = (int(day["inbound_count"]) / max_v) * inner_h
        out_h = (int(day["outbound_count"]) / max_v) * inner_h
        y_in = pad_t + inner_h - in_h
        y_out = pad_t + inner_h - out_h
        parts.append(
            f'<rect x="{x0 + 2:.1f}" y="{y_in:.1f}" width="{bar_w:.1f}" height="{in_h:.1f}" fill="#3f6fff" rx="2">'
            f'<title>{day["day"]} inbound {day["inbound_count"]}</title></rect>'
        )
        parts.append(
            f'<rect x="{x0 + bar_w + 4:.1f}" y="{y_out:.1f}" width="{bar_w:.1f}" height="{out_h:.1f}" fill="#18794e" rx="2">'
            f'<title>{day["day"]} outbound {day["outbound_count"]}</title></rect>'
        )
        if n <= 16 or idx % max(1, n // 8) == 0:
            label = day["day"][5:]  # MM-DD
            parts.append(
                f'<text x="{x0 + slot / 2:.1f}" y="{height - 10}" font-size="9" text-anchor="middle" fill="#617084">{label}</text>'
            )
    parts.append(
        f'<text x="{pad_l}" y="12" font-size="11" fill="#3f6fff">Inbound</text>'
        f'<text x="{pad_l + 70}" y="12" font-size="11" fill="#18794e">Outbound</text>'
    )
    parts.append("</svg>")
    return "".join(parts)