lab1-2021112105/oop.py

import copy
import heapq
import re
import sys
import time
from math import isinf, inf
import networkx as nx
import matplotlib.pyplot as plt
from typing import Dict, Generator, List, Optional
import random


def main():
    # print("Welcome to Lab1")
    print("功能5中输入@停止功能")
    tu = None

    while True:

        i = input("请输入操作（q:退出, 0:读入文件, 1:展示图形, 2:查找桥接词, 3:生成新文本, 4:计算最短路径, 5:随机游走）：")
        if i == "q":
            sys.exit()
        if i == "0":
            tu = Tu()

            file_name = input("请输入文件名称：")

            tu.generate_directed_dict(tu.read_text_file(file_name))
            tu.generate_directed_graph()
            continue
        if tu is None:
            print("未找到有效的图")

            continue

        if i == "1":
            tu.drawDirectedGraph()
        elif i == "2":
            word1 = input("Enter the word1:")
            word2 = input("Enter the word2:")

            tu.queryBridgeWords(word1, word2)
        elif i == "3":
            text = input("Enter the text:")
            tu.generateNewText(text)

        elif i == "4":
            word1 = input("Enter the word1:")
            word2 = input("Enter the word2,just '' is all")
            tu.calcShortestPath(word1, word2)

        elif i == "5":
            tu.randomWalk()

        else:
            print("Invalid input")


