src/validator.rs - master - web-master/IPTV-Aggregator

src/validator.rs
419 строк14 KB
web-master
Первый коммит
28 фев 2026, 18:57
28 фев 2026, 18:57d6f0f0c
//! Валидация IPTV каналов

use crate::config::{
    VAL_CHECK_TIMEOUT_SECS, VAL_MAX_CONCURRENT_CHECKS, VAL_MAX_RETRIES,
    VAL_MIN_CHUNK_SIZE, VAL_MIN_PLAYLIST_SIZE, VAL_QUICK_TIMEOUT_SECS, VAL_REDIRECT_LIMIT,
    VAL_RETRY_DELAY_MS,
};
use crate::types::Channel;
use log::{debug, info, warn};
use reqwest::{Client, StatusCode};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;

/// Результат проверки канала
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ValidationStatus {
    /// Канал работает
    Alive,
    /// Канал не работает (причина)
    Dead(String),
    /// Платный канал
    Paid(String),
    /// Заблокирован по гео
    GeoBlocked(String),
}

/// Статистика валидации
#[derive(Debug, Default)]
pub struct ValidationStats {
    pub alive: usize,
    pub dead: usize,
    pub paid: usize,
    pub geo_blocked: usize,
    pub errors: usize,
}

impl ValidationStats {
    /// Общее количество проверенных каналов
    pub fn total(&self) -> usize {
        self.alive + self.dead + self.paid + self.geo_blocked + self.errors
    }

    /// Процент рабочих каналов
    pub fn alive_percentage(&self) -> f64 {
        let total = self.total();
        if total == 0 {
            0.0
        } else {
            (self.alive as f64 / total as f64) * 100.0
        }
    }
}

/// Проверка HLS плейлиста
fn validate_hls_playlist(text: &str) -> ValidationStatus {
    let has_valid_tags = text.contains("#EXTM3U")
        || text.contains("#EXTINF")
        || text.contains("#EXT-X-STREAM-INF");
    let is_long_enough = text.len() >= VAL_MIN_PLAYLIST_SIZE;

    if has_valid_tags && is_long_enough {
        ValidationStatus::Alive
    } else if !has_valid_tags {
        ValidationStatus::Dead("Invalid HLS tags".to_string())
    } else {
        ValidationStatus::Dead("Playlist too short".to_string())
    }
}

/// Проверка обычного потока
async fn validate_stream_response(mut resp: reqwest::Response) -> ValidationStatus {
    match resp.chunk().await {
        Ok(Some(chunk)) => {
            if chunk.len() > VAL_MIN_CHUNK_SIZE {
                ValidationStatus::Alive
            } else {
                ValidationStatus::Dead("Content too short".to_string())
            }
        }
        Ok(None) => ValidationStatus::Dead("Empty response".to_string()),
        Err(e) => ValidationStatus::Dead(format!("Chunk error: {}", e)),
    }
}

