nouffer/rogaining_srv

Fork 0

Files

Akira 1ba305641e initial setting at 20-Aug-2025

2025-08-20 19:15:19 +09:00

25 KiB

Raw Blame History

統合データベース設計書

1. 概要

1.1 目的

現在運用されているDjango Admin DBとMobServer DBを統合し、一元的なデータ管理システムを構築する。システム停止中であるため、マイグレーション期間を考慮せず、直接統合を実施する。

1.2 基本方針

Django Admin DBをメインデータベースとして位置づけ
MobServerの機能をDjangoベースに統合
PostGISによる地理情報管理を継続
既存データの完全保持

1.3 統合アプローチ

完全統合アプローチ: MobServer DBの全テーブルをDjango管理下に移行
機能重複解消: 同一機能の重複テーブル・フィールドを統一
データ型統一: Django Modelsの型システムに準拠

2. 現行システム分析

2.1 Django Admin DB (rog/models.py)

主要モデル一覧

# ユーザー・認証系
- CustomUser: カスタムユーザーモデル
- UserProfile: ユーザープロフィール

# チーム・参加者管理
- Team: チーム情報
- TeamMember: チームメンバー関係
- Entry: イベント参加エントリー

# イベント・大会管理
- NewEvent2: イベント情報
- Location: 会場・地点情報
- Checkpoint: チェックポイント情報

# GPS・位置情報
- GpsCheckin: GPS位置チェックイン
- GpsLogger: GPS追跡ログ

# スコア・ランキング
- Score: スコア管理
- ResultExport: 結果出力管理

# 外部連携
- S3Upload: AWS S3アップロード管理
- ExternalTeamRegistration: 外部チーム登録

Django DB 特徴

PostGISによる地理情報フィールド対応
Django Admin UI による管理機能
REST Framework による API 提供
多言語対応（i18n）
詳細な権限管理

2.2 MobServer DB (rogaining.sql)

主要テーブル一覧

-- チーム・ユーザー管理
- team_table: チーム基本情報
- user_table: ユーザー情報とチーム関連

-- イベント・チェックポイント
- event_table: イベント基本情報
- checkpoint_table: チェックポイント情報

-- GPS・位置情報
- gps_information: GPS チェックイン情報
- gpslogger_data: GPS ログデータ

-- チャット・コミュニケーション
- chat_log: LINE Bot チャットログ
- chat_status: チャット状態管理

-- スコア・ランキング（VIEW）
- ranking: ランキング計算
- ranking_fix: 修正版ランキング
- cp_counter_*: チェックポイント統計

MobServer DB 特徴

LINE Bot との密接な連携
複雑なビュー構造によるランキング計算
リアルタイム GPS データ処理
チャット履歴の永続化

3. テーブル対応・統合分析

3.1 ユーザー・認証系

統合対象

Django: CustomUser, UserProfile
MobServer: user_table, chat_status

統合方針

統合先: Django CustomUser を拡張

class CustomUser(AbstractUser):
    # 既存フィールド
    username = models.CharField(max_length=150, unique=True)
    email = models.EmailField(unique=True)
    first_name = models.CharField(max_length=150)
    last_name = models.CharField(max_length=150)
    
    # MobServer統合フィールド
    line_user_id = models.CharField(max_length=100, unique=True, null=True)
    chat_status = models.CharField(max_length=50, default='active')
    chat_memory = models.TextField(blank=True)
    
    # 共通フィールド
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

3.2 チーム管理系

統合対象

Django: Team, TeamMember, Entry
MobServer: team_table

統合方針

統合先: Django Team モデルを拡張

class Team(models.Model):
    # 既存フィールド
    name = models.CharField(max_length=100)
    event = models.ForeignKey('NewEvent2', on_delete=models.CASCADE)
    category = models.CharField(max_length=50)
    
    # MobServer統合フィールド
    zekken_number = models.CharField(max_length=20, unique=True)  # team_table.zekken_number
    password = models.CharField(max_length=100)  # team_table.password
    class_name = models.CharField(max_length=100)  # team_table.class_name
    
    # 地理情報
    location = models.PointField(null=True, blank=True)
    
    class Meta:
        unique_together = ['zekken_number', 'event']

