算法实现生成器

函数说明

• 算法名称：dijkstra_shortest_path
• 目标：使用优先队列实现 Dijkstra 最短路径算法，支持有向/无向加权图（权重非负）。可计算：
  • 起点到所有顶点的最短距离
  • 起点到指定终点的最短路并重建路径
• 输入：
  • graph: dict[节点, list[(邻接节点, 权重: float)]]
  • start: 节点标识（字符串或整数，需可哈希）
  • end: 可选节点标识，None 表示计算到所有节点
  • return_path: bool，true 时返回路径（当 end 为 None 时返回前驱表用于路径恢复）
• 输出：
  • 若 end 为 None：
    • return_path = False：返回 dict[节点 -> 最短距离]
    • return_path = True：返回 (dict[节点 -> 最短距离], dict[节点 -> 前驱节点])
  • 若提供 end：
    • return_path = False：返回 distance: float（不可达为 math.inf）
    • return_path = True：返回 (distance: float, path: list[节点])（不可达为 (math.inf, [])）
• 约定：
  • 支持有向/无向图：无向图请在输入中为每条边提供双向邻接即可
  • 权重必须为非负且有限数值
  • 空图、孤立顶点可健壮处理

代码实现（可直接运行与单元测试）

from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
import heapq
import math
import unittest


