はじめに

勤怠管理システムを作ってみたいけど、複数拠点をどう管理するか悩んだことはありませんか？

この記事では、Django + PaSoRi（NFCリーダー）を使って、複数拠点の勤怠データを1台のサーバーに集約するシステムを実装した経験を共有します。

この記事で学べること

🏗️ 分散型システムの基本的な設計パターン
💳 nfcpyを使ったNFCカード読み取りの実装
🔄 Django REST APIでの端末間通信
📊 Djangoモデルでのビジネスロジック実装
⚙️ Management Commandsでのバッチ処理

対象読者

Djangoの基本を理解している方（models, views, urls）
複数デバイス間の通信に興味がある方
実践的なWebアプリケーション設計を学びたい方

完成したコード（GitHub）

🎯 解決したい課題

要件定義

このシステムで実現したかったこと：

各拠点でカード打刻 - 本社、支社、工場など複数の場所で勤怠記録
データの一元管理 - すべてのデータを1つのDBに集約
リアルタイム性 - 打刻後すぐに本社で確認可能
低コスト - 既存のPC + PaSoRiで実現
自動計算 - 勤務時間、残業時間、給与を自動算出

技術的な課題

複数端末から同時にアクセスされるAPI設計
NFCデバイスとの安定した通信
日本の労働法に準拠した計算ロジック
エラーハンドリング（ネットワーク断、カード読み取り失敗）

🏗️ システムアーキテクチャ

全体構成

                    ┌─────────────────────────┐
                    │   Django Server         │
                    │   (本社)                │
                    │   - REST API            │
                    │   - PostgreSQL          │
                    │   - Admin UI            │
                    └────────────┬─────────────┘
                                 │ HTTP/HTTPS
                    ┌────────────┴─────────────┐
                    │   社内LAN / VPN          │
                    └────────────┬─────────────┘
                                 │
         ┌───────────────┬───────┴───────┬────────────────┐
         │               │               │                │
    ┌────▼────┐     ┌────▼────┐    ┌────▼────┐     ┌────▼────┐
    │Terminal │     │Terminal │    │Terminal │     │Terminal │
    │(本社)   │     │(支社)   │     │(工場)   │     │(営業所) │
    │PySide6  │     │PySide6  │    │PySide6  │     │PySide6  │
    │+ nfcpy  │     │+ nfcpy  │    │+ nfcpy  │     │+ nfcpy  │
    └─────────┘     └─────────┘    └─────────┘     └─────────┘

設計のポイント

1. 疎結合な設計

端末とサーバーはHTTP APIのみで通信
端末が落ちてもサーバーは影響を受けない
端末の追加・削除が容易

2. ステートレスなAPI

各リクエストが独立して処理可能
スケーラビリティが高い

3. クライアントサイドUI

PySide6でネイティブアプリケーション
オフライン時のキューイングも可能（今後の拡張）

💻 実装の詳細

1. NFCリーダーの実装（端末側）

nfcpyの基本的な使い方

# nfc_terminal/nfc_kiosk.py（簡略版）

import nfc
import requests
from PySide6.QtCore import QThread, Signal

class NFCReaderThread(QThread):
    card_detected = Signal(str)  # カード検出時のシグナル
    
    def __init__(self):
        super().__init__()
        self.running = True
        self.mode = 'clock_in'  # or 'clock_out'
    
    def run(self):
        """メインループ：NFCカードを待ち受ける"""
        while self.running:
            try:
                # PaSoRiに接続
                clf = nfc.ContactlessFrontend('usb')
                
                # コールバック関数
                def on_connect(tag):
                    # カードの固有IDを取得
                    uid = tag.identifier.hex()
                    
                    # サーバーAPIを呼び出し
                    self.send_to_server(uid)
                    
                    # 接続を終了（次のカード待ち）
                    return False
                
                # カードタッチを待機
                clf.connect(rdwr={'on-connect': on_connect})
                clf.close()
                
            except Exception as e:
                print(f"NFC Error: {e}")
                time.sleep(2)  # エラー時は2秒待機
    
    def send_to_server(self, uid):
        """サーバーAPIに打刻データを送信"""
        try:
            response = requests.post(
                'http://192.168.1.100:8000/api/nfc-clock/',
                json={
                    'uid': uid,
                    'action': self.mode  # clock_in or clock_out
                },
                timeout=5
            )
            
            if response.status_code == 200:
                data = response.json()
                # UIに結果を通知
                self.card_detected.emit(data['message'])
            
        except requests.Timeout:
            self.card_detected.emit('タイムアウト：サーバーに接続できません')
        except requests.ConnectionError:
            self.card_detected.emit('ネットワークエラー')

ポイント解説

