598 lines
20 KiB
Rust
598 lines
20 KiB
Rust
//! Cliente HTTP de coleta com sessão descartável, aquecimento e retry.
|
|
//!
|
|
//! Uma instância de [`HttpSession`] representa uma única identidade lógica:
|
|
//! cookie jar, User-Agent e idioma permanecem estáveis durante
|
|
//! `aquecer -> acessar`. A fábrica nunca compartilha sessões entre jobs.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use rand::Rng;
|
|
use reqwest::header::{
|
|
ACCEPT, ACCEPT_LANGUAGE, CACHE_CONTROL, HeaderMap, HeaderValue, PRAGMA, REFERER,
|
|
};
|
|
use reqwest::{Client, StatusCode};
|
|
use serde::{Deserialize, Serialize};
|
|
use tokio::sync::Semaphore;
|
|
use url::Url;
|
|
|
|
use crate::error::{AppError, AppResult};
|
|
use crate::logs::{info, warn};
|
|
use crate::persona::{MAJOR_VERSIONS, Persona, Platform};
|
|
use crate::proxy::ProxyConfig;
|
|
|
|
const MODULE: &str = "http_client";
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(default)]
|
|
pub struct HttpClientConfig {
|
|
pub connect_timeout_secs: u64,
|
|
pub request_timeout_secs: u64,
|
|
pub queue_timeout_secs: u64,
|
|
pub max_attempts: u32,
|
|
pub retry_base_delay_ms: u64,
|
|
pub max_concurrency: usize,
|
|
pub max_text_bytes: usize,
|
|
pub max_binary_bytes: usize,
|
|
}
|
|
|
|
impl Default for HttpClientConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
connect_timeout_secs: 15,
|
|
request_timeout_secs: 40,
|
|
queue_timeout_secs: 30,
|
|
max_attempts: 4,
|
|
retry_base_delay_ms: 650,
|
|
max_concurrency: 8,
|
|
max_text_bytes: 4 * 1024 * 1024,
|
|
max_binary_bytes: 20 * 1024 * 1024,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct BrowserIdentity {
|
|
pub user_agent: String,
|
|
pub accept_language: String,
|
|
pub sec_ch_ua: String,
|
|
}
|
|
|
|
impl BrowserIdentity {
|
|
pub fn random_pt_br() -> Self {
|
|
const LANGUAGES: &[&str] = &[
|
|
"pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
|
|
"pt-BR,pt;q=0.9,en;q=0.8",
|
|
"pt-BR,pt;q=0.8,en-US;q=0.6,en;q=0.4",
|
|
];
|
|
let mut rng = rand::thread_rng();
|
|
let version = MAJOR_VERSIONS[rng.gen_range(0..MAJOR_VERSIONS.len())] as u16;
|
|
let accept_language = LANGUAGES[rng.gen_range(0..LANGUAGES.len())].to_owned();
|
|
Self {
|
|
user_agent: format!(
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 \
|
|
(KHTML, like Gecko) Chrome/{version}.0.0.0 Safari/537.36"
|
|
),
|
|
accept_language,
|
|
sec_ch_ua: format!(
|
|
"\"Chromium\";v=\"{version}\", \"Google Chrome\";v=\"{version}\", \
|
|
\"Not_A Brand\";v=\"24\""
|
|
),
|
|
}
|
|
}
|
|
|
|
/// Constrói a identidade a partir de uma [`Persona`] sorteada para a
|
|
/// sessão. Garante que a stack `reqwest` exponha o mesmo User-Agent,
|
|
/// Accept-Language e platform Client Hints que a aba do Chromium.
|
|
pub fn from_persona(persona: &Persona) -> Self {
|
|
Self {
|
|
user_agent: persona.user_agent.clone(),
|
|
accept_language: persona.accept_language.clone(),
|
|
sec_ch_ua: persona.sec_ch_ua(),
|
|
}
|
|
}
|
|
|
|
/// Valor do header `sec-ch-ua-platform` coerente com a plataforma da
|
|
/// persona. Usado para alinhar Client Hints do `reqwest` ao Chromium.
|
|
pub fn sec_ch_ua_platform(persona: &Persona) -> &'static str {
|
|
persona.platform.sec_ch_ua_platform()
|
|
}
|
|
}
|
|
|
|
impl Platform {
|
|
pub fn sec_ch_ua_header_value(self) -> &'static str {
|
|
match self {
|
|
Self::Windows => "\"Windows\"",
|
|
Self::MacOs => "\"macOS\"",
|
|
Self::Linux => "\"Linux\"",
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct HttpClientFactory {
|
|
proxy: ProxyConfig,
|
|
config: HttpClientConfig,
|
|
semaphore: Arc<Semaphore>,
|
|
}
|
|
|
|
impl HttpClientFactory {
|
|
pub fn new(proxy: ProxyConfig, config: HttpClientConfig) -> AppResult<Self> {
|
|
if config.max_attempts == 0 || config.max_concurrency == 0 {
|
|
return Err(AppError::Config(
|
|
"max_attempts e max_concurrency devem ser maiores que zero".into(),
|
|
));
|
|
}
|
|
proxy.validate()?;
|
|
Ok(Self {
|
|
proxy,
|
|
semaphore: Arc::new(Semaphore::new(config.max_concurrency)),
|
|
config,
|
|
})
|
|
}
|
|
|
|
pub fn proxy(&self) -> &ProxyConfig {
|
|
&self.proxy
|
|
}
|
|
|
|
/// Cria uma fábrica com a mesma política de proxy, mas com um orçamento
|
|
/// menor para fontes auxiliares. Útil para busca/crawl: uma origem lenta
|
|
/// não deve reter todo o pipeline de contatos por vários minutos.
|
|
pub fn with_request_budget(
|
|
&self,
|
|
request_timeout_secs: u64,
|
|
max_attempts: u32,
|
|
) -> AppResult<Self> {
|
|
let mut config = self.config.clone();
|
|
config.request_timeout_secs = request_timeout_secs.max(1);
|
|
config.max_attempts = max_attempts.max(1);
|
|
Self::new(self.proxy.clone(), config)
|
|
}
|
|
|
|
/// Constrói cookie jar e conexões novos. O retorno não implementa
|
|
/// `Clone` de propósito: uma sessão pertence a uma operação.
|
|
pub fn fresh(&self, request_id: &str) -> AppResult<HttpSession> {
|
|
let identity = BrowserIdentity::random_pt_br();
|
|
let platform_header = "\"Windows\"";
|
|
self.build_session(request_id, identity, platform_header)
|
|
}
|
|
|
|
/// Variação de [`fresh`](Self::fresh) que reutiliza a [`Persona`] sorteada
|
|
/// para a sessão, mantendo a stack `reqwest` coerente com a aba do
|
|
/// Chromium (mesmo UA, Accept-Language e `sec-ch-ua-platform`).
|
|
pub fn fresh_with_persona(
|
|
&self,
|
|
request_id: &str,
|
|
persona: &Persona,
|
|
) -> AppResult<HttpSession> {
|
|
let identity = BrowserIdentity::from_persona(persona);
|
|
let platform_header = persona.platform.sec_ch_ua_header_value();
|
|
self.build_session(request_id, identity, platform_header)
|
|
}
|
|
|
|
fn build_session(
|
|
&self,
|
|
request_id: &str,
|
|
identity: BrowserIdentity,
|
|
platform_header: &'static str,
|
|
) -> AppResult<HttpSession> {
|
|
let mut headers = HeaderMap::new();
|
|
headers.insert(
|
|
ACCEPT_LANGUAGE,
|
|
HeaderValue::from_str(&identity.accept_language)
|
|
.map_err(|error| AppError::Config(error.to_string()))?,
|
|
);
|
|
headers.insert(
|
|
"sec-ch-ua",
|
|
HeaderValue::from_str(&identity.sec_ch_ua)
|
|
.map_err(|error| AppError::Config(error.to_string()))?,
|
|
);
|
|
headers.insert("sec-ch-ua-mobile", HeaderValue::from_static("?0"));
|
|
headers.insert(
|
|
"sec-ch-ua-platform",
|
|
HeaderValue::from_static(platform_header),
|
|
);
|
|
headers.insert(CACHE_CONTROL, HeaderValue::from_static("no-cache"));
|
|
headers.insert(PRAGMA, HeaderValue::from_static("no-cache"));
|
|
|
|
let mut builder = Client::builder()
|
|
.cookie_store(true)
|
|
.default_headers(headers)
|
|
.user_agent(&identity.user_agent)
|
|
.connect_timeout(Duration::from_secs(self.config.connect_timeout_secs))
|
|
.timeout(Duration::from_secs(self.config.request_timeout_secs))
|
|
.pool_max_idle_per_host(0)
|
|
.pool_idle_timeout(Duration::ZERO)
|
|
.redirect(reqwest::redirect::Policy::limited(8));
|
|
if let Some(proxy) = self.proxy.reqwest_proxy()? {
|
|
builder = builder.proxy(proxy);
|
|
}
|
|
let client = builder.build()?;
|
|
info(MODULE, request_id, "nova sessão HTTP isolada criada");
|
|
Ok(HttpSession {
|
|
client,
|
|
identity,
|
|
config: self.config.clone(),
|
|
semaphore: self.semaphore.clone(),
|
|
})
|
|
}
|
|
}
|
|
|
|
pub struct HttpSession {
|
|
client: Client,
|
|
identity: BrowserIdentity,
|
|
config: HttpClientConfig,
|
|
semaphore: Arc<Semaphore>,
|
|
}
|
|
|
|
impl HttpSession {
|
|
pub fn identity(&self) -> &BrowserIdentity {
|
|
&self.identity
|
|
}
|
|
|
|
/// Visita a raiz da origem no mesmo cookie jar antes do recurso final.
|
|
/// Status HTTP de bloqueio não é ignorado, mas falhas transitórias passam
|
|
/// pela mesma política de retry do acesso principal.
|
|
pub async fn warm_up(&self, request_id: &str, target_url: &str) -> AppResult<()> {
|
|
let target = parse_http_url(target_url)?;
|
|
let origin = origin_url(&target)?;
|
|
info(
|
|
MODULE,
|
|
request_id,
|
|
format!("aquecendo sessão em {}", redacted_url(&origin)),
|
|
);
|
|
let _ = self
|
|
.fetch_bytes_with_limit(
|
|
request_id,
|
|
origin.as_str(),
|
|
self.config.max_text_bytes,
|
|
RequestKind::Navigation,
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn warm_then_get_text(
|
|
&self,
|
|
request_id: &str,
|
|
target_url: &str,
|
|
) -> AppResult<HttpTextResponse> {
|
|
self.warm_up(request_id, target_url).await?;
|
|
self.get_text(request_id, target_url).await
|
|
}
|
|
|
|
pub async fn get_text(&self, request_id: &str, url: &str) -> AppResult<HttpTextResponse> {
|
|
self.get_text_with_limit(request_id, url, self.config.max_text_bytes)
|
|
.await
|
|
}
|
|
|
|
pub async fn get_text_with_limit(
|
|
&self,
|
|
request_id: &str,
|
|
url: &str,
|
|
max_bytes: usize,
|
|
) -> AppResult<HttpTextResponse> {
|
|
let payload = self
|
|
.fetch_bytes_with_limit(request_id, url, max_bytes, RequestKind::Navigation)
|
|
.await?;
|
|
let text = String::from_utf8(payload.body).map_err(|_| AppError::External {
|
|
service: service_name(url),
|
|
message: "resposta textual não é UTF-8".into(),
|
|
})?;
|
|
Ok(HttpTextResponse {
|
|
status: payload.status,
|
|
final_url: payload.final_url,
|
|
content_type: payload.content_type,
|
|
text,
|
|
})
|
|
}
|
|
|
|
pub async fn get_bytes(&self, request_id: &str, url: &str) -> AppResult<HttpBytesResponse> {
|
|
self.fetch_bytes_with_limit(
|
|
request_id,
|
|
url,
|
|
self.config.max_binary_bytes,
|
|
RequestKind::Navigation,
|
|
)
|
|
.await
|
|
}
|
|
|
|
pub async fn get_bytes_with_limit(
|
|
&self,
|
|
request_id: &str,
|
|
url: &str,
|
|
max_bytes: usize,
|
|
) -> AppResult<HttpBytesResponse> {
|
|
self.fetch_bytes_with_limit(request_id, url, max_bytes, RequestKind::Navigation)
|
|
.await
|
|
}
|
|
|
|
pub async fn get_xhr_text(
|
|
&self,
|
|
request_id: &str,
|
|
url: &str,
|
|
referer: &str,
|
|
) -> AppResult<HttpTextResponse> {
|
|
let payload = self
|
|
.fetch_bytes_with_limit(
|
|
request_id,
|
|
url,
|
|
self.config.max_text_bytes,
|
|
RequestKind::Xhr { referer },
|
|
)
|
|
.await?;
|
|
let text = String::from_utf8(payload.body).map_err(|_| AppError::External {
|
|
service: service_name(url),
|
|
message: "resposta XHR não é UTF-8".into(),
|
|
})?;
|
|
Ok(HttpTextResponse {
|
|
status: payload.status,
|
|
final_url: payload.final_url,
|
|
content_type: payload.content_type,
|
|
text,
|
|
})
|
|
}
|
|
|
|
async fn fetch_bytes_with_limit(
|
|
&self,
|
|
request_id: &str,
|
|
url: &str,
|
|
max_bytes: usize,
|
|
kind: RequestKind<'_>,
|
|
) -> AppResult<HttpBytesResponse> {
|
|
let parsed = parse_http_url(url)?;
|
|
let safe_log_url = redacted_url(&parsed);
|
|
let service = service_name(url);
|
|
let mut attempt = 1;
|
|
|
|
loop {
|
|
let permit = tokio::time::timeout(
|
|
Duration::from_secs(self.config.queue_timeout_secs),
|
|
self.semaphore.acquire(),
|
|
)
|
|
.await
|
|
.map_err(|_| AppError::Timeout("fila HTTP excedeu o tempo limite".into()))?
|
|
.map_err(|_| AppError::Cancelled)?;
|
|
|
|
let mut request = self.client.get(parsed.clone());
|
|
match kind {
|
|
RequestKind::Navigation => {
|
|
request = request
|
|
.header(
|
|
ACCEPT,
|
|
"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
|
)
|
|
.header("sec-fetch-dest", "document")
|
|
.header("sec-fetch-mode", "navigate")
|
|
.header("sec-fetch-site", "none")
|
|
.header("upgrade-insecure-requests", "1");
|
|
}
|
|
RequestKind::Xhr { referer } => {
|
|
request = request
|
|
.header(ACCEPT, "application/json, text/plain, */*")
|
|
.header(REFERER, referer)
|
|
.header("sec-fetch-dest", "empty")
|
|
.header("sec-fetch-mode", "cors")
|
|
.header("sec-fetch-site", "same-origin");
|
|
}
|
|
}
|
|
|
|
info(
|
|
MODULE,
|
|
request_id,
|
|
format!("GET {safe_log_url} tentativa {attempt}"),
|
|
);
|
|
let send = tokio::time::timeout(
|
|
Duration::from_secs(self.config.request_timeout_secs),
|
|
request.send(),
|
|
)
|
|
.await;
|
|
|
|
let mut response = match send {
|
|
Err(_) if attempt < self.config.max_attempts => {
|
|
drop(permit);
|
|
warn(
|
|
MODULE,
|
|
request_id,
|
|
"timeout HTTP; nova tentativa será feita",
|
|
);
|
|
self.sleep_before_retry(attempt, None, false).await;
|
|
attempt += 1;
|
|
continue;
|
|
}
|
|
Err(_) => {
|
|
return Err(AppError::Timeout(format!(
|
|
"tempo limite acessando {safe_log_url}"
|
|
)));
|
|
}
|
|
Ok(Err(error)) if attempt < self.config.max_attempts && retryable_error(&error) => {
|
|
drop(permit);
|
|
warn(
|
|
MODULE,
|
|
request_id,
|
|
format!("falha transitória de rede: {error}; repetindo"),
|
|
);
|
|
self.sleep_before_retry(attempt, None, false).await;
|
|
attempt += 1;
|
|
continue;
|
|
}
|
|
Ok(Err(error)) => return Err(error.into()),
|
|
Ok(Ok(response)) => response,
|
|
};
|
|
|
|
let status = response.status();
|
|
let retry_after = parse_retry_after(response.headers());
|
|
if retryable_status(status) && attempt < self.config.max_attempts {
|
|
drop(response);
|
|
drop(permit);
|
|
warn(
|
|
MODULE,
|
|
request_id,
|
|
format!("origem respondeu {status}; repetindo com backoff"),
|
|
);
|
|
self.sleep_before_retry(
|
|
attempt,
|
|
retry_after,
|
|
status == StatusCode::TOO_MANY_REQUESTS,
|
|
)
|
|
.await;
|
|
attempt += 1;
|
|
continue;
|
|
}
|
|
if !status.is_success() {
|
|
return Err(if status == StatusCode::NOT_FOUND {
|
|
AppError::NotFound(format!("recurso não encontrado em {safe_log_url}"))
|
|
} else {
|
|
AppError::External {
|
|
service,
|
|
message: format!("status HTTP {status}"),
|
|
}
|
|
});
|
|
}
|
|
|
|
if let Some(length) = response.content_length()
|
|
&& length > max_bytes as u64
|
|
{
|
|
return Err(AppError::Validation(format!(
|
|
"resposta excede o limite de {max_bytes} bytes"
|
|
)));
|
|
}
|
|
let final_url = response.url().to_string();
|
|
let content_type = response
|
|
.headers()
|
|
.get(reqwest::header::CONTENT_TYPE)
|
|
.and_then(|value| value.to_str().ok())
|
|
.map(str::to_owned);
|
|
let mut body = Vec::new();
|
|
while let Some(chunk) = response.chunk().await? {
|
|
if body.len().saturating_add(chunk.len()) > max_bytes {
|
|
return Err(AppError::Validation(format!(
|
|
"resposta excede o limite de {max_bytes} bytes"
|
|
)));
|
|
}
|
|
body.extend_from_slice(&chunk);
|
|
}
|
|
drop(permit);
|
|
return Ok(HttpBytesResponse {
|
|
status: status.as_u16(),
|
|
final_url,
|
|
content_type,
|
|
body,
|
|
});
|
|
}
|
|
}
|
|
|
|
async fn sleep_before_retry(
|
|
&self,
|
|
attempt: u32,
|
|
retry_after_secs: Option<u64>,
|
|
rate_limited: bool,
|
|
) {
|
|
let delay = if rate_limited {
|
|
// 429 é uma indicação explícita para desacelerar. Mesmo que a
|
|
// origem não informe Retry-After, não repetimos em segundos.
|
|
let floor = retry_after_secs.unwrap_or(60).max(60);
|
|
Duration::from_secs(floor.saturating_add(rand::thread_rng().gen_range(0_u64..=120)))
|
|
} else if let Some(seconds) = retry_after_secs {
|
|
Duration::from_secs(seconds.min(30))
|
|
} else {
|
|
let exponent = attempt.saturating_sub(1).min(5);
|
|
let base = self
|
|
.config
|
|
.retry_base_delay_ms
|
|
.saturating_mul(1_u64 << exponent)
|
|
.min(12_000);
|
|
let jitter = rand::thread_rng().gen_range(75_u64..=125_u64);
|
|
Duration::from_millis(base.saturating_mul(jitter) / 100)
|
|
};
|
|
tokio::time::sleep(delay).await;
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct HttpTextResponse {
|
|
pub status: u16,
|
|
pub final_url: String,
|
|
pub content_type: Option<String>,
|
|
pub text: String,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct HttpBytesResponse {
|
|
pub status: u16,
|
|
pub final_url: String,
|
|
pub content_type: Option<String>,
|
|
pub body: Vec<u8>,
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum RequestKind<'a> {
|
|
Navigation,
|
|
Xhr { referer: &'a str },
|
|
}
|
|
|
|
fn parse_http_url(value: &str) -> AppResult<Url> {
|
|
let parsed = Url::parse(value)
|
|
.map_err(|error| AppError::Validation(format!("URL inválida: {error}")))?;
|
|
if !matches!(parsed.scheme(), "http" | "https") || parsed.host_str().is_none() {
|
|
return Err(AppError::Validation(
|
|
"apenas URLs http(s) absolutas são aceitas".into(),
|
|
));
|
|
}
|
|
Ok(parsed)
|
|
}
|
|
|
|
fn origin_url(url: &Url) -> AppResult<Url> {
|
|
let host = url
|
|
.host_str()
|
|
.ok_or_else(|| AppError::Validation("URL sem host".into()))?;
|
|
let mut origin = Url::parse(&format!("{}://{host}", url.scheme()))
|
|
.map_err(|error| AppError::Validation(error.to_string()))?;
|
|
if let Some(port) = url.port() {
|
|
origin
|
|
.set_port(Some(port))
|
|
.map_err(|_| AppError::Validation("porta inválida".into()))?;
|
|
}
|
|
Ok(origin)
|
|
}
|
|
|
|
pub fn redacted_url(url: &Url) -> String {
|
|
let mut clean = url.clone();
|
|
clean.set_query(None);
|
|
clean.set_fragment(None);
|
|
let _ = clean.set_username("");
|
|
let _ = clean.set_password(None);
|
|
clean.to_string()
|
|
}
|
|
|
|
fn service_name(url: &str) -> String {
|
|
Url::parse(url)
|
|
.ok()
|
|
.and_then(|url| url.host_str().map(str::to_owned))
|
|
.unwrap_or_else(|| "origem externa".into())
|
|
}
|
|
|
|
fn retryable_status(status: StatusCode) -> bool {
|
|
matches!(
|
|
status,
|
|
StatusCode::TOO_MANY_REQUESTS
|
|
| StatusCode::UNAUTHORIZED
|
|
| StatusCode::FORBIDDEN
|
|
| StatusCode::REQUEST_TIMEOUT
|
|
| StatusCode::BAD_GATEWAY
|
|
| StatusCode::SERVICE_UNAVAILABLE
|
|
| StatusCode::GATEWAY_TIMEOUT
|
|
) || status.is_server_error()
|
|
}
|
|
|
|
fn retryable_error(error: &reqwest::Error) -> bool {
|
|
error.is_timeout() || error.is_connect() || error.is_request() || error.is_body()
|
|
}
|
|
|
|
fn parse_retry_after(headers: &HeaderMap) -> Option<u64> {
|
|
headers
|
|
.get(reqwest::header::RETRY_AFTER)
|
|
.and_then(|value| value.to_str().ok())
|
|
.and_then(|value| value.parse().ok())
|
|
}
|