3.3 イベント管理系

統合対象

Django: NewEvent2, Location
MobServer: event_table

統合方針

統合先: Django NewEvent2 を拡張

class NewEvent2(models.Model):
    # 既存フィールド
    name = models.CharField(max_length=200)
    description = models.TextField()
    start_date = models.DateTimeField()
    end_date = models.DateTimeField()
    
    # MobServer統合フィールド
    event_code = models.CharField(max_length=50, unique=True)  # event_table.event_code
    start_time = models.CharField(max_length=20)  # event_table.start_time
    event_day = models.CharField(max_length=20)  # event_table.event_day
    
    # 会場情報統合
    venue_location = models.PointField(null=True, blank=True)
    venue_address = models.CharField(max_length=500, blank=True)

3.4 チェックポイント管理系

統合対象

Django: Checkpoint
MobServer: checkpoint_table

統合方針

統合先: Django Checkpoint を拡張

class Checkpoint(models.Model):
    # MobServer完全統合
    cp_number = models.IntegerField()  # checkpoint_table.cp_number
    event = models.ForeignKey('NewEvent2', on_delete=models.CASCADE)
    cp_name = models.CharField(max_length=200)  # checkpoint_table.cp_name
    
    # 位置情報（PostGIS対応）
    location = models.PointField()  # latitude, longitude統合
    
    # ポイント情報
    photo_point = models.IntegerField(default=0)  # checkpoint_table.photo_point
    buy_point = models.IntegerField(default=0)  # checkpoint_table.buy_point
    
    # サンプル・メモ
    sample_photo = models.CharField(max_length=500, blank=True)
    colabo_company_memo = models.TextField(blank=True)
    
    class Meta:
        unique_together = ['cp_number', 'event']

3.5 GPS・位置情報系

統合対象

Django: GpsCheckin, GpsLogger
MobServer: gps_information, gpslogger_data

統合方針

統合先: Django GpsCheckin を拡張

