local_server.py

"""
OpenCrawl Local — 本地轻量版，无需 R2/积分/认证
直接局域网内通信，结果通过 WebSocket 直传
"""
import json
import uuid
import time
import asyncio
from datetime import date, datetime, timedelta
from pathlib import Path
from urllib.parse import urlparse, quote_plus

import re
import random
import httpx
from urllib.parse import parse_qs
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware


async def safe_parse_body(request: Request) -> dict:
    """兼容各种编码和格式的请求体解析
    - JSON (UTF-8 / GBK / Latin-1)
    - x-www-form-urlencoded
    - 纯文本 JSON
    """
    raw = await request.body()
    if not raw:
        return {}

    content_type = request.headers.get("content-type", "")

    # 1. 尝试 form-urlencoded
    if "x-www-form-urlencoded" in content_type:
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            text = raw.decode("latin-1")
        parsed = parse_qs(text)
        return {k: v[0] if len(v) == 1 else v for k, v in parsed.items()}

    # 2. 尝试多种编码解析 JSON
    for encoding in ("utf-8", "utf-8-sig", "gbk", "gb2312", "latin-1"):
        try:
            text = raw.decode(encoding)
            return json.loads(text)
        except (UnicodeDecodeError, json.JSONDecodeError):
            continue

    # 3. 最后兜底：强制 latin-1 解码后尝试
    try:
        return json.loads(raw.decode("latin-1", errors="replace"))
    except Exception:
        return {}

# ============ 配置 ============
HTTP_PORT = 9878  # 本地版用不同端口，避免和云端冲突
TASK_TIMEOUT = 60
MAX_HISTORY = 500
MIN_WORKER_VERSION = "1.2.0"

def version_gte(v: str, min_v: str) -> bool:
    try:
        return tuple(int(x) for x in v.split(".")) >= tuple(int(x) for x in min_v.split("."))
    except Exception:
        return False

# ============ UA 池 ============
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0",
]

# ============ Search 引擎 ============
SEARCH_ENGINES = {
    "duckduckgo": "https://html.duckduckgo.com/html/?q={}",
    "bing": "https://www.bing.com/search?q={}&setlang=en&cc=us",
    "google": "https://www.google.com/search?q={}",
    "baidu": "https://www.baidu.com/s?wd={}",
}

http_client = httpx.AsyncClient(timeout=15, follow_redirects=True)

def _random_headers():
    return {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7",
        "DNT": "1",
    }


def _build_search_url(q: str, engine: str) -> str:
    template = SEARCH_ENGINES.get(engine, SEARCH_ENGINES["duckduckgo"])
    return template.format(quote_plus(q))


# ============ 按时间段搜索 ============
def _parse_date(s: str) -> date:
    return datetime.strptime(s, "%Y-%m-%d").date()


def _next_month(d: date) -> date:
    y, m = d.year, d.month + 1
    if m > 12:
        y, m = y + 1, 1
    return date(y, m, 1)


def _split_time_windows(start: date, end: date, interval: str):
    """把 [start, end] 切成一组窗口。interval 支持 month/week/day 或 Nd（N 天）。"""
    if end < start:
        raise ValueError("to 必须 >= from")

    m = re.fullmatch(r"(\d+)d", interval)
    step_days = None
    if interval == "day":
        step_days = 1
    elif interval == "week":
        step_days = 7
    elif m:
        step_days = int(m.group(1))
        if step_days < 1:
            raise ValueError("interval 天数必须 >= 1")
    elif interval != "month":
        raise ValueError("interval 仅支持 month/week/day 或 Nd")

    windows = []
    cur = start
    while cur <= end:
        if interval == "month":
            win_end = min(_next_month(cur.replace(day=1)) - timedelta(days=1), end)
        else:
            win_end = min(cur + timedelta(days=step_days - 1), end)
        windows.append((cur, win_end))
        cur = win_end + timedelta(days=1)
    return windows