/// Проверка одного канала
async fn check_channel(client: &Client, url: &str, semaphore: Arc<Semaphore>) -> ValidationStatus {
    let _permit = semaphore.acquire().await.unwrap();
    let is_hls = url.ends_with(".m3u8") || url.contains(".m3u8?");

    for attempt in 0..=VAL_MAX_RETRIES {
        if attempt > 0 {
            tokio::time::sleep(Duration::from_millis(VAL_RETRY_DELAY_MS * attempt as u64)).await;
        }

        let result = client
            .get(url)
            .timeout(Duration::from_secs(VAL_CHECK_TIMEOUT_SECS))
            .header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")
            .header("Accept", "*/*")
            .header("Accept-Language", "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7")
            .header("Connection", "keep-alive")
            .header("Referer", if is_hls { url } else { "https://google.com" })
            .send()
            .await;

        match result {
            Ok(resp) => {
                let status = resp.status();
                debug!("Channel {} returned status {}", url, status);

                // Проверка на платные/заблокированные каналы
                if matches!(
                    status,
                    StatusCode::FORBIDDEN
                        | StatusCode::UNAUTHORIZED
                        | StatusCode::PAYMENT_REQUIRED
                ) {
                    return ValidationStatus::Paid(format!("HTTP {}", status));
                }

                // Проверка на geo-blocking
                if status == StatusCode::NOT_FOUND && url.contains("geo") {
                    return ValidationStatus::GeoBlocked("Geo-blocked".to_string());
                }

                if status == StatusCode::OK || status == StatusCode::PARTIAL_CONTENT {
                    if is_hls {
                        match resp.text().await {
                            Ok(text) => return validate_hls_playlist(&text),
                            Err(e) => {
                                warn!("Failed to read HLS response for {}: {}", url, e);
                                return ValidationStatus::Dead(format!("Read error: {}", e));
                            }
                        }
                    } else {
                        return validate_stream_response(resp).await;
                    }
                }

                if status.is_server_error() {
                    debug!("Server error for {}, retrying...", url);
                    continue;
                }

                if status.is_redirection() {
                    return ValidationStatus::Alive;
                }

                if status.is_client_error() {
                    return ValidationStatus::Dead(format!("Client error {}", status));
                }
            }
            Err(e) => {
                if e.is_timeout() {
                    debug!("Timeout for {}, retrying...", url);
                    continue;
                }

                if e.is_connect() {
                    return ValidationStatus::Dead(format!("Connection error: {}", e));
                }

                if e.is_request() {
                    debug!("Request error for {}, retrying...", url);
                    continue;
                }

                return ValidationStatus::Dead(format!("Error: {}", e));
            }
        }
    }

    ValidationStatus::Dead("Max retries exceeded".to_string())
}

/// Параллельная проверка всех каналов с ограничением конкурентности
pub async fn validate_channels(channels: &mut Vec<Channel>) {
    if channels.is_empty() {
        return;
    }

    info!("Starting validation of {} channels...", channels.len());

    let client = Client::builder()
        .timeout(Duration::from_secs(VAL_CHECK_TIMEOUT_SECS))
        .redirect(reqwest::redirect::Policy::limited(VAL_REDIRECT_LIMIT))
        .build()
        .unwrap_or_else(|_| Client::new());

    let semaphore = Arc::new(Semaphore::new(VAL_MAX_CONCURRENT_CHECKS));

    // Создаем задачи для всех каналов
    let mut handles = Vec::with_capacity(channels.len());
    for channel in channels.iter() {
        let url = channel.url.clone();
        let client = client.clone();
        let sem = semaphore.clone();

        let handle = tokio::spawn(async move {
            let status = check_channel(&client, &url, sem).await;
            (url, status)
        });

        handles.push(handle);
    }

    // Собираем результаты
    let mut stats = ValidationStats::default();

    for handle in handles {
        match handle.await {
            Ok((url, status)) => {
                if let Some(channel) = channels.iter_mut().find(|c| c.url == url) {
                    match status {
                        ValidationStatus::Alive => {
                            channel.is_alive = true;
                            stats.alive += 1;
                        }
                        ValidationStatus::Paid(reason) => {
                            channel.is_alive = false;
                            stats.paid += 1;
                            debug!("Paid channel {}: {}", url, reason);
                        }
                        ValidationStatus::GeoBlocked(reason) => {
                            channel.is_alive = false;
                            stats.geo_blocked += 1;
                            debug!("Geo-blocked channel {}: {}", url, reason);
                        }
                        ValidationStatus::Dead(reason) => {
                            channel.is_alive = false;
                            stats.dead += 1;
                            debug!("Dead channel {}: {}", url, reason);
                        }
                    }
                }
            }
            Err(e) => {
                stats.errors += 1;
                warn!("Task failed: {}", e);
            }
        }
    }

    info!(
        "Validation complete: {} alive, {} dead, {} paid, {} geo-blocked, {} errors ({}% alive)",
        stats.alive,
        stats.dead,
        stats.paid,
        stats.geo_blocked,
        stats.errors,
        stats.alive_percentage() as usize
    );
}

