2026年08月03日 谷歌域名防红+QQ微信防红+防反诈屏蔽+APK爆毒 怎么用Python脚本搭建全自动域名检测日志系统？手把手教你每日定时检测、异常告警和历史趋势追踪

不用自动化日志系统，手动跑域名检测到底会漏掉多少关键变化？

手动检测的最大问题不是「不准」，而是「你不可能每天都记得跑」。我们来看一个真实的时间线还原：

怎么用Python写一个能同时检测五大维度的自动化脚本？

下面是一套可以直接部署运行的Python检测脚本。它并行检测谷歌 Safe Browsing、QQ 拦截、微信拦截、运营商反诈DNS屏蔽、APK VirusTotal 五个维度，将结果输出为结构化 JSON 日志。

第一步：安装依赖

pip install requests dnspython schedule

第二步：核心检测逻辑

#!/usr/bin/env python3
"""全平台域名检测日志系统 - 333Check出品"""

import json, os, time, hashlib
from datetime import datetime
from urllib.parse import quote
import requests
import dns.resolver

DOMAINS_FILE = "/etc/domain-check/domains.txt"
LOG_DIR = "/var/log/domain-check"
TELEGRAM_BOT_TOKEN = os.getenv("TG_BOT_TOKEN", "")
TELEGRAM_CHAT_ID = os.getenv("TG_CHAT_ID", "")

def check_google_safe_browsing(domain):
    """谷歌 Safe Browsing API v4 检测"""
    api_key = os.getenv("GOOGLE_SB_API_KEY", "")
    url = "https://safebrowsing.googleapis.com/v4/threatMatches:find?key=" + api_key
    body = {
        "client": {"clientId": "domain-checker", "clientVersion": "1.0"},
        "threatInfo": {
            "threatTypes": ["MALWARE","SOCIAL_ENGINEERING","UNWANTED_SOFTWARE"],
            "platformTypes": ["ANY_PLATFORM"],
            "threatEntryTypes": ["URL"],
            "threatEntries": [{"url": "https://" + domain + "/"}]
        }
    }
    try:
        r = requests.post(url, json=body, timeout=10)
        if r.status_code == 200:
            matches = r.json().get("matches", [])
            return {"status": "blocked" if matches else "clean", "matches": len(matches)}
    except:
        pass
    return {"status": "error", "matches": -1}

def check_qq_wechat_block(domain):
    """QQ/微信 拦截检测（腾讯URL安全API）"""
    api_url = "https://cgi.urlsec.qq.com/index.php?m=url&a=validUrl&url=https://" + domain
    try:
        r = requests.get(api_url, timeout=10,
            headers={"User-Agent": "Mozilla/5.0"})
        data = r.json()
        qq_type = data.get("type", 0)
        # type: 0=安全, 1=未知（审核队列中）, 2=危险
        return {
            "qq": "clean" if qq_type == 0 else ("suspicious" if qq_type == 1 else "blocked"),
            "raw_type": qq_type
        }
    except:
        return {"qq": "error", "raw_type": -1}

def check_antifraud_dns(domain):
    """防反诈DNS屏蔽检测（三大运营商公共DNS对比）"""
    dns_servers = {
        "114DNS": "114.114.114.114",
        "AliDNS": "223.5.5.5",
        "DNSPod": "119.29.29.29"
    }
    results = {}
    for name, ns in dns_servers.items():
        try:
            resolver = dns.resolver.Resolver()
            resolver.nameservers = [ns]
            answers = resolver.resolve(domain, "A", lifetime=5)
            ips = [str(a) for a in answers]
            # 被运营商DNS劫持/拦截的典型特征
            is_blocked = any(
                ip.startswith("127.") or ip.startswith("0.")
                for ip in ips
            )
            results[name] = {
                "status": "blocked" if is_blocked else "clean",
                "resolved_ips": ips
            }
        except dns.resolver.NXDOMAIN:
            results[name] = {"status": "nxdomain"}
        except:
            results[name] = {"status": "error"}
    return results

def check_apk_virustotal(apk_hash):
    """APK VirusTotal检测"""
    api_key = os.getenv("VT_API_KEY", "")
    headers = {"x-apikey": api_key}
    try:
        r = requests.get(
            "https://www.virustotal.com/api/v3/files/" + apk_hash,
            headers=headers, timeout=15)
        if r.status_code == 200:
            stats = r.json()["data"]["attributes"]["last_analysis_stats"]
            return {
                "status": "clean" if stats["malicious"] == 0 else "infected",
                "malicious": stats["malicious"],
                "suspicious": stats["suspicious"],
                "total_engines": sum(stats.values())
            }
    except:
        pass
    return {"status": "error"}

def run_all_checks(domain, apk_hashes=None):
    """对单个域名执行全维度检测"""
    result = {
        "domain": domain,
        "timestamp": datetime.now().isoformat(),
        "ts_unix": int(time.time())
    }

    # 并行检测五大维度
    result["google_sb"] = check_google_safe_browsing(domain)
    result["qq_wechat"] = check_qq_wechat_block(domain)
    result["antifraud_dns"] = check_antifraud_dns(domain)

    if apk_hashes:
        result["apk_vt"] = {}
        for h in apk_hashes:
            result["apk_vt"][h] = check_apk_virustotal(h)

    # 汇总状态
    any_blocked = any([
        result["google_sb"]["status"] == "blocked",
        result["qq_wechat"]["qq"] == "blocked",
        any(v["status"] == "blocked" for v in result["antifraud_dns"].values()),
    ])
    result["summary"] = "BLOCKED" if any_blocked else "clean"

    return result