def _bing_query_hint(start: date, end: date) -> str:
    """Bing 日期过滤 URL 参数已失效，改用查询词偏置。返回附加到 q 的字符串。"""
    if start == end:
        return start.isoformat()
    if start.year == end.year and start.month == end.month:
        return f"{start.year}-{start.month:02d}"
    if start.year == end.year:
        return str(start.year)
    return f"{start.year}..{end.year}"


def _build_search_url_with_range(q: str, engine: str, start: date, end: date) -> str:
    """按引擎追加日期范围参数（Bing 用查询词嵌入代替 URL 参数）。"""
    if engine == "bing":
        hint = _bing_query_hint(start, end)
        return _build_search_url(f"{q} {hint}", "bing")

    base = _build_search_url(q, engine)
    if engine == "google":
        # tbs=cdr:1,cd_min:MM/DD/YYYY,cd_max:MM/DD/YYYY
        cd_min = quote_plus(start.strftime("%m/%d/%Y"))
        cd_max = quote_plus(end.strftime("%m/%d/%Y"))
        return f"{base}&tbs=cdr:1,cd_min:{cd_min},cd_max:{cd_max}"
    if engine == "baidu":
        # gpc=stf=<start_ts>,<end_ts>|stftype=2
        start_ts = int(datetime(start.year, start.month, start.day).timestamp())
        end_dt = datetime(end.year, end.month, end.day) + timedelta(days=1) - timedelta(seconds=1)
        end_ts = int(end_dt.timestamp())
        return f"{base}&gpc=" + quote_plus(f"stf={start_ts},{end_ts}|stftype=2")
    if engine == "duckduckgo":
        # df=YYYY-MM-DD..YYYY-MM-DD
        return f"{base}&df={start.isoformat()}..{end.isoformat()}"
    return base


# ============ 任务 & Worker ============
# task_id -> {url, selector, start_time, future, mode}
tasks: dict = {}
task_history: list = []

# websocket -> {id, join_time, completed, failed, domains, last_pong, active_tasks}
workers: dict = {}


# ============ FastAPI ============
app = FastAPI(title="OpenCrawl Local")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


# ============ 心跳检测 ============
async def heartbeat_checker():
    while True:
        await asyncio.sleep(15)
        now = time.time()
        dead = []
        for ws, info in list(workers.items()):
            if now - info.get("last_pong", info["join_time"]) > 120:
                dead.append((ws, info))
        for ws, info in dead:
            print(f"[Local] Worker {info['id']} 心跳超时，断开")
            workers.pop(ws, None)
            try:
                await ws.close()
            except Exception:
                pass
        if dead:
            await broadcast_status()


