Added check on first frame feed
This commit is contained in:
@@ -0,0 +1,607 @@
|
||||
use crate::{
|
||||
config::Config,
|
||||
terminal::{ScreenState, TerminalUi},
|
||||
};
|
||||
use anyhow::{anyhow, Context, Result};
|
||||
use std::{
|
||||
fs,
|
||||
path::{Path, PathBuf},
|
||||
sync::atomic::{AtomicBool, Ordering},
|
||||
sync::Arc,
|
||||
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, AsyncWriteExt},
|
||||
process::Command,
|
||||
time::{sleep, timeout},
|
||||
};
|
||||
|
||||
const DEFAULT_DONE_SUFFIX: &str = ":done";
|
||||
const DEFAULT_SENTINEL: &str = "__CAMFINDER_DONE__";
|
||||
|
||||
pub async fn run_redis_coordinator(
|
||||
config: Config,
|
||||
ui: Arc<TerminalUi>,
|
||||
_interrupted: Arc<AtomicBool>,
|
||||
) -> Result<()> {
|
||||
let ips = load_ip_list(&config.redis_input_csv)?;
|
||||
if ips.is_empty() {
|
||||
return Err(anyhow!(
|
||||
"nenhum IP encontrado em {}",
|
||||
config.redis_input_csv
|
||||
));
|
||||
}
|
||||
|
||||
let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX);
|
||||
clear_queue(&config.redis_url, &config.redis_queue_name, &done_key).await?;
|
||||
enqueue_ips(
|
||||
&config.redis_url,
|
||||
&config.redis_queue_name,
|
||||
&ips,
|
||||
config.redis_worker_count.max(1),
|
||||
)
|
||||
.await?;
|
||||
mark_done(&config.redis_url, &done_key).await?;
|
||||
|
||||
if ui.is_enabled() {
|
||||
ui.print_banner();
|
||||
ui.render(&ScreenState {
|
||||
current_ip: format!("fila Redis: {}", config.redis_queue_name),
|
||||
tested: ips.len(),
|
||||
total: ips.len(),
|
||||
});
|
||||
ui.finish();
|
||||
} else {
|
||||
println!(
|
||||
"Fila Redis alimentada com {} IPs em {}. Workers esperados: {}",
|
||||
ips.len(),
|
||||
config.redis_queue_name,
|
||||
config.redis_worker_count.max(1)
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn run_redis_worker(
|
||||
config: Config,
|
||||
_ui: Arc<TerminalUi>,
|
||||
interrupted: Arc<AtomicBool>,
|
||||
) -> Result<()> {
|
||||
ensure_output_dirs(&config.snapshot_dir, &config.sqlite_path)?;
|
||||
init_sqlite(&config.sqlite_path).await?;
|
||||
|
||||
let done_key = format!("{}{}", config.redis_queue_name, DEFAULT_DONE_SUFFIX);
|
||||
let started_at = Instant::now();
|
||||
let mut processed = 0usize;
|
||||
let mut captured = 0usize;
|
||||
let paths = normalize_paths(&config.rtsp_paths);
|
||||
|
||||
println!(
|
||||
"worker pronto para {} paths em {}",
|
||||
paths.len(),
|
||||
config.redis_queue_name
|
||||
);
|
||||
|
||||
loop {
|
||||
if interrupted.load(Ordering::SeqCst) {
|
||||
break;
|
||||
}
|
||||
|
||||
match pop_queue_item(&config.redis_url, &config.redis_queue_name).await? {
|
||||
Some(item) if item == DEFAULT_SENTINEL => break,
|
||||
Some(ip) => {
|
||||
for rtsp_path in &paths {
|
||||
if interrupted.load(Ordering::SeqCst) {
|
||||
break;
|
||||
}
|
||||
|
||||
let attempt = build_rtsp_url(&ip, config.port, &rtsp_path);
|
||||
let checked_at = timestamp_millis();
|
||||
let probe = probe_rtsp(&attempt, config.capture_timeout_ms).await;
|
||||
|
||||
let (status, screenshot_path, error) = if probe.is_connectable() {
|
||||
let output_path =
|
||||
build_snapshot_path(&config.snapshot_dir, &ip, &rtsp_path, checked_at);
|
||||
let outcome = capture_feed(&attempt, &output_path, config.capture_timeout_ms)
|
||||
.await;
|
||||
|
||||
match outcome {
|
||||
Ok(()) => {
|
||||
captured += 1;
|
||||
(
|
||||
"captured".to_string(),
|
||||
Some(output_path.to_string_lossy().to_string()),
|
||||
None,
|
||||
)
|
||||
}
|
||||
Err(err) => {
|
||||
let _ = fs::remove_file(&output_path);
|
||||
("failed".to_string(), None, Some(err.to_string()))
|
||||
}
|
||||
}
|
||||
} else {
|
||||
(
|
||||
probe.status,
|
||||
None,
|
||||
probe.error.map(|error| error.to_string()),
|
||||
)
|
||||
};
|
||||
|
||||
insert_sqlite_result(
|
||||
&config.sqlite_path,
|
||||
checked_at,
|
||||
&ip,
|
||||
&rtsp_path,
|
||||
&attempt,
|
||||
status,
|
||||
screenshot_path.as_deref(),
|
||||
error.as_deref(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
processed += 1;
|
||||
println!(
|
||||
"{} -> {}",
|
||||
attempt,
|
||||
screenshot_path.unwrap_or_else(|| "sem imagem".to_string())
|
||||
);
|
||||
}
|
||||
}
|
||||
None => {
|
||||
if redis_key_exists(&config.redis_url, &done_key).await? {
|
||||
break;
|
||||
}
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let elapsed = started_at.elapsed().as_secs_f64();
|
||||
println!(
|
||||
"worker finalizado em {:.2}s, tentativas {}, capturas {}",
|
||||
elapsed, processed, captured
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result<()> {
|
||||
let mut command = Command::new("ffmpeg");
|
||||
command
|
||||
.kill_on_drop(true)
|
||||
.arg("-hide_banner")
|
||||
.arg("-loglevel")
|
||||
.arg("error")
|
||||
.arg("-fflags")
|
||||
.arg("nobuffer")
|
||||
.arg("-flags")
|
||||
.arg("low_delay")
|
||||
.arg("-probesize")
|
||||
.arg("32")
|
||||
.arg("-analyzeduration")
|
||||
.arg("0")
|
||||
.arg("-rtsp_transport")
|
||||
.arg("tcp")
|
||||
.arg("-i")
|
||||
.arg(url)
|
||||
.arg("-frames:v")
|
||||
.arg("1")
|
||||
.arg("-skip_frame")
|
||||
.arg("nokey")
|
||||
.arg("-an")
|
||||
.arg("-sn")
|
||||
.arg("-dn")
|
||||
.arg("-y")
|
||||
.arg(output_path.as_os_str());
|
||||
|
||||
let output = timeout(
|
||||
Duration::from_millis(timeout_ms),
|
||||
async { command.output().await },
|
||||
)
|
||||
.await
|
||||
.map_err(|_| anyhow!("ffmpeg timeout for {url}"))??;
|
||||
|
||||
if output.status.success() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||
if output_path.exists() {
|
||||
let _ = fs::remove_file(output_path);
|
||||
}
|
||||
Err(anyhow!(
|
||||
"ffmpeg failed for {url}: {}",
|
||||
if stderr.is_empty() {
|
||||
"sem stderr".to_string()
|
||||
} else {
|
||||
stderr
|
||||
}
|
||||
))
|
||||
}
|
||||
|
||||
async fn probe_rtsp(url: &str, timeout_ms: u64) -> RtspProbe {
|
||||
let host = extract_host(url);
|
||||
let connect_result = timeout(
|
||||
Duration::from_millis(timeout_ms),
|
||||
tokio::net::TcpStream::connect(host),
|
||||
)
|
||||
.await;
|
||||
|
||||
let mut stream = match connect_result {
|
||||
Ok(Ok(stream)) => stream,
|
||||
Ok(Err(err)) => {
|
||||
return RtspProbe {
|
||||
status: "connect_error".to_string(),
|
||||
error: Some(err.to_string()),
|
||||
};
|
||||
}
|
||||
Err(_) => {
|
||||
return RtspProbe {
|
||||
status: "timeout".to_string(),
|
||||
error: Some("connect timeout".to_string()),
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let request = format!(
|
||||
"DESCRIBE {url} RTSP/1.0\r\nCSeq: 1\r\nUser-Agent: CamFinder/0.1\r\nAccept: application/sdp\r\n\r\n"
|
||||
);
|
||||
match timeout(
|
||||
Duration::from_millis(timeout_ms),
|
||||
stream.write_all(request.as_bytes()),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(err)) => {
|
||||
return RtspProbe {
|
||||
status: "write_error".to_string(),
|
||||
error: Some(err.to_string()),
|
||||
};
|
||||
}
|
||||
Err(_) => {
|
||||
return RtspProbe {
|
||||
status: "timeout".to_string(),
|
||||
error: Some("write timeout".to_string()),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
let mut buf = vec![0u8; 1024];
|
||||
let read_result = timeout(
|
||||
Duration::from_millis(timeout_ms),
|
||||
stream.read(&mut buf),
|
||||
)
|
||||
.await;
|
||||
|
||||
let bytes_read = match read_result {
|
||||
Ok(Ok(0)) => {
|
||||
return RtspProbe {
|
||||
status: "no_response".to_string(),
|
||||
error: Some("empty response".to_string()),
|
||||
};
|
||||
}
|
||||
Ok(Ok(n)) => n,
|
||||
Ok(Err(err)) => {
|
||||
return RtspProbe {
|
||||
status: "read_error".to_string(),
|
||||
error: Some(err.to_string()),
|
||||
};
|
||||
}
|
||||
Err(_) => {
|
||||
return RtspProbe {
|
||||
status: "timeout".to_string(),
|
||||
error: Some("read timeout".to_string()),
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
match parse_rtsp_status(&buf[..bytes_read]) {
|
||||
Ok(Some(200)) => RtspProbe {
|
||||
status: "ok".to_string(),
|
||||
error: None,
|
||||
},
|
||||
Ok(Some(401)) | Ok(Some(403)) => RtspProbe {
|
||||
status: "auth_required".to_string(),
|
||||
error: None,
|
||||
},
|
||||
Ok(Some(code)) => RtspProbe {
|
||||
status: format!("rtsp_{code}"),
|
||||
error: None,
|
||||
},
|
||||
Ok(None) => RtspProbe {
|
||||
status: "bad_response".to_string(),
|
||||
error: Some("missing RTSP status line".to_string()),
|
||||
},
|
||||
Err(err) => RtspProbe {
|
||||
status: "bad_response".to_string(),
|
||||
error: Some(err.to_string()),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct RtspProbe {
|
||||
status: String,
|
||||
error: Option<String>,
|
||||
}
|
||||
|
||||
impl RtspProbe {
|
||||
fn is_connectable(&self) -> bool {
|
||||
matches!(
|
||||
self.status.as_str(),
|
||||
"ok" | "auth_required" | "rtsp_200" | "rtsp_401" | "rtsp_403"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_rtsp_status(response: &[u8]) -> Result<Option<u16>> {
|
||||
let response = std::str::from_utf8(response).context("RTSP response is not valid UTF-8")?;
|
||||
let first_line = response.lines().next().unwrap_or("").trim();
|
||||
if !first_line.starts_with("RTSP/1.") {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let mut parts = first_line.split_whitespace();
|
||||
let _version = parts.next();
|
||||
let code = parts
|
||||
.next()
|
||||
.ok_or_else(|| anyhow!("RTSP response missing status code"))?;
|
||||
let code = code
|
||||
.parse::<u16>()
|
||||
.with_context(|| format!("invalid RTSP status code: {code}"))?;
|
||||
Ok(Some(code))
|
||||
}
|
||||
|
||||
fn extract_host(url: &str) -> String {
|
||||
let rest = url.strip_prefix("rtsp://").unwrap_or(url);
|
||||
let authority = rest.split('/').next().unwrap_or(rest);
|
||||
authority.to_string()
|
||||
}
|
||||
|
||||
async fn pop_queue_item(redis_url: &str, queue_name: &str) -> Result<Option<String>> {
|
||||
let output = redis_cli(redis_url, &["--raw", "LPOP", queue_name]).await?;
|
||||
let value = output.trim();
|
||||
if value.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
Ok(Some(value.to_string()))
|
||||
}
|
||||
|
||||
async fn redis_key_exists(redis_url: &str, key: &str) -> Result<bool> {
|
||||
let output = redis_cli(redis_url, &["--raw", "EXISTS", key]).await?;
|
||||
Ok(output.trim() == "1")
|
||||
}
|
||||
|
||||
async fn clear_queue(redis_url: &str, queue_name: &str, done_key: &str) -> Result<()> {
|
||||
let _ = redis_cli(redis_url, &["DEL", queue_name, done_key]).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn mark_done(redis_url: &str, done_key: &str) -> Result<()> {
|
||||
let _ = redis_cli(redis_url, &["SET", done_key, "1"]).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn enqueue_ips(
|
||||
redis_url: &str,
|
||||
queue_name: &str,
|
||||
ips: &[String],
|
||||
worker_count: usize,
|
||||
) -> Result<()> {
|
||||
for chunk in ips.chunks(400) {
|
||||
let mut args = Vec::with_capacity(chunk.len() + 2);
|
||||
args.push("RPUSH".to_string());
|
||||
args.push(queue_name.to_string());
|
||||
args.extend(chunk.iter().cloned());
|
||||
let args_ref = args.iter().map(String::as_str).collect::<Vec<_>>();
|
||||
let _ = redis_cli(redis_url, &args_ref).await?;
|
||||
}
|
||||
|
||||
let mut sentinels = Vec::with_capacity(worker_count + 2);
|
||||
sentinels.push("RPUSH".to_string());
|
||||
sentinels.push(queue_name.to_string());
|
||||
sentinels.extend(std::iter::repeat_with(|| DEFAULT_SENTINEL.to_string()).take(worker_count));
|
||||
let sentinel_args = sentinels.iter().map(String::as_str).collect::<Vec<_>>();
|
||||
let _ = redis_cli(redis_url, &sentinel_args).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn redis_cli(redis_url: &str, args: &[&str]) -> Result<String> {
|
||||
let mut command = Command::new("redis-cli");
|
||||
command.kill_on_drop(true).arg("-u").arg(redis_url);
|
||||
for arg in args {
|
||||
command.arg(arg);
|
||||
}
|
||||
|
||||
let output = timeout(Duration::from_secs(15), async { command.output().await })
|
||||
.await
|
||||
.context("redis-cli timeout")??;
|
||||
|
||||
if !output.status.success() {
|
||||
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||
return Err(anyhow!(
|
||||
"redis-cli failed{}",
|
||||
if stderr.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
format!(": {stderr}")
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
Ok(String::from_utf8_lossy(&output.stdout).to_string())
|
||||
}
|
||||
|
||||
async fn init_sqlite(sqlite_path: &Path) -> Result<()> {
|
||||
let sql = r#"
|
||||
CREATE TABLE IF NOT EXISTS rtsp_feed_results (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
checked_at INTEGER NOT NULL,
|
||||
target_ip TEXT NOT NULL,
|
||||
rtsp_path TEXT NOT NULL,
|
||||
rtsp_url TEXT NOT NULL,
|
||||
screenshot_path TEXT,
|
||||
status TEXT NOT NULL,
|
||||
error TEXT
|
||||
);
|
||||
"#;
|
||||
sqlite_exec(sqlite_path, sql).await
|
||||
}
|
||||
|
||||
async fn insert_sqlite_result(
|
||||
sqlite_path: &Path,
|
||||
checked_at: i64,
|
||||
target_ip: &str,
|
||||
rtsp_path: &str,
|
||||
rtsp_url: &str,
|
||||
status: String,
|
||||
screenshot_path: Option<&str>,
|
||||
error: Option<&str>,
|
||||
) -> Result<()> {
|
||||
let sql = format!(
|
||||
"INSERT INTO rtsp_feed_results (checked_at, target_ip, rtsp_path, rtsp_url, screenshot_path, status, error) VALUES ({checked_at}, {target_ip}, {rtsp_path}, {rtsp_url}, {screenshot_path}, {status}, {error});",
|
||||
checked_at = checked_at,
|
||||
target_ip = sql_quote(target_ip),
|
||||
rtsp_path = sql_quote(rtsp_path),
|
||||
rtsp_url = sql_quote(rtsp_url),
|
||||
screenshot_path = sql_nullable_quote(screenshot_path),
|
||||
status = sql_quote(&status),
|
||||
error = sql_nullable_quote(error),
|
||||
);
|
||||
sqlite_exec(sqlite_path, &sql).await
|
||||
}
|
||||
|
||||
async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> {
|
||||
let output = timeout(
|
||||
Duration::from_secs(20),
|
||||
async {
|
||||
let mut command = Command::new("sqlite3");
|
||||
command.kill_on_drop(true).arg(sqlite_path).arg(sql);
|
||||
command.output().await
|
||||
},
|
||||
)
|
||||
.await
|
||||
.context("sqlite3 timeout")??;
|
||||
|
||||
if !output.status.success() {
|
||||
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||
return Err(anyhow!(
|
||||
"sqlite3 failed{}",
|
||||
if stderr.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
format!(": {stderr}")
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn load_ip_list(input: &str) -> Result<Vec<String>> {
|
||||
let content = fs::read_to_string(input)
|
||||
.with_context(|| format!("failed to read input CSV {input}"))?;
|
||||
let mut ips = Vec::new();
|
||||
|
||||
for line in content.lines() {
|
||||
let line = line.trim();
|
||||
if line.is_empty() || line.starts_with('#') {
|
||||
continue;
|
||||
}
|
||||
|
||||
let Some(first) = line.split(',').next() else {
|
||||
continue;
|
||||
};
|
||||
let value = strip_quotes(first.trim());
|
||||
if value.is_empty() || is_header(value) {
|
||||
continue;
|
||||
}
|
||||
ips.push(value.to_string());
|
||||
}
|
||||
|
||||
Ok(ips)
|
||||
}
|
||||
|
||||
fn normalize_paths(paths: &[String]) -> Vec<String> {
|
||||
paths
|
||||
.iter()
|
||||
.map(|path| path.trim())
|
||||
.filter(|path| !path.is_empty())
|
||||
.map(|path| {
|
||||
if path.starts_with('/') {
|
||||
path.to_string()
|
||||
} else {
|
||||
format!("/{path}")
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn build_rtsp_url(ip: &str, port: u16, path: &str) -> String {
|
||||
format!("rtsp://{ip}:{port}{path}")
|
||||
}
|
||||
|
||||
fn build_snapshot_path(base_dir: &Path, ip: &str, rtsp_path: &str, checked_at: i64) -> PathBuf {
|
||||
let ip_component = sanitize_component(ip);
|
||||
let path_component = sanitize_component(rtsp_path);
|
||||
base_dir.join(format!("{ip_component}_{checked_at}_{path_component}.png"))
|
||||
}
|
||||
|
||||
fn ensure_output_dirs(snapshot_dir: &Path, sqlite_path: &Path) -> Result<()> {
|
||||
fs::create_dir_all(snapshot_dir)
|
||||
.with_context(|| format!("failed to create snapshot dir {}", snapshot_dir.display()))?;
|
||||
if let Some(parent) = sqlite_path.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.with_context(|| format!("failed to create sqlite dir {}", parent.display()))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn sql_quote(value: &str) -> String {
|
||||
format!("'{}'", value.replace('\'', "''"))
|
||||
}
|
||||
|
||||
fn sql_nullable_quote(value: Option<&str>) -> String {
|
||||
match value {
|
||||
Some(value) if !value.is_empty() => sql_quote(value),
|
||||
_ => "NULL".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn strip_quotes(value: &str) -> &str {
|
||||
value
|
||||
.strip_prefix('"')
|
||||
.and_then(|value| value.strip_suffix('"'))
|
||||
.or_else(|| {
|
||||
value
|
||||
.strip_prefix('\'')
|
||||
.and_then(|value| value.strip_suffix('\''))
|
||||
})
|
||||
.unwrap_or(value)
|
||||
}
|
||||
|
||||
fn is_header(value: &str) -> bool {
|
||||
matches!(
|
||||
value.to_ascii_lowercase().as_str(),
|
||||
"ip" | "host" | "hostname" | "address" | "target" | "camera"
|
||||
)
|
||||
}
|
||||
|
||||
fn sanitize_component(value: &str) -> String {
|
||||
value
|
||||
.chars()
|
||||
.map(|ch| match ch {
|
||||
'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '-' | '_' => ch,
|
||||
_ => '_',
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn timestamp_millis() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| duration.as_millis() as i64)
|
||||
.unwrap_or_default()
|
||||
}
|
||||
Reference in New Issue
Block a user