chendandan_homework/04_work/scripts/test_serpapi.py

from __future__ import annotations
"""直接调用SerpApi REST接口，验证原始JSON返回结构。"""
import asyncio
import sys
from pathlib import Path
from typing import Any
import httpx
from rich.console import Console
from rich.table import Table

# 直接运行本文件时,Python 只把 scripts/ 加入 sys.path,找不到项目根的 app 包。
# 这里把项目根目录(本文件的上级目录)注入 sys.path,保证直接运行文件也能 import app。
# sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
# 使用 `python -m scripts.test_amap_mcp` 运行时则无需此处理,项目根会自动加入 sys.path。


from app.config import get_settings

SERPAPI_URL = "https://serpapi.com/search.json"

console = Console()

async def serpapi_request(
        client:httpx.AsyncClient,
        params:dict[str,Any],
)->dict[str,Any]:
    """调用SerpAPI，并进行统一错误处理。"""
    response = await client.get(SERPAPI_URL,params=params)
    response.raise_for_status()
    data:dict[str,Any] = response.json()

    if error := data.get("error"):
        raise RuntimeError(f"SerpApi 查询失败：{error}")

    status = data.get("search_metadata",{}).get("status")

    # SerpApi 返回的 status 形如 "Success"(首字母大写),这里做大小写不敏感比较
    if status is not None and str(status).lower() != "success":
        raise RuntimeError(f"SerpApi 返回异常状态：{status}")
    return data

def format_flight_route(flight_option:dict[str,Any])->str:
    """把航班列表转换成可读路线"""

    segments = flight_option.get("flights",[])

    if not segments:
        return "未知路线"

    airport_ids:list[str] = []

    for index,segment in enumerate(segments):
        departure = segment.get("departure_airport",{}).get("id","?")
        arrival = segment.get("arrival_airport",{}).get("id","?")

        if index == 0:
            airport_ids.append(departure)
        airport_ids.append(arrival)

    return " -> ".join(airport_ids)


def format_duration(minutes:Any)->str:
    """把分钟数转成可读时长,如 200分钟 -> 3小时20分。"""
    if not isinstance(minutes,(int,float)):
        return "?"
    total = int(minutes)
    hours = total // 60
    mins = total % 60
    if hours == 0:
        return f"{mins}分钟"
    if mins == 0:
        return f"{hours}小时"
    return f"{hours}小时{mins}分"


async def test_flights(client:httpx.AsyncClient,api_key:str)->None:
    """
    测试上海到成都的真实航班查询。
    第一版使用单程搜索。
    正式项目中，去成和返程各调用一次，
    比使用departure_token 的往返联动流程更容易维护。
    """
    console.rule("[bold]1.SerpApi 航班查询")

    params = {
        "engine":"google_flights",
        "api_key":api_key,
        "departure_id":"SHA,PVG",
        "arrival_id":"CTU,TFU",
        "outbound_date":"2026-08-10",
        "type":2, # 单程
        "travel_class":1,
        "currency":"CNY",
        "gl":"cn",
        "hl":"zh-cn",
    }

    data = await serpapi_request(client,params)

    options = [
        *data.get("best_flights",[]),
        *data.get("other_flights",[]),
    ]

    if not options:
        console.print("[yellow]查询成功，但没有返回航班信息。[/yellow]")
        console.print("返回字段：",list(data.keys()))
        return

    table = Table(title="上海->成都航班候选")
    table.add_column("路线")
    table.add_column("总时长")
    table.add_column("价格")
    table.add_column("航班号")
    table.add_column("出发时间")
    table.add_column("到达时间")

    for option in options:
        segments = option.get("flights",[])
        first_segment = segments[0] if segments else {}
        last_segment = segments[-1] if segments else {}

        # 预先取值，避免 f-string 嵌套双引号(Python 3.9 不支持)
        duration = option.get("total_duration","?")
        price = option.get("price","?")
        flight_numbers = ", ".join(
            seg.get("flight_number","?") for seg in segments
        ) or "?"

        table.add_row(
            format_flight_route(option),
            format_duration(duration),
            f"¥{price}",
            flight_numbers,
            str(first_segment.get("departure_airport",{}).get("time","?")),
            str(last_segment.get("arrival_airport",{}).get("time","?")),
        )

    console.print(table)

    metadata = data.get("search_metadata",{})
    console.print(
        {
            "provider":"serpapi_google_flights",
            "status":metadata.get("status"),
            "search_id":metadata.get("id"),
            "result_count":len(options),
        }
    )



async def test_hotels(client:httpx.AsyncClient,api_key:str)->None:
    """测试成都酒店真实查询。"""

    params = {
        "engine":"google_hotels",
        "api_key":api_key,
        "q":"成都 地铁附近 酒店",
        "check_in_date":"2026-08-10",
        "check_out_date":"2026-08-15",
        "adults":2,
        "children":0,
        "currency":"CNY",
        "gl":"cn",
        "hl":"zh-cn",
    }

    data = await serpapi_request(client,params)
    properties = data.get("properties",[])

    if not properties:
        console.print("[yellow]查询成功，但没有返回酒店候选。[/yellow]")
        console.print("返回字段",list(data.keys()))
        return

    table = Table(title="成都酒店候选")
    table.add_column("酒店")
    table.add_column("星级")
    table.add_column("评分")
    table.add_column("每晚最低价")
    table.add_column("附近地标")

    for hotel in properties[:20]:
        rate = hotel.get("rate_per_night",{})

        # 价格:用 extracted_lowest(纯数字)统一加 ¥,避免和 SerpApi 自带的 ¥ 重复
        extracted = rate.get("extracted_lowest")
        price_text = f"¥{extracted}" if extracted is not None else "?"

        # 附近地标:取 nearby_places 第一项的名称 + 交通耗时
        nearby = hotel.get("nearby_places") or []
        landmark_text = "?"
        if nearby:
            first = nearby[0]
            landmark_name = first.get("name","?")
            transports = first.get("transportations") or []
            duration = transports[0].get("duration","") if transports else ""
            landmark_text = f"{landmark_name}({duration})" if duration else str(landmark_name)

        table.add_row(
            str(hotel.get("name","?")),
            str(hotel.get("hotel_class","?")),
            str(hotel.get("overall_rating","?")),
            price_text,
            landmark_text,
        )

    console.print(table)

    metadata = data.get("search_metadata",{})
    console.print(
        {
            "provider":"serpapi_google_hotels",
            "status":metadata.get("status"),
            "search_id":metadata.get("id"),
            "result_count":len(properties),
        }
    )

async def main()->None:
    settings = get_settings()
    api_key = settings.require("serpapi_api_key")

    time_out = httpx.Timeout(
        timeout = settings.request_timeout_seconds,
        connect = 10.0,
    )

    async with httpx.AsyncClient(timeout = time_out, follow_redirects = True) as client:
        await test_flights(client,api_key)
        await test_hotels(client,api_key)


if __name__=="__main__":
    try:
        asyncio.run(main())
    except httpx.HTTPStatusError as e:
        console.print(
            f"[red]HTTP错误：{e.response.status_code} {e.response.text[:500]}[/red]"
        )
        raise SystemExit(1) from e
    except (httpx.RequestError, RuntimeError) as e:
        console.print(f"[red]测试实效：{e}[/red]")
        raise SystemExit(1) from e