分刻みの通知設定

Author: fezzlk

scripts/backfill_notification_schedules.py

@@ -0,0 +1,19 @@
+from src.Infrastructure.Repositories import (
+    line_user_repository,
+    notification_schedule_repository,
+)
+
+
+def main() -> None:
+    line_users = line_user_repository.find()
+    for line_user in line_users:
+        notification_schedule_repository.upsert(
+            line_user_id=line_user.line_user_id,
+            notify_time="12:00",
+            timezone_name="Asia/Tokyo",
+        )
+    print(f"backfilled notification_schedules: {len(line_users)}")
+
+
+if __name__ == "__main__":
+    main()

src/Domains/Entities/NotificationSchedule.py

@@ -0,0 +1,31 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+
+@dataclass()
+class NotificationSchedule:
+    line_user_id: str
+    notify_time: str
+    timezone: str
+    enabled: bool
+    next_notify_at: datetime
+    created_at: datetime
+    updated_at: datetime
+
+    def __init__(
+        self,
+        line_user_id: str = None,
+        notify_time: str = "12:00",
+        timezone: str = "Asia/Tokyo",
+        enabled: bool = True,
+        next_notify_at: datetime = None,
+        created_at: datetime = datetime.now(),
+        updated_at: datetime = datetime.now(),
+    ):
+        self.line_user_id = line_user_id
+        self.notify_time = notify_time
+        self.timezone = timezone
+        self.enabled = enabled
+        self.next_notify_at = next_notify_at
+        self.created_at = created_at
+        self.updated_at = updated_at

src/Infrastructure/Repositories/NotificationScheduleRepository.py

