在旅游行业快速发展的今天如何帮助用户从海量景点信息中找到真正适合自己的目的地成为技术开发者面临的重要挑战。本文将完整实现一个基于Python Flask框架的旅游景点推荐系统涵盖数据爬取、清洗分析、推荐算法建模和可视化展示全流程为计算机专业学生和Web开发爱好者提供可直接复用的毕业设计解决方案。1. 系统架构与技术选型1.1 系统整体架构设计旅游景点推荐系统采用典型的三层架构模式分为数据层、业务逻辑层和表现层。数据层负责景点数据的采集、存储和管理业务逻辑层包含数据处理、推荐算法和用户管理核心功能表现层通过Web界面展示推荐结果和数据分析可视化。系统的工作流程从数据采集开始通过网络爬虫获取多个旅游平台的景点信息经过数据清洗和特征提取后存储到MySQL数据库。当用户访问系统时推荐算法会根据用户的历史行为和偏好生成个性化推荐最终通过Flask框架渲染的Web页面展示给用户。1.2 技术栈选择与优势分析Python作为本项目核心语言其丰富的数据处理库和简洁语法非常适合快速开发。Flask框架轻量灵活适合中小型Web应用开发。主要技术组件包括Web框架Flask 2.0 - 轻量级Web框架易于扩展和定制数据爬取Requests BeautifulSoup - 简单易用的HTTP请求和HTML解析库数据处理Pandas NumPy - 数据清洗、分析和转换推荐算法Scikit-learn - 机器学习算法实现数据可视化ECharts Pyecharts - 交互式图表展示数据库MySQL 8.0 - 关系型数据存储前端界面HTML5 CSS3 JavaScript Layui这种技术组合的优势在于组件成熟稳定、社区支持完善且大多数库都是开源免费的大大降低了开发成本和学习门槛。2. 开发环境搭建2.1 Python环境配置首先需要安装Python 3.8或更高版本。建议使用Anaconda distribution来管理Python环境它可以方便地处理包依赖关系。# 创建专用环境 conda create -n travel-recommendation python3.9 conda activate travel-recommendation # 安装核心依赖包 pip install flask2.3.3 pip install requests2.31.0 pip install beautifulsoup44.12.2 pip install pandas2.0.3 pip install scikit-learn1.3.0 pip install pymysql1.1.0 pip install jieba0.42.12.2 数据库环境配置MySQL数据库安装完成后需要创建系统专用的数据库和用户-- 创建数据库 CREATE DATABASE travel_recommendation CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建专用用户 CREATE USER travel_userlocalhost IDENTIFIED BY secure_password; GRANT ALL PRIVILEGES ON travel_recommendation.* TO travel_userlocalhost; FLUSH PRIVILEGES;2.3 项目目录结构规划合理的目录结构是项目可维护性的基础travel-recommendation-system/ ├── app.py # Flask应用主入口 ├── config.py # 配置文件 ├── requirements.txt # 依赖包列表 ├── static/ # 静态资源 │ ├── css/ │ ├── js/ │ └── images/ ├── templates/ # Jinja2模板 │ ├── base.html │ ├── index.html │ └── recommendation.html ├── spiders/ # 爬虫模块 │ ├── __init__.py │ ├── base_spider.py │ └── ctrip_spider.py ├── models/ # 数据模型 │ ├── __init__.py │ ├── database.py │ └── recommendation.py ├── utils/ # 工具函数 │ ├── __init__.py │ ├── data_clean.py │ └── visualization.py └── data/ # 数据文件 ├── raw/ # 原始数据 └── processed/ # 处理后的数据3. 数据爬取模块实现3.1 爬虫设计原则与伦理规范旅游景点数据爬取需要遵守robots.txt协议和网站使用条款控制请求频率避免对目标服务器造成压力。重要原则包括设置合理的请求间隔建议1-2秒识别并遵守robots.txt限制只爬取公开可访问的数据不爬取个人隐私信息设置超时机制和异常处理3.2 多平台景点数据采集通过分析多个旅游平台的数据结构设计通用的爬虫基类import requests from bs4 import BeautifulSoup import time import json from urllib.parse import urljoin class BaseSpider: def __init__(self, base_url, headersNone): self.base_url base_url self.headers headers or { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } self.session requests.Session() self.session.headers.update(self.headers) def get_page(self, url, paramsNone, timeout10): 获取页面内容包含异常处理 try: response self.session.get(url, paramsparams, timeouttimeout) response.raise_for_status() return response.text except requests.RequestException as e: print(f请求失败: {url}, 错误: {e}) return None def parse_attraction_list(self, html): 解析景点列表页 - 需要子类实现 raise NotImplementedError def parse_attraction_detail(self, html): 解析景点详情页 - 需要子类实现 raise NotImplementedError def save_data(self, data, filename): 保存数据到JSON文件 with open(fdata/raw/{filename}, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) def delay(self, seconds1): 请求延迟避免过于频繁 time.sleep(seconds)3.3 携程景点数据爬取实战以携程旅行网为例实现具体的景点数据采集class CtripSpider(BaseSpider): def __init__(self): super().__init__(https://you.ctrip.com) self.attraction_data [] def get_city_attractions(self, city_id, max_pages10): 获取指定城市的景点列表 for page in range(1, max_pages 1): url f{self.base_url}/sight/{city_id}/s0-p{page}.html html self.get_page(url) if not html: break attractions self.parse_attraction_list(html) for attr in attractions: detail_data self.get_attraction_detail(attr[link]) if detail_data: self.attraction_data.append({**attr, **detail_data}) self.delay(1.5) # 礼貌延迟 def parse_attraction_list(self, html): 解析景点列表页面 soup BeautifulSoup(html, html.parser) attractions [] items soup.select(.sight_item) for item in items: try: name_elem item.select_one(.sight_item_caption a) name name_elem.text.strip() if name_elem else link name_elem[href] if name_elem else rating_elem item.select_one(.score .score_num) rating float(rating_elem.text) if rating_elem else 0 reviews_elem item.select_one(.score .reviews) review_count int(reviews_elem.text.strip(条点评)) if reviews_elem else 0 attractions.append({ name: name, link: urljoin(self.base_url, link), rating: rating, review_count: review_count }) except Exception as e: print(f解析景点项失败: {e}) continue return attractions def get_attraction_detail(self, detail_url): 获取景点详细信息 html self.get_page(detail_url) if not html: return None return self.parse_attraction_detail(html) def parse_attraction_detail(self, html): 解析景点详情页 soup BeautifulSoup(html, html.parser) detail {} # 解析景点描述 desc_elem soup.select_one(.sight_detail_content p) detail[description] desc_elem.text.strip() if desc_elem else # 解析门票价格 price_elem soup.select_one(.price .num) detail[price] float(price_elem.text) if price_elem else 0 # 解析开放时间 time_elem soup.select_one(.baseInfo .item .content) detail[open_time] time_elem.text.strip() if time_elem else # 解析景点标签 tags [tag.text for tag in soup.select(.tag .item)] detail[tags] tags return detail # 使用示例 if __name__ __main__: spider CtripSpider() # 北京城市ID为1爬取前5页数据 spider.get_city_attractions(1, max_pages5) spider.save_data(spider.attraction_data, beijing_attractions.json)4. 数据清洗与特征工程4.1 数据质量评估与清洗原始爬取数据通常包含缺失值、异常值和格式不一致等问题需要进行系统性的数据清洗import pandas as pd import numpy as np import re from sklearn.preprocessing import StandardScaler, LabelEncoder class DataCleaner: def __init__(self, raw_data_path): self.df pd.read_json(raw_data_path) self.cleaned_data None def assess_data_quality(self): 评估数据质量 quality_report { total_records: len(self.df), missing_values: self.df.isnull().sum().to_dict(), duplicate_records: self.df.duplicated().sum(), data_types: self.df.dtypes.to_dict() } return quality_report def handle_missing_values(self): 处理缺失值 # 数值型字段用中位数填充 numeric_cols [rating, price, review_count] for col in numeric_cols: if col in self.df.columns: self.df[col].fillna(self.df[col].median(), inplaceTrue) # 文本字段用默认值填充 text_cols [description, open_time, tags] for col in text_cols: if col in self.df.columns: self.df[col].fillna(, inplaceTrue) def remove_duplicates(self): 去除重复记录 self.df self.df.drop_duplicates(subset[name, link]) def extract_features(self): 特征工程 # 从描述中提取关键词特征 self.df[description_length] self.df[description].apply(len) # 处理标签数据 self.df[tag_count] self.df[tags].apply(lambda x: len(x) if isinstance(x, list) else 0) # 创建价格分段特征 def price_category(price): if price 0: return 免费 elif price 50: return 低价 elif price 100: return 中价 else: return 高价 self.df[price_category] self.df[price].apply(price_category) # 评分分段 self.df[rating_level] pd.cut(self.df[rating], bins[0, 3, 4, 4.5, 5], labels[较差, 一般, 良好, 优秀]) def normalize_features(self): 特征标准化 numeric_features [rating, price, review_count, description_length, tag_count] # 只对存在的数值特征进行标准化 existing_numeric [col for col in numeric_features if col in self.df.columns] if existing_numeric: scaler StandardScaler() scaled_features scaler.fit_transform(self.df[existing_numeric]) self.df[existing_numeric] scaled_features # 对分类特征进行编码 categorical_features [price_category, rating_level] for feature in categorical_features: if feature in self.df.columns: le LabelEncoder() self.df[f{feature}_encoded] le.fit_transform(self.df[feature]) def clean_all(self): 执行完整的数据清洗流程 print(开始数据质量评估...) quality_report self.assess_data_quality() print(quality_report) print(处理缺失值...) self.handle_missing_values() print(去除重复记录...) self.remove_duplicates() print(特征工程...) self.extract_features() print(特征标准化...) self.normalize_features() self.cleaned_data self.df return self.df # 使用示例 cleaner DataCleaner(data/raw/beijing_attractions.json) cleaned_df cleaner.clean_all() cleaned_df.to_csv(data/processed/cleaned_attractions.csv, indexFalse)4.2 文本特征提取与处理景点描述和标签文本包含丰富信息需要专门处理import jieba import jieba.analyse from collections import Counter class TextProcessor: def __init__(self): # 加载旅游领域自定义词典 jieba.load_userdict(data/travel_terms.txt) def extract_keywords(self, text, topk10): 使用TF-IDF提取关键词 if not text or len(text.strip()) 10: return [] keywords jieba.analyse.extract_tags(text, topKtopk, withWeightTrue) return keywords def process_descriptions(self, df): 处理所有景点的描述文本 df[keywords] df[description].apply( lambda x: self.extract_keywords(x, topk5) ) # 提取关键词权重作为特征 df[keyword_features] df[keywords].apply( lambda x: [weight for _, weight in x] if x else [0]*5 ) return df def analyze_sentiment(self, text): 简单情感分析实际项目可使用snownlp等专业库 positive_words [美丽, 壮观, 推荐, 值得, 不错, 精彩] negative_words [拥挤, 失望, 一般, 不值, 普通] if not text: return 0.5 words jieba.lcut(text) pos_count sum(1 for word in words if word in positive_words) neg_count sum(1 for word in words if word in negative_words) total pos_count neg_count if total 0: return 0.5 return pos_count / total # 文本处理应用 processor TextProcessor() df_with_text_features processor.process_descriptions(cleaned_df)5. 推荐算法模型实现5.1 基于内容的推荐算法基于景点特征的相似度计算为用户推荐类似景点from sklearn.metrics.pairwise import cosine_similarity from sklearn.feature_extraction.text import TfidfVectorizer import numpy as np class ContentBasedRecommender: def __init__(self, attractions_df): self.df attractions_df self.feature_matrix None self.similarity_matrix None self._prepare_features() def _prepare_features(self): 准备特征矩阵 # 数值特征 numeric_features [rating, price, review_count, description_length, tag_count] numeric_cols [col for col in numeric_features if col in self.df.columns] # 文本特征景点描述 tfidf TfidfVectorizer(max_features100, stop_words[的, 了, 在]) text_features tfidf.fit_transform(self.df[description].fillna()) # 组合所有特征 if numeric_cols: numeric_matrix self.df[numeric_cols].values self.feature_matrix np.hstack([numeric_matrix, text_features.toarray()]) else: self.feature_matrix text_features.toarray() # 计算相似度矩阵 self.similarity_matrix cosine_similarity(self.feature_matrix) def recommend_similar(self, attraction_id, top_n10): 推荐相似景点 if attraction_id len(self.df): return [] # 获取相似度分数 sim_scores list(enumerate(self.similarity_matrix[attraction_id])) # 按相似度排序 sim_scores sorted(sim_scores, keylambda x: x[1], reverseTrue) # 获取前N个推荐排除自身 sim_scores sim_scores[1:top_n1] # 返回推荐景点索引和相似度 attraction_indices [i[0] for i in sim_scores] return [(self.df.iloc[i][name], sim_scores[j][1]) for j, i in enumerate(attraction_indices)] def recommend_by_features(self, feature_preferences, top_n10): 根据特征偏好推荐 # 创建虚拟用户特征向量 user_vector np.zeros(self.feature_matrix.shape[1]) # 根据偏好调整特征权重这里需要根据实际特征映射 # 简化实现直接计算与所有景点的相似度 similarities cosine_similarity([user_vector], self.feature_matrix) # 获取推荐 sim_scores list(enumerate(similarities[0])) sim_scores sorted(sim_scores, keylambda x: x[1], reverseTrue) return [self.df.iloc[i[0]][name] for i in sim_scores[:top_n]] # 使用示例 recommender ContentBasedRecommender(df_with_text_features) similar_attractions recommender.recommend_similar(0, top_n5) print(相似景点推荐:, similar_attractions)5.2 协同过滤推荐算法基于用户行为的协同过滤推荐from sklearn.decomposition import TruncatedSVD import numpy as np class CollaborativeFiltering: def __init__(self, user_ratings): user_ratings: 用户-景点评分矩阵 self.user_ratings user_ratings self.n_users, self.n_items user_ratings.shape self.predicted_ratings None def matrix_factorization(self, k10, epochs100, alpha0.01, beta0.01): 使用矩阵分解进行评分预测 # 初始化用户和物品特征矩阵 P np.random.normal(0, 0.1, (self.n_users, k)) Q np.random.normal(0, 0.1, (self.n_items, k)) # 训练过程 for epoch in range(epochs): for i in range(self.n_users): for j in range(self.n_items): if self.user_ratings[i, j] 0: # 计算预测误差 error self.user_ratings[i, j] - np.dot(P[i, :], Q[j, :].T) # 更新特征向量 P[i, :] alpha * (2 * error * Q[j, :] - beta * P[i, :]) Q[j, :] alpha * (2 * error * P[i, :] - beta * Q[j, :]) # 计算总误差 total_error 0 for i in range(self.n_users): for j in range(self.n_items): if self.user_ratings[i, j] 0: total_error (self.user_ratings[i, j] - np.dot(P[i, :], Q[j, :].T)) ** 2 total_error (beta/2) * (np.linalg.norm(P[i, :]) ** 2 np.linalg.norm(Q[j, :]) ** 2) if epoch % 10 0: print(fEpoch {epoch}, Error: {total_error}) self.predicted_ratings np.dot(P, Q.T) return self.predicted_ratings def recommend_for_user(self, user_id, top_n10): 为用户生成推荐 if self.predicted_ratings is None: self.matrix_factorization() # 获取用户预测评分 user_ratings self.predicted_ratings[user_id, :] # 排除用户已经评分的项目 rated_items np.where(self.user_ratings[user_id, :] 0)[0] user_ratings[rated_items] -1 # 设置为负值避免被推荐 # 获取Top-N推荐 top_indices np.argsort(user_ratings)[::-1][:top_n] return top_indices, user_ratings[top_indices] # 示例创建模拟用户评分数据 n_users, n_items 100, 50 user_ratings np.random.randint(0, 6, (n_users, n_items)) # 0-5分评分 # 将50%的评分设为0未评分 mask np.random.random((n_users, n_items)) 0.5 user_ratings[mask] 0 cf CollaborativeFiltering(user_ratings) recommendations cf.recommend_for_user(0, top_n5) print(协同过滤推荐:, recommendations)5.3 混合推荐策略结合多种推荐算法的优势class HybridRecommender: def __init__(self, content_recommender, cf_recommender, attractions_df): self.content_rec content_recommender self.cf_rec cf_recommender self.df attractions_df self.weights {content: 0.6, collaborative: 0.4} def hybrid_recommend(self, user_id, attraction_historyNone, top_n10): 混合推荐 recommendations {} # 基于内容的推荐基于用户历史喜好 if attraction_history: content_scores {} for attr_id in attraction_history: similar self.content_rec.recommend_similar(attr_id, top_n*2) for attr_name, score in similar: content_scores[attr_name] content_scores.get(attr_name, 0) score # 协同过滤推荐 cf_indices, cf_scores self.cf_rec.recommend_for_user(user_id, top_n*2) cf_recommendations {self.df.iloc[i][name]: score for i, score in zip(cf_indices, cf_scores)} # 合并推荐结果 for method, recs in [(content, content_scores), (collaborative, cf_recommendations)]: for attr, score in recs.items(): recommendations[attr] recommendations.get(attr, 0) score * self.weights[method] # 排序并返回Top-N sorted_recommendations sorted(recommendations.items(), keylambda x: x[1], reverseTrue)[:top_n] return sorted_recommendations def update_weights(self, user_feedback): 根据用户反馈调整权重简单实现 # 基于用户对推荐结果的点击/评分调整算法权重 content_performance user_feedback.get(content, 0.5) cf_performance user_feedback.get(collaborative, 0.5) total content_performance cf_performance self.weights[content] content_performance / total self.weights[collaborative] cf_performance / total # 混合推荐使用示例 hybrid_rec HybridRecommender(recommender, cf, df_with_text_features) user_history [0, 5, 12] # 用户历史浏览的景点ID hybrid_recommendations hybrid_rec.hybrid_recommend(0, user_history, top_n8) print(混合推荐结果:, hybrid_recommendations)6. Flask Web应用开发6.1 应用结构与路由设计from flask import Flask, render_template, request, jsonify, session import pandas as pd import json app Flask(__name__) app.secret_key your_secret_key_here # 初始化推荐系统组件 df pd.read_csv(data/processed/cleaned_attractions.csv) content_recommender ContentBasedRecommender(df) app.route(/) def index(): 首页 - 展示热门景点和推荐 popular_attractions df.nlargest(8, review_count)[[name, rating, price]].to_dict(records) return render_template(index.html, popular_attractionspopular_attractions) app.route(/recommend, methods[GET, POST]) def recommend(): 景点推荐页面 if request.method POST: # 处理用户偏好表单 preferences { max_price: float(request.form.get(max_price, 200)), min_rating: float(request.form.get(min_rating, 3.5)), tags: request.form.get(tags, ).split(,) } # 基于偏好过滤 filtered df[ (df[price] preferences[max_price]) (df[rating] preferences[min_rating]) ] # 简单的标签匹配实际项目需要更复杂的匹配逻辑 if preferences[tags]: tag_filtered filtered[ filtered[tags].apply(lambda x: any(tag in str(x) for tag in preferences[tags])) ] recommendations tag_filtered.head(10).to_dict(records) else: recommendations filtered.head(10).to_dict(records) return render_template(recommendation.html, recommendationsrecommendations, preferencespreferences) return render_template(recommendation.html) app.route(/attraction/int:attraction_id) def attraction_detail(attraction_id): 景点详情页 if attraction_id len(df): attraction df.iloc[attraction_id].to_dict() # 获取相似推荐 similar content_recommender.recommend_similar(attraction_id, top_n6) # 更新用户浏览历史简化实现 if view_history not in session: session[view_history] [] session[view_history].append(attraction_id) session.modified True return render_template(attraction_detail.html, attractionattraction, similar_attractionssimilar) return 景点不存在, 404 app.route(/api/recommendations) def api_recommendations(): 推荐API接口 user_id request.args.get(user_id, typeint, default0) top_n request.args.get(top_n, typeint, default10) # 获取用户历史简化实现 history session.get(view_history, []) # 生成推荐这里使用基于内容的推荐作为示例 recommendations [] if history: # 基于最近浏览的景点推荐相似景点 recent_attraction history[-1] similar content_recommender.recommend_similar(recent_attraction, top_ntop_n) recommendations [{name: name, score: score} for name, score in similar] else: # 默认推荐热门景点 popular df.nlargest(top_n, review_count)[[name, rating]] recommendations popular.to_dict(records) return jsonify(recommendations) if __name__ __main__: app.run(debugTrue, host0.0.0.0, port5000)6.2 前端界面与可视化使用ECharts实现数据可视化!DOCTYPE html html head meta charsetUTF-8 title旅游景点推荐系统/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script link relstylesheet href{{ url_for(static, filenamecss/style.css) }} /head body div classcontainer header h1智能旅游景点推荐系统/h1 /header div classdashboard div classchart-container div idratingDistribution stylewidth: 600px;height:400px;/div /div div classrecommendation-list h3为您推荐的景点/h3 div idrecommendations {% for attr in recommendations %} div classattraction-card h4{{ attr.name }}/h4 p评分: {{ %.1f|format(attr.rating) }} | 价格: ¥{{ attr.price }}/p /div {% endfor %} /div /div /div /div script // 初始化评分分布图表 var chartDom document.getElementById(ratingDistribution); var myChart echarts.init(chartDom); var option { title: { text: 景点评分分布 }, tooltip: { trigger: axis }, xAxis: { type: category, data: [1-2分, 2-3分, 3-4分, 4-5分] }, yAxis: { type: value }, series: [{ data: [{% for count in rating_counts %}{{ count }},{% endfor %}], type: bar }] }; myChart.setOption(option); // 响应窗口大小变化 window.addEventListener(resize, function() { myChart.resize(); }); /script /body /html7. 系统部署与性能优化7.1 生产环境部署配置使用Gunicorn Nginx部署Flask应用# gunicorn_config.py bind 0.0.0.0:5000 workers 4 worker_class sync worker_connections 1000 timeout 30 max_requests 1000 max_requests_jitter 100 preload_app True# 启动命令 gunicorn -c gunicorn_config.py app:app7.2 数据库优化与缓存策略import redis from functools import wraps import pickle class CacheManager: def __init__(self, hostlocalhost, port6379, db0): self.redis_client redis.Redis(hosthost, portport, dbdb) def cache_result(self, timeout300): 缓存装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 生成缓存键 key f{func.__name__}:{str(args)}:{str(kwargs)} # 尝试从缓存获取 cached self.redis_client.get(key) if cached: return pickle.loads(cached) # 执行函数并缓存结果 result func(*args, **kwargs) self.redis_client.setex(key, timeout, pickle.dumps(result)) return result return wrapper return decorator # 应用缓存 cache CacheManager() app.route(/api/popular) cache.cache_result(timeout600) # 缓存10分钟 def get_popular_attractions(): 获取热门景点带缓存 popular df.nlargest(20, review_count) return jsonify(popular.to_dict(records))7.3 性能监控与日志记录import logging from logging.handlers import RotatingFileHandler import time from functools import wraps def setup_logging(): 配置日志系统 logging.basicConfig(levellogging.INFO) # 文件日志处理器 file_handler RotatingFileHandler(logs/app.log, maxBytes10240, 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) def log_performance(func): 性能监控装饰器 wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) execution_time time.time() - start_time app.logger.info(f{func.__name__} executed in {execution_time:.4f} seconds) # 记录慢查询 if execution_time 1.0: # 超过1秒视为慢查询 app.logger.warning(fSlow performance detected in {func.__name__}) return result return wrapper # 应用性能监控 app.route(/api/search) log_performance def search_attractions(): 景点搜索接口带性能监控 query request.args.get(q, ) # 搜索逻辑... return jsonify(results)8. 系统测试与评估8.1 功能测试用例import unittest from app import app import json class TestTravelRecommendationSystem(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) self.assertIn(b旅游景点推荐系统, response.data) def test_recommendation_api(self): 测试推荐API response self.app.get(/api/recommendations?user_id1top_n5) self