class GpsCheckin(models.Model):
    # MobServer統合
    serial_number = models.AutoField(primary_key=True)
    team = models.ForeignKey('Team', on_delete=models.CASCADE)
    checkpoint = models.ForeignKey('Checkpoint', on_delete=models.CASCADE)
    
    # 画像・証拠
    image_address = models.CharField(max_length=500, blank=True)
    
    # タイミング・ペナルティ
    goal_time = models.CharField(max_length=20, blank=True)
    late_point = models.IntegerField(default=0)
    
    # フラグ
    buy_flag = models.BooleanField(default=False)
    minus_photo_flag = models.BooleanField(default=False)
    
    # 管理情報
    create_user = models.CharField(max_length=100, blank=True)
    update_user = models.CharField(max_length=100, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    # コラボ・メモ
    colabo_company_memo = models.TextField(blank=True)

3.6 チャット・LINE Bot系

統合対象

Django: 新規作成
MobServer: chat_log, chat_status

統合方針

新規作成: LINE Bot 専用モデル

class ChatLog(models.Model):
    serial_number = models.AutoField(primary_key=True)
    user = models.ForeignKey('CustomUser', on_delete=models.CASCADE)
    event = models.ForeignKey('NewEvent2', on_delete=models.CASCADE, null=True)
    
    # LINE Bot情報
    line_user_id = models.CharField(max_length=100)
    message_text = models.TextField()
    message_type = models.CharField(max_length=50, default='text')
    
    # 処理情報
    process_status = models.CharField(max_length=50, default='processed')
    response_text = models.TextField(blank=True)
    
    # タイムスタンプ
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

class ChatStatus(models.Model):
    user = models.OneToOneField('CustomUser', on_delete=models.CASCADE)
    status = models.CharField(max_length=50, default='active')
    memory = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

4. ビューとランキング機能の統合

4.1 MobServer複雑ビュー分析

主要ビュー

-- 基本ランキング
ranking: チーム別総合ポイント計算
ranking_fix: 修正版ランキング
ranking_fix_for_fc_gifu: 地域特化版

-- チェックポイント統計
cp_counter_*: クラス別チェックポイント通過統計
cp_point_usage_summary: ポイント使用状況

-- GPS詳細
gps_detail: GPS + チェックポイント結合
gps_detail_fix: 修正版GPS詳細

4.2 Django実装方針

カスタムマネージャー実装

class RankingManager(models.Manager):
    def get_team_ranking(self, event):
        """チーム別ランキング計算"""
        return self.filter(
            team__event=event
        ).values(
            'team__zekken_number',
            'team__name', 
            'team__class_name'
        ).annotate(
            total_point=models.Sum(
                models.F('checkpoint__photo_point') + 
                models.F('checkpoint__buy_point') - 
                models.F('late_point')
            ),
            cp_count=models.Count('checkpoint'),
            late_point_total=models.Sum('late_point')
        ).order_by('-total_point')
    
    def get_checkpoint_statistics(self, event, class_name=None):
        """チェックポイント統計"""
        qs = self.filter(team__event=event)
        if class_name:
            qs = qs.filter(team__class_name=class_name)
        
        return qs.values(
            'checkpoint__cp_number',
            'checkpoint__cp_name'
        ).annotate(
            visit_count=models.Count('team', distinct=True)
        ).order_by('checkpoint__cp_number')

class GpsCheckin(models.Model):
    # フィールド定義...
    
    objects = RankingManager()

キャッシュ戦略

from django.core.cache import cache

class RankingService:
    @staticmethod
    def get_cached_ranking(event_id, class_name=None):
        cache_key = f"ranking_{event_id}_{class_name or 'all'}"
        ranking = cache.get(cache_key)
        
        if ranking is None:
            ranking = GpsCheckin.objects.get_team_ranking(
                event_id=event_id
            )
            if class_name:
                ranking = ranking.filter(team__class_name=class_name)
            
            cache.set(cache_key, list(ranking), 300)  # 5分キャッシュ
        
        return ranking

5. データ移行戦略

5.1 移行手順

Step 1: Django Model拡張

既存Django Modelsにフィールド追加
マイグレーションファイル生成・実行
インデックス最適化

Step 2: データ移行スクリプト

# management/commands/migrate_mobserver_data.py
from django.core.management.base import BaseCommand
from django.db import transaction
import psycopg2

class Command(BaseCommand):
    def handle(self, *args, **options):
        with transaction.atomic():
            self.migrate_users()
            self.migrate_events()
            self.migrate_teams()
            self.migrate_checkpoints()
            self.migrate_gps_data()
            self.migrate_chat_data()
    
    def migrate_users(self):
        """user_table -> CustomUser"""
        # MobServer DBからデータ取得
        # Django Modelに挿入
        pass
    
    def migrate_teams(self):
        """team_table -> Team"""
        pass
    
    # 他の移行メソッド...

Step 3: 機能統合テスト

API エンドポイント動作確認
ランキング計算精度検証
LINE Bot 連携テスト

5.2 データ整合性保証

制約条件維持

class Team(models.Model):
    # 既存フィールド...
    
    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=['zekken_number', 'event'],
                name='unique_team_per_event'
            ),
            models.CheckConstraint(
                check=models.Q(zekken_number__isnull=False),
                name='zekken_number_required'
            )
        ]

外部キー関係

# MobServer の参照整合性をDjangoで再現
class GpsCheckin(models.Model):
    team = models.ForeignKey(
        'Team', 
        on_delete=models.CASCADE,
        db_constraint=True  # 外部キー制約有効
    )
    checkpoint = models.ForeignKey(
        'Checkpoint',
        on_delete=models.CASCADE,
        db_constraint=True
    )

6. API統合設計

6.1 Django REST Framework統合

シリアライザー統合

class TeamSerializer(serializers.ModelSerializer):
    """MobServer API互換性維持"""
    zekken_number = serializers.CharField()
    team_name = serializers.CharField(source='name')
    class_name = serializers.CharField()
    
    # ランキング情報追加
    total_point = serializers.IntegerField(read_only=True)
    cp_count = serializers.IntegerField(read_only=True)
    
    class Meta:
        model = Team
        fields = [
            'zekken_number', 'team_name', 'class_name',
            'total_point', 'cp_count'
        ]