# ============ WebSocket Worker 连接 ============
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    worker_id = uuid.uuid4().hex[:8]
    workers[ws] = {
        "id": worker_id,
        "join_time": time.time(),
        "completed": 0,
        "failed": 0,
        "active_tasks": 0,
        "domains": {},
        "last_pong": time.time(),
    }
    print(f"[Local] Worker {worker_id} connected, total: {len(workers)}")
    await broadcast_status()

    try:
        while True:
            data = await ws.receive_text()
            msg = json.loads(data)

            if msg.get("type") == "register":
                client_version = msg.get("version", "0.0.0")
                if not version_gte(client_version, MIN_WORKER_VERSION):
                    await ws.send_text(json.dumps({
                        "type": "update_required",
                        "current": client_version,
                        "required": MIN_WORKER_VERSION,
                    }))
                    workers.pop(ws, None)
                    await broadcast_status()
                    continue

                workers[ws]["version"] = client_version
                client_worker_id = msg.get("workerId")
                if client_worker_id:
                    workers[ws]["client_id"] = client_worker_id
                    for old_ws, old_info in list(workers.items()):
                        if old_ws is not ws and old_info.get("client_id") == client_worker_id:
                            print(f"[Local] 移除幽灵 Worker {old_info['id']}")
                            workers.pop(old_ws, None)
                            # 不主动 close 旧连接，避免触发扩展端连锁重连
                            # 旧连接会在心跳超时后自然断开
                print(f"[Local] Worker {worker_id} registered")

            elif msg.get("type") == "taskComplete" and msg.get("taskId") in tasks:
                task_id = msg["taskId"]
                task = tasks.pop(task_id)
                worker = workers.get(ws, {})
                if worker:
                    worker["active_tasks"] = max(0, worker.get("active_tasks", 0) - 1)
                duration = time.time() - task["start_time"]

                entry = {
                    "taskId": task_id,
                    "url": task["url"],
                    "status": "failed" if msg.get("error") else "success",
                    "error": msg.get("error"),
                    "workerId": worker.get("id"),
                    "startTime": task["start_time"],
                    "duration": round(duration * 1000),
                }

                if msg.get("error"):
                    worker["failed"] = worker.get("failed", 0) + 1
                    task["future"].set_result({"error": msg["error"]})
                else:
                    # 本地版：直接从 WebSocket 消息获取数据
                    result_data = msg.get("data")
                    if result_data:
                        worker["completed"] = worker.get("completed", 0) + 1
                        task["future"].set_result({"data": result_data})
                    else:
                        entry["status"] = "failed"
                        entry["error"] = "Worker 未返回数据"
                        worker["failed"] = worker.get("failed", 0) + 1
                        task["future"].set_result({"error": "Worker 未返回数据"})

                task_history.append(entry)
                if len(task_history) > MAX_HISTORY:
                    task_history.pop(0)
                await broadcast_status()

            elif msg.get("type") == "heartbeat":
                if ws in workers:
                    workers[ws]["last_pong"] = time.time()
                await ws.send_text(json.dumps({"type": "pong"}))

    except WebSocketDisconnect:
        pass
    except Exception as e:
        print(f"[Local] Worker {worker_id} error: {e}")
    finally:
        workers.pop(ws, None)
        print(f"[Local] Worker {worker_id} disconnected, total: {len(workers)}")
        await broadcast_status()


def select_worker(target_domain: str):
    best = None
    best_score = float("inf")
    for ws, info in workers.items():
        active = info.get("active_tasks", 0)
        domain_count = info["domains"].get(target_domain, 0)
        score = active * 100 + domain_count
        if score < best_score:
            best_score = score
            best = ws
    return best


async def dispatch(ws: WebSocket, task_id, url, selector, mode="full", link_filter=None):
    worker = workers.get(ws)
    if worker:
        worker["active_tasks"] = worker.get("active_tasks", 0) + 1

    # 本地版：不发 uploadUrl，Worker 会直接通过 WS 回传数据
    await ws.send_text(json.dumps({
        "type": "task", "taskId": task_id,
        "url": url, "selector": selector, "mode": mode,
        "linkFilter": link_filter,
    }))
    if worker:
        domain = urlparse(url).hostname
        worker["domains"][domain] = worker["domains"].get(domain, 0) + 1


async def broadcast_status():
    msg = json.dumps({
        "type": "status",
        "workers": len(workers),
        "activeTasks": [
            {"taskId": tid, "url": t["url"], "startTime": t["start_time"]}
            for tid, t in tasks.items()
        ],
        "totalCompleted": sum(1 for h in task_history if h["status"] == "success"),
        "totalFailed": sum(1 for h in task_history if h["status"] == "failed"),
        "recentHistory": task_history[-20:],
    })
    for ws in list(workers.keys()):
        try:
            await ws.send_text(msg)
        except Exception:
            pass


async def crawl(url: str, selector: str = None, mode: str = "full", link_filter: str = None):
    if not workers:
        raise HTTPException(503, detail="没有可用的 Worker")

    domain = urlparse(url).hostname
    ws = select_worker(domain)
    if not ws:
        raise HTTPException(503, detail="没有可用的 Worker")

    task_id = uuid.uuid4().hex
    future = asyncio.get_event_loop().create_future()
    timeout = 30 if mode == "lite" else (120 if mode == "download" else TASK_TIMEOUT)

    tasks[task_id] = {
        "url": url, "selector": selector,
        "start_time": time.time(), "future": future,
        "mode": mode,
    }

    await dispatch(ws, task_id, url, selector, mode, link_filter)

    try:
        result = await asyncio.wait_for(future, timeout=timeout)
    except asyncio.TimeoutError:
        tasks.pop(task_id, None)
        raise HTTPException(504, detail="任务超时")

    if "error" in result:
        raise HTTPException(500, detail=result["error"])

    return result