def save_log(domain, result):
    """保存JSON日志 + 对比上一次状态"""
    os.makedirs(LOG_DIR, exist_ok=True)
    date_str = datetime.now().strftime("%Y-%m-%d")
    log_path = os.path.join(LOG_DIR, date_str + ".json")

    logs = []
    if os.path.exists(log_path):
        with open(log_path) as f:
            logs = json.load(f)

    # 检测状态变化
    if logs:
        prev = logs[-1]
        changes = []
        if result["summary"] != prev["summary"]:
            changes.append("汇总: " + prev["summary"] + " → " + result["summary"])

        if result["google_sb"]["status"] != prev.get("google_sb",{}).get("status"):
            changes.append("谷歌: " + prev.get("google_sb",{}).get("status","?") + " → " + result["google_sb"]["status"])
        if result["qq_wechat"]["qq"] != prev.get("qq_wechat",{}).get("qq"):
            changes.append("QQ/微信: " + prev.get("qq_wechat",{}).get("qq","?") + " → " + result["qq_wechat"]["qq"])

        if changes:
            result["status_changed"] = True
            result["changes"] = changes
            send_telegram_alert(domain, changes)
        else:
            result["status_changed"] = False

    logs.append(result)
    # 只保留最近30天日志（每6小时一条 × 4条/天 × 30天）
    logs = logs[-2880:]
    with open(log_path, "w") as f:
        json.dump(logs, f, ensure_ascii=False, indent=2)

def send_telegram_alert(domain, changes):
    """发送Telegram告警"""
    msg = "🚨 *域名状态变化警报*\n"
    msg += "🌐 域名: `" + domain + "`\n"
    msg += "⏰ 时间: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "\n"
    for c in changes:
        msg += "• " + c + "\n"
    msg += "\n🔗 [333Check快速复检](/fanghong)"
    try:
        requests.post(
            "https://api.telegram.org/bot" + TELEGRAM_BOT_TOKEN + "/sendMessage",
            json={"chat_id": TELEGRAM_CHAT_ID, "text": msg, "parse_mode": "Markdown"},
            timeout=5)
    except:
        pass

if __name__ == "__main__":
    with open(DOMAINS_FILE) as f:
        domains = [line.strip() for line in f if line.strip() and not line.startswith("#")]

    for domain in domains:
        print("[" + datetime.now().strftime("%H:%M:%S") + "] 检测 " + domain + "...")
        result = run_all_checks(domain)
        save_log(domain, result)
        status_emoji = "🟢" if result["summary"]=="clean" else "🔴"
        print("  " + status_emoji + " " + result["summary"])

cron定时任务怎么配置才能平衡检测频率和API配额消耗？

检测频率需要拿捏：太频繁会烧API配额（谷歌Safe Browsing免费配额每月10万次），太低则发现延迟太长。下面四种配置方案覆盖从「省钱」到「高频」的需求：

# 域名检测脚本 - 每6小时执行（标准型）
0 */6 * * * /usr/bin/python3 /opt/domain-check/checker.py >> /var/log/domain-check/runner.log 2>&1

# 每日凌晨3点生成趋势报告
0 3 * * * /usr/bin/python3 /opt/domain-check/report.py

# 每周一凌晨4点清理30天前的旧日志
0 4 * * 1 find /var/log/domain-check/ -name "*.json" -mtime +30 -delete

域名检测日志数据怎么分析才能提前预警「即将被红」的趋势信号？

自动化日志系统的真正价值不只是「发现已经被红了」，而是通过历史日志的趋势分析预警「即将被红」。以下是三个能从日志数据中提取的早期预警信号：

怎么搭建检测日志的可视化趋势仪表盘？

JSON日志文件虽然结构化但不够直观。以下是一个单HTML文件的趋势仪表盘，通过读取日志目录的JSON文件生成30天趋势折线图——纯前端，不需要后端框架，直接把HTML放到Nginx目录里就行：

<!-- trends.html — 域名检测趋势仪表盘 -->
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>域名检测趋势 · 333Check</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
  body{font-family:sans-serif;background:#f8fafd;padding:24px}
  .card{background:#fff;border-radius:14px;padding:24px;margin:16px 0;box-shadow:0 2px 8px rgba(0,0,0,0.04)}
  canvas{max-height:300px}
  .alert-row{display:flex;align-items:center;gap:12px;padding:8px 0;border-bottom:1px solid #eee}
  .dot{width:10px;height:10px;border-radius:50%}
  .dot.red{background:#ef4444}.dot.yellow{background:#f59e0b}.dot.green{background:#059669}
</style>
</head>
<body>
<h1>📊 域名检测趋势仪表盘</h1>
<div class="card">
  <h3>过去30天域名状态分布</h3>
  <canvas id="trendChart"></canvas>
</div>
<div class="card">
  <h3>检测告警时间线</h3>
  <div id="alerts">加载中...</div>
</div>
<script>
// 从API加载JSON日志
fetch('/api/domain-logs?days=30')
  .then(r => r.json())
  .then(logs => {
    const dates = [...new Set(logs.map(l => l.date))];
    const cleanCounts = dates.map(d => logs.filter(l => l.date===d && l.summary==='clean').length);
    const blockedCounts = dates.map(d => logs.filter(l => l.date===d && l.summary==='BLOCKED').length);

    new Chart(document.getElementById('trendChart'), {
      type: 'line',
      data: {
        labels: dates,
        datasets: [
          {label: '干净域名', data: cleanCounts, borderColor: '#059669', tension:0.3},
          {label: '被红域名', data: blockedCounts, borderColor: '#ef4444', tension:0.3}
        ]
      }
    });
  });
</script>
</body>
</html>