Compare commits

...
2 Commits
Author SHA1 Message Date
wjy c350d88b01 Merge remote-tracking branch 'origin/main' 2026-06-30 19:55:25 +08:00
wjy 225f66a490 house_service3.0.py 2026-06-30 19:55:10 +08:00
3 changed files with 830 additions and 0 deletions
+77
View File
@@ -0,0 +1,77 @@
import json
import os
#v1
HOUSE_FILE = "./data/houses.json"
def init_db():
"""初始化房源数据文件"""
os.makedirs("./data", exist_ok=True)
if not os.path.exists(HOUSE_FILE):
with open(HOUSE_FILE, 'w', encoding='utf-8') as f:
json.dump([], f, ensure_ascii=False, indent=4)
def read_houses():
"""读取所有房源"""
init_db()
with open(HOUSE_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
def write_houses(houses):
"""写入房源"""
with open(HOUSE_FILE, 'w', encoding='utf-8') as f:
json.dump(houses, f, ensure_ascii=False, indent=4)
#v2
STACK_FILE = "./data/stack_data.json"
def init_stack_db():
"""初始化栈文件"""
os.makedirs("./data", exist_ok=True)
if not os.path.exists(STACK_FILE):
init_data = {"operation_stack": [], "browse_stack": []}
with open(STACK_FILE, 'w', encoding='utf-8') as f:
json.dump(init_data, f, ensure_ascii=False, indent=4)
def read_stack(stack_type):
"""读取指定栈"""
init_stack_db()
with open(STACK_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
return data.get(f"{stack_type}_stack", [])
def write_stack(stack_type, new_stack):
"""写入指定栈"""
init_stack_db()
with open(STACK_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
data[f"{stack_type}_stack"] = new_stack
with open(STACK_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=4)
# v3
MATRIX_FILE = "./data/matrix.json"
def init_matrix_db():
"""初始化小区邻接矩阵文件"""
os.makedirs("./data", exist_ok=True)
if not os.path.exists(MATRIX_FILE):
init_data = {
"community_list": [], # 小区名称列表(一维数组)
"distance_matrix": [] # 小区距离二维数组(邻接矩阵)
}
with open(MATRIX_FILE, 'w', encoding='utf-8') as f:
json.dump(init_data, f, ensure_ascii=False, indent=4)
def read_matrix_data():
"""读取小区列表和距离邻接矩阵"""
init_matrix_db()
with open(MATRIX_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
def write_matrix_data(matrix_dict):
"""保存小区列表和距离邻接矩阵"""
init_matrix_db()
with open(MATRIX_FILE, 'w', encoding='utf-8') as f:
json.dump(matrix_dict, f, ensure_ascii=False, indent=4)
+466
View File
@@ -0,0 +1,466 @@
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
+287
View File
@@ -0,0 +1,287 @@
from house_service import *
def print_menu():
print("\n===== 房屋出租系统 V5.0 =====")
print("1. 新增房源")
print("2. 删除房源")
print("3. 修改房源")
print("4. 查询房源")
print("5. 查看/管理操作记录")
print("6. 查看/管理浏览历史")
print("7. 地址字符串解析测试")
print("8. 按区域关键词模糊查询")
print("9. 初始化小区邻接矩阵")
print("10. 更新小区间距离")
print("11. 房源二叉树范围查询(租金/面积)")
print("12. 房源冒泡排序(租金/面积)")
print("13. 房源快速排序(租金/面积)")
#V5.0 新增菜单
print("14. 可视化小区距离图(邻接矩阵)")
print("15. 计算小区间最短路径(Dijkstra)")
print("16. 基于最短路径推荐房源")
print("17. 退出系统")
print("============================")
# V2原有操作记录子菜单
def op_record_menu():
while True:
print("\n----- 操作记录管理 -----")
print("1. 查看最近10条")
print("2. 取出最新一条")
print("3. 清空记录")
print("4. 返回")
c = input("请选择:")
if c == "1":
lst = get_recent_records("operation", 10)
print(lst if lst else "暂无记录")
elif c == "2":
print(pop_operation_stack() or "暂无记录")
elif c == "3":
clear_stack("operation")
elif c == "4":
break
else:
print("输入有误")
# V2浏览历史子菜单
def browse_history_menu():
while True:
print("\n----- 浏览历史管理 -----")
print("1. 查看最近10条")
print("2. 取出最新一条")
print("3. 清空历史")
print("4. 返回")
c = input("请选择:")
if c == "1":
id_list = get_recent_records("browse", 10)
if not id_list:
print("暂无浏览历史")
continue
for idx, hid in enumerate(id_list, 1):
info = query_house("id", hid)
if info:
h = info[0]
print(f"{idx}. ID:{hid} 地址:{h['address']}")
else:
print(f"{idx}. ID:{hid} 房源已不存在")
elif c == "2":
hid = pop_browse_stack()
if not hid:
print("暂无浏览历史")
continue
info = query_house("id", hid)
print(f"最近浏览ID:{hid}")
elif c == "3":
clear_stack("browse")
elif c == "4":
break
else:
print("输入有误")
def main():
while True:
print_menu()
choice = input("请输入功能编号:")
# ========== V1 原有功能==========
if choice == "1":
addr = input("输入房源地址:")
try:
price = float(input("输入月租金:"))
area = float(input("输入面积:"))
except:
print("租金面积必须是数字!")
continue
htype = input("输入户型:")
if add_house(addr, price, area, htype):
new_id = get_next_house_id() - 1
push_operation_stack(f"新增房源ID:{new_id} 地址:{addr}")
elif choice == "2":
try:
hid = int(input("输入要删除房源ID"))
res = query_house("id", hid)
addr = res[0]["address"] if res else "未知地址"
if delete_house(hid):
push_operation_stack(f"删除房源ID:{hid} 地址:{addr}")
except:
print("ID必须是整数")
elif choice == "3":
try:
hid = int(input("输入要修改房源ID"))
edit = {}
a = input("新地址(回车不修改)")
if a: edit["address"] = a
p = input("新租金(回车不修改)")
if p: edit["price"] = float(p)
ar = input("新面积(回车不修改)")
if ar: edit["area"] = float(ar)
t = input("新户型(回车不修改)")
if t: edit["house_type"] = t
if edit:
if update_house(hid, edit):
push_operation_stack(f"修改房源ID:{hid} 变更:{edit}")
else:
print("未填写任何修改内容")
except:
print("输入格式错误")
elif choice == "4":
print("1-ID 2-地址 3-租金 4-户型")
sel = input("选择查询类型:")
map_dic = {"1": "id", "2": "address", "3": "price", "4": "house_type"}
if sel not in map_dic:
print("选择无效")
continue
val = input("输入查询值:")
res_list = query_house(map_dic[sel], val)
if not res_list:
print("未找到房源")
continue
for h in res_list:
print(f"ID:{h['id']} 地址:{h['address']} 租金:{h['price']}")
if map_dic[sel] == "id":
push_browse_stack(h["id"])
# ========== V2 原有功能 ==========
elif choice == "5":
op_record_menu()
elif choice == "6":
browse_history_menu()
# ========== V3 原有功能==========
elif choice == "7":
full_addr = input("输入完整地址(如:四川省德阳市什邡市XX小区):")
res = parse_address_string(full_addr)
print("解析结果:", res)
elif choice == "8":
keyword = input("输入区域关键词(如:什邡、德阳):")
res_list = fuzzy_query_by_district(keyword)
if not res_list:
print("未找到该区域房源")
else:
for h in res_list:
print(f"ID:{h['id']} 地址:{h['address']}")
elif choice == "9":
community_input = input("输入所有小区名称,用英文逗号分隔:")
community_list = community_input.split(",")
init_community_matrix(community_list)
elif choice == "10":
c1 = input("输入小区A名称:")
c2 = input("输入小区B名称:")
try:
dis = float(input("输入两小区距离:"))
update_matrix_distance(c1, c2, dis)
except:
print("距离必须是数字")
# ========== V4 原有功能==========
elif choice == "11":
print("请选择查询维度:1-租金 2-面积")
dim_choice = input("输入维度编号:")
key_field = "price" if dim_choice == "1" else "area"
field_name = "租金" if dim_choice == "1" else "面积"
try:
min_val = float(input(f"输入{field_name}最小值:"))
max_val = float(input(f"输入{field_name}最大值:"))
except:
print("数值必须是数字!")
continue
all_houses = read_houses()
if not all_houses:
print("暂无房源数据!")
continue
bst_root = build_house_bst(all_houses, key_field)
result = search_bst_range(bst_root, min_val, max_val, key_field)
if not result:
print(f"未找到{field_name}{min_val}-{max_val}之间的房源")
else:
print(f"\n{field_name}{min_val}-{max_val}之间的房源:")
for h in result:
print(f"ID:{h['id']} 地址:{h['address']} {field_name}:{h[key_field]}")
elif choice == "12":
print("请选择排序维度:1-租金 2-面积")
dim_choice = input("输入维度编号:")
sort_field = "price" if dim_choice == "1" else "area"
field_name = "租金" if dim_choice == "1" else "面积"
print("请选择排序方式:1-升序 2-降序")
sort_way = input("输入方式编号:")
reverse = (sort_way == "2")
all_houses = read_houses()
if not all_houses:
print("暂无房源数据!")
continue
sorted_houses = bubble_sort_houses(all_houses, sort_field, reverse)
print(f"\n{field_name}{'降序' if reverse else '升序'}排序结果(冒泡排序):")
for h in sorted_houses:
print(f"ID:{h['id']} 地址:{h['address']} {field_name}:{h[sort_field]}")
elif choice == "13":
print("请选择排序维度:1-租金 2-面积")
dim_choice = input("输入维度编号:")
sort_field = "price" if dim_choice == "1" else "area"
field_name = "租金" if dim_choice == "1" else "面积"
print("请选择排序方式:1-升序 2-降序")
sort_way = input("输入方式编号:")
reverse = (sort_way == "2")
all_houses = read_houses()
if not all_houses:
print("暂无房源数据!")
continue
sorted_houses = quick_sort_houses(all_houses, sort_field, reverse)
print(f"\n{field_name}{'降序' if reverse else '升序'}排序结果(快速排序):")
for h in sorted_houses:
print(f"ID:{h['id']} 地址:{h['address']} {field_name}:{h[sort_field]}")
# ========== V5 新增功能分支 ==========
elif choice == "14":
# 可视化小区图
print_community_graph()
elif choice == "15":
# 计算最短路径
start_comm = input("输入起始小区名称:")
end_comm = input("输入目标小区名称:")
shortest_dist, shortest_path = dijkstra_shortest_path(start_comm, end_comm)
if shortest_dist is not None:
print(f"\n{start_comm}{end_comm} 的最短距离:{shortest_dist:.1f}")
print(f"最短路径:{' -> '.join(shortest_path)}")
elif choice == "16":
# 基于路径推荐房源
target_comm = input("输入目标小区名称:")
try:
max_dist = float(input("输入最大推荐距离:"))
except:
print("距离必须是数字!")
continue
recommend_list = recommend_houses_by_path(target_comm, max_dist)
if not recommend_list:
print(f"未找到{target_comm}周边{max_dist}范围内的房源")
else:
print(f"\n{target_comm}周边{max_dist}范围内的推荐房源:")
for h in recommend_list:
print(f"ID:{h['id']} 地址:{h['address']} 租金:{h['price']} 面积:{h['area']}")
elif choice == "17":
print("系统退出成功!")
break
else:
print("请输入有效编号")
if __name__ == "__main__":
main()