class GpsCheckinSerializer(serializers.ModelSerializer):
    """GPS チェックイン統合"""
    cp_number = serializers.IntegerField(source='checkpoint.cp_number')
    cp_name = serializers.CharField(source='checkpoint.cp_name')
    photo_point = serializers.IntegerField(source='checkpoint.photo_point')
    buy_point = serializers.IntegerField(source='checkpoint.buy_point')
    
    class Meta:
        model = GpsCheckin
        fields = [
            'serial_number', 'cp_number', 'cp_name',
            'photo_point', 'buy_point', 'late_point',
            'buy_flag', 'goal_time', 'image_address'
        ]

ビューセット統合

class RankingViewSet(viewsets.ReadOnlyModelViewSet):
    """ランキングAPI - MobServer互換"""
    
    def list(self, request):
        event_code = request.query_params.get('event_code')
        class_name = request.query_params.get('class_name')
        
        # キャッシュされたランキング取得
        ranking = RankingService.get_cached_ranking(
            event_code, class_name
        )
        
        serializer = TeamSerializer(ranking, many=True)
        return Response(serializer.data)

class CheckpointStatsViewSet(viewsets.ReadOnlyModelViewSet):
    """チェックポイント統計API"""
    
    def list(self, request):
        event_code = request.query_params.get('event_code')
        stats = GpsCheckin.objects.get_checkpoint_statistics(
            event_code
        )
        return Response(stats)

6.2 LINE Bot API統合

Django統合LINE Bot

# views_apis/line_bot.py
from linebot import LineBotApi, WebhookHandler
from linebot.models import TextMessage, QuickReply

class LineWebhookView(APIView):
    def post(self, request):
        """LINE Bot Webhook処理"""
        signature = request.META.get('HTTP_X_LINE_SIGNATURE')
        body = request.body.decode('utf-8')
        
        try:
            handler.handle(body, signature)
        except InvalidSignatureError:
            return Response(status=400)
        
        return Response({'status': 'ok'})

@handler.add(MessageEvent, message=TextMessage)
def handle_text_message(event):
    """テキストメッセージ処理"""
    user_id = event.source.user_id
    message_text = event.message.text
    
    # Django User取得/作成
    user, created = CustomUser.objects.get_or_create(
        line_user_id=user_id,
        defaults={'username': f'line_{user_id[:8]}'}
    )
    
    # チャットログ保存
    ChatLog.objects.create(
        user=user,
        line_user_id=user_id,
        message_text=message_text
    )
    
    # 応答処理
    response = process_line_message(user, message_text)
    line_bot_api.reply_message(
        event.reply_token,
        TextMessage(text=response)
    )

7. パフォーマンス最適化

7.1 データベースインデックス

統合DB最適化インデックス

-- 統合後の重要インデックス
CREATE INDEX idx_team_event_zekken ON rog_team(event_id, zekken_number);
CREATE INDEX idx_gps_team_checkpoint ON rog_gpscheckin(team_id, checkpoint_id);
CREATE INDEX idx_gps_created_at ON rog_gpscheckin(created_at);
CREATE INDEX idx_checkpoint_event_cp ON rog_checkpoint(event_id, cp_number);

-- PostGIS空間インデックス
CREATE INDEX idx_checkpoint_location ON rog_checkpoint USING GIST(location);
CREATE INDEX idx_team_location ON rog_team USING GIST(location);

-- ランキング計算高速化
CREATE INDEX idx_gps_ranking ON rog_gpscheckin(team_id, buy_flag, late_point);

7.2 キャッシュ戦略

Redis統合キャッシュ

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
        }
    }
}

# キャッシュキー設計
CACHE_KEYS = {
    'ranking': 'ranking:event_{event_id}:class_{class_name}',
    'checkpoint_stats': 'cp_stats:event_{event_id}',
    'team_detail': 'team:event_{event_id}:zekken_{zekken_number}',
}

7.3 Celery非同期処理

8. セキュリティ・権限統合

8.1 Django認証統合

統合認証システム