# ============ HTTP API ============
@app.get("/", response_class=HTMLResponse)
async def index():
    return HTMLResponse("""
    <html><head><meta charset="UTF-8"><title>OpenCrawl Local</title>
    <style>body{background:#0d1117;color:#e6edf3;font-family:system-ui;display:flex;justify-content:center;align-items:center;min-height:100vh}
    .c{text-align:center}h1{font-size:2em}h1 span{color:#58a6ff}.sub{color:#8b949e;margin-top:8px}
    </style></head>
    <body><div class="c"><h1><span>Open</span>Crawl Local</h1><p class="sub">本地轻量版运行中</p></div></body></html>
    """)


@app.post("/api/crawl")
async def api_crawl_post(request: Request):
    body = await safe_parse_body(request)
    if not body.get("url"):
        return JSONResponse({"success": False, "error": "缺少 url"}, 400)

    url = body["url"]
    mode = body.get("mode", "full")
    selector = body.get("selector")
    link_filter = body.get("linkFilter")

    result = await crawl(url, selector, mode, link_filter)
    return _format_crawl_response(result["data"], url, mode)


@app.get("/api/crawl")
async def api_crawl_get(request: Request, url: str = None, selector: str = None, mode: str = "full", linkFilter: str = None):
    if not url:
        return JSONResponse({"success": False, "error": "缺少 url"}, 400)

    result = await crawl(url, selector, mode, linkFilter)
    return _format_crawl_response(result["data"], url, mode)


def _format_crawl_response(data, url, mode):
    """统一格式化 crawl 响应，data 始终为字符串，links 单独字段"""
    # data 可能是纯文本字符串，也可能是 JSON 字符串 {"text": ..., "links": [...]}
    parsed = None
    if isinstance(data, str):
        try:
            parsed = json.loads(data)
        except (json.JSONDecodeError, ValueError):
            pass
    elif isinstance(data, dict):
        parsed = data

    if isinstance(parsed, dict) and "text" in parsed:
        resp = {"success": True, "url": url, "mode": mode, "data": parsed["text"]}
        if parsed.get("links"):
            resp["links"] = parsed["links"]
        return resp

    # 纯文本或其他格式，保持原样
    text = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False) if data else ""
    return {"success": True, "url": url, "mode": mode, "data": text}


# ============ Search API ============
def _merge_results(all_results, sources) -> list:
    seen_urls = set()
    merged = []
    for results, source in zip(all_results, sources):
        for r in results:
            url = r.get("url", "")
            norm = url.rstrip("/").lower().split("?")[0]
            if norm in seen_urls or not url:
                continue
            seen_urls.add(norm)
            r["source"] = source
            merged.append(r)
    return merged


async def _do_search(q: str, mode: str):
    if mode == "full":
        engines = ["duckduckgo", "bing", "google", "baidu"]
        coros = []
        for eng in engines:
            url = _build_search_url(q, eng)
            coros.append(crawl(url, "__search__", "full"))

        try:
            results_raw = await asyncio.wait_for(
                asyncio.gather(*coros, return_exceptions=True),
                timeout=45
            )
        except asyncio.TimeoutError:
            results_raw = [TimeoutError("总超时")] * len(engines)

        all_results = []
        for i, r in enumerate(results_raw):
            if isinstance(r, (Exception, BaseException)):
                print(f"[Local] Search {engines[i]} failed: {type(r).__name__}: {r}")
                all_results.append([])
            else:
                data_str = r.get("data", "")
                try:
                    items = json.loads(data_str) if isinstance(data_str, str) and data_str.startswith("[") else (data_str if isinstance(data_str, list) else [])
                except Exception:
                    items = []
                print(f"[Local] Search {engines[i]}: {len(items)} results")
                all_results.append(items)

        merged = _merge_results(all_results, engines)
        return merged, engines
    else:
        url = _build_search_url(q, "duckduckgo")
        result = await crawl(url, "__search__", "full")
        data_str = result.get("data", "")
        try:
            items = json.loads(data_str) if isinstance(data_str, str) and data_str.startswith("[") else (data_str if isinstance(data_str, list) else [])
        except Exception:
            items = []
        return items, ["duckduckgo"]


