Python游客行为分析：Django+Scrapy数据可视化全栈项目实战

这次我们来看一个完整的Python游客行为分析可视化项目，这个项目结合了Django框架、Scrapy爬虫、数据分析和可视化技术，非常适合作为计算机专业毕业设计或实际应用开发。

这个项目的核心价值在于它提供了一个从数据采集到分析展示的完整解决方案。使用Scrapy框架进行数据爬取，Django作为Web应用框架，结合Python的数据分析库（如Pandas、Matplotlib、Seaborn等）进行数据处理，最终通过可视化图表展示游客行为分析结果。整个项目架构清晰，技术栈实用，具有很强的学习和参考价值。

1. 核心能力速览

2. 适用场景与使用边界

这个项目特别适合以下场景：

• 计算机专业毕业设计：展示完整的技术栈应用能力
• 数据分析学习项目：实践从数据采集到分析展示的全流程
• 旅游行业应用：为景区管理提供游客行为分析参考
• Python全栈开发练习：涵盖前后端和数据处理多个环节

使用边界需要注意：

• 爬虫数据采集需遵守网站robots协议和相关法律法规
• 商业使用需确保数据来源的合法性
• 分析结果仅供参考，重大决策需结合更多数据源
• 项目规模适合中小型数据集，海量数据需优化架构

3. 环境准备与前置条件

3.1 基础软件环境

• 操作系统：Windows 10/11、macOS、Linux均可
• Python版本：Python 3.8及以上（推荐3.9）
• 包管理工具：pip或conda

3.2 开发工具推荐

• 代码编辑器：VS Code、PyCharm
• 数据库：SQLite（开发）、MySQL/PostgreSQL（生产）
• 版本控制：Git

3.3 环境检查清单

在开始前，请确认以下环境已就绪：

# 检查Python版本 python --version # 检查pip版本 pip --version # 检查虚拟环境工具 python -m venv --help

4. 项目架构设计

4.1 整体架构流程

数据采集(Scrapy) → 数据清洗(Pandas) → 数据存储(数据库) → 数据分析(统计计算) → 可视化展示(Django模板) → 前端交互(Chart.js/ECharts)

4.2 目录结构设计

tourist_analysis/ ├── scrapy_crawler/ # Scrapy爬虫项目 │ ├── spiders/ # 爬虫文件 │ ├── items.py # 数据模型 │ ├── pipelines.py # 数据处理管道 │ └── settings.py # 爬虫配置 ├── django_app/ # Django Web应用 │ ├── manage.py │ ├── requirements.txt │ ├── tourist_analysis/ # 主应用 │ │ ├── settings.py │ │ ├── urls.py │ │ └── wsgi.py │ └── analysis/ # 数据分析应用 │ ├── models.py │ ├── views.py │ ├── urls.py │ └── templates/ └── data/ # 数据文件目录 ├── raw/ # 原始数据 ├── processed/ # 处理后的数据 └── results/ # 分析结果

5. 核心模块实现详解

5.1 Scrapy爬虫模块实现

爬虫配置文件（settings.py）：

# scrapy_crawler/settings.py BOT_NAME = 'tourist_crawler' SPIDER_MODULES = ['scrapy_crawler.spiders'] NEWSPIDER_MODULE = 'scrapy_crawler.spiders' # 遵守robots协议 ROBOTSTXT_OBEY = True # 下载延迟设置 DOWNLOAD_DELAY = 2 # 管道配置 ITEM_PIPELINES = { 'scrapy_crawler.pipelines.DataCleaningPipeline': 300, 'scrapy_crawler.pipelines.DatabasePipeline': 400, } # 数据库配置 DATABASE = { 'drivername': 'sqlite', 'database': '../data/tourist_data.db' }

爬虫核心代码示例：

# scrapy_crawler/spiders/tourist_spider.py import scrapy from scrapy_crawler.items import TouristItem from datetime import datetime class TouristBehaviorSpider(scrapy.Spider): name = "tourist_behavior" allowed_domains = ["example-tourism-site.com"] start_urls = ["http://example-tourism-site.com/visitor-data"] def parse(self, response): # 解析游客数据页面 data_sections = response.css('.visitor-data-section') for section in data_sections: item = TouristItem() # 提取游客基本信息 item['timestamp'] = datetime.now() item['visitor_count'] = section.css('.count::text').get() item['age_group'] = section.css('.age-group::text').get() item['stay_duration'] = section.css('.duration::text').get() item['consumption'] = section.css('.consumption::text').get() item['activity_type'] = section.css('.activity::text').get() yield item

5.2 数据清洗与处理管道