class RogainingPermission(BasePermission):
    """ロゲイニング専用権限"""
    
    def has_permission(self, request, view):
        if not request.user.is_authenticated:
            return False
        
        # イベント主催者権限
        if hasattr(request.user, 'is_event_organizer'):
            return request.user.is_event_organizer
        
        # チーム所属権限
        return request.user.teams.exists()
    
    def has_object_permission(self, request, view, obj):
        # チーム関連オブジェクトの権限
        if hasattr(obj, 'team'):
            return obj.team.members.filter(user=request.user).exists()
        
        return False

8.2 API認証統合

トークン認証

class APIKeyAuthentication(BaseAuthentication):
    """API キー認証（MobServer互換）"""
    
    def authenticate(self, request):
        api_key = request.META.get('HTTP_X_API_KEY')
        if not api_key:
            return None
        
        try:
            # API キー検証
            user = CustomUser.objects.get(api_key=api_key)
            return (user, None)
        except CustomUser.DoesNotExist:
            raise AuthenticationFailed('Invalid API key')

9. 監視・ログ統合

9.1 統合ログシステム

構造化ログ

import structlog

logger = structlog.get_logger()

class LoggingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        start_time = time.time()
        
        response = self.get_response(request)
        
        logger.info(
            "api_request",
            method=request.method,
            path=request.path,
            status_code=response.status_code,
            duration=time.time() - start_time,
            user_id=request.user.id if request.user.is_authenticated else None
        )
        
        return response

9.2 メトリクス収集

Django統合メトリクス

# monitoring/metrics.py
from prometheus_client import Counter, Histogram

# メトリクス定義
api_requests_total = Counter(
    'api_requests_total',
    'Total API requests',
    ['method', 'endpoint', 'status']
)

ranking_calculation_duration = Histogram(
    'ranking_calculation_duration_seconds',
    'Ranking calculation duration'
)

def track_api_request(method, endpoint, status):
    api_requests_total.labels(
        method=method,
        endpoint=endpoint,
        status=status
    ).inc()

10. 運用・保守計画

10.1 データバックアップ戦略

統合バックアップシステム

#!/bin/bash
# backup_integrated_db.sh

BACKUP_DIR="/backup/rogaining_integrated"
DATE=$(date +%Y%m%d_%H%M%S)

# PostgreSQL + PostGIS バックアップ
pg_dump \
  --host=localhost \
  --port=5432 \
  --username=admin \
  --dbname=rogaining_integrated \
  --format=custom \
  --file="${BACKUP_DIR}/rogaining_${DATE}.dump"

# S3アップロード
aws s3 cp "${BACKUP_DIR}/rogaining_${DATE}.dump" \
  s3://rogaining-backups/db/

# 古いバックアップ削除（30日以上）
find $BACKUP_DIR -name "*.dump" -mtime +30 -delete

10.2 健全性チェック

データ整合性検証

# management/commands/check_data_integrity.py
class Command(BaseCommand):
    def handle(self, *args, **options):
        self.check_team_integrity()
        self.check_gps_integrity()
        self.check_ranking_consistency()
    
    def check_team_integrity(self):
        """チーム情報整合性確認"""
        # zekken_number重複チェック
        duplicates = Team.objects.values(
            'zekken_number', 'event'
        ).annotate(
            count=Count('id')
        ).filter(count__gt=1)
        
        if duplicates.exists():
            self.stdout.write(
                self.style.ERROR(f'Team duplicates found: {duplicates}')
            )
    
    def check_gps_integrity(self):
        """GPS データ整合性確認"""
        # 孤児GPSレコードチェック
        orphan_gps = GpsCheckin.objects.filter(
            team__isnull=True
        ).count()
        
        if orphan_gps > 0:
            self.stdout.write(
                self.style.WARNING(f'Orphan GPS records: {orphan_gps}')
            )

10.3 パフォーマンス監視

自動パフォーマンス監視