@app.post("/api/search")
async def api_search_post(request: Request):
    body = await safe_parse_body(request)
    q = body.get("q", "").strip()
    if not q:
        return JSONResponse({"success": False, "error": "缺少搜索词 q"}, 400)

    mode = body.get("mode", "lite")
    results, engines = await _do_search(q, mode)

    return {
        "success": True,
        "query": q,
        "type": "search",
        "mode": mode,
        "engines": engines,
        "web": {"results": results[:30]},
    }


@app.get("/api/search")
async def api_search_get(request: Request, q: str = None, mode: str = "lite"):
    if not q:
        return JSONResponse({"success": False, "error": "缺少搜索词 q"}, 400)

    results, engines = await _do_search(q, mode)

    return {
        "success": True,
        "query": q,
        "type": "search",
        "mode": mode,
        "engines": engines,
        "web": {"results": results[:30]},
    }


@app.get("/api/download")
async def api_download(url: str = None):
    """浏览器通道文件下载（PDF 等）：由扩展在 Chrome 上下文 fetch，返回 base64。
    用于服务端直连被 WAF/反爬拦截（403 等）的场景。"""
    if not url:
        return JSONResponse({"success": False, "error": "缺少 url"}, 400)
    try:
        result = await crawl(url, "__download__", "download")
    except HTTPException as e:
        return JSONResponse({"success": False, "error": str(e.detail)}, e.status_code)
    if result.get("error"):
        return JSONResponse({"success": False, "error": result["error"]})
    try:
        payload = json.loads(result.get("data") or "{}")
    except Exception:
        payload = {}
    if not payload.get("b64"):
        return JSONResponse({"success": False, "error": "Worker 未返回文件数据"})
    return {
        "success": True,
        "content_type": payload.get("contentType", ""),
        "size": payload.get("size", 0),
        "data_b64": payload["b64"],
    }


DEFAULT_RANGE_ENGINES = ["duckduckgo", "bing", "google", "baidu"]
RANGE_CONCURRENCY_DEFAULT = 6
RANGE_MAX_WINDOWS = 60


async def _search_one_engine(sem: asyncio.Semaphore, q: str, engine: str, start: date, end: date):
    """单窗口 × 单引擎，返回 (engine, items, error)。异常吃掉。"""
    url = _build_search_url_with_range(q, engine, start, end)
    async with sem:
        try:
            result = await crawl(url, "__search__", "full")
        except HTTPException as e:
            return engine, [], f"{e.status_code}: {e.detail}"
        except Exception as e:
            return engine, [], f"{type(e).__name__}: {e}"

    data_str = result.get("data", "")
    try:
        if isinstance(data_str, str) and data_str.startswith("["):
            items = json.loads(data_str)
        elif isinstance(data_str, list):
            items = data_str
        else:
            items = []
    except Exception:
        items = []
    return engine, items, None


async def _do_search_range(q: str, from_d: date, to_d: date, interval: str, engines: list, concurrency: int):
    windows = _split_time_windows(from_d, to_d, interval)
    if len(windows) > RANGE_MAX_WINDOWS:
        raise HTTPException(400, detail=f"时间窗数量 {len(windows)} 超过上限 {RANGE_MAX_WINDOWS}，请增大 interval 或缩小范围")

    sem = asyncio.Semaphore(max(1, concurrency))
    # 扁平化所有 (window_idx, engine) 任务，全局并发受 sem 控制
    task_meta = []  # (wi, engine)
    coros = []
    for wi, (start, end) in enumerate(windows):
        for eng in engines:
            task_meta.append((wi, eng))
            coros.append(_search_one_engine(sem, q, eng, start, end))

    flat_results = await asyncio.gather(*coros)

    # 按窗口聚合
    per_window = {wi: [] for wi in range(len(windows))}
    for (wi, _eng_meta), (eng, items, err) in zip(task_meta, flat_results):
        per_window[wi].append((eng, items, err))

    buckets = []
    for wi, (start, end) in enumerate(windows):
        engine_counts = {}
        errors = {}
        per_engine_lists = []
        source_labels = []
        for eng, items, err in per_window[wi]:
            engine_counts[eng] = len(items)
            if err:
                errors[eng] = err
            per_engine_lists.append(items)
            source_labels.append(eng)
        merged = _merge_results(per_engine_lists, source_labels)
        bucket = {
            "from": start.isoformat(),
            "to": end.isoformat(),
            "engineCounts": engine_counts,
            "results": merged,
        }
        if errors:
            bucket["errors"] = errors
        buckets.append(bucket)

    return buckets


