"""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__)

# Postfix Queue ID formats:
#   - Legacy (< 3.5):  hex uppercase, e.g. "1A2B3C4D5E6F"
#   - Long (>= 3.5):   base57 alphanumeric (A-Z a-z 0-9 except I O l),
#                      enabled by default with compatibility_level >= 3.6.
# Use [A-Za-z0-9]+ to match both formats.
QID_FROM_RE = re.compile(
    r"(?P<qid>[A-Za-z0-9]+):\s+from=<(?P<from>[^>]*)>,\s+size=(?P<size>\d+)",
)
QID_TO_RE = re.compile(
    r"(?P<qid>[A-Za-z0-9]+):\s+to=<(?P<to>[^>]*)>.*?status=(?P<status>\w+)",
)
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 _unwrap_srs(address: str) -> str:
    addr = address.lower()
    if addr.startswith("srs0=") or addr.startswith("srs1="):
        parts = addr.split("=")
        if len(parts) >= 5:
            user_host = parts[-1].split("@")
            user = user_host[0]
            domain = parts[3]
            return f"{user}@{domain}"
    return address


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@-",
        "-t", "postfix",
        "-t", "postfix/qmgr",
        "-t", "postfix/lmtp",
        "-t", "postfix/smtp",
        "-t", "postfix/smtpd",
        "-t", "postfix/cleanup",
        "-t", "postfix/master",
        "--no-pager",
        "-o", "short-iso-precise",
        "-n", str(max_lines),
    ]
    if since:
        cmd.extend(["--since", since if "UTC" in since or "ago" in since else f"{since} UTC"])
    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 UTC")])
    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()
        # Fallback if unit/tag filter returned no output
        if not lines:
            fallback_cmd = ["journalctl", "SYSLOG_FACILITY=9", "--no-pager", "-o", "short-iso-precise", "-n", str(max_lines)]
            if since:
                fallback_cmd.extend(["--since", since if "UTC" in since or "ago" in since else f"{since} UTC"])
            elif state.last_run_at:
                fallback_cmd.extend(["--since", (state.last_run_at - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S UTC")])
            else:
                fallback_cmd.extend(["--since", "7 days ago"])
            f_result = subprocess.run(fallback_cmd, capture_output=True, text=True, timeout=120, check=False)
            lines = (f_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"))
            raw_sender = (from_match.group("from") or "").strip().lower()
            sender = _unwrap_srs(raw_sender)
            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()
        # "sent": relay to remote MTA; "delivered": local LMTP delivery.
        # "queued"/"forwarded": local virtual delivery or aliases.
        if status not in {"sent", "delivered", "queued", "forwarded"}:
            continue
        qid = to_match.group("qid").upper()
        rcpt = (to_match.group("to") or "").strip().lower()
        size = sizes.get(qid, 0)
        sender = senders.get(qid, "")

        # Normalize subaddresses (e.g. user+tag@domain.com -> user@domain.com)
        clean_rcpt = rcpt.split("+")[0] + "@" + rcpt.split("@")[-1] if ("+" in rcpt and "@" in rcpt) else rcpt
        clean_sender = sender.split("+")[0] + "@" + sender.split("@")[-1] if ("+" in sender and "@" in sender) else sender

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

        # Outbound: local mailbox or local domain is the envelope sender.
        if sender in mailboxes or clean_sender in mailboxes or (sender and "@" in sender and sender.split("@")[-1] in domains):
            target_box = clean_sender if clean_sender in mailboxes else sender
            key = _event_key("outbound", target_box, qid, size)
            events.append(("outbound", target_box, 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)