FCMNotifier class

sqlite

import sqlite3
import firebase_admin
from firebase_admin import credentials
from firebase_admin import messaging

class FCMNotifier:
    def __init__(self, db_path, firebase_config):
        # Firebase Admin SDK 초기화
        cred = credentials.Certificate(firebase_config)
        firebase_admin.initialize_app(cred)

        # SQLite 데이터베이스 경로 저장
        self.db_path = db_path

    def get_device_tokens(self):
        """SQLite에서 장치 토큰 가져오기"""
        connection = sqlite3.connect(self.db_path)
        cursor = connection.cursor()

        # 토큰을 가져오는 쿼리
        query = "SELECT token FROM device_tokens"  # device_tokens 테이블에서 token을 선택
        cursor.execute(query)

        tokens = [row[0] for row in cursor.fetchall()]  # 결과를 리스트로 변환

        cursor.close()
        connection.close()

        return tokens

    def send_fcm_messages(self, device_tokens, title, body, image_url, company_name):
        """FCM 메시지 전송"""
        if not device_tokens:
            print("등록된 장치 토큰이 없습니다.")
            return

        # 푸시 알림 메시지 생성
        message = messaging.MulticastMessage(
            notification=messaging.Notification(
                title=title,
                body=body
            ),
            data={
                'image': image_url,  # 이미지 URL 
                'click_action': 'FLUTTER_NOTIFICATION_CLICK',  # 클릭 시 동작 정의
                'companyName': company_name  # 회사명 추가
            },
            tokens=device_tokens,  # 여러 개의 토큰을 리스트로 전달
        )

        # 메시지 전송
        try:
            response = messaging.send_multicast(message)
            print('Successfully sent messages:', response)
        except Exception as e:
            print('Error sending messages:', e)

# 예제 사용
if __name__ == "__main__":
    db_path = './path/to/your/database.db'  # SQLite 데이터베이스 파일 경로
    firebase_config = './path/to/your/serviceAccountKey.json'
    
    notifier = FCMNotifier(db_path, firebase_config)

    device_tokens = notifier.get_device_tokens()  # DB에서 토큰 리스트 가져오기
    image_url = 'https://example.com/path/to/your/image.jpg'  # 실제 이미지 URL로 대체
    company_name = '회사명'  # 실제 회사명으로 대체

    notifier.send_fcm_messages(device_tokens, "안녕하세요", "FCM HTTP v1 푸시 알림 테스트입니다.", image_url, company_name)

mysql

import mysql.connector
import firebase_admin
from firebase_admin import credentials
from firebase_admin import messaging

class FCMNotifier:
    def __init__(self, db_config, firebase_config):
        # Firebase Admin SDK 초기화
        cred = credentials.Certificate(firebase_config)
        firebase_admin.initialize_app(cred)

        # MySQL 데이터베이스 연결 정보 저장
        self.db_config = db_config

    def get_device_tokens(self):
        """MySQL에서 장치 토큰 가져오기"""
        connection = mysql.connector.connect(**self.db_config)
        cursor = connection.cursor()

        # 토큰을 가져오는 쿼리
        query = "SELECT token FROM device_tokens"  # device_tokens 테이블에서 token을 선택
        cursor.execute(query)

        tokens = [row[0] for row in cursor.fetchall()]  # 결과를 리스트로 변환

        cursor.close()
        connection.close()

        return tokens

    def send_fcm_messages(self, device_tokens, title, body, image_url, company_name):
        """FCM 메시지 전송"""
        if not device_tokens:
            print("등록된 장치 토큰이 없습니다.")
            return

        # 푸시 알림 메시지 생성
        message = messaging.MulticastMessage(
            notification=messaging.Notification(
                title=title,
                body=body
            ),
            data={
                'image': image_url,  # 이미지 URL 
                'click_action': 'FLUTTER_NOTIFICATION_CLICK',  # 클릭 시 동작 정의
                'companyName': company_name  # 회사명 추가
            },
            tokens=device_tokens,  # 여러 개의 토큰을 리스트로 전달
        )

        # 메시지 전송
        try:
            response = messaging.send_multicast(message)
            print('Successfully sent messages:', response)
        except Exception as e:
            print('Error sending messages:', e)

# 예제 사용
if __name__ == "__main__":
    db_config = {
        'host': 'YOUR_DB_HOST',
        'user': 'YOUR_DB_USER',
        'password': 'YOUR_DB_PASSWORD',
        'database': 'YOUR_DATABASE_NAME'
    }
    firebase_config = './path/to/your/serviceAccountKey.json'
    
    notifier = FCMNotifier(db_config, firebase_config)

    device_tokens = notifier.get_device_tokens()  # DB에서 토큰 리스트 가져오기
    image_url = 'https://example.com/path/to/your/image.jpg'  # 실제 이미지 URL로 대체
    company_name = '회사명'  # 실제 회사명으로 대체

    notifier.send_fcm_messages(device_tokens, "안녕하세요", "FCM HTTP v1 푸시 알림 테스트입니다.", image_url, company_name)