@app.post("/api/search_range")
async def api_search_range(request: Request):
    body = await safe_parse_body(request)
    q = (body.get("q") or "").strip()
    if not q:
        return JSONResponse({"success": False, "error": "缺少搜索词 q"}, 400)

    try:
        to_d = _parse_date(body["to"]) if body.get("to") else date.today()
        from_d = _parse_date(body["from"]) if body.get("from") else date(to_d.year - 1, to_d.month, 1)
    except ValueError as e:
        return JSONResponse({"success": False, "error": f"日期格式错误（YYYY-MM-DD）: {e}"}, 400)

    interval = (body.get("interval") or "month").strip()
    engines = body.get("engines") or DEFAULT_RANGE_ENGINES
    if isinstance(engines, str):
        engines = [e.strip() for e in engines.split(",") if e.strip()]
    engines = [e for e in engines if e in SEARCH_ENGINES]
    if not engines:
        return JSONResponse({"success": False, "error": "engines 无效"}, 400)

    try:
        concurrency = int(body.get("concurrency") or RANGE_CONCURRENCY_DEFAULT)
    except (TypeError, ValueError):
        concurrency = RANGE_CONCURRENCY_DEFAULT

    if not workers:
        return JSONResponse({"success": False, "error": "没有可用的 Worker"}, 503)

    try:
        buckets = await _do_search_range(q, from_d, to_d, interval, engines, concurrency)
    except HTTPException as e:
        return JSONResponse({"success": False, "error": e.detail}, e.status_code)
    except ValueError as e:
        return JSONResponse({"success": False, "error": str(e)}, 400)

    notes = []
    if "bing" in engines:
        notes.append("bing 的 URL 日期过滤已失效，改用把年月/日期嵌入查询词做软偏置，结果非严格按时间过滤")

    return {
        "success": True,
        "query": q,
        "from": from_d.isoformat(),
        "to": to_d.isoformat(),
        "interval": interval,
        "engines": engines,
        "concurrency": concurrency,
        "notes": notes,
        "buckets": buckets,
    }


@app.get("/api/status")
async def api_status():
    worker_list = []
    for info in workers.values():
        worker_list.append({
            "id": info["id"],
            "completed": info["completed"],
            "failed": info["failed"],
            "uptime": round(time.time() - info["join_time"]),
        })
    return {
        "workers": len(workers),
        "workerList": worker_list,
        "activeTasks": len(tasks),
        "totalCompleted": sum(1 for h in task_history if h["status"] == "success"),
        "totalFailed": sum(1 for h in task_history if h["status"] == "failed"),
        "history": task_history[-50:],
    }


# ============ 启动 ============
@app.on_event("startup")
async def startup():
    print(f"[OpenCrawl Local] 本地版运行中")
    print(f"[OpenCrawl Local] API: http://0.0.0.0:{HTTP_PORT}")
    print(f"[OpenCrawl Local] WebSocket: ws://0.0.0.0:{HTTP_PORT}/ws")
    print()
    print("[OpenCrawl Local] API (无需认证):")
    print("  POST /api/crawl        {{url, selector?, mode?, linkFilter?}}")
    print("  POST /api/search       {{q, mode?}}")
    print("  POST /api/search_range {{q, from?, to?, interval?, engines?, concurrency?}}")
    print("  GET  /api/status       平台状态")
    asyncio.create_task(heartbeat_checker())


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=HTTP_PORT)