我将为您创建一个智能旅行规划系统,基于性价比算法推荐高性价比的酒店和景点组合。
项目结构
travel_planner_ai/
├── main.py # 主程序入口
├── budget_analyzer.py # 预算分析模块
├── hotel_recommender.py # 酒店推荐模块
├── attraction_recommender.py # 景点推荐模块
├── route_optimizer.py # 路线优化模块
├── utils.py # 工具函数
├── data_loader.py # 数据加载模块
├── requirements.txt # 依赖包列表
├── README.md # 项目说明
└── knowledge_cards.md # 核心知识点卡片
1. requirements.txt
pandas>=1.3.0
numpy>=1.21.0
scikit-learn>=1.0.0
geopy>=2.2.0
matplotlib>=3.5.0
seaborn>=0.11.0
requests>=2.26.0
json>=2.0.9
datetime>=4.3
typing-extensions>=4.0.0
2. utils.py - 工具函数模块
import json
import math
from datetime import datetime, timedelta
from typing import List, Dict, Any, Tuple
import pandas as pd
import numpy as np
class TravelUtils:
"""旅行规划工具类"""
@staticmethod
def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""
计算两个地理坐标之间的距离(公里)
使用Haversine公式
Args:
lat1, lon1: 第一个点的经纬度
lat2, lon2: 第二个点的经纬度
Returns:
两点间的距离(公里)
"""
# 将十进制度数转化为弧度
lat1_rad = math.radians(lat1)
lon1_rad = math.radians(lon1)
lat2_rad = math.radians(lat2)
lon2_rad = math.radians(lon2)
# Haversine公式
dlon = lon2_rad - lon1_rad
dlat = lat2_rad - lat1_rad
a = math.sin(dlat/2)**2 + math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(dlon/2)**2
c = 2 * math.asin(math.sqrt(a))
# 地球半径(公里)
r = 6371.0
return c * r
@staticmethod
def format_currency(amount: float, currency: str = "CNY") -> str:
"""
格式化货币显示
Args:
amount: 金额
currency: 货币符号
Returns:
格式化后的货币字符串
"""
if currency == "CNY":
return f"¥{amount:.2f}"
elif currency == "USD":
return f"${amount:.2f}"
else:
return f"{currency}{amount:.2f}"
@staticmethod
def validate_budget(budget: float, min_budget: float = 500) -> Tuple[bool, str]:
"""
验证预算合理性
Args:
budget: 用户输入的预算
min_budget: 最低预算要求
Returns:
(是否有效, 错误信息)
"""
if budget <= 0:
return False, "预算必须大于0"
if budget < min_budget:
return False, f"预算过低,建议至少{min_budget}元以确保基本旅行体验"
return True, ""
@staticmethod
def validate_duration(days: int) -> Tuple[bool, str]:
"""
验证旅行天数合理性
Args:
days: 旅行天数
Returns:
(是否有效, 错误信息)
"""
if days <= 0:
return False, "旅行天数必须大于0"
if days > 30:
return False, "单次旅行不建议超过30天"
return True, ""
@staticmethod
def calculate_daily_budget(total_budget: float, days: int,
accommodation_ratio: float = 0.4,
attraction_ratio: float = 0.3,
food_ratio: float = 0.2,
transport_ratio: float = 0.1) -> Dict[str, float]:
"""
计算各项支出的日均预算
Args:
total_budget: 总预算
days: 旅行天数
accommodation_ratio: 住宿预算占比
attraction_ratio: 景点预算占比
food_ratio: 餐饮预算占比
transport_ratio: 交通预算占比
Returns:
各项日均预算字典
"""
daily_total = total_budget / days
return {
"accommodation": daily_total * accommodation_ratio,
"attraction": daily_total * attraction_ratio,
"food": daily_total * food_ratio,
"transport": daily_total * transport_ratio,
"total": daily_total
}
@staticmethod
def get_season_info(date: datetime = None) -> Dict[str, Any]:
"""
获取指定日期的季节信息
Args:
date: 指定日期,默认为今天
Returns:
季节信息字典
"""
if date is None:
date = datetime.now()
month = date.month
if month in [3, 4, 5]:
season = "春季"
weather = "温和"
crowd_level = "中等"
elif month in [6, 7, 8]:
season = "夏季"
weather = "炎热"
crowd_level = "较高"
elif month in [9, 10, 11]:
season = "秋季"
weather = "凉爽"
crowd_level = "中等"
else:
season = "冬季"
weather = "寒冷"
crowd_level = "较低"
return {
"season": season,
"weather": weather,
"crowd_level": crowd_level,
"month": month
}
@staticmethod
def save_travel_plan(plan: Dict[str, Any], filename: str = None) -> str:
"""
保存旅行计划到JSON文件
Args:
plan: 旅行计划字典
filename: 文件名,默认为自动生成
Returns:
保存的文件路径
"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"travel_plan_{timestamp}.json"
# 确保目录存在
import os
os.makedirs("plans", exist_ok=True)
filepath = os.path.join("plans", filename)
# 添加保存时间
plan["saved_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(plan, f, ensure_ascii=False, indent=2)
return filepath
@staticmethod
def load_sample_data() -> Dict[str, Any]:
"""
加载示例数据
Returns:
包含示例酒店和景点数据的字典
"""
return {
"hotels": [
{
"id": 1,
"name": "如家快捷酒店",
"city": "北京",
"price": 180,
"rating": 4.2,
"location": [39.9042, 116.4074],
"amenities": ["免费WiFi", "24小时前台", "空调"],
"distance_to_center": 5.2
},
{
"id": 2,
"name": "汉庭酒店",
"city": "北京",
"price": 220,
"rating": 4.4,
"location": [39.9163, 116.3972],
"amenities": ["免费WiFi", "早餐", "健身房"],
"distance_to_center": 3.8
},
{
"id": 3,
"name": "7天连锁酒店",
"city": "上海",
"price": 160,
"rating": 4.0,
"location": [31.2304, 121.4737],
"amenities": ["免费WiFi", "24小时前台"],
"distance_to_center": 4.5
},
{
"id": 4,
"name": "锦江之星",
"city": "上海",
"price": 190,
"rating": 4.3,
"location": [31.2222, 121.4581],
"amenities": ["免费WiFi", "早餐", "商务中心"],
"distance_to_center": 2.1
}
],
"attractions": [
{
"id": 1,
"name": "故宫博物院",
"city": "北京",
"ticket_price": 60,
"rating": 4.8,
"location": [39.9163, 116.3972],
"category": "历史文化",
"visit_duration": 4
},
{
"id": 2,
"name": "天安门广场",
"city": "北京",
"ticket_price": 0,
"rating": 4.6,
"location": [39.9031, 116.3974],
"category": "历史地标",
"visit_duration": 2
},
{
"id": 3,
"name": "外滩",
"city": "上海",
"ticket_price": 0,
"rating": 4.7,
"location": [31.2397, 121.4916],
"category": "城市风光",
"visit_duration": 3
},
{
"id": 4,
"name": "东方明珠",
"city": "上海",
"ticket_price": 180,
"rating": 4.4,
"location": [31.2397, 121.4999],
"category": "现代建筑",
"visit_duration": 2
}
]
}
class DataValidator:
"""数据验证器类"""
@staticmethod
def validate_hotel_data(hotel: Dict[str, Any]) -> Tuple[bool, List[str]]:
"""
验证酒店数据完整性
Args:
hotel: 酒店数据字典
Returns:
(是否有效, 错误信息列表)
"""
errors = []
required_fields = ["name", "city", "price", "rating", "location"]
for field in required_fields:
if field not in hotel:
errors.append(f"缺少必需字段: {field}")
if "price" in hotel and hotel["price"] < 0:
errors.append("价格不能为负数")
if "rating" in hotel and (hotel["rating"] < 0 or hotel["rating"] > 5):
errors.append("评分必须在0-5之间")
if "location" in hotel and len(hotel["location"]) != 2:
errors.append("位置必须是[纬度, 经度]格式")
return len(errors) == 0, errors
@staticmethod
def validate_attraction_data(attraction: Dict[str, Any]) -> Tuple[bool, List[str]]:
"""
验证景点数据完整性
Args:
attraction: 景点数据字典
Returns:
(是否有效, 错误信息列表)
"""
errors = []
required_fields = ["name", "city", "ticket_price", "rating", "location"]
for field in required_fields:
if field not in attraction:
errors.append(f"缺少必需字段: {field}")
if "ticket_price" in attraction and attraction["ticket_price"] < 0:
errors.append("门票价格不能为负数")
if "rating" in attraction and (attraction["rating"] < 0 or attraction["rating"] > 5):
errors.append("评分必须在0-5之间")
if "visit_duration" in attraction and attraction["visit_duration"] <= 0:
errors.append("游览时长必须大于0")
return len(errors) == 0, errors
3. data_loader.py - 数据加载模块
import pandas as pd
import json
import os
from typing import Dict, List, Any, Optional
from utils import TravelUtils, DataValidator
class DataLoader:
"""数据加载器类"""
def __init__(self, data_dir: str = "data"):
"""
初始化数据加载器
Args:
data_dir: 数据目录路径
"""
self.data_dir = data_dir
self.hotels_data = []
self.attractions_data = []
self.cities_data = {}
# 确保数据目录存在
os.makedirs(data_dir, exist_ok=True)
def load_hotels_from_csv(self, filename: str) -> List[Dict[str, Any]]:
"""
从CSV文件加载酒店数据
Args:
filename: CSV文件名
Returns:
酒店数据列表
"""
filepath = os.path.join(self.data_dir, filename)
try:
df = pd.read_csv(filepath, encoding='utf-8')
hotels = []
for _, row in df.iterrows():
hotel = {
"id": int(row.get("id", len(hotels) + 1)),
"name": row["name"],
"city": row["city"],
"price": float(row["price"]),
"rating": float(row["rating"]),
"location": [float(row["latitude"]), float(row["longitude"])],
"amenities": row.get("amenities", "").split(","),
"distance_to_center": float(row.get("distance_to_center", 0))
}
# 验证数据
is_valid, errors = DataValidator.validate_hotel_data(hotel)
if is_valid:
hotels.append(hotel)
else:
print(f"跳过无效酒店数据: {hotel['name']}, 错误: {errors}")
self.hotels_data = hotels
print(f"成功加载 {len(hotels)} 条酒店数据")
return hotels
except FileNotFoundError:
print(f"文件未找到: {filepath}")
return []
except Exception as e:
print(f"加载酒店数据时出错: {e}")
return []
def load_attractions_from_csv(self, filename: str) -> List[Dict[str, Any]]:
"""
从CSV文件加载景点数据
Args:
filename: CSV文件名
Returns:
景点数据列表
"""
filepath = os.path.join(self.data_dir, filename)
try:
df = pd.read_csv(filepath, encoding='utf-8')
attractions = []
for _, row in df.iterrows():
attraction = {
"id": int(row.get("id", len(attractions) + 1)),
"name": row["name"],
"city": row["city"],
"ticket_price": float(row["ticket_price"]),
"rating": float(row["rating"]),
"location": [float(row["latitude"]), float(row["longitude"])],
"category": row.get("category", "其他"),
"visit_duration": int(row.get("visit_duration", 2))
}
# 验证数据
is_valid, errors = DataValidator.validate_attraction_data(attraction)
if is_valid:
attractions.append(attraction)
else:
print(f"跳过无效景点数据: {attraction['name']}, 错误: {errors}")
self.attractions_data = attractions
print(f"成功加载 {len(attractions)} 条景点数据")
return attractions
except FileNotFoundError:
print(f"文件未找到: {filepath}")
return []
except Exception as e:
print(f"加载景点数据时出错: {e}")
return []
def load_data_from_json(self, filename: str) -> Dict[str, Any]:
"""
从JSON文件加载数据
Args:
filename: JSON文件名
Returns:
数据字典
"""
filepath = os.path.join(self.data_dir, filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
if "hotels" in data:
self.hotels_data = data["hotels"]
if "attractions" in data:
self.attractions_data = data["attractions"]
print(f"成功从JSON文件加载数据")
return data
except FileNotFoundError:
print(f"文件未找到: {filepath}")
return {}
except Exception as e:
print(f"加载JSON数据时出错: {e}")
return {}
def save_data_to_json(self, data: Dict[str, Any], filename: str) -> bool:
"""
保存数据到JSON文件
Args:
data: 要保存的数据
filename: 文件名
Returns:
是否保存成功
"""
filepath = os.path.join(self.data_dir, filename)
try:
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"数据已保存至: {filepath}")
return True
except Exception as e:
print(f"保存数据时出错: {e}")
return False
def get_hotels_by_city(self, city: str) -> List[Dict[str, Any]]:
"""
根据城市筛选酒店
Args:
city: 城市名称
Returns:
该城市的酒店列表
"""
return [hotel for hotel in self.hotels_data if hotel["city"] == city]
def get_attractions_by_city(self, city: str) -> List[Dict[str, Any]]:
"""
根据城市筛选景点
Args:
city: 城市名称
Returns:
该城市的景点列表
"""
return [attraction for attraction in self.attractions_data if attraction["city"] == city]
def create_sample_csv_files(self):
"""创建示例CSV文件"""
# 示例酒店数据
hotels_df = pd.DataFrame([
{
"id": 1, "name": "如家快捷酒店", "city": "北京", "price": 180,
"rating": 4.2, "latitude": 39.9042, "longitude": 116.4074,
"amenities": "免费WiFi,24小时前台,空调", "distance_to_center": 5.2
},
{
"id": 2, "name": "汉庭酒店", "city": "北京", "price": 220,
"rating": 4.4, "latitude": 39.9163, "longitude": 116.3972,
"amenities": "免费WiFi,早餐,健身房", "distance_to_center": 3.8
},
{
"id": 3, "name": "7天连锁酒店", "city": "上海", "price": 160,
"rating": 4.0, "latitude": 31.2304, "longitude": 121.4737,
"amenities": "免费WiFi,24小时前台", "distance_to_center": 4.5
},
{
"id": 4, "name": "锦江之星", "city": "上海", "price": 190,
"rating": 4.3, "latitude": 31.2222, "longitude": 121.4581,
"amenities": "免费WiFi,早餐,商务中心", "distance_to_center": 2.1
},
{
"id": 5, "name": "维也纳酒店", "city": "广州", "price": 280,
"rating": 4.6, "latitude": 23.1291, "longitude": 113.2644,
"amenities": "免费WiFi,早餐,健身房,游泳池", "distance_to_center": 6.8
}
])
# 示例景点数据
attractions_df = pd.DataFrame([
{
"id": 1, "name": "故宫博物院", "city": "北京", "ticket_price": 60,
"rating": 4.8, "latitude": 39.9163, "longitude": 116.3972,
"category": "历史文化", "visit_duration": 4
},
{
"id": 2, "name": "天安门广场", "city": "北京", "ticket_price": 0,
"rating": 4.6, "latitude": 39.9031, "longitude": 116.3974,
"category": "历史地标", "visit_duration": 2
},
{
"id": 3, "name": "颐和园", "city": "北京", "ticket_price": 30,
"rating": 4.7, "latitude": 39.9999, "longitude": 116.2750,
"category": "皇家园林", "visit_duration": 5
},
{
"id": 4, "name": "外滩", "city": "上海", "ticket_price": 0,
"rating": 4.7, "latitude": 31.2397, "longitude": 121.4916,
"category": "城市风光", "visit_duration": 3
},
{
"id": 5, "name": "东方明珠", "city": "上海", "ticket_price": 180,
"rating": 4.4, "latitude": 31.2397, "longitude": 121.4999,
"category": "现代建筑", "visit_duration": 2
},
{
"id": 6, "name": "豫园", "city": "上海", "ticket_price": 40,
"rating": 4.3, "latitude": 31.2278, "longitude": 121.4926,
"category": "古典园林", "visit_duration": 2
}
])
# 保存CSV文件
hotels_df.to_csv(os.path.join(self.data_dir, "hotels.csv"), index=False, encoding='utf-8')
attractions_df.to_csv(os.path.join(self.data_dir, "attractions.csv"), index=False, encoding='utf-8')
print("示例CSV文件已创建在 data/ 目录下")
# 示例使用
if __name__ == "__main__":
loader = DataLoader()
loader.create_sample_csv_files()
# 加载数据
hotels = loader.load_hotels_from_csv("hotels.csv")
attractions = loader.load_attractions_from_csv("attractions.csv")
print(f"加载了 {len(hotels)} 家酒店和 {len(attractions)} 个景点")
4. budget_analyzer.py - 预算分析模块
import numpy as np
from typing import Dict, List, Tuple
from utils import Trav
关注我,有更多实用程序等着你!