以下是对原代码的重构版本（Python 3），按建议拆分为 load_and_parse、fetch_scores（通过 ScoreClient 实现批量获取）、rank_top；移除全局 CACHE，引入可替换的缓存接口；将网络请求抽象为客户端以便替换与测试。
重构后代码：
#!/usr/bin/env python3
import json
import logging
from dataclasses import dataclass
from typing import List, Dict, Optional, Sequence, Protocol
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from urllib.parse import quote
from concurrent.futures import ThreadPoolExecutor, as_completed
@dataclass(frozen=True)
class Item:
name: str
value: int
@dataclass(frozen=True)
class ScoredItem:
name: str
value: int
score: int
class Cache(Protocol):
def get(self, key: str) -> Optional[object]:
...
def set(self, key: str, value: object) -> None:
...
class MemoryCache:
def init(self):
self._store: Dict[str, object] = {}
def get(self, key: str) -> Optional[object]:
return self._store.get(key)
def set(self, key: str, value: object) -> None:
self._store[key] = value
def load_and_parse(path: str, cache: Optional[Cache] = None) -> List[Item]:
"""
读取并解析 JSON 文件，归一化为 Item 列表。
通过 cache（可选）缓存解析结果，避免重复 I/O 与解析开销。
"""
cache_key = f"file:{path}"
if cache:
cached = cache.get(cache_key)
if isinstance(cached, list):
return cached
try:
with open(path, "r", encoding="utf-8") as f:
raw = f.read()
data = json.loads(raw)
except Exception as e:
logging.warning("Failed to read/parse JSON at %s: %s", path, e)
return []
items: List[Item] = []
if isinstance(data, list):
for x in data:
if isinstance(x, dict) and x.get("name"):
name = str(x["name"]).strip()
try:
value = int(x.get("value", 0))
except Exception:
value = 0
items.append(Item(name=name, value=value))
else:
logging.warning("JSON root is not a list at %s", path)
if cache:
cache.set(cache_key, items)
return items
class ScoreClient:
"""
可替换的打分客户端，支持：
- requests.Session 连接复用与重试策略
- 去重后并发请求
- 可选的本地缓存（按 name.lower() 缓存）
"""
def init(self, base_url: str, session: Optional[requests.Session] = None,
timeout: float = 3.0, retries: int = 3, backoff: float = 0.2,
cache: Optional[Cache] = None, max_workers: int = 8):
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.cache = cache
self.max_workers = max_workers
self.session = session or requests.Session()
adapter = HTTPAdapter(max_retries=Retry(
total=retries,
backoff_factor=backoff,
status_forcelist=[429, 500, 502, 503, 504],
))
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
def get_scores(self, names: Sequence[str]) -> Dict[str, int]:
# 去重（按小写），并利用缓存减少请求数量
uniques = {}
for name in names:
lower = name.lower()
if lower not in uniques:
uniques[lower] = name
scores: Dict[str, int] = {}
to_fetch: List[str] = []
if self.cache:
for lower, original in uniques.items():
cached = self.cache.get(f"score:{lower}")
if isinstance(cached, int):
scores[lower] = cached
else:
to_fetch.append(original)
else:
to_fetch = list(uniques.values())
# 并发获取未命中缓存的条目
if to_fetch:
workers = min(len(to_fetch), max(1, self.max_workers))
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(self._fetch_score, name): name for name in to_fetch}
for fut in as_completed(futures):
original = futures[fut]
lower = original.lower()
try:
score = fut.result()
except Exception as e:
logging.warning("Score request failed for %s: %s", original, e)
score = 0
scores[lower] = score
if self.cache:
self.cache.set(f"score:{lower}", score)
return scores
def _fetch_score(self, name: str) -> int:
url = f"{self.base_url}?q={quote(name)}"
try:
resp = self.session.get(url, timeout=self.timeout)
resp.raise_for_status()
payload = resp.json()
score = int(payload.get("score", 0))
except Exception as e:
logging.debug("Error fetching score for %s: %s", name, e)
score = 0
return score
def rank_top(items: List[Item], scores: Dict[str, int], topk: int) -> List[ScoredItem]:
"""
合并本地值与远端分数，排序并去重，输出前 topk。
"""
enriched: List[ScoredItem] = [
ScoredItem(name=i.name, value=i.value, score=scores.get(i.name.lower(), 0))
for i in items
]
enriched.sort(key=lambda x: (-x.score, -x.value, x.name.lower()))
seen = set()
out: List[ScoredItem] = []
for r in enriched:
key = r.name.lower()
if key in seen:
continue
seen.add(key)
out.append(r)
if len(out) >= topk:
break
return out
def run(path: str, url: str, topk: int = 5,
data_cache: Optional[Cache] = None,
score_cache: Optional[Cache] = None,
client: Optional[ScoreClient] = None,
max_workers: int = 8,
timeout: float = 3.0,
verbose: bool = False) -> List[ScoredItem]:
"""
组合管线：加载→打分→排名。将 I/O、网络、计算解耦，支持依赖注入。
"""
items = load_and_parse(path, cache=data_cache)
if not client:
client = ScoreClient(base_url=url, cache=score_cache, max_workers=max_workers, timeout=timeout)
names = [i.name for i in items]
scores = client.get_scores(names)
top = rank_top(items, scores, topk)
if verbose:
for r in top:
print(r.name, r.value, r.score)
return top
if name == "main":
logging.basicConfig(level=logging.INFO)
data_cache = MemoryCache()
score_cache = MemoryCache()
res = run(
"data.json",
"https://api.example.com/search",
topk=10,
data_cache=data_cache,
score_cache=score_cache,
verbose=True,
)
主要重构点与改进效果：
-
拆分职责（可维护性、可读性）
- 将原本“读取文件+解析+网络请求+排序+打印”混在一起的 run 函数拆分为：
- load_and_parse：只负责文件读取与 JSON 解析，并将数据归一化。
- ScoreClient.get_scores（对应 fetch_scores(batch)）：抽象网络请求客户端，负责批量获取分数。
- rank_top：只做排序和去重。
- 好处：
- 单一职责更清晰，易于理解与修改。
- 各函数易于单元测试（例如 rank_top 可独立测试排序逻辑）。
- 通过依赖注入（缓存与客户端）替换实现更加容易。
-
引入缓存接口，移除全局变量（可维护性、可读性）
- 定义 Cache 协议与 MemoryCache 实现，替代全局 CACHE。
- load_and_parse 使用 data_cache 缓存解析后的结构化数据；ScoreClient 使用 score_cache 缓存分数。
- 好处：
- 消除隐式全局状态，避免难以追踪的副作用。
- 更易注入不同缓存策略（内存、磁盘、Redis），便于横向扩展与测试。
-
网络层性能优化（性能优化）
- 连接复用：使用 requests.Session，减少 TLS 握手与建立连接成本。
- 重试与退避：HTTPAdapter + Retry，对 429/5xx 做带退避的重试，提高鲁棒性，降低偶发失败影响。
- 并发请求：ThreadPoolExecutor 并发请求未命中缓存的名称，显著降低总体墙钟时间。
- 去重请求：按 name.lower() 去重后才请求，避免重复调用同一个查询。
- URL 编码：quote 防止特殊字符导致的请求失败。
- 结果缓存：命中 score_cache 时不触发网络。
- 去除 time.sleep(0.01)：避免无意义的延迟。
-
计算与排名逻辑改进（可读性、性能）
- 使用 dataclass（Item/ScoredItem）明确数据结构，字段含义更直观。
- 排序键增加第三项 x.name.lower()，使排序稳定且更可预测。
- 将异常处理集中在数据加载与网络层，计算层保持纯逻辑，便于测试与复用。
-
错误处理与日志（可维护性）
- 使用 logging.warning / logging.debug 替代 print，便于在生产环境控制日志级别。
- 对非列表 JSON 根、数值解析失败等情况做健壮处理。
-
类型标注与文档（可读性、可维护性）
- 全面添加类型注解与简短文档字符串，提升 IDE 支持与阅读体验。
复杂度与性能对比（定性）：
- 原实现：每个条目都发起一次网络请求，即使名称重复；顺序串行，存在不必要的 sleep。
- 新实现：
- 请求数量由 n 降至 u（u 为去重后的唯一名称数），并发度为 w（max_workers），总体墙钟时间约为 O(u / w)。
- 通过缓存可进一步将请求数量降至近似 0（命中时）。
- 连接复用与重试减少失败及重建连接开销。
测试与可扩展性：
- 可在测试中传入假的 ScoreClient（返回固定分数），或将 score_cache 预置数据，避免真实网络。
- Cache 接口可替换为持久化实现（如文件或 Redis）。
- run 支持 verbose 控制输出，核心逻辑不耦合打印，便于在不同环境中集成。
如需进一步优化（可选）：
- 在高并发场景下引入异步（aiohttp）版本的 ScoreClient。
- 为 MemoryCache 增加 TTL 与清理策略。
- 在 load_and_parse 中对 name 预先去重，减少后续计算开销。