# monitoring/performance.py
class PerformanceMonitor:
    @staticmethod
    def check_slow_queries():
        """遅いクエリの検出"""
        with connection.cursor() as cursor:
            cursor.execute("""
                SELECT query, calls, total_time, mean_time
                FROM pg_stat_statements
                WHERE mean_time > 1000  -- 1秒以上
                ORDER BY mean_time DESC
                LIMIT 10;
            """)
            
            slow_queries = cursor.fetchall()
            if slow_queries:
                logger.warning("Slow queries detected", queries=slow_queries)
    
    @staticmethod
    def check_cache_hit_rate():
        """キャッシュヒット率確認"""
        hit_rate = cache.get_stats().get('hit_rate', 0)
        if hit_rate < 0.8:  # 80%未満
            logger.warning(f"Low cache hit rate: {hit_rate}")

11. 結論

11.1 統合効果

システム統一によるメリット

管理効率向上: 単一のDjango Admin UIでの一元管理
開発効率向上: 統一されたDjango開発環境
保守性向上: コードベース統一とドキュメント整備
スケーラビリティ: Django + Celery によるスケーラブルな非同期処理

データ管理改善

データ整合性: Django ORMによる制約管理
バックアップ統一: 単一データベースの包括的バックアップ
パフォーマンス: 最適化されたインデックスとキャッシュ戦略

11.2 今後の拡張計画

機能拡張ロードマップ

Phase 1: 基本統合（4週間）
- データ移行完了
- 基本API動作確認
- LINE Bot統合
Phase 2: 機能強化（4週間）
- リアルタイムランキング
- 高度な統計機能
- モバイルアプリ対応
Phase 3: 運用最適化（4週間）
- パフォーマンス調整
- 監視システム構築
- 自動化システム

11.3 リスク管理

主要リスク項目

データ移行リスク: 段階的移行とバックアップによる対応
パフォーマンスリスク: 事前負荷テストとキャッシュ戦略
機能互換リスク: API互換性維持とテストケース網羅

対応策

詳細な移行計画と検証手順
ロールバック計画の準備
段階的リリースとモニタリング

統合データベース設計により、Django AdminとMobServerの機能を完全に統合し、効率的で拡張可能なシステムを構築します。

25 KiB Raw Blame History Unescape Escape

統合データベース設計書

1. 概要

1.1 目的

1.2 基本方針

1.3 統合アプローチ

2. 現行システム分析

2.1 Django Admin DB (rog/models.py)

主要モデル一覧

Django DB 特徴

2.2 MobServer DB (rogaining.sql)

主要テーブル一覧

MobServer DB 特徴

3. テーブル対応・統合分析

3.1 ユーザー・認証系

統合対象

統合方針

3.2 チーム管理系

統合対象

統合方針

3.3 イベント管理系

統合対象

統合方針

3.4 チェックポイント管理系

統合対象

統合方針

3.5 GPS・位置情報系

統合対象

統合方針

3.6 チャット・LINE Bot系

統合対象

統合方針

4. ビューとランキング機能の統合

4.1 MobServer複雑ビュー分析

主要ビュー

4.2 Django実装方針

カスタムマネージャー実装

キャッシュ戦略

5. データ移行戦略

5.1 移行手順

Step 1: Django Model拡張

Step 2: データ移行スクリプト

Step 3: 機能統合テスト

5.2 データ整合性保証

制約条件維持

外部キー関係

6. API統合設計

6.1 Django REST Framework統合

シリアライザー統合

ビューセット統合

6.2 LINE Bot API統合

Django統合LINE Bot

7. パフォーマンス最適化

7.1 データベースインデックス

統合DB最適化インデックス

7.2 キャッシュ戦略

Redis統合キャッシュ

7.3 Celery非同期処理

ランキング計算の非同期化

8. セキュリティ・権限統合

8.1 Django認証統合

統合認証システム

8.2 API認証統合

トークン認証

9. 監視・ログ統合

9.1 統合ログシステム

構造化ログ

9.2 メトリクス収集

Django統合メトリクス

10. 運用・保守計画

10.1 データバックアップ戦略

統合バックアップシステム

10.2 健全性チェック

データ整合性検証

10.3 パフォーマンス監視

自動パフォーマンス監視

11. 結論

11.1 統合効果

システム統一によるメリット

データ管理改善

25 KiB

Raw Blame History