LINE機器人自動備份群組照片

發表於 2024-08-03 分類於專案成果閱讀次數：

前言

在執行工程時，經常會為了聯絡相關事宜建立LINE群組來進行交談，通常施工照片也會在這個群裡面進行上傳讓大家可以確認內容，上傳照片到LINE群組的時候除了直接上傳外，也可以透過建立相簿將照片永久保存。

在LINE群組的檔案機制，照片或文件會有一定時間的存活期，當過了存活期之後照片就會呈現無法下載的情形，因此將照片做一個額外存放的動作算是滿重要的一個手段，而這件事除了繁瑣也很可能會因為再次請求上傳的過程多存了不少張重複的照片，而這個部分相信大多數人都是透過人眼來進行判別，火眼金睛用久也是會累的。

希望能夠打造一個當群組內判定到照片資料時，能夠自動備份到指定硬碟並汰除重複照片的機器人，減少這些瑣事的發生。

至於照片後續要進行處理就可以利用施工照片VBA進行資料排序、批次改名、報表輸出。

操作畫面

Fig1.LINE群組畫面

Fig2.本地端儲存畫面

基本工具

LINEBOT
- MessengerAPI
ngrok
- 主要是用來建立https連線給LINEBOT的webhook使用
python
- flask
- linebotsdk

相關連結

實作邏輯

預先準備內容

建立flask並啟用app
啟用ngrok
複製將ngrok上面的對外連線IP
貼到LINEBOT的webhook

訊息接收過程

LINEBOT在接受到群組內的訊息時
判定訊息屬性是否為Image
圖片內容會先暫時放在LINE機房並且生成一組token
收到訊息的同時會執行API(Webhook機制)
API中會將訊息做過濾，藉由token取得照片到本地端
本地端資料庫會記錄已經存放的照片跟Image_hash
判定是否有重複的image_hash，如有則回傳"圖片已存在"
通過重複汰除機制過濾完成後可將照片存放到指定位置

程式碼

webhook

app_local.py

# 存放在本地端sqlite+本地端硬碟

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, ImageMessage, TextSendMessage
import os
import uuid
import hashlib

from sql_utils import get_db, save_photo_to_db, is_image_hash_exist

app = Flask(__name__)

# 設定你的 Channel Access Token 和 Channel Secret
CHANNEL_ACCESS_TOKEN = 'YOUR_CHANNEL_ACCESS_TOKEN'
CHANNEL_SECRET = 'YOUR_CHANNEL_SECRET'

line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)

PHOTOS_DIR = 'photos'
if not os.path.exists(PHOTOS_DIR):
    os.makedirs(PHOTOS_DIR)

@app.route("/callback", methods=['POST'])
def callback():
    # 獲取 X-Line-Signature header
    signature = request.headers['X-Line-Signature']
    
    # 獲取請求體
    body = request.get_data(as_text=True)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'

@handler.add(MessageEvent, message=ImageMessage)
def handle_image_message(event):
    # 獲取圖像內容
    message_content = line_bot_api.get_message_content(event.message.id)
    image_data = message_content.content
    image_hash = calculate_image_hash(image_data)
    user_id = event.source.user_id
    group_id = event.source.group_id
    photo_properties = {"edit": "false"}
    photo_url=None

    with next(get_db()) as db:

        # 檢查圖片是否已存在
        if is_image_hash_exist(db, image_hash):
            reply_text = "圖片已存在，未保存。"
        else:
            # 保存圖片到本地資料夾
            file_name = str(uuid.uuid4()) + '.jpg'
            file_path = os.path.join(PHOTOS_DIR, file_name)
            with open(file_path, 'wb') as f:
                for chunk in message_content.iter_content():
                    f.write(chunk)

            # 保存圖片到資料庫
            saved_photo = save_photo_to_db(db, user_id, group_id, file_name, photo_url, photo_properties, image_hash)
            print("Photo saved successfully:", saved_photo)

            reply_text = "圖片已保存到本地端！"

    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=reply_text)
    )

def calculate_image_hash(image_data):
    hasher = hashlib.sha256()
    hasher.update(image_data)
    return hasher.hexdigest()

@handler.add(MessageEvent)
def handle_message(event):
    if not isinstance(event.message, ImageMessage):
        reply_text = "請傳送圖片訊息。"
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text=reply_text)
        )

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

sql_utils.py

from sqlalchemy import create_engine, Column, Integer, String, DateTime, JSON
from sqlalchemy.orm import sessionmaker,declarative_base
from datetime import datetime
from sqlalchemy.orm import Session

DATABASE_URL="sqlite:///photos.db"
engine = create_engine(DATABASE_URL)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 定義數據庫模型
class Photo(Base):
    __tablename__ = 'photos_backup'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(String)
    group_id = Column(String)
    photo_name = Column(String)
    photo_url = Column(String)
    photo_thumbnail_url = Column(String)
    created_time = Column(DateTime, default=datetime.utcnow)
    photo_properties = Column(JSON)
    photo_image_hash = Column(String)#, unique=True, nullable=False)

Base.metadata.create_all(bind=engine)

# 獲取數據庫會話
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def save_photo_to_db(db, user_id,group_id, photo_name, photo_url, photo_properties,image_hash):
    new_photo = Photo(
        user_id=user_id,
        group_id=group_id,
        photo_name=photo_name,
        photo_url=photo_url,
        photo_properties=photo_properties,
        photo_image_hash=image_hash
    )
    db.add(new_photo)
    db.commit()
    db.refresh(new_photo)
    return new_photo

def get_photo_by_id(db: Session, photo_id: int):
    return db.query(Photo).filter(Photo.id == photo_id).first()

def fetch_all_photos(db: Session):
    return db.query(Photo).all()

def delete_photo_from_db(db: Session, photo_id: int):
    photo = db.query(Photo).filter(Photo.id == photo_id).first()
    if photo:
        db.delete(photo)
        db.commit()
        print(f"Photo with ID {photo_id} deleted successfully from database.")
    else:
        print(f"No photo found with ID {photo_id}.")

def is_image_hash_exist(db, image_hash):
    return db.query(Photo).filter(Photo.photo_image_hash == image_hash).first() is not None

結語

如果您本身執行工程專案時有大量的照片需要進行自動備份，同時又有一台電腦可以不關機當作server用途，可以將他建立好python執行環境並參考上述的內容搭配使用，如有不懂的內容也歡迎聯繫作者來協助架設LINE群組機器人，自動備份工程群組內部照片省時又省力。

0%