"""命令行入口（typer）：version / fetch / bench / serve-stub。
这一层已经写好，它只负责"读参数 → 调 engine/report → 打印"。
所以在 client.py / engine.py / report.py 写完之前，fetch 与 bench 会抛 NotImplementedError。
用法：
uv run aggregator --help
uv run aggregator serve-stub --port 8001 --delay 0.1 --fail-rate 0.2
uv run aggregator fetch --count 50 --limit 8 --timeout 2 --output summary.json
uv run aggregator bench --count 30
"""
import asyncio
import time
from collections.abc import Callable
from pathlib import Path
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from aggregator import __version__
from aggregator.client import make_async_client, make_sync_client
from aggregator.engine import (
build_urls,
fetch_all_async,
fetch_all_sequential,
fetch_all_threaded,
)
from aggregator.logging_config import configure_logging, get_logger
from aggregator.models import FetchResult, Summary
from aggregator.report import (
results_table,
summarize,
summary_table,
write_csv,
write_json,
)
from aggregator.settings import Settings, get_settings
from aggregator.stub_server import serve
app = typer.Typer(
help="aggregator：并发抓取聚合器（第 3 周小项目）", no_args_is_help=True
)
console = Console()
logger = get_logger(__name__)
CountOpt = Annotated[int | None, typer.Option("--count", "-n", help="抓多少条")]
LimitOpt = Annotated[int | None, typer.Option("--limit", "-l", help="最大并发数")]
TimeoutOpt = Annotated[float | None, typer.Option("--timeout", help="单条超时（秒）")]
RetriesOpt = Annotated[int | None, typer.Option("--retries", help="重试次数")]
BaseUrlOpt = Annotated[str | None, typer.Option("--base-url", help="桩服务器地址")]
OutputOpt = Annotated[
Path | None, typer.Option("--output", "-o", help="导出到 .json 或 .csv")
]
@app.callback()
def main() -> None:
"""aggregator：并发抓取聚合器。"""
@app.command()
def version() -> None:
"""打印版本号。"""
typer.echo(__version__)
@app.command()
def fetch(
count: CountOpt = None,
limit: LimitOpt = None,
timeout: TimeoutOpt = None,
retries: RetriesOpt = None,
base_url: BaseUrlOpt = None,
output: OutputOpt = None,
) -> None:
"""并发抓一批 URL，打印表格与汇总，可选导出 JSON/CSV。"""
settings = get_settings(
count=count,
limit=limit,
timeout=timeout,
retries=retries,
base_url=base_url,
)
configure_logging(settings.log_level)
urls = build_urls(settings.base_url, settings.count)
started = time.perf_counter()
try:
results = asyncio.run(_fetch_async(settings, urls))
except KeyboardInterrupt:
console.print("[yellow]已取消（Ctrl+C）[/yellow]")
raise typer.Exit(code=130) from None
elapsed_ms = (time.perf_counter() - started) * 1000
summary = summarize(results, elapsed_ms)
console.print(results_table(results, limit=min(settings.count, 10)))
console.print(summary_table(summary))
if output is not None:
_export(output, summary, results)
console.print(f"已写入 {output}")
@app.command()
def bench(
count: CountOpt = None,
limit: LimitOpt = None,
timeout: TimeoutOpt = None,
retries: RetriesOpt = None,
base_url: BaseUrlOpt = None,
) -> None:
"""对比三种实现（顺序 / 线程池 / asyncio）的耗时。"""
settings = get_settings(
count=count,
limit=limit,
timeout=timeout,
retries=retries,
base_url=base_url,
)
configure_logging(settings.log_level)
urls = build_urls(settings.base_url, settings.count)
rows: list[tuple[str, float, int]] = []
with make_sync_client(settings) as client:
rows.append(
_timed(
"sequential",
lambda: fetch_all_sequential(client, urls, retries=settings.retries),
)
)
rows.append(
_timed(
"threaded",
lambda: fetch_all_threaded(
client,
urls,
max_workers=settings.max_workers,
retries=settings.retries,
),
)
)
started = time.perf_counter()
async_results = asyncio.run(_fetch_async(settings, urls))
rows.append(
(
"async",
(time.perf_counter() - started) * 1000,
sum(1 for r in async_results if r.ok),
)
)
baseline = next(ms for name, ms, _ in rows if name == "sequential")
table = Table(title=f"bench：{len(urls)} 条，limit={settings.limit}")
table.add_column("实现")
table.add_column("耗时 ms", justify="right")
table.add_column("成功", justify="right")
table.add_column("相对顺序", justify="right")
for name, ms, ok in rows:
speedup = baseline / ms if ms else 0.0
table.add_row(name, f"{ms:.0f}", str(ok), f"{speedup:.1f}×")
console.print(table)
@app.command(name="serve-stub")
def serve_stub(
port: Annotated[int, typer.Option("--port", "-p", help="端口，0 表示随机")] = 8001,
delay: Annotated[float, typer.Option("--delay", help="每个请求的延迟（秒）")] = 0.1,
fail_rate: Annotated[float, typer.Option("--fail-rate", help="失败率 0..1")] = 0.0,
) -> None:
"""起一个本地桩服务器（Ctrl+C 退出），演示与基准都打它。"""
serve(port=port, delay=delay, fail_rate=fail_rate)
async def _fetch_async(settings: Settings, urls: list[str]) -> list[FetchResult]:
"""asyncio 版抓取：客户端建一次，用完关掉。"""
async with make_async_client(settings) as client:
return await fetch_all_async(
client, urls, limit=settings.limit, retries=settings.retries
)
def _timed(
name: str, runner: Callable[[], list[FetchResult]]
) -> tuple[str, float, int]:
"""跑一种实现并计时，返回 (名字, 毫秒, 成功数)。"""
started = time.perf_counter()
results = runner()
elapsed_ms = (time.perf_counter() - started) * 1000
return name, elapsed_ms, sum(1 for r in results if r.ok)
def _export(output: Path, summary: Summary, results: list[FetchResult]) -> None:
"""按后缀导出：.json 走 write_json，.csv 走 write_csv。"""
suffix = output.suffix.lower()
if suffix == ".json":
write_json(output, summary, results)
elif suffix == ".csv":
write_csv(output, results)
else:
raise typer.BadParameter(f"只支持 .json 或 .csv，收到 {output.suffix}")
if __name__ == "__main__":
app()