#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
跨境链路质量探针 v1.0
采集 ICMP / TCP / HTTP 延迟、抖动、丢包，输出 CSV 与图表
依赖: pip install ping3 requests pandas matplotlib
"""

import csv
import os
import socket
import statistics
import subprocess
import sys
import time
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import List, Optional

import ping3
import requests
import pandas as pd
import matplotlib.pyplot as plt

# ---------- 配置 ----------
TARGETS = [
    {"name": "HK-IEPL",   "host": "1.1.1.1",   "port": 443, "url": "https://1.1.1.1"},
    {"name": "JP-BGP",    "host": "8.8.8.8",   "port": 443, "url": "https://8.8.8.8"},
    {"name": "LA-CN2GIA", "host": "9.9.9.9",   "port": 443, "url": "https://9.9.9.9"},
]
SAMPLE_COUNT = 20          # 每轮采样次数
INTERVAL_SEC = 0.5         # 采样间隔
ROUND_INTERVAL_SEC = 300   # 轮次间隔（5 分钟）
OUTPUT_DIR = "./probe_data"

# ---------- 数据结构 ----------
@dataclass
class SampleResult:
    timestamp: str
    target: str
    icmp_avg_ms: Optional[float]
    icmp_jitter_ms: Optional[float]
    icmp_loss_pct: float
    tcp_connect_ms: Optional[float]
    http_ttfb_ms: Optional[float]
    http_status: Optional[int]

# ---------- ICMP 探测 ----------
def probe_icmp(host: str, count: int = SAMPLE_COUNT) -> tuple:
    """返回 (avg_ms, jitter_ms, loss_pct)"""
    rtts: List[float] = []
    lost = 0
    for _ in range(count):
        try:
            r = ping3.ping(host, timeout=2, unit="ms")
            if r is None or r is False:
                lost += 1
            else:
                rtts.append(r)
        except Exception:
            lost += 1
        time.sleep(INTERVAL_SEC)

    loss_pct = lost / count * 100
    if not rtts:
        return None, None, loss_pct

    avg = statistics.mean(rtts)
    if len(rtts) > 1:
        diffs = [abs(rtts[i] - rtts[i-1]) for i in range(1, len(rtts))]
        jitter = statistics.mean(diffs)
    else:
        jitter = 0.0
    return avg, jitter, loss_pct

# ---------- TCP 握手探测 ----------
def probe_tcp(host: str, port: int, timeout: float = 3.0) -> Optional[float]:
    """返回 TCP 三次握手耗时 (ms)"""
    try:
        start = time.perf_counter()
        with socket.create_connection((host, port), timeout=timeout):
            elapsed = (time.perf_counter() - start) * 1000
        return elapsed
    except Exception:
        return None

# ---------- HTTP TTFB 探测 ----------
def probe_http(url: str, timeout: float = 5.0) -> tuple:
    """返回 (ttfb_ms, status_code)"""
    try:
        start = time.perf_counter()
        r = requests.get(url, timeout=timeout, stream=True)
        ttfb = (time.perf_counter() - start) * 1000
        r.close()
        return ttfb, r.status_code
    except Exception:
        return None, None

# ---------- 单轮采集 ----------
def run_round() -> List[SampleResult]:
    results: List[SampleResult] = []
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    for t in TARGETS:
        print(f"[{ts}] 探测 {t['name']} ({t['host']}) ...")
        icmp_avg, icmp_jitter, icmp_loss = probe_icmp(t["host"])
        tcp_ms = probe_tcp(t["host"], t["port"])
        ttfb_ms, status = probe_http(t["url"])

        results.append(SampleResult(
            timestamp=ts,
            target=t["name"],
            icmp_avg_ms=round(icmp_avg, 2) if icmp_avg else None,
            icmp_jitter_ms=round(icmp_jitter, 2) if icmp_jitter else None,
            icmp_loss_pct=round(icmp_loss, 2),
            tcp_connect_ms=round(tcp_ms, 2) if tcp_ms else None,
            http_ttfb_ms=round(ttfb_ms, 2) if ttfb_ms else None,
            http_status=status,
        ))
    return results

# ---------- 持久化 ----------
def append_csv(results: List[SampleResult]):
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    path = os.path.join(OUTPUT_DIR, "probe.csv")
    file_exists = os.path.isfile(path)
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(asdict(results[0]).keys()))
        if not file_exists:
            writer.writeheader()
        for r in results:
            writer.writerow(asdict(r))

# ---------- 可视化 ----------
def plot_metrics():
    path = os.path.join(OUTPUT_DIR, "probe.csv")
    if not os.path.isfile(path):
        print("无数据，跳过绘图")
        return
    df = pd.read_csv(path, parse_dates=["timestamp"])
    df["hour"] = df["timestamp"].dt.hour

    fig, axes = plt.subplots(3, 1, figsize=(14, 12))

    for target, group in df.groupby("target"):
        axes[0].plot(group["timestamp"], group["icmp_avg_ms"], label=target, marker=".", linewidth=1)
        axes[1].plot(group["timestamp"], group["icmp_jitter_ms"], label=target, marker=".", linewidth=1)
        axes[2].plot(group["timestamp"], group["icmp_loss_pct"], label=target, marker=".", linewidth=1)

    axes[0].set_title("ICMP 平均延迟 (ms)")
    axes[1].set_title("ICMP 抖动 (ms)")
    axes[2].set_title("ICMP 丢包率 (%)")
    for ax in axes:
        ax.legend()
        ax.grid(alpha=0.3)
        ax.set_xlabel("时间")

    plt.tight_layout()
    out = os.path.join(OUTPUT_DIR, "metrics.png")
    plt.savefig(out, dpi=120)
    print(f"图表已保存: {out}")

# ---------- 主循环 ----------
def main():
    print("=" * 60)
    print("跨境链路质量探针启动")
    print(f"目标数: {len(TARGETS)}, 轮次间隔: {ROUND_INTERVAL_SEC}s")
    print("=" * 60)
    round_no = 0
    try:
        while True:
            round_no += 1
            print(f"\n>>> 第 {round_no} 轮采集")
            results = run_round()
            append_csv(results)
            for r in results:
                print(f"  {r.target}: ICMP {r.icmp_avg_ms}ms / "
                      f"抖动 {r.icmp_jitter_ms}ms / 丢包 {r.icmp_loss_pct}% / "
                      f"TCP {r.tcp_connect_ms}ms / TTFB {r.http_ttfb_ms}ms")
            if round_no % 12 == 0:  # 每 12 轮（1 小时）绘图一次
                plot_metrics()
            time.sleep(ROUND_INTERVAL_SEC)
    except KeyboardInterrupt:
        print("\n采集停止，生成最终图表...")
        plot_metrics()

if __name__ == "__main__":
    main()

# /etc/systemd/system/probe.service
[Unit]
Description=Cross-border Link Probe
After=network.target

[Service]
Type=simple
User=probe
WorkingDirectory=/opt/probe
ExecStart=/usr/bin/python3 /opt/probe/probe.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

import pandas as pd

df = pd.read_csv("probe_data/probe.csv", parse_dates=["timestamp"])
df["hour"] = df["timestamp"].dt.hour

# 晚高峰（19-23点）vs 白天（10-16点）
peak = df[df["hour"].between(19, 23)]
day  = df[df["hour"].between(10, 16)]

summary = df.groupby("target").apply(lambda g: pd.Series({
    "全天均值": g["icmp_avg_ms"].mean(),
    "晚高峰均值": g[g["hour"].between(19,23)]["icmp_avg_ms"].mean(),
    "晚高峰抖动P95": g[g["hour"].between(19,23)]["icmp_jitter_ms"].quantile(0.95),
    "晚高峰丢包均值": g[g["hour"].between(19,23)]["icmp_loss_pct"].mean(),
}))
print(summary)

# 临时生效
sysctl -w net.ipv4.tcp_keepalive_time=60
sysctl -w net.ipv4.tcp_keepalive_intvl=15
sysctl -w net.ipv4.tcp_keepalive_probes=3

# 永久生效
cat >> /etc/sysctl.conf <<EOF
net.ipv4.tcp_keepalive_time = 60
net.ipv4.tcp_keepalive_intvl = 15
net.ipv4.tcp_keepalive_probes = 3
net.ipv4.tcp_fin_timeout = 15
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
EOF
sysctl -p

import asyncio

async def heartbeat(writer, interval=20):
    """每 20 秒发送一个心跳包"""
    while True:
        try:
            writer.write(b"\x00")  # 或协议特定的心跳帧
            await writer.drain()
        except Exception:
            break
        await asyncio.sleep(interval)