Skip to content

Commit

Permalink
Optimise format
Browse files Browse the repository at this point in the history
  • Loading branch information
@yihong1120
yihong1120 committed Oct 15, 2024
1 parent ad261b6 commit 3d9830a
Show file tree
Hide file tree
Showing 2 changed files with 280 additions and 121 deletions.
131 changes: 88 additions & 43 deletions src/notifiers/line_notifier_message_api.py
View file
Original file line number Diff line number Diff line change
@@ -1,19 +1,23 @@
import os
from __future__ import annotations

import json
import cloudinary
import cloudinary.uploader
import os
from datetime import datetime
from datetime import timedelta
from typing import TypedDict

import cloudinary.api
import cloudinary.uploader
import cv2
import numpy as np
import requests
from dotenv import load_dotenv
from typing import TypedDict
import numpy as np
import cv2
from datetime import datetime, timedelta

# JSON file to store image records
IMAGE_RECORD_FILE = 'config/image_records.json'
CHECK_INTERVAL_DAYS = 1 # Check for old images once a day


class InputData(TypedDict):
message: str
image: bytes | None
Expand All @@ -25,31 +29,42 @@ class ResultData(TypedDict):

class LineMessenger:
"""
A class for managing notifications sent via the LINE Messaging API and Cloudinary image upload.
A class for managing notifications sent
via the LINE Messaging API and Cloudinary image upload.
"""

def __init__(self, channel_access_token: str | None = None, recipient_id: str | None = None):
def __init__(
self,
channel_access_token: str | None = None,
recipient_id: str | None = None,
) -> None:
"""
Initialises the LineMessenger instance.
Args:
channel_access_token (Optional[str]): The LINE Messaging API channel access token.
channel_access_token (Optional[str]): The LINE Messaging API
channel access token.
"""
load_dotenv()
self.channel_access_token = channel_access_token or os.getenv('LINE_CHANNEL_ACCESS_TOKEN')
self.channel_access_token = channel_access_token or os.getenv(
'LINE_CHANNEL_ACCESS_TOKEN',
)
self.recipient_id = recipient_id

if not self.channel_access_token:
raise ValueError('LINE_CHANNEL_ACCESS_TOKEN not provided or in environment variables.')

raise ValueError(
'LINE_CHANNEL_ACCESS_TOKEN not provided '
'or in environment variables.',
)

if not self.recipient_id:
raise ValueError('Recipient ID not provided.')

# Configure Cloudinary
cloudinary.config(
cloud_name=os.getenv('CLOUDINARY_CLOUD_NAME'),
api_key=os.getenv('CLOUDINARY_API_KEY'),
api_secret=os.getenv('CLOUDINARY_API_SECRET')
api_secret=os.getenv('CLOUDINARY_API_SECRET'),
)

# Load upload records
Expand All @@ -61,7 +76,7 @@ def load_image_records(self) -> dict:
"""
try:
if os.path.exists(IMAGE_RECORD_FILE):
with open(IMAGE_RECORD_FILE, 'r') as file:
with open(IMAGE_RECORD_FILE) as file:
return json.load(file)
except Exception as e:
print(f"Failed to load image records: {e}")
Expand All @@ -77,31 +92,36 @@ def save_image_records(self) -> None:
except Exception as e:
print(f"Failed to save image records: {e}")

def push_message(self, message: str, image_bytes: bytes | None = None) -> int:
def push_message(
self,
message: str,
image_bytes: bytes | None = None,
) -> int:
"""
Sends a message via LINE Messaging API, optionally including an image.
Args:
recipient_id (str): The recipient ID.
message (str): The message to send.
image_bytes (Optional[bytes]): The image bytes to upload to Cloudinary. Defaults to None.
image_bytes (Optional[bytes]): The image bytes
to upload to Cloudinary. Defaults to None.
Returns:
response.status_code (int): The status code of the response.
"""
headers = {
'Content-Type': 'application/json',
'Authorization': f"Bearer {self.channel_access_token}"
'Authorization': f"Bearer {self.channel_access_token}",
}

message_data = {
"to": self.recipient_id,
"messages": [
'to': self.recipient_id,
'messages': [
{
"type": "text",
"text": message
}
]
'type': 'text',
'text': message,
},
],
}

public_id = None
Expand All @@ -110,13 +130,13 @@ def push_message(self, message: str, image_bytes: bytes | None = None) -> int:
# Upload image to Cloudinary
image_url, public_id = self.upload_image_to_cloudinary(image_bytes)
if not image_url:
raise ValueError("Failed to upload image to Cloudinary")
raise ValueError('Failed to upload image to Cloudinary')

print(f"Image URL: {image_url}")
message_data["messages"].append({
"type": "image",
"originalContentUrl": image_url,
"previewImageUrl": image_url
message_data['messages'].append({
'type': 'image',
'originalContentUrl': image_url,
'previewImageUrl': image_url,
})

# Send message via LINE API
Expand All @@ -128,7 +148,7 @@ def push_message(self, message: str, image_bytes: bytes | None = None) -> int:

# Print response for debugging
if response.status_code == 200:
print(f"Message sent successfully.")
print('Message sent successfully.')
else:
print(f"Error: {response.status_code}, {response.text}")

Expand All @@ -151,11 +171,13 @@ def upload_image_to_cloudinary(self, image_data: bytes) -> tuple[str, str]:
tuple: The URL of the uploaded image and its public_id.
"""
try:
response = cloudinary.uploader.upload(image_data, resource_type="image")
response = cloudinary.uploader.upload(
image_data, resource_type='image',
)
return response['secure_url'], response['public_id']
except Exception as e:
print(f"Failed to upload image to Cloudinary: {e}")
return "", ""
return '', ''