@@ -0,0 +1,149 @@
+from datetime import datetime, timedelta, timezone
+from typing import List, Optional
+from zoneinfo import ZoneInfo
+
+from google.cloud import firestore
+
+from src.Domains.Entities.NotificationSchedule import NotificationSchedule
+from src.firestore_client import firestore_client
+
+
+class NotificationScheduleRepository:
+    _collection_name = "notification_schedules"
+    _default_notify_time = "12:00"
+    _default_timezone = "Asia/Tokyo"
+
+    def _collection(self):
+        return firestore_client.collection(self._collection_name)
+
+    def find_due(self, now_utc: datetime, limit: int = 500) -> List[NotificationSchedule]:
+        query = (
+            self._collection()
+            .where("enabled", "==", True)
+            .where("next_notify_at", "<=", now_utc)
+            .order_by("next_notify_at", direction=firestore.Query.ASCENDING)
+            .limit(limit)
+        )
+        schedules: List[NotificationSchedule] = []
+        for doc in query.stream():
+            data = doc.to_dict() or {}
+            data["line_user_id"] = doc.id
+            schedules.append(NotificationSchedule(**data))
+        return schedules
+
+    def find_by_line_user_id(self, line_user_id: str) -> Optional[NotificationSchedule]:
+        snapshot = self._collection().document(line_user_id).get()
+        if not snapshot.exists:
+            return None
+        data = snapshot.to_dict() or {}
+        data["line_user_id"] = snapshot.id
+        return NotificationSchedule(**data)
+
+    def compute_next_notify_at(
+        self,
+        notify_time: Optional[str],
+        timezone_name: Optional[str],
+        base_time_utc: datetime,
+    ) -> datetime:
+        safe_notify_time = notify_time or self._default_notify_time
+        safe_timezone = timezone_name or self._default_timezone
+        try:
+            hour, minute = [int(x) for x in safe_notify_time.split(":", 1)]
+        except (ValueError, TypeError):
+            hour, minute = [int(x) for x in self._default_notify_time.split(":", 1)]
+
+        local_tz = ZoneInfo(safe_timezone)
+        local_now = base_time_utc.astimezone(local_tz)
+        try:
+            next_local = local_now.replace(
+                hour=hour,
+                minute=minute,
+                second=0,
+                microsecond=0,
+            )
+        except ValueError:
+            default_hour, default_minute = [int(x) for x in self._default_notify_time.split(":", 1)]
+            next_local = local_now.replace(
+                hour=default_hour,
+                minute=default_minute,
+                second=0,
+                microsecond=0,
+            )
+        if next_local <= local_now:
+            next_local += timedelta(days=1)
+        return next_local.astimezone(timezone.utc)
+
+    def claim_and_schedule_next(self, line_user_id: str, now_utc: datetime) -> bool:
+        doc_ref = self._collection().document(line_user_id)
+        transaction = firestore_client.transaction()
+
+        @firestore.transactional
+        def _claim(tx):
+            snapshot = doc_ref.get(transaction=tx)
+            if not snapshot.exists:
+                return False
+
+            data = snapshot.to_dict() or {}
+            if not data.get("enabled", True):
+                return False
+
+            next_notify_at = data.get("next_notify_at")
+            if next_notify_at is None:
+                return False
+            if next_notify_at > now_utc:
+                return False
+
+            next_at = self.compute_next_notify_at(
+                notify_time=data.get("notify_time"),
+                timezone_name=data.get("timezone"),
+                base_time_utc=now_utc,
+            )
+            tx.update(
+                doc_ref,
+                {
+                    "next_notify_at": next_at,
+                    "updated_at": now_utc,
+                },
+            )
+            return True
+
+        return bool(_claim(transaction))
+
+    def upsert(
+        self,
+        line_user_id: str,
+        notify_time: str = "12:00",
+        timezone_name: str = "Asia/Tokyo",
+        enabled: Optional[bool] = None,
+        base_time_utc: Optional[datetime] = None,
+    ) -> NotificationSchedule:
+        now_utc = base_time_utc or datetime.now(timezone.utc)
+        next_notify_at = self.compute_next_notify_at(
+            notify_time=notify_time,
+            timezone_name=timezone_name,
+            base_time_utc=now_utc,
+        )
+        doc_ref = self._collection().document(line_user_id)
+        snapshot = doc_ref.get()
+        data = snapshot.to_dict() or {}
+        merged_enabled = data.get("enabled", True) if enabled is None else enabled
+        created_at = data.get("created_at", now_utc)
+        payload = {
+            "notify_time": notify_time,
+            "timezone": timezone_name,
+            "enabled": merged_enabled,
+            "next_notify_at": next_notify_at,
+            "updated_at": now_utc,
+        }
+        if not snapshot.exists:
+            payload["created_at"] = now_utc
+        doc_ref.set(payload, merge=True)
+        return NotificationSchedule(
+            line_user_id=line_user_id,
+            notify_time=notify_time,
+            timezone=timezone_name,
+            enabled=merged_enabled,
+            next_notify_at=next_notify_at,
+            created_at=created_at,
+            updated_at=now_utc,
+        )

src/Infrastructure/Repositories/__init__.py

@@ -1,4 +1,5 @@
 from .LineUserRepository import LineUserRepository
+from .NotificationScheduleRepository import NotificationScheduleRepository
 from .StockRepository import StockRepository
 from .WebUserRepository import WebUserRepository
 from .HabitTaskRepository import HabitTaskRepository
@@ -7,6 +8,7 @@
 
 web_user_repository = WebUserRepository()
 line_user_repository = LineUserRepository()
+notification_schedule_repository = NotificationScheduleRepository()
 stock_repository = StockRepository()
 habit_task_repository = HabitTaskRepository()
 habit_task_log_repository = HabitTaskLogRepository()

src/UseCases/Line/CheckExpiredStockUseCase.py

@@ -1,44 +1,58 @@
+from datetime import datetime, timezone
+
 from src import config
