domainCheck/app/detectors/wayback_detector.py

# -*- coding: UTF-8 -*-
'''
@Project :domainScanDemo
@File    :wayback_detector.py
@IDE     :PyCharm
@Author  :梦伴
@Date    :2026/4/8 23:57
@explain : Wayback检测器
'''

import requests
from app.detectors.base import BaseDetector


class WaybackDetector(BaseDetector):
    """
    Wayback检测器
    """
    
    def __init__(self):
        """
        初始化Wayback检测器
        """
        super().__init__()
        self.cdx_api_url = 'https://web.archive.org/cdx/search/cdx'
        self.snapshot_url = 'https://web.archive.org/web/{timestamp}/{domain}'
    
    def check_domain(self, domain):
        """
        检测域名
        
        :param domain: 域名
        :return: dict - 检测结果
        """
        try:
            # 获取快照年份
            years = self.get_snapshot_years(domain)
            
            # 检查是否包含敏感内容
            has_sensitive = self.has_sensitive_content(domain)
            
            return {
                'snapshot_years': years,
                'has_sensitive_content': has_sensitive
            }
        except Exception as e:
            return self._handle_exception(e, domain)
    
    def get_snapshot_years(self, domain):
        """
        获取快照年份
        
        :param domain: 域名
        :return: list - 快照年份列表
        """
        try:
            params = {
                'url': domain,
                'output': 'json',
                'fl': 'timestamp',
                'filter': 'statuscode:200'
            }
            
            response = requests.get(self.cdx_api_url, params=params, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                years = set()
                
                # 跳过表头
                for item in data[1:]:
                    timestamp = item[0]
                    if len(timestamp) >= 4:
                        year = int(timestamp[:4])
                        years.add(year)
                
                return sorted(years)
            else:
                self._log_warning(f"获取快照年份失败: {response.status_code}")
                return []
        except Exception as e:
            self._handle_exception(e, domain)
            return []
    
    def has_sensitive_content(self, domain):
        """
        检查是否包含敏感内容
        
        :param domain: 域名
        :return: bool - 是否包含敏感内容
        """
        try:
            # 获取最近的快照
            params = {
                'url': domain,
                'output': 'json',
                'fl': 'timestamp',
                'filter': 'statuscode:200',
                'limit': '1'
            }
            
            response = requests.get(self.cdx_api_url, params=params, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                if len(data) > 1:
                    timestamp = data[1][0]
                    snapshot_url = self.snapshot_url.format(timestamp=timestamp, domain=domain)
                    
                    # 获取快照内容
                    snapshot_response = requests.get(snapshot_url, timeout=10)
                    if snapshot_response.status_code == 200:
                        content = snapshot_response.text
                        return self._check_sensitive_words(content)
            
            return False
        except Exception as e:
            self._handle_exception(e, domain)
            return False
    
    def _check_sensitive_words(self, content):
        """
        检查敏感词
        
        :param content: 内容
        :return: bool - 是否包含敏感词
        """
        # 敏感词列表
        sensitive_words = [
            '色情', '赌博', '博彩', '毒品', '暴力', '诈骗',
            '私服', '外挂', '破解', '盗版', '黄色', '反动'
        ]
        
        for word in sensitive_words:
            if word in content:
                return True
        
        return False
    
    def get_backlink_count(self, domain):
        """
        获取友情链接数量
        
        :param domain: 域名
        :return: int - 友情链接数量
        """
        try:
            # 获取最近的快照
            params = {
                'url': domain,
                'output': 'json',
                'fl': 'timestamp',
                'filter': 'statuscode:200',
                'limit': '1'
            }
            
            response = requests.get(self.cdx_api_url, params=params, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                if len(data) > 1:
                    timestamp = data[1][0]
                    snapshot_url = self.snapshot_url.format(timestamp=timestamp, domain=domain)
                    
                    # 获取快照内容
                    snapshot_response = requests.get(snapshot_url, timeout=10)
                    if snapshot_response.status_code == 200:
                        content = snapshot_response.text
                        return self._count_backlinks(content)
            
            return 0
        except Exception as e:
            self._handle_exception(e, domain)
            return 0
    
    def _count_backlinks(self, content):
        """
        统计友情链接数量
        
        :param content: 内容
        :return: int - 友情链接数量
        """
        # 简单的友情链接检测
        import re
        links = re.findall(r'<a\s+href=["\'](https?://[^"\']+)["\']', content)
        
        # 过滤掉同一域名的链接
        domain_links = set()
        for link in links:
            if 'http' in link:
                domain_links.add(link)
        
        return len(domain_links)