def _recipient_dict(r) -> dict:
    return {"Email": r.email, "FirstName": r.first_name or "", "LastName": r.last_name or "",
            "CompanyName": r.company_name or "", "CompanyPhone": r.company_phone or "",
            "CustomerEmail": r.customer_email or r.email}


def _plain_fallback(html: str) -> str:
    import html as _html
    text = _re.sub(r"<style.*?</style>", "", html, flags=_re.S | _re.I)
    text = _re.sub(r"<[^>]+>", " ", text)
    return _html.unescape(_re.sub(r"\s+", " ", text)).strip()[:20000]


def _unsubscribe_url(recipient, campaign) -> str:
    base = settings.UNSUBSCRIBE_BASE or settings.BASE_URL
    if recipient and recipient.unsub_token:
        return f"{base}/unsubscribe?token={recipient.unsub_token}"
    return f"{base}/unsubscribe"


def _sanitize_response(resp) -> str:
    s = str(resp)
    for w in ("refresh_token", "access_token", "client_secret"):
        s = s.replace(w, "[REDACTED]")
    return s[:1000]


def _rate_window(key: str, interval: float):
    """Per-key fixed-interval limiter in Redis (SET NX + compare)."""
    if interval <= 0 or rdb is None:
        return True, 0
    try:
        now_ts = time.time()
        k = f"rlw:{key}"
        pipe = rdb.pipeline()
        pipe.set(k, now_ts, nx=True, ex=max(60, int(interval * 4)))
        res = pipe.execute()
        if res[0]:  # newly set -> we own this window
            return True, 0
        cur = float(rdb.get(k) or 0)
        elapsed = now_ts - cur
        if elapsed >= interval:
            rdb.set(k, now_ts, ex=max(60, int(interval * 4)))
            return True, 0
        return False, max(0.5, interval - elapsed)
    except Exception:
        return True, 0


def _reschedule(db, task, campaign_id, cr, job, delay: timedelta, reason: str):
    cr.status = "queued"
    cr.last_error = reason
    if job:
        job.status = "retrying"
        job.retry_after = now() + delay
    db.commit()
    task.apply_async(args=[campaign_id, cr.id],
                     countdown=max(1, int(delay.total_seconds()) + 1))


def _handle_send_error(db, task, campaign, cr, job, account, attempt, error: str):
    err_class = classify_error(error)
    max_retries = campaign.max_retries if campaign.max_retries is not None else settings.DEFAULT_RETRIES
    delay = int(campaign.retry_delay_seconds or settings.DEFAULT_RETRY_DELAY_SECONDS)

    cr.last_error = error[:1000]
    account.error_count += 1
    low = (error or "").lower()

    if err_class == "permanent":
        attempt.outcome = "perm_failed"
        attempt.failure_reason = error[:1000]
        attempt.error_class = err_class
        cr.status = "perm_failed"
        campaign.failed_count += 1
        if job:
            job.status = "failed"
        db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                          gmail_account_id=account.id, event_type="failed", detail=error[:500]))
        if any(p in low for p in ("invalid grant", "401", "unauthorized", "forbidden")):
            account.status = "expired"
            campaign.status = "paused"
        elif any(p in low for p in ("quota", "rate limit", "429", "resourceexhausted")):
            account.cooldown_until = now() + timedelta(minutes=30)
        db.commit()
        _publish_progress(campaign.id)
        return {"ok": False, "status": "perm_failed", "reason": error[:200]}

    if cr.attempt_count >= max_retries + 1:
        # Dead-letter: never retry indefinitely
        attempt.outcome = "temp_failed"
        attempt.failure_reason = error[:1000]
        attempt.error_class = err_class
        cr.status = "temp_failed"
        if job:
            job.status = "dead"
        db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                          gmail_account_id=account.id, event_type="failed",
                          detail=f"dead-letter after {cr.attempt_count} attempts: {error[:300]}"))
        campaign.failed_count += 1
        db.commit()
        _push_dlq(campaign.id, cr.id, error)
        _publish_progress(campaign.id)
        return {"ok": False, "status": "dead_letter", "reason": error[:200]}

    # temporary -> exponential backoff with jitter
    attempt.outcome = "temp_failed"
    attempt.failure_reason = error[:1000]
    attempt.error_class = err_class
    backoff = delay * (2 ** max(0, cr.attempt_count - 1)) + random.uniform(0, 5)
    db.commit()
    _reschedule(db, task, campaign.id, cr, job, timedelta(seconds=backoff),
                f"temporary error: {error[:120]}")
    return {"ok": False, "status": "retrying", "reason": error[:200]}


def _push_dlq(campaign_id, cr_id, error):
    if rdb is None:
        return
    try:
        rdb.rpush("dlq:send", f"{campaign_id}:{cr_id}:{error[:200]}")
    except Exception:
        pass


def _publish_control(campaign_id, action, reason=""):
    if rdb is None:
        return
    try:
        rdb.publish(f"campaign:{campaign_id}:control", f"{action}|{reason}")
    except Exception:
        pass


def _publish_progress(campaign_id):
    if rdb is None:
        return
    try:
        rdb.publish(f"campaign:{campaign_id}:progress", "tick")
    except Exception:
        pass