def dijkstra_shortest_path(
    graph: Dict[Any, Iterable[Tuple[Any, Union[int, float]]]],
    start: Any,
    end: Optional[Any] = None,
    return_path: bool = False,
):
    """
    使用优先队列的 Dijkstra 最短路径算法。

    参数:
        graph: 字典，key 为节点（可哈希），value 为邻接列表 (邻居, 权重)。
               示例: {"A":[("B",1.5),("C",2.0)], "B":[("C",1.0)]}
               注意：无向图请提供双向边。
        start: 起点（可哈希）
        end:   终点（可选，None 表示计算到所有节点）
        return_path: 若为 True：
            - 当 end is None：返回 (distances, predecessors)
            - 当 end is not None：返回 (distance, path)

    返回:
        - 如果 end is None:
            - return_path=False -> dict[node] = 最短距离
            - return_path=True  -> (dict[node] = 最短距离, dict[node] = 前驱节点)
        - 如果 end is not None:
            - return_path=False -> distance: float（不可达为 math.inf）
            - return_path=True  -> (distance: float, path: list[节点]；不可达为 (math.inf, []))

    健壮性:
        - 空图处理: 若图为空，仅返回与 start 相关的结果（到自身 0，其余不存在）
        - 孤立顶点: 允许 start 不在图中，视为孤立点；若 end==start，距离为 0，否则不可达
        - 校验:
            - 节点需可哈希
            - 权重需为非负且有限的数值

    时间复杂度:
        - O(E log V)，其中 V 为顶点数，E 为边数（使用最小堆）
    空间复杂度:
        - O(V + E) 用于距离、前驱、邻接表和堆

    """
    # ------------------------
    # 输入校验与规范化
    # ------------------------
    if graph is None:
        raise TypeError("graph 不能为 None，请提供字典形式的邻接表。")
    if not isinstance(graph, dict):
        raise TypeError("graph 必须为 dict[节点 -> 邻接列表]。")

    # 校验并收集节点集；邻接表标准化为 dict[node] -> List[(neighbor, weight)]
    adjacency: Dict[Any, List[Tuple[Any, float]]] = {}
    nodes = set()

    # 校验 start/end 可哈希
    try:
        hash(start)
    except Exception as e:
        raise TypeError(f"start 节点不可哈希: {start!r}") from e
    if end is not None:
        try:
            hash(end)
        except Exception as e:
            raise TypeError(f"end 节点不可哈希: {end!r}") from e

    for u, edges in graph.items():
        # 校验节点可哈希
        try:
            hash(u)
        except Exception as e:
            raise TypeError(f"节点不可哈希: {u!r}") from e

        nodes.add(u)
        # 支持空/None 邻接列表（视为空）
        if edges is None:
            edges = []
        # 校验邻接列表类型
        if not isinstance(edges, (list, tuple)):
            raise TypeError(f"节点 {u!r} 的邻接列表必须为 list/tuple[(邻居, 权重)]。")

        normalized_edges: List[Tuple[Any, float]] = []
        for item in edges:
            if not isinstance(item, (list, tuple)) or len(item) != 2:
                raise TypeError(f"邻接项需为 (neighbor, weight)，收到: {item!r}")
            v, w = item
            try:
                hash(v)
            except Exception as e:
                raise TypeError(f"邻居节点不可哈希: {v!r}") from e
            # 权重校验：数值、有限、非负
            if not isinstance(w, (int, float)):
                raise TypeError(f"权重必须为数值 (int/float)，收到: {w!r}")
            if not math.isfinite(w):
                raise ValueError(f"权重必须为有限数值，收到: {w!r}")
            if w < 0:
                raise ValueError(f"权重必须非负，收到: {w!r}")

            nodes.add(v)
            normalized_edges.append((v, float(w)))

        adjacency[u] = normalized_edges

    # 将 start/end 纳入节点集合（允许孤立点）
    nodes.add(start)
    if end is not None:
        nodes.add(end)

    # 确保每个节点在 adjacency 中有条目
    for n in nodes:
        adjacency.setdefault(n, [])

    # ------------------------
    # Dijkstra 主过程
    # ------------------------
    # 初始化距离与前驱
    dist: Dict[Any, float] = {n: math.inf for n in nodes}
    dist[start] = 0.0
    prev: Dict[Any, Any] = {}  # 记录最优前驱

    # 小顶堆：(距离, 节点)
    heap: List[Tuple[float, Any]] = [(0.0, start)]

    # 若 end 指定，可提前终止：当 end 首次弹出且距离最小，即已确定最短路
    target = end

    while heap:
        d_u, u = heapq.heappop(heap)
        if d_u > dist[u]:
            # 堆中旧条目，忽略
            continue

        if target is not None and u == target:
            # 最短路径已确定，提前结束
            break

        # 松弛
        for v, w in adjacency[u]:
            nd = d_u + w
            if nd < dist[v]:
                dist[v] = nd
                prev[v] = u
                heapq.heappush(heap, (nd, v))

    # ------------------------
    # 构造返回值
    # ------------------------
    def reconstruct_path(predecessor: Dict[Any, Any], s: Any, t: Any) -> List[Any]:
        if s == t:
            return [s]
        if dist.get(t, math.inf) == math.inf:
            return []
        path = []
        cur = t
        while cur != s:
            path.append(cur)
            cur = predecessor.get(cur)
            if cur is None:
                # 理论上不会发生；保护性处理
                return []
        path.append(s)
        path.reverse()
        return path

    if end is None:
        # 需求：返回到所有节点的最短距离；可选返回前驱表
        return (dist, prev) if return_path else dist
    else:
        # 需求：返回到 end 的距离；可选返回路径
        distance = dist.get(end, math.inf)
        if return_path:
            path = reconstruct_path(prev, start, end)
            return (distance, path)
        else:
            return distance


# ------------------------
# 示例调用
# ------------------------
def _demo():
    example_graph = {
        "A": [("B", 1.5), ("C", 2.0)],
        "B": [("C", 1.0)],
        # 无向图示例：请提供双向边
        # "C": [("A", 2.0), ("B", 1.0)]
    }

    print("示例：起点到所有节点的距离")
    dist_all = dijkstra_shortest_path(example_graph, start="A", end=None, return_path=False)
    print(dist_all)  # {'A': 0.0, 'B': 1.5, 'C': 2.5}

    print("示例：起点到终点的最短路与路径")
    distance, path = dijkstra_shortest_path(example_graph, start="A", end="C", return_path=True)
    print(distance, path)  # 2.5 ['A', 'B', 'C']