QThreadの使用
- NFCの待機はブロッキング処理なので、別スレッドで実行
- メインスレッドのUIがフリーズしない
コールバックパターン
- on_connectでカード検出時の処理を定義
- return Falseで接続を終了し、次のカードを待機
エラーハンドリング
- ネットワーク断、タイムアウトを考慮
- エラー時は適切なメッセージをUIに表示

2. Django API実装（サーバー側）

エンドポイントの実装

# employees/views.py

from django.views.decorators.csrf import csrf_exempt
from django.http import JsonResponse
from django.utils import timezone
import json

@csrf_exempt  # 端末からのPOSTを許可（注意：本番ではトークン認証推奨）
def nfc_clock_api(request):
    """
    NFC打刻API
    
    Request:
        POST /api/nfc-clock/
        {
            "uid": "0123456789abcdef",
            "action": "clock_in"  # or "clock_out"
        }
    
    Response:
        {
            "success": true,
            "message": "田中太郎さん、おはようございます",
            "employee": "田中太郎",
            "time": "09:05"
        }
    """
    
    if request.method != 'POST':
        return JsonResponse({'success': False, 'message': 'POST only'}, status=405)
    
    try:
        # リクエストボディをパース
        data = json.loads(request.body)
        uid = data.get('uid')
        action = data.get('action')
        
        # バリデーション
        if not uid or not action:
            return JsonResponse({
                'success': False,
                'message': 'UIDとactionが必要です'
            }, status=400)
        
        # UIDから社員を検索
        try:
            employee = Employee.objects.get(nfc_id=uid, status='active')
        except Employee.DoesNotExist:
            return JsonResponse({
                'success': False,
                'message': '登録されていないカードです'
            }, status=404)
        
        # 今日の勤怠レコードを取得または作成
        today = timezone.now().date()
        attendance, created = Attendance.objects.get_or_create(
            employee=employee,
            date=today,
            defaults={'source': 'nfc'}
        )
        
        # 出勤処理
        if action == 'clock_in':
            if attendance.clock_in_time:
                return JsonResponse({
                    'success': False,
                    'message': f'{employee.name}さんは既に出勤済みです'
                })
            
            attendance.clock_in_time = timezone.now().time()
            attendance.save()
            
            return JsonResponse({
                'success': True,
                'message': f'{employee.name}さん、おはようございます',
                'employee': employee.name,
                'action': 'clock_in',
                'time': attendance.clock_in_time.strftime('%H:%M')
            })
        
        # 退勤処理
        elif action == 'clock_out':
            if not attendance.clock_in_time:
                return JsonResponse({
                    'success': False,
                    'message': '出勤記録がありません'
                })
            
            if attendance.clock_out_time:
                return JsonResponse({
                    'success': False,
                    'message': '既に退勤済みです'
                })
            
            attendance.clock_out_time = timezone.now().time()
            attendance.save()
            
            return JsonResponse({
                'success': True,
                'message': f'{employee.name}さん、お疲れ様でした',
                'employee': employee.name,
                'action': 'clock_out',
                'time': attendance.clock_out_time.strftime('%H:%M')
            })
        
    except Exception as e:
        return JsonResponse({
            'success': False,
            'message': f'エラー: {str(e)}'
        }, status=500)

設計のポイント

冪等性の考慮
- 同じリクエストが複数回来ても安全
- 「既に出勤済み」などのチェック
明示的なエラーレスポンス
- HTTPステータスコードを適切に使用
- エラーメッセージをJSON形式で返却
トランザクション
- get_or_createで競合を防ぐ
- （本番ではさらにselect_for_update()を検討）

3. ビジネスロジックの実装

Djangoモデルのsave()オーバーライド

# attendance/models.py

from django.db import models
from datetime import datetime, timedelta
from django.conf import settings