# scrapy_crawler/pipelines.py import pandas as pd from scrapy.exceptions import DropItem class DataCleaningPipeline: def process_item(self, item, spider): # 数据清洗逻辑 if item['visitor_count'] is None: raise DropItem("Missing visitor count") # 转换数据类型 try: item['visitor_count'] = int(item['visitor_count']) item['consumption'] = float(item['consumption'].replace('¥', '')) except (ValueError, TypeError): raise DropItem("Invalid data format") return item class DatabasePipeline: def __init__(self, database_url): self.database_url = database_url @classmethod def from_crawler(cls, crawler): return cls( database_url=crawler.settings.get('DATABASE') ) def open_spider(self, spider): # 初始化数据库连接 import sqlite3 self.conn = sqlite3.connect(self.database_url['database']) self.create_table() def create_table(self): create_table_sql = """ CREATE TABLE IF NOT EXISTS tourist_behavior ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME, visitor_count INTEGER, age_group TEXT, stay_duration TEXT, consumption REAL, activity_type TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """ self.conn.execute(create_table_sql) self.conn.commit() def process_item(self, item, spider): # 插入数据到数据库 insert_sql = """ INSERT INTO tourist_behavior (timestamp, visitor_count, age_group, stay_duration, consumption, activity_type) VALUES (?, ?, ?, ?, ?, ?) """ values = ( item['timestamp'], item['visitor_count'], item['age_group'], item['stay_duration'], item['consumption'], item['activity_type'] ) self.conn.execute(insert_sql, values) self.conn.commit() return item def close_spider(self, spider): self.conn.close()

5.3 Django模型设计

# django_app/analysis/models.py from django.db import models from django.utils import timezone class TouristBehavior(models.Model): AGE_GROUP_CHOICES = [ ('0-18', '0-18岁'), ('19-30', '19-30岁'), ('31-45', '31-45岁'), ('46-60', '46-60岁'), ('60+', '60岁以上'), ] timestamp = models.DateTimeField('记录时间') visitor_count = models.IntegerField('游客数量') age_group = models.CharField('年龄段', max_length=10, choices=AGE_GROUP_CHOICES) stay_duration = models.CharField('停留时长', max_length=50) consumption = models.DecimalField('消费金额', max_digits=10, decimal_places=2) activity_type = models.CharField('活动类型', max_length=100) created_at = models.DateTimeField('创建时间', default=timezone.now) class Meta: db_table = 'tourist_behavior' ordering = ['-timestamp'] def __str__(self): return f"{self.timestamp} - {self.visitor_count}人"

5.4 数据分析核心逻辑

# django_app/analysis/analytics.py import pandas as pd import numpy as np from django.db import connection from datetime import datetime, timedelta class TouristAnalytics: def __init__(self): self.df = self.load_data() def load_data(self): """从数据库加载数据到DataFrame""" query = "SELECT * FROM tourist_behavior" return pd.read_sql_query(query, connection) def time_analysis(self): """时间维度分析""" df = self.df.copy() df['hour'] = pd.to_datetime(df['timestamp']).dt.hour df['day_of_week'] = pd.to_datetime(df['timestamp']).dt.dayofweek hourly_visitors = df.groupby('hour')['visitor_count'].sum() daily_pattern = df.groupby('day_of_week')['visitor_count'].mean() return { 'hourly_distribution': hourly_visitors.to_dict(), 'daily_pattern': daily_pattern.to_dict() } def consumption_analysis(self): """消费行为分析""" df = self.df.copy() age_consumption = df.groupby('age_group')['consumption'].agg(['mean', 'sum', 'count']) activity_consumption = df.groupby('activity_type')['consumption'].mean() return { 'age_consumption': age_consumption.to_dict('index'), 'activity_consumption': activity_consumption.to_dict() } def behavioral_clustering(self): """游客行为聚类分析""" from sklearn.cluster import KMeans from sklearn.preprocessing import StandardScaler # 准备特征数据 features = self.df[['visitor_count', 'consumption', 'stay_duration']].copy() features['stay_duration'] = features['stay_duration'].str.extract('(\d+)').astype(float) features = features.fillna(0) # 数据标准化 scaler = StandardScaler() scaled_features = scaler.fit_transform(features) # K-means聚类 kmeans = KMeans(n_clusters=3, random_state=42) clusters = kmeans.fit_predict(scaled_features) self.df['cluster'] = clusters return { 'clusters': clusters.tolist(), 'cluster_centers': kmeans.cluster_centers_.tolist() }

5.5 Django视图与API接口

