#抓取监控看板 - 爬虫系统实时监控与告警详解

# scrapy_project/middlewares.py
from prometheus_client import start_http_server, Counter, Gauge, Histogram
import time

# ---------- 全局指标定义 ----------
REQUESTS = Counter(
    'scrapy_requests_total', '总请求数',
    ['spider', 'status']
)
ERRORS = Counter(
    'scrapy_errors_total', '总错误数',
    ['spider', 'error_type']
)
ACTIVE_CONCURRENCY = Gauge(
    'scrapy_active_concurrency', '当前活跃请求数',
    ['spider']
)
RESPONSE_TIME = Histogram(
    'scrapy_response_time_seconds', '响应时间分布',
    ['spider'],
    buckets=[0.1, 0.5, 1, 3, 10]
)

class PrometheusMetricsMiddleware:
    """
    Scrapy 下载器中间件，用于向 Prometheus 暴露抓取指标。
    启动一个 HTTP 服务，专门让 Prometheus server 拉取。
    """
    def __init__(self, settings):
        self.port = settings.getint('PROMETHEUS_METRICS_PORT', 8000)
        # 启动指标暴露服务（与 Scrapy 进程共享同一进程）
        start_http_server(self.port)

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings)

    def process_request(self, request, spider):
        ACTIVE_CONCURRENCY.labels(spider=spider.name).inc()
        request.meta['start_time'] = time.time()

    def process_response(self, request, response, spider):
        if 'start_time' in request.meta:
            duration = time.time() - request.meta['start_time']
            RESPONSE_TIME.labels(spider=spider.name).observe(duration)
        ACTIVE_CONCURRENCY.labels(spider=spider.name).dec()
        REQUESTS.labels(
            spider=spider.name, status=str(response.status)
        ).inc()
        return response

    def process_exception(self, request, exception, spider):
        ACTIVE_CONCURRENCY.labels(spider=spider.name).dec()
        error_type = type(exception).__name__
        ERRORS.labels(spider=spider.name, error_type=error_type).inc()

DOWNLOADER_MIDDLEWARES = {
    'scrapy_project.middlewares.PrometheusMetricsMiddleware': 100,
}
PROMETHEUS_METRICS_PORT = 8001

# diagnostic_tools.py
import psutil
import requests
import socket
import time
from datetime import datetime
import logging
from typing import Dict, List, Optional
import json

class DiagnosticTools:
    """爬虫故障诊断工具（轻量版）"""

    def __init__(self):
        self.logger = logging.getLogger(__name__)

    def check_system_resources(self) -> Dict:
        """检查系统资源，超过 85% 自动标记告警"""
        cpu = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory().percent
        disk = psutil.disk_usage('/').percent
        procs = len(psutil.pids())

        warnings = []
        if cpu > 85:
            warnings.append('CPU 使用率过高')
        if mem > 85:
            warnings.append('内存使用率过高')
        if disk > 85:
            warnings.append('磁盘空间不足')

        return {
            'cpu_percent': cpu,
            'memory_percent': mem,
            'disk_percent': disk,
            'process_count': procs,
            'warnings': warnings
        }

    def check_network_connectivity(self, urls: List[str]) -> Dict:
        """测试目标站点或中间件的网络可达性"""
        results = {}
        for url in urls:
            try:
                start = time.time()
                resp = requests.head(url, timeout=5, allow_redirects=True)
                results[url] = {
                    'status': 'success',
                    'code': resp.status_code,
                    'latency_seconds': round(time.time() - start, 3)
                }
            except Exception as e:
                results[url] = {
                    'status': 'failed',
                    'reason': str(e)
                }
        return results

    def check_port_availability(self, host: str, ports: List[int]) -> Dict:
        """检查关键端口（Scrapyd 6800，Redis 6379，指标 8001 等）"""
        status = {}
        for port in ports:
            sock = None
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(2)
                sock.connect((host, port))
                status[port] = 'open'
            except Exception:
                status[port] = 'closed'
            finally:
                if sock:
                    sock.close()
        return status

    def run_full_diagnosis(self, spider_hosts: List[str] = None) -> str:
        """生成完整的可读诊断报告"""
        report = [f"# 爬虫诊断报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
        report.append("\n## 系统资源")
        sys_res = self.check_system_resources()
        report.append(json.dumps(sys_res, indent=2, ensure_ascii=False))
        report.append("\n## 关键端口（本地）")
        port_res = self.check_port_availability('localhost', [6800, 6379, 8001])
        report.append(json.dumps(port_res, indent=2))
        return "\n".join(report)

# 快速命令行自检
if __name__ == "__main__":
    diag = DiagnosticTools()
    print(diag.run_full_diagnosis())