# -*- coding: UTF-8 -*- ''' @Project :domainScanDemo @File :detect_engine.py @IDE :PyCharm @Author :梦伴 @Date :2026/4/8 23:52 @explain : 检测引擎 ''' import time from loguru import logger from app.utils.database import Database from app.utils.detection_results import normalize_detector_result from app.detectors.rdap_detector import RDAPDetector from app.detectors.wayback_detector import WaybackDetector from app.detectors.baidu_detector import BaiduDetector from app.detectors.qihu360_detector import Qihu360Detector from app.detectors.google_detector import GoogleDetector from app.detectors.chinaz_detector import ChinazDetector from app.detectors.aizhan_detector import AizhanDetector from app.detectors.juziseo_detector import JuziseoDetector from app.detectors.jucha_detector import JuchaDetector from app.utils.status_codes import ( DETECT_STATUS_BLACKLISTED, DETECT_STATUS_COMPLETED, DETECT_STATUS_FAILED, DETECT_STATUS_RUNNING, ) class DetectEngine: """ 检测引擎 """ OUTCOME_SUCCESS = "success" OUTCOME_BLACKLISTED = "blacklisted" OUTCOME_FAILED = "failed" def __init__(self): """ 初始化检测引擎 """ self.db = Database() self.rdap_detector = RDAPDetector() self.wayback_detector = WaybackDetector() self.baidu_detector = BaiduDetector() self.qihu360_detector = Qihu360Detector() self.google_detector = GoogleDetector() self.chinaz_detector = ChinazDetector() self.aizhan_detector = AizhanDetector() self.juziseo_detector = JuziseoDetector() self.jucha_detector = JuchaDetector() def detect_domain(self, domain_id): return self._detect_domain_with_outcome(domain_id) == self.OUTCOME_SUCCESS def _detect_domain_with_outcome(self, domain_id): """ 检测域名 :param domain_id: 域名ID :return: bool - 是否检测成功 """ try: # 获取域名信息 domain_info = self.db.get_domain_by_id(domain_id) if not domain_info: logger.error(f"域名不存在: {domain_id}") return False domain = domain_info['domain'] logger.info(f"开始检测域名: {domain}") # 更新检测状态为检测中 self.db.update_domain_detect_status(domain_id, DETECT_STATUS_RUNNING) # 1. 基础检测 basic_outcome = self._basic_detect(domain_id, domain) if basic_outcome != self.OUTCOME_SUCCESS: logger.info(f"基础检测失败，停止后续检测: {domain}") if basic_outcome == self.OUTCOME_FAILED: self.db.update_domain_detect_status(domain_id, DETECT_STATUS_FAILED) return basic_outcome # 2. 深度检测 deep_outcome = self._deep_detect(domain_id, domain) if deep_outcome != self.OUTCOME_SUCCESS: logger.info(f"深度检测失败: {domain}") if deep_outcome == self.OUTCOME_FAILED: self.db.update_domain_detect_status(domain_id, DETECT_STATUS_FAILED) return deep_outcome # 更新检测状态为正常 self.db.update_domain_detect_status(domain_id, DETECT_STATUS_COMPLETED) logger.info(f"域名检测完成: {domain}") return self.OUTCOME_SUCCESS except Exception as e: logger.error(f"检测域名出错: {e}") # 更新检测状态为检测失败 self.db.update_domain_detect_status(domain_id, DETECT_STATUS_FAILED) return self.OUTCOME_FAILED def _basic_detect(self, domain_id, domain): """ 基础检测 :param domain_id: 域名ID :param domain: 域名 :return: str - 检测结果 """ # 1. 检查是否为一口价域名 is_ykj = self.db.is_ykj_domain(domain_id) # 2. 注册状态检测（一口价域名跳过） if not is_ykj: register_status = self.rdap_detector.check_register_status(domain) self.db.update_domain_register_status(domain_id, register_status) # 3. 黑名单缓存检查 if self.db.is_blacklisted(domain): logger.info(f"域名在黑名单中: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) return self.OUTCOME_BLACKLISTED # 4. 时光机快照年份采集 snapshot_years = self.wayback_detector.get_snapshot_years(domain) if snapshot_years: self.db.update_domain_snapshot_years(domain_id, ','.join(map(str, snapshot_years))) # 5. 时光机正文抽样与敏感词匹配 if self.wayback_detector.has_sensitive_content(domain): logger.info(f"域名包含敏感词: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "快照包含敏感词") return self.OUTCOME_BLACKLISTED return self.OUTCOME_SUCCESS def _deep_detect(self, domain_id, domain): """ 深度检测 :param domain_id: 域名ID :param domain: 域名 :return: str - 检测结果 """ detector_results = {} detector_steps = [ ("baidu_history", lambda: self.baidu_detector.check_history(domain)), ("baidu_site", lambda: self.baidu_detector.check_site(domain)), ("qihu360_site", lambda: self.qihu360_detector.check_site(domain)), ("google_site", lambda: self.google_detector.check_site(domain)), ("chinaz_info", lambda: self.chinaz_detector.check_domain(domain)), ("aizhan_info", lambda: self.aizhan_detector.check_domain(domain)), ("juziseo_info", lambda: self.juziseo_detector.check_domain(domain)), ("jucha_info", lambda: self.jucha_detector.check_domain(domain)), ] for detector_name, runner in detector_steps: detector_results[detector_name] = normalize_detector_result(detector_name, runner()) detector_error = detector_results[detector_name].get("error") if detector_error: logger.error( f"深度检测存在第三方检测错误: {domain}, detector={detector_name}, error={detector_error}" ) return self.OUTCOME_FAILED if self._check_risk( domain_id, domain, detector_results.get("baidu_history"), detector_results.get("baidu_site"), detector_results.get("qihu360_site"), detector_results.get("google_site"), detector_results.get("chinaz_info"), detector_results.get("aizhan_info"), detector_results.get("juziseo_info"), detector_results.get("jucha_info"), ): self._persist_detection_results(domain_id, detector_results) return self.OUTCOME_BLACKLISTED # 保存检测结果 persisted = self._persist_detection_results(domain_id, detector_results) if not persisted: logger.error(f"保存检测结果失败: {domain}") return self.OUTCOME_FAILED return self.OUTCOME_SUCCESS def _check_risk(self, domain_id, domain, baidu_history, baidu_site, qihu360_site, google_site, chinaz_info, aizhan_info, juziseo_info, jucha_info): """ 检查风险 :param domain_id: 域名ID :param domain: 域名 :param baidu_history: 百度历史 :param baidu_site: 百度site :param qihu360_site: 360 site :param google_site: Google site :param chinaz_info: 站长之家信息 :param aizhan_info: 爱站网信息 :param juziseo_info: 桔子SEO信息 :param jucha_info: 聚查信息 :return: bool - 是否有风险 """ # 检查百度历史过灰 if ( isinstance(baidu_history, dict) and baidu_history.get('has_gray') ) or (baidu_history and '灰' in str(baidu_history)): logger.info(f"百度历史过灰: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "百度历史过灰") return True # 检查标题敏感词 if chinaz_info and 'title' in chinaz_info: if self._contains_sensitive_words(chinaz_info['title']): logger.info(f"标题包含敏感词: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "标题包含敏感词") return True # 检查子域名 if baidu_site and 'subdomains' in baidu_site: subdomains = baidu_site['subdomains'] # 排除 www, @, m valid_subdomains = [sub for sub in subdomains if sub not in ['www', '@', 'm']] if valid_subdomains: logger.info(f"存在子域名: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "存在子域名") return True # 检查风险提示 if aizhan_info and 'risk' in aizhan_info: if aizhan_info['risk'] in ['低风险', '疑似色情博彩风险', '严重影响权重']: logger.info(f"风险提示: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, f"风险提示: {aizhan_info['risk']}") return True # 检查WHOIS状态 if jucha_info and 'whois' in jucha_info: if jucha_info['whois'].get('whois_status') in ['clientHold', 'serverHold']: logger.info(f"WHOIS状态异常: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "WHOIS状态异常") return True # 检查拦截检测 if jucha_info and 'intercept' in jucha_info: if not jucha_info['intercept'].get('normal', True): logger.info(f"拦截检测异常: {domain}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_BLACKLISTED) self.db.add_to_blacklist(domain, "拦截检测异常") return True return False def _persist_detection_results(self, domain_id, detector_results): return self.db.add_detection_result( domain_id, detector_results.get("baidu_history"), detector_results.get("baidu_site"), detector_results.get("qihu360_site"), detector_results.get("google_site"), detector_results.get("chinaz_info"), detector_results.get("aizhan_info"), detector_results.get("juziseo_info"), detector_results.get("jucha_info"), ) def _contains_sensitive_words(self, text): """ 检查文本是否包含敏感词 :param text: 文本 :return: bool - 是否包含敏感词 """ sensitive_words = self.db.get_all_sensitive_words() for word in sensitive_words: if word in text: return True return False def process_task(self, task_id): """ 处理检测任务 :param task_id: 任务ID :return: bool - 是否处理成功 """ try: # 获取任务信息 task = self.db.get_task_by_id(task_id) if not task: logger.error(f"任务不存在: {task_id}") return False domain_id = task['domain_id'] # 更新任务状态为执行中 self.db.update_task_status(task_id, 1) # 1 表示执行中 # 执行检测 outcome = self._detect_domain_with_outcome(domain_id) # 更新任务状态 if outcome in (self.OUTCOME_SUCCESS, self.OUTCOME_BLACKLISTED): self.db.update_task_status(task_id, 2) # 2 表示完成 return True else: # 增加重试次数 retry_count = task.get('retry_count', 0) + 1 if retry_count < 3: self.db.update_task_retry_count(task_id, retry_count) self.db.update_task_status(task_id, 0) # 0 表示待执行 else: self.db.update_task_status(task_id, 3) # 3 表示失败 return False except Exception as e: logger.error(f"处理任务出错: {e}") # 更新任务状态为失败 self.db.update_task_status(task_id, 3) # 3 表示失败 return False