# django_app/analysis/views.py from django.shortcuts import render from django.http import JsonResponse from django.views import View from .models import TouristBehavior from .analytics import TouristAnalytics import json class DashboardView(View): def get(self, request): """主仪表板视图""" analytics = TouristAnalytics() context = { 'time_analysis': analytics.time_analysis(), 'consumption_analysis': analytics.consumption_analysis(), 'total_records': TouristBehavior.objects.count(), 'recent_data': TouristBehavior.objects.all()[:10] } return render(request, 'analysis/dashboard.html', context) class AnalyticsAPIView(View): def get(self, request): """提供分析数据的API接口""" analytics = TouristAnalytics() analysis_type = request.GET.get('type', 'time') if analysis_type == 'time': data = analytics.time_analysis() elif analysis_type == 'consumption': data = analytics.consumption_analysis() elif analysis_type == 'clustering': data = analytics.behavioral_clustering() else: data = {'error': 'Invalid analysis type'} return JsonResponse(data) class DataUploadView(View): def post(self, request): """数据上传接口""" try: data = json.loads(request.body) for record in data: TouristBehavior.objects.create( timestamp=record['timestamp'], visitor_count=record['visitor_count'], age_group=record['age_group'], stay_duration=record['stay_duration'], consumption=record['consumption'], activity_type=record['activity_type'] ) return JsonResponse({'status': 'success', 'records_added': len(data)}) except Exception as e: return JsonResponse({'status': 'error', 'message': str(e)})

5.6 前端可视化实现

基础HTML模板：