class Attendance(models.Model):
    employee = models.ForeignKey('employees.Employee', on_delete=models.CASCADE)
    date = models.DateField('日付')
    clock_in_time = models.TimeField('出勤時刻', blank=True, null=True)
    clock_out_time = models.TimeField('退勤時刻', blank=True, null=True)
    working_hours = models.DecimalField('勤務時間', max_digits=4, decimal_places=2, default=0)
    overtime_hours = models.DecimalField('残業時間', max_digits=4, decimal_places=2, default=0)
    late_flag = models.BooleanField('遅刻', default=False)
    
    class Meta:
        unique_together = ['employee', 'date']
    
    def save(self, *args, **kwargs):
        """保存時に勤務時間を自動計算"""
        
        # 出勤・退勤の両方が記録されている場合のみ計算
        if self.clock_in_time and self.clock_out_time:
            # datetime型に変換
            clock_in = datetime.combine(self.date, self.clock_in_time)
            clock_out = datetime.combine(self.date, self.clock_out_time)
            
            # 日付跨ぎ対応（深夜勤務）
            if clock_out < clock_in:
                clock_out += timedelta(days=1)
            
            # 総時間を計算（時間単位）
            total_hours = (clock_out - clock_in).total_seconds() / 3600
            
            # 休憩時間を控除（6時間超勤務の場合は1時間休憩）
            if total_hours > 6:
                total_hours -= 1
            
            self.working_hours = round(total_hours, 2)
            
            # 残業時間計算（標準労働時間を超えた分）
            standard_hours = settings.STANDARD_WORK_HOURS  # 8時間
            if self.working_hours > standard_hours:
                self.overtime_hours = round(
                    self.working_hours - standard_hours, 2
                )
            else:
                self.overtime_hours = 0
            
            # 遅刻判定
            work_start = datetime.strptime(
                settings.WORK_START_TIME, '%H:%M'
            ).time()  # 09:00
            
            if self.clock_in_time > work_start:
                self.late_flag = True
        
        super().save(*args, **kwargs)

なぜsave()に書くのか？

メリット：

✅ ビジネスロジックがモデルに集約される
✅ 管理画面からの更新でも自動計算される
✅ テストが書きやすい

注意点：

⚠️ 複雑すぎるロジックはサービス層に分離を検討
⚠️ 大量更新時はbulk_create()で回避可能

4. 給与計算のバッチ処理

Management Commandの実装

# payroll/management/commands/generate_payroll.py

from django.core.management.base import BaseCommand
from django.db.models import Sum
from employees.models import Employee
from attendance.models import Attendance
from payroll.models import Payroll
from datetime import datetime
from decimal import Decimal

class Command(BaseCommand):
    help = '月次給与計算を実行'
    
    def add_arguments(self, parser):
        parser.add_argument(
            '--month',
            type=str,
            help='対象月 (YYYY-MM形式)',
        )
    
    def handle(self, *args, **options):
        # 対象月の決定
        target_month = options.get('month')
        if not target_month:
            # 指定がなければ前月
            from dateutil.relativedelta import relativedelta
            last_month = datetime.now().date() - relativedelta(months=1)
            target_month = last_month.strftime('%Y-%m')
        
        year, month = map(int, target_month.split('-'))
        
        self.stdout.write(f'給与計算開始: {target_month}')
        
        # アクティブな全社員を取得
        active_employees = Employee.objects.filter(status='active')
        
        for employee in active_employees:
            # 対象月の勤怠データを集計
            attendances = Attendance.objects.filter(
                employee=employee,
                date__year=year,
                date__month=month
            )
            
            stats = attendances.aggregate(
                total_overtime=Sum('overtime_hours')
            )
            
            overtime_hours = stats['total_overtime'] or Decimal('0')
            
            # 基本給
            base_salary = employee.base_salary
            
            # 時給計算（月160時間基準）
            hourly_rate = employee.hourly_rate or (base_salary / Decimal('160'))
            
            # 残業代（1.25倍）
            overtime_pay = (
                overtime_hours * 
                hourly_rate * 
                Decimal('1.25')
            )
            
            # 給与レコードを作成または更新
            payroll, created = Payroll.objects.update_or_create(
                employee=employee,
                month=target_month,
                defaults={
                    'base_salary': base_salary,
                    'overtime_pay': overtime_pay,
                    'bonus': Decimal('0'),
                    'status': 'calculated'
                }
            )
            
            # 控除額を計算（別メソッド）
            payroll.calculate_deductions()
            payroll.save()
            
            action = '作成' if created else '更新'
            self.stdout.write(
                self.style.SUCCESS(
                    f'{employee.name}の給与を{action}: ¥{payroll.total_salary:,.0f}'
                )
            )
        
        self.stdout.write(
            self.style.SUCCESS(f'給与計算完了: {target_month}')
        )

実行方法

# 前月分を自動計算
python manage.py generate_payroll

# 特定月を指定
python manage.py generate_payroll --month=2025-01

# Cronで自動化
0 0 1 * * cd /path/to/project && python manage.py generate_payroll

🧪 テストの書き方

モデルのテスト例

# attendance/tests.py

from django.test import TestCase
from django.contrib.auth.models import User
from employees.models import Employee
from attendance.models import Attendance
from datetime import date, time

