spam_filter.py - master - Eugeny/MailReceiptScripts_Python

spam_filter.py
105 строк4 KB
Eugeny
update spam_filter.py
21 дек 2025, 12:55
21 дек 2025, 12:55ae0a712
import subprocess
import re
import os
import tempfile
from typing import Tuple, Optional
from config import Config
from logger import logger

class SpamAssassinChecker:
    def __init__(self):
        self.available = self._check_availability()
        if self.available:
            logger.info(f"SpamAssassin доступен: {Config.SPAMASSASSIN_PATH}")
        else:
            logger.warning("SpamAssassin недоступен, используется fallback метод")
    
    def _check_availability(self) -> bool:
        if not Config.USE_SPAMASSASSIN:
            return False
        try:
            result = subprocess.run(
                [Config.SPAMASSASSIN_PATH, '--version'],
                capture_output=True, timeout=5
            )
            return result.returncode == 0
        except (subprocess.TimeoutExpired, FileNotFoundError, Exception) as e:
            logger.debug(f"SpamAssassin не найден: {e}")
            return False
    
    def check_spam_with_spamc(self, message_bytes: bytes) -> Tuple[bool, float, str]:
        try:
            result = subprocess.run(
                [Config.SPAMASSASSIN_PATH],
                input=message_bytes, capture_output=True, timeout=30
            )
            output = result.stdout.decode('utf-8', errors='ignore')
            return self._parse_output(output)
        except Exception as e:
            logger.error(f"Ошибка spamc: {e}")
            return False, 0.0, ""
    
    def check_spam_with_spamassassin(self, message_bytes: bytes) -> Tuple[bool, float, str]:
        try:
            with tempfile.NamedTemporaryFile(mode='wb', delete=False) as tmp:
                tmp.write(message_bytes)
                tmp_path = tmp.name
            try:
                result = subprocess.run(
                    [Config.SPAMASSASSIN_PATH, '-t', tmp_path],
                    capture_output=True, timeout=30
                )
                output = result.stdout.decode('utf-8', errors='ignore')
                return self._parse_output(output)
            finally:
                if os.path.exists(tmp_path):
                    os.unlink(tmp_path)
        except Exception as e:
            logger.error(f"Ошибка spamassassin: {e}")
            return False, 0.0, ""

    def _parse_output(self, output: str) -> Tuple[bool, float, str]:
        score = 0.0
        is_spam = False
        for line in output.split('\n'):
            if line.startswith('X-Spam-Status:'):
                if 'Yes' in line:
                    is_spam = True
                score_match = re.search(r'score=([\d.]+)', line)
                if score_match:
                    score = float(score_match.group(1))
        return is_spam, score, output
    
    def check(self, message_bytes: bytes) -> Tuple[bool, float]:
        if not self.available:
            return False, 0.0
        
        if 'spamc' in Config.SPAMASSASSIN_PATH:
            is_spam, score, _ = self.check_spam_with_spamc(message_bytes)
        else:
            is_spam, score, _ = self.check_spam_with_spamassassin(message_bytes)
        
        return score >= Config.SPAM_THRESHOLD, score

# Инициализаци¤ чекера
spam_checker = SpamAssassinChecker()

def is_spam_fallback(subject: str, body: str) -> bool:
    text = f"{(subject or '')} {(body or '')}".lower()
    if any(word.lower() in text for word in Config.SPAM_KEYWORDS):
        logger.info("Спам по ключевым словам")
        return True
    if any(pattern.search(text) for pattern in Config.SPAM_PATTERNS):
        logger.info("Спам по регулярным выражениям")
        return True
    return False

def is_spam(message_bytes: bytes, subject: str, body: str) -> Tuple[bool, Optional[float]]:
    if spam_checker.available:
        is_spam_result, score = spam_checker.check(message_bytes)
        if is_spam_result:
            logger.info(f"Спам SpamAssassin (score: {score})")
        return is_spam_result, score
    
    is_spam_result = is_spam_fallback(subject, body)
    return is_spam_result, None