/// Фильтрация нерабочих каналов
pub fn filter_dead_channels(channels: &mut Vec<Channel>) -> usize {
    let initial_len = channels.len();
    channels.retain(|ch| ch.is_alive);
    let removed = initial_len - channels.len();

    if removed > 0 {
        info!("Filtered out {} dead channels", removed);
    }

    removed
}

/// Быстрая проверка канала (HEAD запрос)
pub async fn quick_validate_channel(url: &str) -> bool {
    let client = Client::new();

    match client
        .head(url)
        .timeout(Duration::from_secs(VAL_QUICK_TIMEOUT_SECS))
        .header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36")
        .send()
        .await
    {
        Ok(resp) => {
            let status = resp.status();
            matches!(
                status,
                StatusCode::OK | StatusCode::PARTIAL_CONTENT
            ) || status.is_redirection()
        }
        Err(_) => false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_validation_stats() {
        let mut stats = ValidationStats::default();
        stats.alive = 80;
        stats.dead = 15;
        stats.paid = 3;
        stats.geo_blocked = 2;

        assert_eq!(stats.total(), 100);
        assert!((stats.alive_percentage() - 80.0).abs() < 0.01);
    }

    #[test]
    fn test_validation_stats_empty() {
        let stats = ValidationStats::default();
        assert_eq!(stats.total(), 0);
        assert_eq!(stats.alive_percentage(), 0.0);
    }

    #[test]
    fn test_validate_hls_playlist_valid() {
        // Создаем контент достаточного размера (VAL_MIN_PLAYLIST_SIZE = 150)
        let content = "#EXTM3U\n#EXTINF:-1,Test Channel\nhttp://example.com/stream.ts\n".repeat(3);
        let status = validate_hls_playlist(&content);
        assert_eq!(status, ValidationStatus::Alive);
    }

    #[test]
    fn test_validate_hls_playlist_invalid_tags() {
        let content = "invalid content without tags";
        let status = validate_hls_playlist(content);
        assert!(matches!(status, ValidationStatus::Dead(_)));
    }

    #[test]
    fn test_validate_hls_playlist_too_short() {
        let content = "#EXTM3U\n#EXTINF:-1,Test\nhttp://x";
        let status = validate_hls_playlist(content);
        assert!(matches!(status, ValidationStatus::Dead(_)));
    }

    #[test]
    fn test_validation_status_variants() {
        let alive = ValidationStatus::Alive;
        let dead = ValidationStatus::Dead("error".to_string());
        let paid = ValidationStatus::Paid("403".to_string());
        let geo = ValidationStatus::GeoBlocked("geo".to_string());

        assert_ne!(alive, dead);
        assert_ne!(paid, geo);
    }

    #[test]
    fn test_filter_dead_channels() {
        let mut channels = vec![
            Channel {
                name: "Alive".to_string(),
                url: "http://alive.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: true,
            },
            Channel {
                name: "Dead".to_string(),
                url: "http://dead.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: false,
            },
        ];

        let removed = filter_dead_channels(&mut channels);
        assert_eq!(removed, 1);
        assert_eq!(channels.len(), 1);
        assert!(channels[0].is_alive);
    }

    #[test]
    fn test_filter_dead_channels_all_alive() {
        let mut channels = vec![
            Channel {
                name: "Alive1".to_string(),
                url: "http://alive1.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: true,
            },
            Channel {
                name: "Alive2".to_string(),
                url: "http://alive2.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: true,
            },
        ];

        let removed = filter_dead_channels(&mut channels);
        assert_eq!(removed, 0);
        assert_eq!(channels.len(), 2);
    }

    #[test]
    fn test_filter_dead_channels_all_dead() {
        let mut channels = vec![
            Channel {
                name: "Dead1".to_string(),
                url: "http://dead1.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: false,
            },
            Channel {
                name: "Dead2".to_string(),
                url: "http://dead2.com".to_string(),
                group: "Test".to_string(),
                logo: None,
                is_alive: false,
            },
        ];

        let removed = filter_dead_channels(&mut channels);
        assert_eq!(removed, 2);
        assert!(channels.is_empty());
    }
}