十个互联网医院平台横向对比是怎么做的：指标、采样频次与代码骨架

from __future__ import annotations

import asyncio
import logging
import random
from dataclasses import dataclass, field
from datetime import datetime
from typing import Callable

import httpx

logger = logging.getLogger("probe")


@dataclass(slots=True)
class ProbeResult:
    platform: str
    endpoint: str
    ok: bool
    latency_ms: float
    attempts: int
    detail: str = ""
    sampled_at: str = field(
        default_factory=lambda: datetime.now().isoformat(timespec="seconds")
    )

def setup_logging() -> None:
    fmt = "%(asctime)s %(levelname)s %(name)s: %(message)s"
    logging.basicConfig(
        level=logging.INFO,
        format=fmt,
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler("probe.log", encoding="utf-8"),
        ],
    )

RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})


class ProxyPool:
    """轮询代理列表，每次请求取一个出口 IP。"""

    def __init__(self, proxies: list[str]) -> None:
        self._proxies = proxies
        self._cursor = 0

    def next(self) -> str | None:
        if not self._proxies:
            return None
        proxy = self._proxies[self._cursor % len(self._proxies)]
        self._cursor += 1
        return proxy


async def get_with_retry(
    client: httpx.AsyncClient,
    url: str,
    *,
    pool: ProxyPool | None = None,
    max_attempts: int = 3,
    base_delay: float = 0.5,
) -> httpx.Response:
    last_error: httpx.HTTPError | None = None
    for attempt in range(1, max_attempts + 1):
        proxy = pool.next() if pool else None
        try:
            resp = await client.get(url, timeout=httpx.Timeout(8.0), proxy=proxy)
            if resp.status_code in RETRYABLE_STATUS:
                raise httpx.HTTPStatusError(
                    f"HTTP {resp.status_code}", request=resp.request, response=resp
                )
            return resp
        except httpx.HTTPError as exc:
            last_error = exc
            logger.warning("第 %d 次请求 %s 失败（代理 %s）：%s", attempt, url, proxy, exc)
            if attempt == max_attempts:
                break
            delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.3)
            await asyncio.sleep(delay)
    if last_error is not None:
        raise last_error
    raise RuntimeError("unreachable")

def parse_online_count(data: dict) -> str:
    total = data.get("data", {}).get("total", 0)
    return f"当前在线医生 {total} 人"


async def probe_platform(
    client: httpx.AsyncClient,
    name: str,
    url: str,
    parse: Callable[[dict], str],
    sem: asyncio.Semaphore,
) -> ProbeResult:
    async with sem:
        started = asyncio.get_running_loop().time()
        try:
            resp = await get_with_retry(client, url)
            detail = parse(resp.json())
            elapsed = (asyncio.get_running_loop().time() - started) * 1000
            return ProbeResult(name, url, True, elapsed, 1, detail)
        except (httpx.HTTPError, ValueError, KeyError) as exc:
            return ProbeResult(name, url, False, 0.0, 3, str(exc))

import csv
from dataclasses import asdict, fields


def save_csv(results: list[ProbeResult], path: str) -> None:
    if not results:
        return
    with open(path, "w", newline="", encoding="utf-8-sig") as fh:
        writer = csv.DictWriter(fh, fieldnames=[f.name for f in fields(ProbeResult)])
        writer.writeheader()
        for r in results:
            writer.writerow(asdict(r))

async def main() -> None:
    setup_logging()
    configs = [
        ("微医", "https://api.example.com/v1/online_doctors", parse_online_count),
        ("好大夫在线", "https://api.example.com/v1/online_doctors", parse_online_count),
    ]
    sem = asyncio.Semaphore(2)  # 同一时刻最多 2 个在途请求
    async with httpx.AsyncClient(
        headers={"User-Agent": "personal-research/0.1"},
        follow_redirects=True,
        timeout=httpx.Timeout(8.0),
    ) as client:
        tasks = [
            probe_platform(client, name, url, parse, sem)
            for name, url, parse in configs
        ]
        results = await asyncio.gather(*tasks)
    save_csv(results, "probe_results.csv")
    for r in results:
        state = "正常" if r.ok else "失败"
        logger.info("%s：%s，耗时 %.0f ms，%s", r.platform, state, r.latency_ms, r.detail)


if __name__ == "__main__":
    asyncio.run(main())

def load_proxies(extract_url: str, *, timeout: float = 10.0) -> list[str]:
    resp = httpx.get(extract_url, timeout=timeout)
    resp.raise_for_status()
    proxies = [line.strip() for line in resp.text.splitlines() if line.strip()]
    logger.info("提取代理 %d 个", len(proxies))
    return proxies

EXTRACT_URL = "https://api.example.com/extract?num=20"  # 换成服务商给的提取地址


async def demo(client: httpx.AsyncClient, url: str) -> httpx.Response:
    pool = ProxyPool(load_proxies(EXTRACT_URL))
    return await get_with_retry(client, url, pool=pool)