75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""本地 JSON 客户端，模拟 APIClient 的分页接口，从落盘的 JSON 回放数据。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Iterable, Tuple
|
||
|
||
from api.client import APIClient
|
||
from utils.json_store import endpoint_to_filename
|
||
|
||
|
||
class LocalJsonClient:
|
||
"""
|
||
读取 RecordingAPIClient 生成的 JSON，提供 iter_paginated/get_paginated 接口。
|
||
"""
|
||
|
||
def __init__(self, base_dir: str | Path):
|
||
self.base_dir = Path(base_dir)
|
||
if not self.base_dir.exists():
|
||
raise FileNotFoundError(f"JSON 目录不存在: {self.base_dir}")
|
||
|
||
def iter_paginated(
|
||
self,
|
||
endpoint: str,
|
||
params: dict | None,
|
||
page_size: int = 200,
|
||
page_field: str = "page",
|
||
size_field: str = "limit",
|
||
data_path: tuple = ("data",),
|
||
list_key: str | None = None,
|
||
) -> Iterable[Tuple[int, list, dict, dict]]:
|
||
file_path = self.base_dir / endpoint_to_filename(endpoint)
|
||
if not file_path.exists():
|
||
raise FileNotFoundError(f"未找到匹配的 JSON 文件: {file_path}")
|
||
|
||
with file_path.open("r", encoding="utf-8") as fp:
|
||
payload = json.load(fp)
|
||
|
||
pages = payload.get("pages")
|
||
if not isinstance(pages, list) or not pages:
|
||
pages = [{"page": 1, "request": params or {}, "response": payload}]
|
||
|
||
for idx, page in enumerate(pages, start=1):
|
||
response = page.get("response", {})
|
||
request_params = page.get("request") or {}
|
||
page_no = page.get("page") or idx
|
||
records = APIClient._extract_list(response, data_path, list_key) # type: ignore[attr-defined]
|
||
yield page_no, records, request_params, response
|
||
|
||
def get_paginated(
|
||
self,
|
||
endpoint: str,
|
||
params: dict,
|
||
page_size: int = 200,
|
||
page_field: str = "page",
|
||
size_field: str = "limit",
|
||
data_path: tuple = ("data",),
|
||
list_key: str | None = None,
|
||
) -> tuple[list, list]:
|
||
records: list = []
|
||
pages_meta: list = []
|
||
for page_no, page_records, request_params, response in self.iter_paginated(
|
||
endpoint=endpoint,
|
||
params=params,
|
||
page_size=page_size,
|
||
page_field=page_field,
|
||
size_field=size_field,
|
||
data_path=data_path,
|
||
list_key=list_key,
|
||
):
|
||
records.extend(page_records)
|
||
pages_meta.append({"page": page_no, "request": request_params, "response": response})
|
||
return records, pages_meta
|