-from datetime import datetime
-from src.UseCases.Interface.IUseCase import IUseCase
-from src.Domains.IRepositories.ILineUserRepository import ILineUserRepository
 from src.Domains.IRepositories.IStockRepository import IStockRepository
 from src.Domains.IRepositories.IWebUserRepository import IWebUserRepository
 from src.UseCases.Interface.ILineResponseService import ILineResponseService
+from src.UseCases.Interface.IUseCase import IUseCase
 
 
 class CheckExpiredStockUseCase(IUseCase):
     def __init__(
         self,
-        line_user_repository: ILineUserRepository,
+        notification_schedule_repository,
         web_user_repository: IWebUserRepository,
         stock_repository: IStockRepository,
         line_response_service: ILineResponseService,
     ):
-        self._line_user_repository = line_user_repository
+        self._notification_schedule_repository = notification_schedule_repository
         self._web_user_repository = web_user_repository
         self._stock_repository = stock_repository
         self._line_response_service = line_response_service
 
     def execute(self) -> None:
-        line_users = self._line_user_repository.find()
-        for line_user in line_users:
-            web_users = self._web_user_repository.find(
-                {
-                    "linked_line_user_id": line_user.line_user_id,
-                    "is_linked_line_user": True,
-                }
+        now_utc = datetime.now(timezone.utc)
+        due_schedules = self._notification_schedule_repository.find_due(now_utc=now_utc)
+        if len(due_schedules) == 0:
+            return
+
+        due_line_user_ids = [s.line_user_id for s in due_schedules]
+        linked_web_users = self._web_user_repository.find(
+            {
+                "linked_line_user_id__in": due_line_user_ids,
+                "is_linked_line_user": True,
+            }
+        )
+        web_user_id_map = {user.linked_line_user_id: user._id for user in linked_web_users}
+        today = datetime.now().date()
+
+        for schedule in due_schedules:
+            claimed = self._notification_schedule_repository.claim_and_schedule_next(
+                line_user_id=schedule.line_user_id,
+                now_utc=now_utc,
             )
-            web_user_id = "" if len(web_users) == 0 else web_users[0]._id
+            if not claimed:
+                continue
+
+            web_user_id = web_user_id_map.get(schedule.line_user_id, "")
             stocks = self._stock_repository.find(
                 {
-                    "owner_id__in": [line_user.line_user_id, web_user_id],
+                    "owner_id__in": [schedule.line_user_id, web_user_id],
                     "status": 1,
                 }
             )
             near_due_stocks = []
             notify_on_items = []
-            today = datetime.now().date()
             for stock in stocks:
                 if stock.notify_enabled:
                     notify_on_items.append(stock.item_name)
@@ -73,5 +87,4 @@ def execute(self) -> None:
                 self._line_response_service.add_message(
                     "通知ONのアイテム:\n" + "\n".join(notify_on_items)
                 )
-
-            self._line_response_service.push(to=line_user.line_user_id)
+            self._line_response_service.push(to=schedule.line_user_id)

src/UseCases/Line/FollowUseCase.py

@@ -12,10 +12,12 @@ def __init__(
         line_request_service: ILineRequestService,
         line_response_service: ILineResponseService,
         line_user_service: ILineUserService,
+        notification_schedule_repository=None,
     ):
         self._line_request_service = line_request_service
         self._line_response_service = line_response_service
         self._line_user_service = line_user_service
+        self._notification_schedule_repository = notification_schedule_repository
 
     def execute(self) -> None:
         name = self._line_request_service.req_line_user_name
@@ -24,6 +26,12 @@ def execute(self) -> None:
             line_user_id=self._line_request_service.req_line_user_id,
         )
         self._line_user_service.find_or_create(new_line_user=new_line_user)
+        if self._notification_schedule_repository is not None:
+            self._notification_schedule_repository.upsert(
+                line_user_id=new_line_user.line_user_id,
+                notify_time="12:00",
+                timezone_name="Asia/Tokyo",
+            )
         self._line_response_service.add_message(
             f"{name}さん、友達登録ありがとうございます！"
         )