def _validate_value(key: str, value) -> tuple: d = SETTING_DEFS.get(key) if not d: return False, f"Unknown setting key: {key}" t = d["type"] try: if t == "float": v = float(value) if "min" in d and v < d["min"]: return False, f"Value below minimum {d['min']}" if "max" in d and v > d["max"]: return False, f"Value above maximum {d['max']}" return True, v if t == "int": v = int(value) if "min" in d and v < d["min"]: return False, f"Value below minimum {d['min']}" if "max" in d and v > d["max"]: return False, f"Value above maximum {d['max']}" return True, v if t == "bool": if isinstance(value, str): v = value.strip().lower() in ("1", "true", "yes", "on") else: v = bool(value) return True, v if t == "str": v = str(value).strip() if "max_len" in d and len(v) > d["max_len"]: return False, f"Max length {d['max_len']}" if key == "attachment_kind" and v not in ("invoice", "order", "receipt", "document"): return False, "attachment_kind must be invoice|order|receipt|document" if key == "invoice_format" and ("{" not in v and v): return False, "Format must contain tokens like {prefix}{R#}{seq4}" return True, v except (TypeError, ValueError): return False, f"Value must be {t}" return False, "Invalid value" @router.get("/list") ===EXPORT=== async def export_logs(campaign_id: int = Query(0), outcome: str = Query(""), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """CSV export of send attempts for a campaign (all pages).""" q = (select(SendAttempt) .join(Campaign, Campaign.id == SendAttempt.campaign_id) .where(Campaign.user_id == user.id)) if campaign_id: q = q.where(SendAttempt.campaign_id == campaign_id) if outcome: q = q.where(SendAttempt.outcome == outcome) rows = (await db.execute(q.order_by(SendAttempt.id.asc()))).scalars().all() def generate(): buf = io.StringIO() w = csv.writer(buf) w.writerow(["attempt_id", "campaign_id", "email", "attempt_number", "outcome", "error_class", "error_message", "message_id", "method", "provider_response", "timestamp"]) yield buf.getvalue() buf.seek(0); buf.truncate(0) for a in rows: w.writerow([a.id, a.campaign_id, a.recipient_email, (a.retry_no or 0) + 1, a.outcome, a.error_class, (a.failure_reason or "")[:500], a.message_id, a.method, (a.provider_response or "")[:500], a.created_at.isoformat() if a.created_at else ""]) yield buf.getvalue() buf.seek(0); buf.truncate(0) return StreamingResponse(generate(), media_type="text/csv", headers={"Content-Disposition": f"attachment; filename=send_logs{('_' + str(campaign_id)) if campaign_id else 's'}.csv"})