def record_image_upload(self, public_id: str) -> None:
"""
Expand All @@ -166,11 +188,24 @@ def record_image_upload(self, public_id: str) -> None:

def delete_old_images_with_interval(self) -> None:
"""
Deletes images that have been uploaded for more than 7 days, and only checks every day.
Deletes images that have been uploaded for more than 7 days,
and only checks every day.
"""
last_checked = self.image_records.get('last_checked')
if last_checked is None or datetime.now() - datetime.fromisoformat(last_checked) > timedelta(days=CHECK_INTERVAL_DAYS):
# Record check time
if last_checked is None:
# If last_checked is not recorded, check immediately
should_check = True
else:
try:
last_checked_time = datetime.fromisoformat(last_checked)
now = datetime.now()
interval = timedelta(days=CHECK_INTERVAL_DAYS)
should_check = now - last_checked_time > interval
except ValueError:
# If last_checked is not a valid datetime, check immediately
should_check = True

if should_check:
self.image_records['last_checked'] = datetime.now().isoformat()
self.delete_old_images()
self.save_image_records()
Expand All @@ -182,12 +217,14 @@ def delete_old_images(self) -> None:
now = datetime.now()
expired_public_ids = [
public_id for public_id, upload_time in self.image_records.items()
if public_id != 'last_checked' and now - datetime.fromisoformat(upload_time) > timedelta(days=7)
if public_id != 'last_checked'
and now - datetime.fromisoformat(upload_time) > timedelta(days=7)
]

for public_id in expired_public_ids:
self.delete_image_from_cloudinary(public_id)
del self.image_records[public_id] # Remove expired image from records
# Remove expired image from records
del self.image_records[public_id]

# Save records after deletion
self.save_image_records()
Expand All @@ -202,23 +239,31 @@ def delete_image_from_cloudinary(self, public_id: str) -> None:
try:
response = cloudinary.uploader.destroy(public_id)
if response.get('result') == 'ok':
print(f"Image with public_id {public_id} successfully deleted from Cloudinary.")
print(
f"Image with public_id {public_id} successfully "
'deleted from Cloudinary.',
)
else:
print(f"Failed to delete image with public_id {public_id}. Response: {response}")
print(
f"Failed to delete image with public_id {public_id}. "
f"Response: {response}",
)
except Exception as e:
print(f"Error deleting image from Cloudinary: {e}")


# Example usage
def main():
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', 'YOUR_LINE_CHANNEL')
channel_access_token = os.getenv(
'LINE_CHANNEL_ACCESS_TOKEN', 'YOUR_LINE_CHANNEL',
)
recipient_id = os.getenv('LINE_RECIPIENT_ID', 'YOUR_RECIPIENT_ID')

messenger = LineMessenger(
channel_access_token=channel_access_token,
recipient_id=recipient_id
channel_access_token=channel_access_token,
recipient_id=recipient_id,
)

message = 'Hello, LINE Messaging API!'

# Create a black image for testing
Expand Down
Loading

0 comments on commit 3d9830a

Please sign in to comment.