class AttendanceCalculationTest(TestCase):
    def setUp(self):
        """テスト用のユーザーと社員を作成"""
        user = User.objects.create_user(
            username='test001',
            password='testpass'
        )
        self.employee = Employee.objects.create(
            user=user,
            employee_no='TEST001',
            name='テスト太郎',
            department='開発部',
            position='一般',
            hire_date=date(2024, 1, 1),
            base_salary=300000
        )
    
    def test_working_hours_calculation(self):
        """勤務時間が正しく計算されるか"""
        attendance = Attendance.objects.create(
            employee=self.employee,
            date=date.today(),
            clock_in_time=time(9, 0),
            clock_out_time=time(18, 0)
        )
        
        # 9:00-18:00 = 9時間 - 休憩1時間 = 8時間
        self.assertEqual(float(attendance.working_hours), 8.0)
        self.assertEqual(float(attendance.overtime_hours), 0.0)
    
    def test_overtime_calculation(self):
        """残業時間が正しく計算されるか"""
        attendance = Attendance.objects.create(
            employee=self.employee,
            date=date.today(),
            clock_in_time=time(9, 0),
            clock_out_time=time(21, 0)
        )
        
        # 9:00-21:00 = 12時間 - 休憩1時間 = 11時間
        # 残業 = 11 - 8 = 3時間
        self.assertEqual(float(attendance.working_hours), 11.0)
        self.assertEqual(float(attendance.overtime_hours), 3.0)
    
    def test_late_flag(self):
        """遅刻フラグが正しく設定されるか"""
        attendance = Attendance.objects.create(
            employee=self.employee,
            date=date.today(),
            clock_in_time=time(9, 30),  # 9:00より遅い
            clock_out_time=time(18, 30)
        )
        
        self.assertTrue(attendance.late_flag)

🔐 セキュリティの考慮事項

本番環境での改善点

1. API認証の追加

# トークンベース認証の例
from rest_framework.authentication import TokenAuthentication
from rest_framework.permissions import IsAuthenticated

@api_view(['POST'])
@authentication_classes([TokenAuthentication])
@permission_classes([IsAuthenticated])
def nfc_clock_api(request):
    # ...

2. HTTPS必須

Let's Encryptで無料SSL証明書
nginxでリバースプロキシ設定

3. レート制限

# django-ratelimitの使用
from ratelimit.decorators import ratelimit

@ratelimit(key='ip', rate='10/m')
def nfc_clock_api(request):
    # ...

4. ログ記録

import logging

logger = logging.getLogger(__name__)

@csrf_exempt
def nfc_clock_api(request):
    logger.info(f'NFC clock attempt from {request.META.get("REMOTE_ADDR")}')
    # ...

📊 パフォーマンス最適化

N+1問題の回避

# 悪い例
for attendance in Attendance.objects.all():
    print(attendance.employee.name)  # 毎回DBアクセス

# 良い例
for attendance in Attendance.objects.select_related('employee'):
    print(attendance.employee.name)  # 1回のJOINで取得

インデックスの追加

class Attendance(models.Model):
    employee = models.ForeignKey(Employee, on_delete=models.CASCADE)
    date = models.DateField('日付', db_index=True)  # 検索頻度が高い
    
    class Meta:
        indexes = [
            models.Index(fields=['employee', 'date']),  # 複合インデックス
        ]

🎓 学んだこと・詰まったポイント

1. NFCデバイスの安定性

問題: PaSoRiが不定期に接続を失う

解決:

try-exceptで再接続を試みる
USB電源供給の確認（セルフパワーUSBハブ推奨）

2. タイムゾーンの扱い

問題: datetime.now()とtimezone.now()の違い

解決:

Djangoでは常にtimezone.now()を使用
settings.USE_TZ = Trueを設定

3. Decimalの精度

問題: 給与計算で小数点以下の誤差

解決:

floatではなくDecimalを使用
.quantize(Decimal('0.01'))で丸め処理

🔮 今後の改善案

技術的な拡張

WebSocket導入
- リアルタイムダッシュボード更新
- Django Channels使用
キャッシュ戦略
- Redis導入
- 頻繁にアクセスされるデータをキャッシュ
非同期処理
- Celeryで給与計算をバックグラウンド実行
- メール通知の非同期化
モバイル対応
- PWA化
- REST APIのバージョニング

💪 まとめ

このプロジェクトを通じて学んだこと：

✅ 分散システムの基本 - 複数デバイスからの同時アクセス処理
✅ ハードウェア連携 - nfcpyを使ったデバイス制御
✅ ビジネスロジック - Djangoモデルでのドメイン実装
✅ バッチ処理 - Management Commandsの活用
✅ エラーハンドリング - 実運用を想定した設計

特に 「どこにロジックを書くべきか」 という設計判断が重要でした。

モデルのsave(): データ整合性に関わるロジック
ビュー: リクエスト/レスポンス処理
Management Command: バッチ処理
サービス層: 複雑なビジネスロジック（今後の拡張）

📚 参考資料

質問やフィードバックがあれば、GitHubのIssueまたはコメント欄でお気軽にどうぞ！

#Django #Python #NFC #PySide6 #勤怠管理 #分散システム #初心者向け