<!-- django_app/analysis/templates/analysis/dashboard.html --> <!DOCTYPE html> <html> <head> <title>游客行为分析系统</title> <script src="https://cdn.jsdelivr.net/npm/chart.js"></script> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <style> .dashboard-container { max-width: 1200px; margin: 0 auto; } .chart-container { margin: 20px 0; padding: 20px; border: 1px solid #ddd; } .stats-cards { display: flex; justify-content: space-between; margin: 20px 0; } .stat-card { flex: 1; padding: 15px; margin: 0 10px; background: #f5f5f5; } </style> </head> <body> <div class="dashboard-container"> <h1>游客行为分析可视化系统</h1> <div class="stats-cards"> <div class="stat-card"> <h3>总记录数</h3> <p id="total-records">{{ total_records }}</p> </div> <div class="stat-card"> <h3>数据分析时间</h3> <p id="analysis-time">{{ analysis_time }}</p> </div> </div> <div class="chart-container"> <h2>游客时间分布</h2> <div id="time-chart" style="height: 400px;"></div> </div> <div class="chart-container"> <h2>消费行为分析</h2> <div id="consumption-chart" style="height: 400px;"></div> </div> </div> <script> // 初始化时间分布图表 function initTimeChart() { const chart = echarts.init(document.getElementById('time-chart')); fetch('/api/analytics/?type=time') .then(response => response.json()) .then(data => { const option = { title: { text: '游客时间分布分析' }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: Object.keys(data.hourly_distribution) }, yAxis: { type: 'value' }, series: [{ data: Object.values(data.hourly_distribution), type: 'line', smooth: true }] }; chart.setOption(option); }); } // 初始化消费分析图表 function initConsumptionChart() { const chart = echarts.init(document.getElementById('consumption-chart')); fetch('/api/analytics/?type=consumption') .then(response => response.json()) .then(data => { const option = { title: { text: '各年龄段消费分析' }, tooltip: { trigger: 'item' }, series: [{ type: 'pie', data: Object.entries(data.age_consumption).map(([age, stats]) => ({ name: age, value: stats.mean })) }] }; chart.setOption(option); }); } // 页面加载完成后初始化图表 document.addEventListener('DOMContentLoaded', function() { initTimeChart(); initConsumptionChart(); }); </script> </body> </html>

6. 项目部署与运行

6.1 环境配置与依赖安装

requirements.txt：

Django==4.2.7 Scrapy==2.11.0 pandas==2.0.3 numpy==1.24.3 scikit-learn==1.3.0 matplotlib==3.7.2 seaborn==0.12.2 sqlite3 python-dateutil==2.8.2

安装命令：

# 创建虚拟环境 python -m venv tourist_analysis_env # 激活虚拟环境 # Windows: tourist_analysis_env\Scripts\activate # Linux/macOS: source tourist_analysis_env/bin/activate # 安装依赖 pip install -r requirements.txt

6.2 数据库初始化

# 进入Django项目目录 cd django_app # 创建数据库迁移 python manage.py makemigrations python manage.py migrate # 创建超级用户（可选） python manage.py createsuperuser

6.3 启动服务

# 启动Django开发服务器 python manage.py runserver # 在另一个终端启动Scrapy爬虫 cd scrapy_crawler scrapy crawl tourist_behavior

6.4 生产环境部署配置

Django生产设置：

# django_app/tourist_analysis/settings_production.py DEBUG = False ALLOWED_HOSTS = ['your-domain.com', 'localhost'] DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': 'tourist_analysis', 'USER': 'your_username', 'PASSWORD': 'your_password', 'HOST': 'localhost', 'PORT': '5432', } } # 静态文件配置 STATIC_ROOT = '/var/www/tourist_analysis/static/'

7. 功能测试与验证

7.1 爬虫功能测试

# tests/test_crawler.py import unittest from scrapy.crawler import CrawlerProcess from scrapy.utils.project import get_project_settings from scrapy_crawler.spiders.tourist_spider import TouristBehaviorSpider class TestCrawler(unittest.TestCase): def test_spider_initialization(self): spider = TouristBehaviorSpider() self.assertEqual(spider.name, 'tourist_behavior') self.assertTrue(len(spider.start_urls) > 0) def test_data_processing(self): # 测试数据清洗管道 from scrapy_crawler.pipelines import DataCleaningPipeline pipeline = DataCleaningPipeline() test_item = { 'visitor_count': '100', 'consumption': '¥500.50' } processed_item = pipeline.process_item(test_item, None) self.assertEqual(processed_item['visitor_count'], 100) self.assertEqual(processed_item['consumption'], 500.50)

7.2 Django功能测试

# tests/test_views.py from django.test import TestCase, Client from django.urls import reverse from analysis.models import TouristBehavior class ViewTests(TestCase): def setUp(self): self.client = Client() # 创建测试数据 TouristBehavior.objects.create( visitor_count=50, age_group='19-30', consumption=200.0, activity_type='观光' ) def test_dashboard_view(self): response = self.client.get(reverse('dashboard')) self.assertEqual(response.status_code, 200) self.assertContains(response, '游客行为分析系统') def test_api_view(self): response = self.client.get('/api/analytics/?type=time') self.assertEqual(response.status_code, 200) self.assertIn('hourly_distribution', response.json())

7.3 数据分析验证

# tests/test_analytics.py import unittest import pandas as pd from analysis.analytics import TouristAnalytics class TestAnalytics(unittest.TestCase): def setUp(self): # 创建测试数据 self.test_data = pd.DataFrame({ 'timestamp': pd.date_range('2024-01-01', periods=100, freq='H'), 'visitor_count': range(100), 'age_group': ['19-30'] * 100, 'consumption': range(100, 200), 'activity_type': ['观光'] * 100 }) def test_time_analysis(self): analytics = TouristAnalytics() analytics.df = self.test_data result = analytics.time_analysis() self.assertIn('hourly_distribution', result) self.assertIn('daily_pattern', result)

8. 性能优化与扩展

8.1 数据库查询优化

# 使用select_related和prefetch_related优化查询 def optimized_queryset(): return TouristBehavior.objects.select_related( 'related_model' ).prefetch_related( 'many_to_many_relation' ).only( 'timestamp', 'visitor_count', 'consumption' ) # 添加数据库索引 class Migration(migrations.Migration): dependencies = [ ('analysis', '0001_initial'), ] operations = [ migrations.AddIndex( model_name='touristbehavior', index=models.Index(fields=['timestamp'], name='timestamp_idx'), ), migrations.AddIndex( model_name='touristbehavior', index=models.Index(fields=['age_group', 'activity_type'], name='age_activity_idx'), ), ]

8.2 缓存策略

# django_app/analysis/views.py from django.core.cache import cache from django.views.decorators.cache import cache_page @cache_page(60 * 15) # 缓存15分钟 def cached_dashboard(request): analytics = TouristAnalytics() # ... 视图逻辑 # 手动缓存数据分析结果 def get_cached_analysis(): cache_key = 'tourist_analysis_results' result = cache.get(cache_key) if not result: analytics = TouristAnalytics() result = { 'time_analysis': analytics.time_analysis(), 'consumption_analysis': analytics.consumption_analysis() } cache.set(cache_key, result, 60 * 10) # 缓存10分钟 return result

8.3 异步任务处理

# 使用Celery处理耗时任务 from celery import shared_task @shared_task def process_large_dataset(dataset_id): """异步处理大数据集""" from .analytics import TouristAnalytics analytics = TouristAnalytics() # 执行耗时分析任务 result = analytics.behavioral_clustering() return result # 在视图中调用异步任务 def start_analysis(request): task = process_large_dataset.delay(dataset_id=1) return JsonResponse({'task_id': task.id})

9. 常见问题与解决方案

9.1 爬虫相关问题

问题1：爬虫被网站屏蔽

# 解决方案：添加请求头和使用代理 custom_settings = { 'DOWNLOAD_DELAY': 3, 'CONCURRENT_REQUESTS': 1, 'HEADERS': { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }, 'PROXIES': [ 'http://proxy1.example.com:8080', 'http://proxy2.example.com:8080', ] }

问题2：数据解析失败

# 解决方案：增加错误处理和备用解析方案 def parse_with_fallback(self, response): try: # 主要解析逻辑 data = response.css('.primary-selector::text').get() if data: return data except Exception as e: self.logger.warning(f"Primary parsing failed: {e}") # 备用解析方案 fallback_data = response.xpath('//div[@class="fallback"]/text()').get() return fallback_data

9.2 Django相关问题

问题1：静态文件无法加载

# settings.py中正确配置静态文件 STATIC_URL = '/static/' STATICFILES_DIRS = [BASE_DIR / "static"] STATIC_ROOT = BASE_DIR / "staticfiles" # 生产环境需要收集静态文件 python manage.py collectstatic

问题2：数据库连接问题

# 使用连接池和重试机制 DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': 'tourist_analysis', 'CONN_MAX_AGE': 600, # 连接存活10分钟 'OPTIONS': { 'connect_timeout': 10, } } }

9.3 数据分析相关问题

问题1：内存不足处理大数据

# 使用分块处理大数据集 def process_large_data_chunked(file_path, chunk_size=10000): chunk_iterator = pd.read_csv(file_path, chunksize=chunk_size) for i, chunk in enumerate(chunk_iterator): # 处理每个数据块 processed_chunk = process_chunk(chunk) # 保存或进一步处理 save_chunk(processed_chunk, i)

问题2：分析结果不准确

# 增加数据质量检查 def validate_data_quality(df): checks = { 'missing_values': df.isnull().sum(), 'data_types': df.dtypes, 'value_ranges': df.describe(), 'duplicates': df.duplicated().sum() } for check_name, result in checks.items(): if isinstance(result, pd.Series) and result.sum() > len(df) * 0.1: raise ValueError(f"Data quality issue: {check_name}")

10. 项目扩展与进阶功能

10.1 实时数据流处理

# 使用WebSocket实现实时数据更新 # django_app/analysis/consumers.py import json from channels.generic.websocket import WebsocketConsumer from .analytics import TouristAnalytics class AnalyticsConsumer(WebsocketConsumer): def connect(self): self.accept() def receive(self, text_data): data = json.loads(text_data) analysis_type = data.get('type', 'time') analytics = TouristAnalytics() if analysis_type == 'real_time': result = analytics.get_real_time_analysis() self.send(text_data=json.dumps(result))

10.2 机器学习预测功能

# 游客数量预测模型 from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split class TouristPredictor: def __init__(self): self.model = RandomForestRegressor(n_estimators=100, random_state=42) def prepare_features(self, df): """准备特征数据""" df['hour'] = pd.to_datetime(df['timestamp']).dt.hour df['day_of_week'] = pd.to_datetime(df['timestamp']).dt.dayofweek df['month'] = pd.to_datetime(df['timestamp']).dt.month df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int) return df[['hour', 'day_of_week', 'month', 'is_weekend']] def train_model(self, df): """训练预测模型""" X = self.prepare_features(df) y = df['visitor_count'] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2) self.model.fit(X_train, y_train) return self.model.score(X_test, y_test) # 返回模型得分 def predict(self, future_dates): """预测未来游客数量""" features = self.prepare_features(future_dates) return self.model.predict(features)

10.3 多数据源集成

# 支持多种数据源接入 class MultiSourceDataLoader: def __init__(self): self.sources = { 'api': APIDataLoader(), 'database': DatabaseLoader(), 'file': FileDataLoader(), 'web': WebScraperLoader() } def load_data(self, source_type, **kwargs): loader = self.sources.get(source_type) if loader: return loader.load(**kwargs) else: raise ValueError(f"Unsupported data source: {source_type}") class APIDataLoader: def load(self, endpoint, params=None): import requests response = requests.get(endpoint, params=params) return response.json()

这个Python游客行为分析可视化项目展示了完整的数据处理流程，从爬虫数据采集到Django Web展示，涵盖了现代Web开发的核心技术栈。项目结构清晰，代码规范，既适合作为学习项目，也具备实际应用的价值。

在实际部署时，建议先从小型数据集开始测试，逐步优化性能。对于生产环境使用，需要考虑数据安全、用户权限管理、系统监控等企业级功能。项目的模块化设计使得扩展新功能变得相对容易，可以根据具体需求添加更多的分析维度和可视化图表。