466 lines
14 KiB
Python
466 lines
14 KiB
Python
from house_operation import *
|
||
from datetime import datetime
|
||
|
||
|
||
# v1
|
||
def get_next_house_id():
|
||
houses = read_houses()
|
||
if not houses:
|
||
return 1
|
||
max_id = max(house["id"] for house in houses)
|
||
return max_id + 1
|
||
|
||
|
||
def add_house(address, price, area, house_type):
|
||
if not address or not house_type:
|
||
print("错误：地址和户型不能为空！")
|
||
return False
|
||
if not isinstance(price, (int, float)) or price <= 0:
|
||
print("错误：租金必须是正数！")
|
||
return False
|
||
if not isinstance(area, (int, float)) or area <= 0:
|
||
print("错误：面积必须是正数！")
|
||
return False
|
||
|
||
new_house = {
|
||
"id": get_next_house_id(),
|
||
"address": address,
|
||
"price": price,
|
||
"area": area,
|
||
"house_type": house_type
|
||
}
|
||
houses = read_houses()
|
||
houses.append(new_house)
|
||
write_houses(houses)
|
||
print(f"房源新增成功！房源ID：{new_house['id']}")
|
||
return True
|
||
|
||
|
||
def delete_house(house_id):
|
||
houses = read_houses()
|
||
for index, house in enumerate(houses):
|
||
if house["id"] == house_id:
|
||
del houses[index]
|
||
write_houses(houses)
|
||
print(f"房源ID {house_id} 删除成功！")
|
||
return True
|
||
print(f"错误：未找到房源ID {house_id}！")
|
||
return False
|
||
|
||
|
||
def update_house(house_id, new_info):
|
||
houses = read_houses()
|
||
for house in houses:
|
||
if house["id"] == house_id:
|
||
for key, value in new_info.items():
|
||
if key in ["address", "price", "area", "house_type"]:
|
||
if key in ["price", "area"]:
|
||
if not isinstance(value, (int, float)) or value <= 0:
|
||
print(f"错误：{key}必须是正数！")
|
||
return False
|
||
house[key] = value
|
||
write_houses(houses)
|
||
print(f"房源ID {house_id} 修改成功！")
|
||
return True
|
||
print(f"错误：未找到房源ID {house_id}！")
|
||
return False
|
||
|
||
|
||
def query_house(condition_type, condition_value):
|
||
houses = read_houses()
|
||
result = []
|
||
for house in houses:
|
||
if condition_type == "id":
|
||
if house["id"] == int(condition_value):
|
||
result.append(house)
|
||
elif condition_type == "address":
|
||
if condition_value in house["address"]:
|
||
result.append(house)
|
||
elif condition_type == "house_type":
|
||
if house["house_type"] == condition_value:
|
||
result.append(house)
|
||
elif condition_type == "price":
|
||
if house["price"] == float(condition_value):
|
||
result.append(house)
|
||
else:
|
||
print("错误：不支持的查询条件类型！")
|
||
return []
|
||
return result
|
||
|
||
|
||
# v2
|
||
MAX_STACK_LENGTH = 100
|
||
|
||
|
||
def push_operation_stack(operation_info):
|
||
time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
full_info = f"[{time_str}] {operation_info}"
|
||
stack = read_stack("operation")
|
||
stack.append(full_info)
|
||
if len(stack) > MAX_STACK_LENGTH:
|
||
stack.pop(0)
|
||
write_stack("operation", stack)
|
||
|
||
|
||
def push_browse_stack(house_id):
|
||
stack = read_stack("browse")
|
||
if stack and stack[-1] == house_id:
|
||
return
|
||
stack.append(house_id)
|
||
if len(stack) > MAX_STACK_LENGTH:
|
||
stack.pop(0)
|
||
write_stack("browse", stack)
|
||
|
||
|
||
def pop_operation_stack():
|
||
stack = read_stack("operation")
|
||
if not stack:
|
||
return None
|
||
latest_op = stack.pop()
|
||
write_stack("operation", stack)
|
||
return latest_op
|
||
|
||
|
||
def pop_browse_stack():
|
||
stack = read_stack("browse")
|
||
if not stack:
|
||
return None
|
||
latest_hid = stack.pop()
|
||
write_stack("browse", stack)
|
||
return latest_hid
|
||
|
||
|
||
def get_recent_records(stack_type, limit=10):
|
||
stack = read_stack(stack_type)
|
||
return stack[-limit:][::-1]
|
||
|
||
|
||
def clear_stack(stack_type):
|
||
write_stack(stack_type, [])
|
||
print(f"{stack_type}记录已清空！")
|
||
|
||
|
||
#v3
|
||
def parse_address_string(full_address):
|
||
"""解析完整地址字符串：拆分省、市、区、小区"""
|
||
parts = full_address.split("市")
|
||
province_city = parts[0] + "市" if "市" in full_address else parts[0]
|
||
if "省" in province_city:
|
||
province, city = province_city.split("省")
|
||
province += "省"
|
||
city += "市"
|
||
else:
|
||
province = "未知省份"
|
||
city = province_city
|
||
|
||
if len(parts) > 1:
|
||
district_part = parts[1]
|
||
if "区" in district_part or "县" in district_part:
|
||
district, community = district_part.split("区") if "区" in district_part else district_part.split("县")
|
||
district += "区" if "区" in district_part else "县"
|
||
else:
|
||
district = "未知区县"
|
||
community = district_part
|
||
else:
|
||
district = "未知区县"
|
||
community = "未知小区"
|
||
|
||
return {
|
||
"province": province,
|
||
"city": city,
|
||
"district": district,
|
||
"community": community
|
||
}
|
||
|
||
|
||
def fuzzy_query_by_district(keyword):
|
||
"""根据区域关键词模糊匹配所有房源"""
|
||
houses = read_houses()
|
||
result = []
|
||
for house in houses:
|
||
addr_info = parse_address_string(house["address"])
|
||
if keyword in addr_info["district"] or keyword in addr_info["community"]:
|
||
result.append(house)
|
||
return result
|
||
|
||
|
||
def init_community_matrix(community_name_list):
|
||
"""初始化小区距离邻接矩阵"""
|
||
INF = float("inf")
|
||
n = len(community_name_list)
|
||
distance_matrix = [[INF] * n for _ in range(n)]
|
||
for i in range(n):
|
||
distance_matrix[i][i] = 0
|
||
|
||
matrix_data = {
|
||
"community_list": community_name_list,
|
||
"distance_matrix": distance_matrix
|
||
}
|
||
write_matrix_data(matrix_data)
|
||
print("邻接矩阵初始化完成！")
|
||
|
||
|
||
def update_matrix_distance(community_a, community_b, distance):
|
||
"""更新两个小区之间的距离"""
|
||
matrix_data = read_matrix_data()
|
||
community_list = matrix_data["community_list"]
|
||
distance_matrix = matrix_data["distance_matrix"]
|
||
|
||
try:
|
||
idx_a = community_list.index(community_a)
|
||
idx_b = community_list.index(community_b)
|
||
distance_matrix[idx_a][idx_b] = distance
|
||
distance_matrix[idx_b][idx_a] = distance
|
||
write_matrix_data(matrix_data)
|
||
print(f"{community_a} <-> {community_b} 距离更新成功！")
|
||
except ValueError:
|
||
print("小区名称不存在！")
|
||
#v4
|
||
class HouseBSTNode:
|
||
"""
|
||
二叉查找树节点：按指定字段（租金/面积）作为键值
|
||
"""
|
||
|
||
def __init__(self, house_data, key_field="price"):
|
||
self.key = house_data[key_field] # 排序键（租金/面积）
|
||
self.house = house_data # 房源完整数据
|
||
self.left = None # 左子节点（键值更小）
|
||
self.right = None # 右子节点（键值更大）
|
||
|
||
|
||
# V4.0 新增：二叉查找树核心操作
|
||
def build_house_bst(houses, key_field="price"):
|
||
"""
|
||
构建房源二叉查找树（按租金/面积）
|
||
时间复杂度：O(n log n)（最优/平均），O(n²)（最坏，有序数据）
|
||
"""
|
||
if not houses:
|
||
return None
|
||
# 选第一个元素作为根节点
|
||
root = HouseBSTNode(houses[0], key_field)
|
||
# 逐个插入剩余节点
|
||
for house in houses[1:]:
|
||
insert_bst_node(root, house, key_field)
|
||
return root
|
||
|
||
|
||
def insert_bst_node(root, house_data, key_field="price"):
|
||
"""
|
||
向二叉查找树插入节点
|
||
时间复杂度：O(log n)（平均），O(n)（最坏）
|
||
"""
|
||
current = root
|
||
while True:
|
||
# 键值小于当前节点，走左子树
|
||
if house_data[key_field] < current.key:
|
||
if current.left is None:
|
||
current.left = HouseBSTNode(house_data, key_field)
|
||
break
|
||
else:
|
||
current = current.left
|
||
# 键值大于等于当前节点，走右子树
|
||
else:
|
||
if current.right is None:
|
||
current.right = HouseBSTNode(house_data, key_field)
|
||
break
|
||
else:
|
||
current = current.right
|
||
|
||
|
||
def inorder_bst_traversal(root, result_list):
|
||
"""
|
||
二叉查找树中序遍历（输出有序列表）
|
||
时间复杂度：O(n)（遍历所有节点）
|
||
"""
|
||
if root is not None:
|
||
inorder_bst_traversal(root.left, result_list)
|
||
result_list.append(root.house)
|
||
inorder_bst_traversal(root.right, result_list)
|
||
|
||
|
||
def search_bst_range(root, min_val, max_val, key_field="price", result_list=None):
|
||
"""
|
||
二叉查找树范围查询（如：租金5000-8000）
|
||
时间复杂度：O(log n + k)（k为符合条件的节点数）
|
||
"""
|
||
if result_list is None:
|
||
result_list = []
|
||
if root is None:
|
||
return result_list
|
||
|
||
# 若当前键值大于最小值，遍历左子树
|
||
if root.key > min_val:
|
||
search_bst_range(root.left, min_val, max_val, key_field, result_list)
|
||
# 若当前键值在范围内，加入结果
|
||
if min_val <= root.key <= max_val:
|
||
result_list.append(root.house)
|
||
# 若当前键值小于最大值，遍历右子树
|
||
if root.key < max_val:
|
||
search_bst_range(root.right, min_val, max_val, key_field, result_list)
|
||
return result_list
|
||
|
||
def bubble_sort_houses(houses, sort_field="price", reverse=False):
|
||
"""
|
||
V4.0新增 排序知识点
|
||
冒泡排序：按指定字段（租金/面积）排序
|
||
时间复杂度：O(n²)（稳定排序）
|
||
"""
|
||
n = len(houses)
|
||
# 深拷贝避免修改原列表
|
||
sorted_houses = [h.copy() for h in houses]
|
||
for i in range(n):
|
||
swapped = False
|
||
for j in range(0, n - i - 1):
|
||
if sorted_houses[j][sort_field] > sorted_houses[j + 1][sort_field]:
|
||
# 交换元素
|
||
sorted_houses[j], sorted_houses[j + 1] = sorted_houses[j + 1], sorted_houses[j]
|
||
swapped = True
|
||
# 无交换则提前退出
|
||
if not swapped:
|
||
break
|
||
# 降序反转
|
||
if reverse:
|
||
sorted_houses = sorted_houses[::-1]
|
||
return sorted_houses
|
||
|
||
|
||
def quick_sort_houses(houses, sort_field="price", reverse=False):
|
||
"""
|
||
V4新增 排序知识点
|
||
快速排序：按指定字段（租金/面积）排序
|
||
时间复杂度：O(n log n)（平均），O(n²)（最坏）
|
||
"""
|
||
if len(houses) <= 1:
|
||
return houses
|
||
# 深拷贝避免修改原列表
|
||
sorted_houses = [h.copy() for h in houses]
|
||
# 选第一个元素作为基准
|
||
pivot = sorted_houses[0][sort_field]
|
||
left = [h for h in sorted_houses[1:] if h[sort_field] <= pivot]
|
||
right = [h for h in sorted_houses[1:] if h[sort_field] > pivot]
|
||
# 递归排序
|
||
result = quick_sort_houses(left, sort_field) + [sorted_houses[0]] + quick_sort_houses(right, sort_field)
|
||
# 降序反转
|
||
if reverse:
|
||
result = result[::-1]
|
||
return result
|
||
|
||
#V5
|
||
def print_community_graph():
|
||
"""
|
||
文本形式可视化小区图（邻接矩阵）
|
||
时间复杂度：O(n²)，n为小区数量
|
||
"""
|
||
matrix_data = read_matrix_data()
|
||
community_list = matrix_data["community_list"]
|
||
distance_matrix = matrix_data["distance_matrix"]
|
||
|
||
if not community_list:
|
||
print("暂无小区图数据！请先初始化邻接矩阵。")
|
||
return
|
||
|
||
# 打印表头
|
||
print("\n===== 小区距离图（邻接矩阵） =====")
|
||
print(" " + " ".join([c[:4].ljust(4) for c in community_list]))
|
||
# 打印每行数据
|
||
for i, community in enumerate(community_list):
|
||
row_str = community[:4].ljust(4)
|
||
for j in range(len(community_list)):
|
||
val = distance_matrix[i][j]
|
||
if val == float("inf"):
|
||
row_str += " ∞ "
|
||
else:
|
||
row_str += f"{val:5.1f}"
|
||
print(row_str)
|
||
|
||
|
||
def dijkstra_shortest_path(start_community, end_community):
|
||
"""
|
||
Dijkstra算法：计算两个小区间的最短路径
|
||
时间复杂度：O(n²)（邻接矩阵实现），n为小区数量
|
||
return: (最短距离, 路径列表)
|
||
"""
|
||
matrix_data = read_matrix_data()
|
||
community_list = matrix_data["community_list"]
|
||
distance_matrix = matrix_data["distance_matrix"]
|
||
|
||
# 校验小区是否存在
|
||
if start_community not in community_list or end_community not in community_list:
|
||
print("起始/目标小区不存在！")
|
||
return (None, None)
|
||
|
||
n = len(community_list)
|
||
start_idx = community_list.index(start_community)
|
||
end_idx = community_list.index(end_community)
|
||
|
||
# 初始化距离数组和前驱节点数组
|
||
INF = float("inf")
|
||
dist = [INF] * n # 各节点到起点的距离
|
||
visited = [False] * n # 节点是否已访问
|
||
prev = [-1] * n # 前驱节点索引
|
||
|
||
dist[start_idx] = 0 # 起点到自己的距离为0
|
||
|
||
# 核心Dijkstra循环
|
||
for _ in range(n):
|
||
# 找到未访问的距离最小节点
|
||
min_dist = INF
|
||
u = -1
|
||
for i in range(n):
|
||
if not visited[i] and dist[i] < min_dist:
|
||
min_dist = dist[i]
|
||
u = i
|
||
|
||
if u == -1 or dist[u] == INF:
|
||
break # 无可达路径
|
||
visited[u] = True
|
||
|
||
# 松弛操作：更新邻接节点的距离
|
||
for v in range(n):
|
||
if not visited[v] and distance_matrix[u][v] != INF:
|
||
if dist[v] > dist[u] + distance_matrix[u][v]:
|
||
dist[v] = dist[u] + distance_matrix[u][v]
|
||
prev[v] = u
|
||
|
||
# 回溯路径
|
||
if dist[end_idx] == INF:
|
||
print(f"{start_community} 到 {end_community} 无可达路径！")
|
||
return (None, None)
|
||
|
||
# 从终点回溯到起点
|
||
path = []
|
||
current = end_idx
|
||
while current != -1:
|
||
path.append(community_list[current])
|
||
current = prev[current]
|
||
path.reverse() # 反转得到正序路径
|
||
|
||
return (dist[end_idx], path)
|
||
|
||
|
||
def recommend_houses_by_path(target_community, max_distance):
|
||
"""
|
||
推荐目标小区最短距离≤max_distance的周边房源
|
||
"""
|
||
matrix_data = read_matrix_data()
|
||
community_list = matrix_data["community_list"]
|
||
if target_community not in community_list:
|
||
print("目标小区不存在！")
|
||
return []
|
||
|
||
# 计算目标小区到所有小区的最短路径
|
||
recommend_communities = []
|
||
for comm in community_list:
|
||
dist, _ = dijkstra_shortest_path(target_community, comm)
|
||
if dist is not None and dist <= max_distance:
|
||
recommend_communities.append(comm)
|
||
|
||
# 查询这些小区的房源
|
||
all_houses = read_houses()
|
||
recommend_houses = []
|
||
for house in all_houses:
|
||
addr_info = parse_address_string(house["address"])
|
||
if addr_info["community"] in recommend_communities:
|
||
recommend_houses.append(house)
|
||
|
||
return recommend_houses |