在旅游信息化快速发展的今天如何从海量旅游数据中挖掘有价值信息并为用户提供个性化推荐成为旅游平台提升用户体验的关键技术挑战。本文将完整实现一个基于Python Flask框架的旅游景点推荐系统涵盖数据爬取、存储、分析、推荐算法和可视化全流程为计算机专业毕业设计提供完整解决方案。1. 系统架构与技术选型1.1 系统整体架构设计旅游景点推荐系统采用典型的三层架构设计包括数据层、业务逻辑层和表现层。数据层负责景点数据的采集、清洗和存储业务逻辑层实现数据分析、推荐算法和用户管理等功能表现层通过Web界面展示推荐结果和可视化分析。系统技术栈选择基于Python生态主要考虑Python在数据处理和机器学习领域的丰富库支持。Flask作为轻量级Web框架适合快速开发原型系统同时具备良好的扩展性。数据库选用MySQL满足结构化数据存储需求。1.2 核心技术组件介绍Flask框架作为系统的Web应用框架Flask提供了路由、模板渲染、请求处理等核心功能。其轻量级特性使得系统启动快速开发效率高。爬虫技术使用requests库发送HTTP请求lxml库解析HTML页面实现旅游网站数据的自动化采集。爬虫模块需要遵守robots协议控制访问频率避免对目标网站造成压力。数据处理库pandas用于数据清洗和预处理numpy提供数值计算支持jieba用于中文文本分词为后续的文本分析做准备。推荐算法采用协同过滤和基于内容的推荐相结合的方式。协同过滤基于用户行为数据基于内容的推荐利用景点特征信息两者结合提升推荐准确性。可视化库ECharts用于生成交互式图表WordCloud生成关键词词云matplotlib用于静态图表绘制。2. 开发环境搭建2.1 Python环境配置首先需要安装Python 3.7及以上版本。建议使用Anaconda发行版它集成了常用的数据科学库方便环境管理。# 创建虚拟环境 conda create -n travel_recommend python3.8 conda activate travel_recommend # 安装核心依赖 pip install flask2.0.1 pip install requests2.25.1 pip install pandas1.3.3 pip install pymysql1.0.2 pip install jieba0.42.1 pip install scikit-learn0.24.2 pip install wordcloud1.8.12.2 数据库环境配置系统使用MySQL 5.7或8.0版本。创建数据库和用户CREATE DATABASE travel_recommend CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER travel_userlocalhost IDENTIFIED BY password123; GRANT ALL PRIVILEGES ON travel_recommend.* TO travel_userlocalhost; FLUSH PRIVILEGES;2.3 项目目录结构建立清晰的目录结构有助于代码维护travel_recommend_system/ ├── app.py # Flask主程序 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 ├── spider/ # 爬虫模块 │ ├── __init__.py │ ├── base_spider.py │ └── travel_spider.py ├── models/ # 数据模型 │ ├── __init__.py │ ├── database.py │ └── entities.py ├── recommender/ # 推荐算法 │ ├── __init__.py │ ├── collaborative_filtering.py │ └── content_based.py ├── static/ # 静态资源 │ ├── css/ │ ├── js/ │ └── images/ └── templates/ # 模板文件 ├── base.html ├── index.html └── recommend.html3. 数据爬取模块实现3.1 爬虫设计原则旅游数据爬取需要遵循合法合规原则严格控制访问频率设置合理的User-Agent避免对目标网站造成负担。实现时加入随机延时、异常重试等机制提高爬虫稳定性。import requests import time import random from lxml import etree import json class TravelSpider: def __init__(self): self.session requests.Session() self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } self.session.headers.update(self.headers) def get_html(self, url, delayTrue): 获取页面HTML内容 try: if delay: time.sleep(random.uniform(1, 3)) # 随机延时1-3秒 response self.session.get(url, timeout10) response.encoding utf-8 if response.status_code 200: return response.text else: print(f请求失败状态码{response.status_code}) return None except Exception as e: print(f请求异常{e}) return None def parse_scenic_info(self, html): 解析景点详细信息 if not html: return None try: selector etree.HTML(html) scenic_data {} # 解析景点名称 name selector.xpath(//h1[classscenic-name]/text()) scenic_data[name] name[0].strip() if name else # 解析评分 rating selector.xpath(//span[classrating]/text()) scenic_data[rating] float(rating[0]) if rating else 0.0 # 解析地址 address selector.xpath(//span[classaddress]/text()) scenic_data[address] address[0].strip() if address else # 解析门票价格 price selector.xpath(//span[classprice]/text()) scenic_data[price] price[0].strip() if price else # 解析景点介绍 description selector.xpath(//div[classdescription]/text()) scenic_data[description] description[0].strip() if description else return scenic_data except Exception as e: print(f解析异常{e}) return None3.2 数据存储设计爬取的数据需要结构化存储到MySQL数据库设计合理的表结构-- 景点基本信息表 CREATE TABLE scenic_spots ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(200) NOT NULL, address VARCHAR(500), price DECIMAL(10,2), rating DECIMAL(3,1), description TEXT, city VARCHAR(100), province VARCHAR(100), tags VARCHAR(500), created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_city (city), INDEX idx_rating (rating) ); -- 用户行为表 CREATE TABLE user_behavior ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, scenic_id INT NOT NULL, behavior_type ENUM(view, collect, comment, rating), behavior_value TEXT, created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (scenic_id) REFERENCES scenic_spots(id), INDEX idx_user_behavior (user_id, scenic_id) ); -- 用户评分表 CREATE TABLE user_ratings ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, scenic_id INT NOT NULL, rating TINYINT NOT NULL CHECK (rating 1 AND rating 5), created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY unique_user_scenic (user_id, scenic_id), FOREIGN KEY (scenic_id) REFERENCES scenic_spots(id) );4. 数据处理与分析模块4.1 数据清洗与预处理爬取的原始数据往往包含噪声和缺失值需要进行数据清洗import pandas as pd import numpy as np import jieba import jieba.analyse from sklearn.feature_extraction.text import TfidfVectorizer class DataProcessor: def __init__(self): self.stop_words self.load_stop_words() def load_stop_words(self): 加载停用词表 try: with open(data/stopwords.txt, r, encodingutf-8) as f: return set([line.strip() for line in f]) except: return set() def clean_data(self, df): 数据清洗 # 处理缺失值 df[price] df[price].fillna(0) df[rating] df[rating].fillna(df[rating].mean()) df[description] df[description].fillna() # 价格格式统一 df[price] df[price].apply(self.standardize_price) # 去除重复数据 df df.drop_duplicates(subset[name, address]) return df def standardize_price(self, price_str): 标准化价格格式 if isinstance(price_str, (int, float)): return float(price_str) try: # 处理免费、100等格式 if 免费 in str(price_str): return 0.0 price_str str(price_str).replace(, ).replace(元, ) return float(price_str) except: return 0.0 def extract_tags(self, text): 从文本中提取关键词标签 if not text: return [] # 使用jieba提取关键词 tags jieba.analyse.extract_tags(text, topK10, withWeightFalse) return [tag for tag in tags if tag not in self.stop_words]4.2 特征工程为推荐算法准备特征数据from sklearn.preprocessing import MinMaxScaler from sklearn.metrics.pairwise import cosine_similarity class FeatureEngineer: def __init__(self): self.scaler MinMaxScaler() self.vectorizer TfidfVectorizer(max_features1000) def prepare_scenic_features(self, df): 准备景点特征矩阵 # 数值特征标准化 numerical_features df[[rating, price]].fillna(0) numerical_scaled self.scaler.fit_transform(numerical_features) # 文本特征向量化 text_features df[description].fillna() df[tags].fillna() text_vectors self.vectorizer.fit_transform(text_features) # 组合特征 from scipy.sparse import hstack feature_matrix hstack([numerical_scaled, text_vectors]) return feature_matrix def calculate_similarity_matrix(self, feature_matrix): 计算景点相似度矩阵 similarity_matrix cosine_similarity(feature_matrix) return similarity_matrix5. 推荐算法实现5.1 协同过滤推荐基于用户的协同过滤算法import numpy as np from scipy.sparse.linalg import svds from sklearn.metrics.pairwise import cosine_similarity class CollaborativeFiltering: def __init__(self, n_factors50, n_similar_users20): self.n_factors n_factors self.n_similar_users n_similar_users self.user_factors None self.item_factors None def fit(self, user_item_matrix): 训练矩阵分解模型 # 矩阵分解 U, sigma, Vt svds(user_item_matrix, kself.n_factors) sigma np.diag(sigma) self.user_factors U self.item_factors Vt.T return self def predict_ratings(self, user_id, n_recommendations10): 为用户预测评分并生成推荐 user_index user_id - 1 # 假设用户ID从1开始 # 计算用户对所有景点的预测评分 user_predicted self.user_factors[user_index].dot(self.item_factors.T) # 获取Top-N推荐 top_indices np.argsort(user_predicted)[::-1][:n_recommendations] return top_indices, user_predicted[top_indices] def find_similar_users(self, user_id, user_item_matrix): 查找相似用户 user_index user_id - 1 user_vector user_item_matrix[user_index] # 计算用户相似度 similarities cosine_similarity(user_vector, user_item_matrix)[0] # 排除自己获取最相似的用户 similar_indices np.argsort(similarities)[::-1][1:self.n_similar_users1] return similar_indices, similarities[similar_indices]5.2 基于内容的推荐class ContentBasedRecommender: def __init__(self, similarity_matrix): self.similarity_matrix similarity_matrix def recommend(self, scenic_id, visited_scenics, n_recommendations10): 基于内容相似度推荐 scenic_index scenic_id - 1 # 获取相似度分数 similarities self.similarity_matrix[scenic_index] # 排除已访问的景点 visited_indices [sid-1 for sid in visited_scenics if sid ! scenic_id] similarities[visited_indices] -1 # 设置为负值避免被选中 # 获取Top-N推荐 recommended_indices np.argsort(similarities)[::-1][:n_recommendations] recommended_scores similarities[recommended_indices] return recommended_indices, recommended_scores def hybrid_recommend(self, user_id, collaborative_scores, content_scores, alpha0.7): 混合推荐结合协同过滤和基于内容的方法 # 加权融合 hybrid_scores alpha * collaborative_scores (1 - alpha) * content_scores # 排序获取最终推荐 final_indices np.argsort(hybrid_scores)[::-1] return final_indices, hybrid_scores[final_indices]6. Flask Web应用开发6.1 应用配置和路由设计from flask import Flask, render_template, request, jsonify, session from flask_cors import CORS import pymysql from config import DevelopmentConfig app Flask(__name__) app.config.from_object(DevelopmentConfig) CORS(app) # 数据库连接 def get_db_connection(): return pymysql.connect( hostapp.config[DB_HOST], userapp.config[DB_USER], passwordapp.config[DB_PASSWORD], databaseapp.config[DB_NAME], charsetutf8mb4 ) app.route(/) def index(): 首页 return render_template(index.html) app.route(/recommend, methods[POST]) def get_recommendations(): 获取推荐结果 try: data request.get_json() user_id data.get(user_id) city data.get(city, ) preferences data.get(preferences, {}) # 调用推荐算法 recommendations generate_recommendations(user_id, city, preferences) return jsonify({ success: True, data: recommendations }) except Exception as e: return jsonify({ success: False, message: str(e) }), 500 app.route(/scenic/int:scenic_id) def scenic_detail(scenic_id): 景点详情页 conn get_db_connection() try: with conn.cursor() as cursor: sql SELECT * FROM scenic_spots WHERE id %s cursor.execute(sql, (scenic_id,)) scenic cursor.fetchone() if scenic: return render_template(scenic_detail.html, scenicscenic) else: return 景点不存在, 404 finally: conn.close()6.2 用户界面设计使用Bootstrap和ECharts构建响应式前端界面!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title旅游景点推荐系统/title link hrefhttps://cdn.jsdelivr.net/npm/bootstrap5.1.3/dist/css/bootstrap.min.css relstylesheet script srchttps://cdn.jsdelivr.net/npm/echarts5.4.2/dist/echarts.min.js/script /head body nav classnavbar navbar-expand-lg navbar-dark bg-primary div classcontainer a classnavbar-brand href/旅游推荐系统/a /div /nav div classcontainer mt-4 div classrow div classcol-md-3 div classcard div classcard-header筛选条件/div div classcard-body form idrecommendForm div classmb-3 label classform-label目的地城市/label input typetext classform-control idcity namecity /div div classmb-3 label classform-label价格范围/label select classform-select idpriceRange namepriceRange option value0不限/option option value1免费/option option value20-100元/option option value3100-300元/option /select /div button typesubmit classbtn btn-primary w-100获取推荐/button /form /div /div /div div classcol-md-9 div classcard div classcard-header推荐结果/div div classcard-body div idrecommendations classrow !-- 推荐结果将通过JavaScript动态加载 -- /div /div /div /div /div /div script srchttps://cdn.jsdelivr.net/npm/bootstrap5.1.3/dist/js/bootstrap.bundle.min.js/script script src/static/js/recommend.js/script /body /html7. 数据可视化实现7.1 景点分布可视化使用ECharts实现景点地理位置分布图// 景点分布地图 function initScenicMap() { const chart echarts.init(document.getElementById(scenic-map)); fetch(/api/scenic/distribution) .then(response response.json()) .then(data { const option { title: { text: 景点地理分布, left: center }, tooltip: { trigger: item, formatter: function(params) { return ${params.name}br/景点数量: ${params.value}; } }, visualMap: { min: 0, max: 100, text: [高, 低], calculable: true, inRange: { color: [#4575b4, #74add1, #abd9e9, #e0f3f8, #ffffbf, #fee090, #fdae61, #f46d43, #d73027] } }, series: [{ name: 景点分布, type: map, map: china, roam: true, emphasis: { label: { show: true } }, data: data.distribution }] }; chart.setOption(option); }); }7.2 推荐结果可视化from wordcloud import WordCloud import matplotlib.pyplot as plt import base64 from io import BytesIO def generate_wordcloud(tags_data): 生成景点标签词云 wordcloud WordCloud( font_pathstatic/fonts/simhei.ttf, width800, height400, background_colorwhite, max_words100 ).generate_from_frequencies(tags_data) # 转换为base64编码图片 buffer BytesIO() plt.figure(figsize(10, 5)) plt.imshow(wordcloud, interpolationbilinear) plt.axis(off) plt.tight_layout() plt.savefig(buffer, formatpng, dpi300, bbox_inchestight) plt.close() image_base64 base64.b64encode(buffer.getvalue()).decode() return fdata:image/png;base64,{image_base64} app.route(/api/visualization/wordcloud) def get_wordcloud(): 获取词云可视化数据 conn get_db_connection() try: with conn.cursor() as cursor: # 获取景点标签频率 sql SELECT tags, COUNT(*) as count FROM scenic_spots GROUP BY tags cursor.execute(sql) tags_data {} for row in cursor.fetchall(): tags row[tags].split(,) if row[tags] else [] for tag in tags: tag tag.strip() if tag: tags_data[tag] tags_data.get(tag, 0) 1 wordcloud_image generate_wordcloud(tags_data) return jsonify({ wordcloud: wordcloud_image }) finally: conn.close()8. 系统测试与优化8.1 功能测试用例设计完整的测试用例确保系统稳定性import unittest from app import app import json class TestTravelRecommendSystem(unittest.TestCase): def setUp(self): self.app app.test_client() self.app.testing True def test_home_page(self): 测试首页访问 response self.app.get(/) self.assertEqual(response.status_code, 200) def test_recommendation_api(self): 测试推荐API test_data { user_id: 1, city: 北京, preferences: {price_range: 0-100} } response self.app.post(/recommend, datajson.dumps(test_data), content_typeapplication/json) self.assertEqual(response.status_code, 200) data json.loads(response.data) self.assertTrue(data[success]) self.assertIn(data, data) def test_scenic_detail(self): 测试景点详情页 response self.app.get(/scenic/1) self.assertEqual(response.status_code, 200) def test_invalid_user(self): 测试无效用户处理 test_data {user_id: 999999} response self.app.post(/recommend, datajson.dumps(test_data), content_typeapplication/json) self.assertEqual(response.status_code, 200) data json.loads(response.data) # 系统应该能够处理无效用户返回默认推荐 self.assertTrue(data[success]) if __name__ __main__: unittest.main()8.2 性能优化策略数据库优化为常用查询字段建立索引使用连接池管理数据库连接对大数据量表进行分表处理缓存策略from flask_caching import Cache cache Cache(config{CACHE_TYPE: SimpleCache}) cache.init_app(app) app.route(/api/scenic/hot) cache.cached(timeout300) # 缓存5分钟 def get_hot_scenics(): 获取热门景点带缓存 # 查询逻辑... return jsonify(hot_scenics)异步处理 对于耗时的推荐计算任务使用Celery进行异步处理from celery import Celery def make_celery(app): celery Celery( app.import_name, backendapp.config[CELERY_RESULT_BACKEND], brokerapp.config[CELERY_BROKER_URL] ) class ContextTask(celery.Task): def __call__(self, *args, **kwargs): with app.app_context(): return self.run(*args, **kwargs) celery.Task ContextTask return celery celery make_celery(app) celery.task def generate_recommendations_async(user_id, preferences): 异步生成推荐结果 # 推荐计算逻辑... return recommendations9. 部署与运维9.1 生产环境部署使用Gunicorn作为WSGI服务器Nginx作为反向代理# 安装Gunicorn pip install gunicorn # 启动命令 gunicorn -w 4 -b 0.0.0.0:8000 app:app # Nginx配置 server { listen 80; server_name your_domain.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } location /static { alias /path/to/your/static; } }9.2 监控与日志配置日志记录和性能监控import logging from logging.handlers import RotatingFileHandler def setup_logging(app): 配置日志系统 if not app.debug: # 生产环境日志配置 file_handler RotatingFileHandler( logs/travel_recommend.log, maxBytes1024*1024, backupCount10 ) file_handler.setFormatter(logging.Formatter( %(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d] )) file_handler.setLevel(logging.INFO) app.logger.addHandler(file_handler) app.logger.setLevel(logging.INFO)10. 常见问题与解决方案10.1 爬虫被封禁问题问题现象爬虫频繁被目标网站封禁IP解决方案使用代理IP池轮换请求设置合理的请求间隔时间模拟真实浏览器行为User-Agent轮换遵守robots.txt协议class ProxyManager: def __init__(self): self.proxies self.load_proxies() self.current_index 0 def get_proxy(self): 获取代理IP if not self.proxies: return None proxy self.proxies[self.current_index] self.current_index (self.current_index 1) % len(self.proxies) return proxy def load_proxies(self): 加载代理IP列表 # 从文件或API获取代理IP try: with open(proxies.txt, r) as f: return [line.strip() for line in f if line.strip()] except: return []10.2 推荐冷启动问题问题现象新用户或新景点缺乏历史数据推荐效果差解决方案新用户基于人口统计学信息推荐热门景点新景点基于内容相似度进行推荐采用混合推荐策略平衡准确性和多样性10.3 系统性能瓶颈问题现象用户量增大时系统响应变慢解决方案使用Redis缓存热门推荐结果对推荐算法进行离线计算采用分布式计算框架处理大数据量数据库查询优化和索引优化本系统完整实现了旅游景点推荐的全流程从数据采集到推荐生成再到可视化展示。系统采用模块化设计便于扩展和维护为计算机毕业设计提供了完整的技术解决方案。在实际部署时需要根据具体需求调整参数配置并持续优化推荐算法以提升用户体验。