# ------------------------
# 单元测试
# ------------------------
class TestDijkstraShortestPath(unittest.TestCase):
    def test_directed_basic(self):
        g = {
            "A": [("B", 1.5), ("C", 2.0)],
            "B": [("C", 1.0)],
        }
        distance, path = dijkstra_shortest_path(g, "A", "C", return_path=True)
        self.assertEqual(distance, 2.5)
        self.assertEqual(path, ["A", "B", "C"])

    def test_all_distances(self):
        g = {
            "A": [("B", 2), ("C", 5)],
            "B": [("C", 1)],
            "C": [],
            "D": []  # 孤立节点
        }
        dist = dijkstra_shortest_path(g, "A", end=None, return_path=False)
        self.assertEqual(dist["A"], 0.0)
        self.assertEqual(dist["B"], 2.0)
        self.assertEqual(dist["C"], 3.0)
        self.assertTrue(math.isinf(dist["D"]))

        dist2, prev = dijkstra_shortest_path(g, "A", end=None, return_path=True)
        self.assertEqual(dist2["C"], 3.0)
        # 路径恢复 A->B->C
        path_c = []
        cur = "C"
        while cur in prev:
            path_c.append(cur)
            cur = prev[cur]
        path_c.append("A")
        path_c.reverse()
        self.assertEqual(path_c, ["A", "B", "C"])

    def test_unreachable(self):
        g = {
            1: [(2, 1.0)],
            2: [],
            3: []  # 不可达
        }
        distance = dijkstra_shortest_path(g, 1, end=3, return_path=False)
        self.assertTrue(math.isinf(distance))
        distance, path = dijkstra_shortest_path(g, 1, end=3, return_path=True)
        self.assertTrue(math.isinf(distance))
        self.assertEqual(path, [])

    def test_empty_graph(self):
        g = {}
        # 起点到自身为 0，其他不存在
        dist = dijkstra_shortest_path(g, "S", end=None, return_path=False)
        self.assertEqual(dist["S"], 0.0)
        # 指定终点且不可达
        d, p = dijkstra_shortest_path(g, "S", end="T", return_path=True)
        self.assertTrue(math.isinf(d))
        self.assertEqual(p, [])

    def test_isolated_start(self):
        g = {"A": [("B", 2.0)], "B": []}
        # start 不在图中，视为孤立点
        d = dijkstra_shortest_path(g, "Z", end=None, return_path=False)
        self.assertEqual(d["Z"], 0.0)
        self.assertTrue(math.isinf(d["A"]))
        self.assertTrue(math.isinf(d["B"]))
        # end==start
        distance, path = dijkstra_shortest_path(g, "Z", end="Z", return_path=True)
        self.assertEqual(distance, 0.0)
        self.assertEqual(path, ["Z"])

    def test_invalid_weight(self):
        g_neg = {"A": [("B", -1.0)]}
        with self.assertRaises(ValueError):
            dijkstra_shortest_path(g_neg, "A", end=None)

        g_inf = {"A": [("B", float("inf"))]}
        with self.assertRaises(ValueError):
            dijkstra_shortest_path(g_inf, "A", end=None)

    def test_non_hashable_node(self):
        g_bad = {["A"]: [("B", 1.0)]}  # 列表不可哈希
        with self.assertRaises(TypeError):
            dijkstra_shortest_path(g_bad, "A", end=None)


if __name__ == "__main__":
    _demo()
    unittest.main(argv=["-v"], exit=False)

简要实现思路与逻辑说明

• 使用最小堆（heapq）维护当前已知的最短距离候选，弹出距离最小的节点进行松弛，复杂度 O(E log V)。
• 使用 dist 记录起点到各节点的最短距离；使用 prev 记录到达每个节点的最佳前驱，用于路径重建。
• 当指定了 end 时，首次从堆中弹出 end 即确定其最短路径，可提前结束。
• 输入校验涵盖：
  • graph 类型与邻接项格式
  • 节点可哈希性与权重非负、有限
  • 空图与孤立顶点处理（将 start/end 纳入节点集合，保证算法健壮性）
• 无向图支持通过在输入中提供双向边实现；函数不自动补全双向边以避免误判图类型。