Django Daisy

以下是 ​基于 Django Daisy 的数据平台完整案例教程，涵盖数据采集、分析、可视化及同步功能，结合国内企业级数据平台需求设计：

一、环境准备与项目初始化

1. 创建虚拟环境并安装依赖

python -m venv venv
source venv/bin/activate  # macOS/Linux
venv\Scripts\activate     # Windows

pip install django django-daisy django-plotly-dash mysqlclient celery redis

2. 创建 Django 项目和应用

django-admin startproject data_platform
cd data_platform
python manage.py startapp analytics

3. 配置 settings.py

INSTALLED_APPS = [
    'django_daisy',  # 放在最前面
    'django.contrib.admin',
    'django_plotly_dash.apps.DjangoPlotlyDashConfig',
    'analytics',
    # ...
]

# Daisy 配置
DAISY_SETTINGS = {
    'SITE_TITLE': '企业数据平台',
    'THEME_COLOR': '#10B981',
    'enable_dark_mode': False,
}

# 数据库配置（MySQL + Redis）
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'data_platform',
        'USER': 'root',
        'PASSWORD': 'password',
        'HOST': 'localhost',
        'PORT': '3306',
    },
    'analytics_db': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'analytics',
        'USER': 'root',
        'PASSWORD': 'password',
        'HOST': 'localhost',
        'PORT': '3306',
    }
}

# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

二、数据模型设计

1. 定义核心数据模型（analytics/models.py）

from django.db import models
from django.contrib.auth.models import User

class DataCollection(models.Model):
    SOURCE_CHOICES = [
        ('erp', 'ERP系统'),
        ('crm', 'CRM系统'),
        ('iot', 'IoT设备'),
    ]
    
    source = models.CharField(max_length=10, choices=SOURCE_CHOICES)
    data = models.JSONField(verbose_name='原始数据')
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = 'data_collection'

class DataAnalysis(models.Model):
    analysis_type = models.CharField(max_length=50, verbose_name='分析类型')
    result = models.JSONField(verbose_name='分析结果')
    status = models.IntegerField(
        choices=[(0, '待处理'), (1, '处理中'), (2, '完成')],
        default=0
    )
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = 'data_analysis'

2. 同步数据库

python manage.py makemigrations
python manage.py migrate

三、数据采集与同步

1. 配置多数据库路由（data_platform/routers.py）

class AnalyticsRouter:
    def db_for_read(self, model, **hints):
        if model._meta.app_label == 'analytics':
            return 'analytics_db'
        return 'default'

    def db_for_write(self, model, **hints):
        if model._meta.app_label == 'analytics':
            return 'analytics_db'
        return 'default'

    def allow_relation(self, obj1, obj2, **hints):
        return True

2. 配置信号实现实时同步（analytics/signals.py）

from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import DataCollection

@receiver(post_save, sender=DataCollection)
def sync_to_analytics(sender, instance, **kwargs):
    # 将数据写入分析数据库
    from analytics.models import DataAnalysis
    DataAnalysis.objects.using('analytics_db').create(
        analysis_type='realtime',
        result={'value': instance.data['value']}
    )

3. 启动 Celery 定时任务（analytics/tasks.py）

from celery import shared_task
from .models import DataCollection

@shared_task
def batch_sync_data():
    # 批量同步历史数据
    from django.db import connections
    with connections['analytics_db'].cursor() as cursor:
        cursor.execute("INSERT INTO data_analysis (analysis_type, result) SELECT 'batch', data FROM data_collection")

四、数据分析与可视化

1. 集成 Plotly Dash（analytics/dash_apps.py）

from django_plotly_dash import DashApp
import plotly.express as px
import pandas as pd

class SalesDashboard(DashApp):
    def initialize(self, request):
        self.dash_app = self.build_dash_app()
    
    def build_dash_app(self):
        # 从分析数据库读取数据
        from django.db import connections
        with connections['analytics_db'].cursor() as cursor:
            cursor.execute("SELECT date, SUM(value) as total FROM data_analysis GROUP BY date")
            data = cursor.fetchall()
        
        df = pd.DataFrame(data, columns=['date', 'total'])
        fig = px.line(df, x='date', y='total', title='销售趋势分析')
        
        return dash.Dash(__name__).layout(fig)

2. 配置 URL 路由（data_platform/urls.py）

from django.urls import path, include
from django_plotly_dash.views import add_to_session
from analytics.dash_apps import SalesDashboard

urlpatterns = [
    path('admin/', admin.site.urls),
    path('plotly_dash/', add_to_session, name='plotly_dash'),
    path('dashboard/', SalesDashboard.as_view(), name='sales_dashboard'),
]

五、Django Daisy Admin 配置

1. 注册模型到 Admin（analytics/admin.py）

from django.contrib import admin
from django_daisy.admin import DaisyModelAdmin
from .models import DataCollection, DataAnalysis

class DataCollectionAdmin(DaisyModelAdmin):
    list_display = ('source', 'created_at')
    search_fields = ('source',)
    actions_on_top = True

class DataAnalysisAdmin(DaisyModelAdmin):
    list_display = ('analysis_type', 'status')
    list_filter = ('status',)
    actions_on_top = True

admin.site.register(DataCollection, DataCollectionAdmin)
admin.site.register(DataAnalysis, DataAnalysisAdmin)

2. 自定义 Admin 首页（templates/admin/index.html）

{% extends "admin/base_site.html" %}
{% block content %}
    <div class="stats-grid">
        <div class="stat-card">
            <h3>实时数据量</h3>
            <p>{{ real_time_count }}</p>
        </div>
        <div class="stat-card">
            <h3>分析完成率</h3>
            <p>{{ completion_rate }}</p>
        </div>
    </div>
    {{ block.super }}
{% endblock %}

六、部署与优化

1. 收集静态文件

python manage.py collectstatic

2. 使用 Gunicorn + Nginx 部署

# 安装 Gunicorn
pip install gunicorn

# 启动命令
gunicorn data_platform.wsgi:application --bind 0.0.0.0:8000 --workers 4

3. Nginx 配置示例

server {
    listen 80;
    server_name data-platform.com;

    location /static/ {
        alias /path/to/staticfiles/;
    }

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

七、功能扩展建议

1. 数据质量监控

# 添加数据校验模型
class DataValidation(models.Model):
    collection = models.ForeignKey(DataCollection, on_delete=models.CASCADE)
    error_type = models.CharField(max_length=100)
    severity = models.IntegerField(choices=[(1, '低'), (2, '中'), (3, '高')])

2. 权限精细化控制

# 自定义权限组
from django.contrib.auth.models import Group, Permission

group, created = Group.objects.get_or_create(name='数据分析师')
permissions = Permission.objects.filter(codename__startswith='analytics')
group.permissions.set(permissions)

八、实测数据与性能对比

九、总结

本教程通过 ​Django Daisy + Plotly Dash + Celery​ 的组合，构建了一个完整的企业级数据平台，涵盖数据采集、分析、可视化及同步全流程。关键优势包括：

1. ​现代化界面​：Daisy 的响应式设计提升用户体验
2. ​高效数据处理​：异步任务队列避免阻塞
3. ​扩展性强​：模块化架构支持快速迭代

建议根据实际业务需求，扩展数据清洗、API 集成等模块。