class Tu:
    def __init__(self):
        self.dict = {}
        self.graph = nx.MultiDiGraph()

    @staticmethod
    def read_text_file(filename):
        try:
            with open(filename, 'r', encoding="utf-8") as file:
                text = file.read()
                # 用正则表达式将非字母字符替换为空格，并将换行符也替换为空格
                text = re.sub(r'[^a-zA-Z\n]+', ' ', text)
                # 将文本转换为小写，并按空格分割成单词列表
                word_sequence = text.lower().split()
                print(word_sequence)
                return word_sequence
        except FileNotFoundError:
            print("File not found.")
            return []

    def generate_directed_dict(self, word_sequence):
        graph = self.dict
        for i in range(len(word_sequence) - 1):
            current_word = word_sequence[i]
            next_word = word_sequence[i + 1]
            # 忽略空字符串
            if current_word == '' or next_word == '':
                continue
            # 将当前单词和下一个单词添加到图中
            if current_word not in graph:
                graph[current_word] = {}
            if next_word not in graph[current_word]:
                graph[current_word][next_word] = 1
            else:
                graph[current_word][next_word] += 1

        # draw_directed_graph(graph)
        return graph

    def generate_directed_graph(self):
        G = self.graph  # wufuzhi,yiqigai
        graph = self.dict
        # 添加节点和边
        for node, neighbors in graph.items():
            for neighbor, weight in neighbors.items():
                G.add_edge(node, neighbor, weight=weight)
                # 如果存在相反方向的边，则再画一个箭头
                # if neighbor in graph and node in graph[neighbor]:
                # G.add_edge(neighbor, node, weight=graph[neighbor][node])
        return G

    def draw_directed_graph(self):
        G = self.graph
        pos = nx.spring_layout(G)
        plt.rcParams['figure.figsize'] = (12.8, 7.2)
        nx.draw_networkx(G, pos, with_labels=True, node_size=1000,
                         node_color="skyblue", edge_color="black", font_size=10,
                         font_weight="bold", arrows=True, connectionstyle='arc3,rad=0.2')
        edge_labels = nx.get_edge_attributes(G, 'weight')
        # 调整箭头位置
        for edge, weight in edge_labels.items():
            if G.has_edge(*edge[::-1]):  # 检查反向边是否存在于图中
                dx = pos[edge[0]][0] - pos[edge[1]][0]
                dy = pos[edge[0]][1] - pos[edge[1]][1]
                plt.annotate(weight, (
                    (pos[edge[0]][0] + pos[edge[1]][0]) / 2 + dy * 0.1,
                    (pos[edge[0]][1] + pos[edge[1]][1]) / 2 - dx * 0.1))
            else:  # 反向边不存在
                plt.annotate(weight, ((pos[edge[0]][0] + pos[edge[1]][0]) / 2,
                                      (pos[edge[0]][1] + pos[edge[1]][1]) / 2))
        plt.show()

    def drawDirectedGraph(self):
        self.draw_directed_graph()

    def find_bridge_words(self, word1, word2):

        graph = self.dict
        if word1 not in graph or word2 not in graph:
            print("No", word1, "or", word2, "in the graph!")
            return []

        bridge_words = [
            word for word in graph[word1]
            if word2 in graph.get(word, {})
        ]

        return bridge_words

    @staticmethod
    def print_bridge_words(bridge_words, word1, word2):

        if not bridge_words:
            print("No bridge words from", word1, "to", word2, "!")
            return []
        else:
            bridge_words_str = ", ".join(bridge_words)
            print("The bridge words from", word1, "to", word2, "are:", bridge_words_str + ".")

            return bridge_words

    def queryBridgeWords(self, word1, word2):
        word1 = self.input_check(word1)
        word2 = self.input_check(word2)
        bridge_words = self.find_bridge_words(word1, word2)
        self.print_bridge_words(bridge_words, word1, word2)

    def insert_bridge_words(self, text):

        words = text.split()

        new_text = []

        for i in range(len(words) - 1):

            word1 = words[i].lower()
            word2 = words[i + 1].lower()

            new_text.append(words[i])
            bridge_words = self.find_bridge_words(word1, word2)

            if len(bridge_words) > 0:
                random_bridge_word = random.choice(bridge_words)
                new_text.append(random_bridge_word)

        new_text.append(words[-1])

        return ' '.join(new_text)

    def generateNewText(self, text):

        (text_re, text_ren) = re.subn(r'[^a-zA-Z\n]+', ' ', text)
        if text_ren > 0:
            print("There are illegal signs in your input")
        if not text_re.islower():
            print("There are upper characters in your input")
        print(self.insert_bridge_words(text_re))

    def find_shortest_path_old(self, word1, word2):

        graph = self.graph
        try:
            shortest_path_generator = (
                nx.all_shortest_paths(graph, source=word1, target=word2, weight='weight'))
            shortest_length = (
                nx.shortest_path_length(graph, source=word1, target=word2, weight='weight'))
            return shortest_path_generator, shortest_length
        except nx.NetworkXNoPath:
            return None, None

    def find_shortest_path(self, word1, word2):

        try:

            (shortest_path_list, shortest_length) = self.calc_shortest_path_len(word1, word2)
            return shortest_path_list, shortest_length
        except nx.NetworkXNoPath:
            return None, None

    def find_shortest_path_ex(self, word1, word2):

        try:
            shortest_path_generator = self.all_simple_paths_graph(word1, word2)
            shortest_length = self.calc_shortest_path_len(word1, word2)

            shortest_path_list = []

            for shortest_path in shortest_path_generator:

                if shortest_path.__len__() == shortest_length:
                    shortest_path_list.append(shortest_path)
            return shortest_path_list, shortest_length
        except nx.NetworkXNoPath:
            return None, None

    def draw_shortest_path(self, shortest_path_list):
        graph = self.graph
        plt.figure(figsize=(10, 6))

        # !!! HUATU ROLL YANSE

        pos = nx.spring_layout(graph)
        nx.draw_networkx_nodes(graph, pos, node_size=1000, node_color="skyblue")
        nx.draw_networkx_labels(graph, pos, font_size=10, font_weight="bold")

        for shortest_path in shortest_path_list:
            colorslist = '0123456789ABCDEF'
            num = "#"
            for i in range(6):
                num += random.choice(colorslist)

            for edge in graph.edges():

                # print(shortest_path)

                if (edge in zip(shortest_path[:-1], shortest_path[1:]) or
                        edge in zip(shortest_path[1:], shortest_path[:-1])):

                    nx.draw_networkx_edges(graph, pos, edgelist=[edge], width=2.0, edge_color=num,
                                           arrows=True, arrowstyle="->", arrowsize=30,
                                           connectionstyle='arc3,rad=0.2')
                else:
                    nx.draw_networkx_edges(graph, pos, edgelist=[edge], width=1.0,
                                           edge_color="black",
                                           arrows=True, arrowstyle="->", arrowsize=30,
                                           connectionstyle='arc3,rad=0.2')
        plt.show()
        time.sleep(1)

    def calcShortestPath(self, word1, word2):
        if word2 == "":
            word1 = self.input_check(word1)
            for wordtmp in self.dict:
                if wordtmp == word1:
                    continue
                else:
                    self.calc_shortest_path(word1, wordtmp)

            pass
        else:
            word1 = self.input_check(word1)
            word2 = self.input_check(word2)
            self.calc_shortest_path(word1, word2)
            pass

    def calc_shortest_path_old(self, word1, word2):

        shortest_path_generator, shortest_length = self.find_shortest_path_old(word1, word2)
        shortest_path_list = []

        for shortest_path in shortest_path_generator:

            if shortest_path:
                print("Shortest path:", '→'.join(shortest_path))
                print("Length of shortest path:", shortest_length)
                shortest_path_list.append(shortest_path)
        if shortest_path_list.__len__() > 0:
            self.draw_shortest_path(shortest_path_list)
        else:
            print("No path exists between", word1, "and", word2)

    def calc_shortest_path(self, word1, word2):

        shortest_path_list, shortest_length = self.find_shortest_path_ex(word1, word2)

        for shortest_path in shortest_path_list:

            if shortest_path:
                print("Shortest path:", '→'.join(shortest_path))
                print("Length of shortest path:", shortest_length)

        if shortest_path_list.__len__() > 0:
            self.draw_shortest_path(shortest_path_list)
        else:
            print("No path exists between", word1, "and", word2)

    def random_traversal(self):

        # 用户输入文件名
        filename = input("Enter the filename to save traversal results: ")
        filename += ".txt"

        graph = copy.deepcopy(self.dict)
        start_node = random.choice(list(graph.keys()))
        visited_nodes = set()
        visited_edges = []
        current_node = start_node

        with open(filename, 'w') as file:
            while True:
                visited_nodes.add(current_node)
                file.write(current_node + '\n')
                print(current_node)
                if input() == "@":
                    file.close()
                    break
                if graph.get(current_node) is not None:
                    pass
                else:
                    break
                neighbors = list(graph[current_node].keys())
                if not neighbors:
                    break
                next_node = random.choice(neighbors)
                visited_edges.append((current_node, next_node))
                graph[current_node].pop(next_node)
                current_node = next_node

        file.close()
        print("Visited nodes:", visited_nodes)
        print("Visited edges:", visited_edges)
        return visited_nodes, visited_edges

    def randomWalk(self):
        self.random_traversal()

    @staticmethod
    def input_check(input_word):
        (input_word_re, input_word_ren) = re.subn(r'[^a-zA-Z\n]', ' ', input_word)

        if input_word_ren > 0:
            print("There are illegal signs in your input,the amount is " + str(input_word_ren))

        if not input_word_re.islower():
            print("There are upper characters in your input")
        tmp = input_word_re.lower().split()
        if len(tmp) > 1:
            print("There are more than one word in your input")

        return tmp[0]

    def all_simple_paths_graph(self, source: str, targets: str) -> Generator[List[str], None, None]:
        G = self.graph
        cutoff = len(G) - 1  # 设置路径的最大深度，防止无限循环。
        visited = dict.fromkeys([source])  # 使用字典跟踪访问过的节点
        stack = [iter(G[source])]  # 使用栈保存当前路径中的节点迭代器
        while stack:
            children = stack[-1]  # 获取当前栈顶节点的迭代器
            child = next(children, None)

            if child is None:
                stack.pop()
                visited.popitem()  # 移除最近访问的节点
            elif len(visited) < cutoff:
                if child in visited:
                    continue
                if child == targets:
                    yield list(visited) + [child]
                visited[child] = None
                if {targets} - set(visited.keys()):  # 扩展栈直到找到目标
                    stack.append(iter(G[child]))
                else:
                    visited.popitem()  # 可能有到达子节点的其他路径
            else:  # 达到最大深度:
                for target in ({targets} & (set(children) | {child})) - set(visited.keys()):
                    yield list(visited) + [target]
                stack.pop()
                visited.popitem()

    def calc_shortest_path_len(self, word1: str, word2: str) -> int:

        if word1 not in self.graph.nodes or word2 not in self.graph.nodes:
            return 0

        distances = {node: inf for node in self.graph.nodes}
        # 存储word1到所有结点的距离，初始化为无穷大
        previous_nodes: Dict[str, Optional[str]] = {node: None for node in
                                                    self.graph.nodes}  # 存储每个节点最短路径中的前一个节点，初始化为None
        distances[word1] = 0  # 距离初始化为0
        priority_queue = [(0, word1)]  # 优先级队列，用于按照从小到大获取节点

        while priority_queue:
            current_distance, current_node = heapq.heappop(priority_queue)  # 优先弹出距离最小的节点

            if current_node == word2:
                break

            if current_distance > distances[current_node]:
                continue

            for neighbor, attributes in self.graph[current_node].items():  # 遍历所有邻居节点及属性
                weight = attributes.get('weight', 1)
                distance = current_distance + weight  # 更新距离，当前距离加入权重

                if distance < distances[neighbor]:
                    distances[neighbor] = distance
                    previous_nodes[neighbor] = current_node
                    heapq.heappush(priority_queue, (distance, neighbor))

        path = []
        current_node = word2
        while prev := previous_nodes[current_node]:  # 从目标节点回溯直到起始节点

            path.insert(0, current_node)
            current_node = prev
        if path:
            path.insert(0, current_node)

        if isinf(distances[word2]):  # 目标节点不可达
            return 0
        return path.__len__()
        # return (' '.join(path),path.__len__())


if __name__ == '__main__':
    main()