시놀로지 Telegram Bot으로 공유 링크 생성하기

안녕하세요!

이번 포스팅에서는, 저번 시간에 다룬 Telegram Bot을 활용하여
특정 공유 폴더 내에 있는 파일을 검색하고, 공유 링크를 생성하는 방법에 대해 다루어 보겠습니다.

먼저 Telegram Bot이 없으신 분들은 이전 포스팅을 참고해서 생성해 주시기 바랍니다.

Docker 경로 생성

먼저 docker 경로에 python > scripts 폴더를 생성합니다.

이후 docker-compose.yml 파일을 생성하고, 아래의 내용을 붙여 넣습니다.

version: '3.8'

services:
  telegram-vn-share:
    image: python:3.13-slim
    container_name: telegram-vn-share
    restart: unless-stopped
    network_mode: host
    volumes:
      - ./scripts:/scripts
    working_dir: /app
    command: >
      sh -c "pip install --no-cache-dir requests python-telegram-bot &&
             python /scripts/search_and_share.py"
    env_file:
      - .env
    environment:
      - PYTHONUNBUFFERED=1
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

.env 파일을 생성하고, 본인 환경에 맞게 수정하여 아래 내용을 붙여 넣습니다.

# Synology NAS 설정
NAS_URL=http://localhost:5000
NAS_USERNAME=아이디
NAS_PASSWORD=비밀번호
NAS_FOLDER=검색을 사용할 경로

# Telegram Bot 설정
TELEGRAM_BOT_TOKEN=텔레그램 봇 토큰

검색을 사용할 경로는 /volume1을 제외하고 /로 시작해야 합니다.

• 예시
  • /Archive-Main/Telegram_Share

Python Script

import requests
from telegram import Update
from telegram.ext import Application, CommandHandler, CallbackContext
from datetime import datetime, timedelta
import urllib.parse
import json
import os

class SynologyAPI:
    def __init__(self, nas_url, username, password):
        self.nas_url = nas_url
        self.username = username
        self.password = password
        self.sid = None

    def login(self):
        login_url = f"{self.nas_url}/webapi/auth.cgi"
        payload = {
            "api": "SYNO.API.Auth",
            "version": "3",
            "method": "login",
            "account": self.username,
            "passwd": self.password,
            "session": "FileStation",
            "format": "sid"
        }
        response = requests.get(login_url, params=payload, timeout=10)
        response.raise_for_status()
        data = response.json()

        if data.get("success"):
            self.sid = data["data"]["sid"]
            print(f"로그인 성공: SID={self.sid}")
        else:
            error_code = data.get('error', {}).get('code', 'Unknown')
            raise Exception(f"로그인 실패: 오류 코드 {error_code}")

    def list_files(self, folder_path="/", recursive=False):
        """폴더 내의 파일을 검색하고 하위 폴더까지 재귀적으로 검색"""
        if not self.sid:
            raise Exception("로그인되지 않았습니다. 먼저 login()을 호출하세요.")

        files_to_search = []

        # 폴더 경로 인코딩 처리
        list_url = f"{self.nas_url}/webapi/entry.cgi"
        params = {
            "api": "SYNO.FileStation.List",
            "version": "2",
            "method": "list",
            "folder_path": folder_path,
            "_sid": self.sid,
            "additional": "time"
        }

        print(f"폴더 검색 중: {folder_path}")
        response = requests.get(list_url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()

        if data.get("success"):
            files = data["data"]["files"]
            for file in files:
                if file.get("isdir", False):
                    if recursive:
                        # 하위 폴더를 재귀적으로 검색
                        subfolder_path = file["path"]
                        try:
                            folder_files = self.list_files(subfolder_path, recursive=True)
                            files_to_search.extend(folder_files)
                        except Exception as e:
                            print(f"하위 폴더 검색 실패 {subfolder_path}: {e}")
                else:
                    files_to_search.append({
                        "name": file["name"],
                        "path": file["path"],
                        "modified": file.get("additional", {}).get("time", {}).get("mtime", 0)
                    })
        else:
            error_code = data.get('error', {}).get('code', 'Unknown')
            raise Exception(f"파일 목록 가져오기 실패: 오류 코드 {error_code}")

        return files_to_search

    def create_share_link(self, file_path):
        """파일 공유 링크 생성 (1일 후 만료)"""
        if not self.sid:
            raise Exception("로그인되지 않았습니다. 먼저 login()을 호출하세요.")

        share_url = f"{self.nas_url}/webapi/entry.cgi"

        # 만료일 계산 (1일 후)
        expiration_date = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")

        # PowerShell 코드처럼 따옴표로 감싸서 전달
        params = {
            "api": "SYNO.FileStation.Sharing",
            "version": "3",
            "method": "create",
            "path": file_path,
            "_sid": self.sid,
            "date_expired": f'"{expiration_date}"',  # 따옴표로 감싸기
            "format": "sid"
        }

        print(f"공유 링크 생성 중: {file_path}, 만료일: {expiration_date}")
        response = requests.get(share_url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()

        if data.get("success"):
            links = data["data"]["links"]
            if links:
                link_id = links[0]["id"]

                # 생성된 링크 정보 조회
                info_params = {
                    "api": "SYNO.FileStation.Sharing",
                    "version": "3",
                    "method": "getinfo",
                    "id": link_id,
                    "_sid": self.sid,
                    "format": "sid"
                }

                info_response = requests.get(share_url, params=info_params, timeout=10)
                info_data = info_response.json()

                if info_data.get("success"):
                    share_data = info_data["data"]
                    url = share_data.get("url", links[0]["url"])

                    # ✅ 프로토콜을 https로 변경하고, 포트를 5000에서 5001로 변경
                    modified_url = url.replace("http:", "https:").replace(":5000", ":5001")

                    return {
                        "url": modified_url, # 최종 수정된 URL 반환
                        "expires": share_data.get("date_expired", expiration_date),
                        "id": link_id,
                        "name": share_data.get("name", file_path.split("/")[-1])
                    }
                else:
                    # getinfo 실패 시 기본 정보 반환
                    url = links[0]["url"]

                    # ✅ 여기도 동일하게 프로토콜과 포트 모두 변경
                    modified_url = url.replace("http:", "https:").replace(":5000", ":5001")

                    return {
                        "url": modified_url, # 최종 수정된 URL 반환
                        "expires": expiration_date,
                        "id": link_id
                    }
            else:
                raise Exception("링크 생성 실패: 링크가 반환되지 않았습니다.")
        else:
            error_code = data.get('error', {}).get('code', 'Unknown')
            error_msg = data.get('error', {}).get('errors', '')
            raise Exception(f"링크 생성 실패: 오류 코드 {error_code}, {error_msg}")

    def logout(self):
        if not self.sid:
            print("로그인되어 있지 않아 로그아웃할 필요가 없습니다.")
            return

        logout_url = f"{self.nas_url}/webapi/auth.cgi"
        params = {
            "api": "SYNO.API.Auth",
            "version": "3",
            "method": "logout",
            "session": "FileStation",
            "_sid": self.sid
        }

        try:
            response = requests.get(logout_url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()

            if data.get("success"):
                print("로그아웃 성공")
                self.sid = None
            else:
                print(f"로그아웃 실패: {data.get('error')}")
        except Exception as e:
            print(f"로그아웃 중 오류 발생: {e}")
            self.sid = None

# 검색 결과를 저장할 딕셔너리 (사용자 ID별로 저장)
search_results = {}

# 환경 변수에서 설정 가져오기
NAS_URL = os.getenv('NAS_URL')
NAS_USERNAME = os.getenv('NAS_USERNAME')
NAS_PASSWORD = os.getenv('NAS_PASSWORD')
NAS_FOLDER = os.getenv('NAS_FOLDER')
BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')

# /search 명령어 처리
async def search_command(update: Update, context: CallbackContext):
    if not context.args:
        await update.message.reply_text("검색어를 입력해주세요. 예: /search test.txt")
        return

    search_query = " ".join(context.args)
    user_id = update.effective_user.id
    await update.message.reply_text(f"'{search_query}' 검색 중... 잠시만 기다려주세요.")

    try:
        synology = SynologyAPI(NAS_URL, NAS_USERNAME, NAS_PASSWORD)
        synology.login()

        # 파일 검색
        files = synology.list_files(folder_path=NAS_FOLDER, recursive=True)
        matching_files = [file for file in files if search_query.lower() in file["name"].lower()]

        if matching_files:
            # 사용자별 검색 결과 저장
            search_results[user_id] = matching_files

            # 검색 결과 메시지 구성
            if len(matching_files) > 20:
                response_message = f"검색 결과: {len(matching_files)}개 파일 발견 (상위 20개만 표시)\n\n"
                display_files = matching_files[:20]
            else:
                response_message = f"검색 결과: {len(matching_files)}개 파일 발견\n\n"
                display_files = matching_files

            for i, file in enumerate(display_files, 1):
                response_message += f"{i}. 📄 {file['name']}\n"
                response_message += f"   경로: {file['path']}\n\n"

            response_message += "💡 사용법: /share [번호] 또는 /share [번호1,번호2,...]\n"
            response_message += "예: /share 1 또는 /share 1,3,5"
        else:
            response_message = f"'{search_query}'와 일치하는 파일을 찾을 수 없습니다."
            # 검색 결과가 없으면 저장된 결과도 초기화
            if user_id in search_results:
                del search_results[user_id]

        await update.message.reply_text(response_message)

    except Exception as e:
        await update.message.reply_text(f"❌ 오류 발생: {str(e)}")
        print(f"검색 오류: {e}")

    finally:
        if 'synology' in locals():
            synology.logout()

# /share 명령어 처리
async def share_command(update: Update, context: CallbackContext):
    user_id = update.effective_user.id

    # 저장된 검색 결과가 있는지 확인
    if user_id not in search_results or not search_results[user_id]:
        await update.message.reply_text(
            "❌ 먼저 /search 명령어로 파일을 검색해주세요.\n"
            "예: /search report.pdf"
        )
        return

    if not context.args:
        await update.message.reply_text(
            "공유할 파일 번호를 입력해주세요.\n"
            "예: /share 1 또는 /share 1,3,5"
        )
        return

    # 번호 파싱
    try:
        numbers_str = context.args[0]
        numbers = [int(n.strip()) for n in numbers_str.split(',')]

        # 유효한 번호 범위 확인
        max_number = len(search_results[user_id])
        invalid_numbers = [n for n in numbers if n < 1 or n > max_number]

        if invalid_numbers:
            await update.message.reply_text(
                f"❌ 잘못된 번호: {invalid_numbers}\n"
                f"유효한 범위: 1~{max_number}"
            )
            return

    except ValueError:
        await update.message.reply_text("❌ 올바른 번호 형식이 아닙니다. 예: /share 1 또는 /share 1,3,5")
        return

    await update.message.reply_text(f"🔄 {len(numbers)}개 파일의 공유 링크 생성 중...")

    try:
        synology = SynologyAPI(NAS_URL, NAS_USERNAME, NAS_PASSWORD)
        synology.login()

        # 결과 메시지 초기화
        success_count = 0
        response_messages = []

        # 선택된 각 파일에 대해 공유 링크 생성
        for num in numbers:
            file_info = search_results[user_id][num - 1]  # 인덱스는 0부터 시작
            file_path = file_info['path']
            file_name = file_info['name']

            try:
                # 파일 공유 링크 생성
                share_info = synology.create_share_link(file_path)

                response_messages.append(
                    f"✅ #{num}. {file_name}\n"
                    f"🔗 링크: {share_info['url']}\n"
                    f"⏰ 만료: {share_info['expires']} (1일)"
                )
                success_count += 1

            except Exception as e:
                response_messages.append(
                    f"❌ #{num}. {file_name}\n"
                    f"   오류: {str(e)}"
                )

        # 최종 결과 메시지
        final_message = f"📊 결과: {success_count}/{len(numbers)}개 성공\n\n"
        final_message += "\n\n".join(response_messages)

        # 메시지가 너무 길면 분할
        if len(final_message) > 4000:
            # 첫 번째 메시지: 요약
            await update.message.reply_text(f"📊 결과: {success_count}/{len(numbers)}개 성공")

            # 개별 결과 메시지
            for msg in response_messages:
                await update.message.reply_text(msg)
        else:
            await update.message.reply_text(final_message)

    except Exception as e:
        await update.message.reply_text(f"❌ 오류 발생: {str(e)}")
        print(f"공유 링크 생성 오류: {e}")

    finally:
        if 'synology' in locals():
            synology.logout()

# /list 명령어 처리 (현재 저장된 검색 결과 다시 보기)
async def list_command(update: Update, context: CallbackContext):
    user_id = update.effective_user.id

    if user_id not in search_results or not search_results[user_id]:
        await update.message.reply_text("📭 저장된 검색 결과가 없습니다. /search 명령어로 먼저 검색해주세요.")
        return

    files = search_res

    if len(files) > 20:
        response_message = f"📋 저장된 검색 결과: {len(files)}개 (상위 20개만 표시)\n\n"
        display_files = files[:20]
    else:
        response_message = f"📋 저장된 검색 결과: {len(files)}개\n\n"
        display_files = files

    for i, file in enumerate(display_files, 1):
        response_message += f"{i}. 📄 {file['name']}\n"

    response_message += "💡 사용법: /share [번호] 또는 /share [번호1,번호2,...]"

    await update.message.reply_text(response_message)

# /clear 명령어 처리 (검색 결과 초기화)
async def clear_command(update: Update, context: CallbackContext):
    user_id = update.effective_user.id

    if user_id in search_results:
        del search_results[user_id]
        await update.message.reply_text("🧹 저장된 검색 결과가 초기화되었습니다.")
    else:
        await update.message.reply_text("📭 초기화할 검색 결과가 없습니다.")

# /help 명령어 처리
async def help_command(update: Update, context: CallbackContext):
    help_text = """
🤖 Synology NAS 파일 관리 봇

사용 가능한 명령어:

📍 /search [검색어]
   - NAS에서 파일을 검색합니다
   - 예: /search report.pdf

📍 /share [번호]
   - 검색 결과에서 번호로 파일을 선택하여 공유 링크를 생성합니다
   - 여러 파일 동시 선택 가능 (쉼표로 구분)
   - 모든 링크는 1일 후 자동 만료됩니다
   - 예: /share 1
   - 예: /share 1,3,5

📍 /list
   - 현재 저장된 검색 결과를 다시 표시합니다

📍 /clear
   - 저장된 검색 결과를 초기화합니다

📍 /help
   - 이 도움말을 표시합니다

💡 사용 순서:
1. /search로 파일 검색
2. 검색 결과에서 번호 확인
3. /share로 원하는 파일의 공유 링크 생성
"""
    await update.message.reply_text(help_text)

def main():
    # 환경 변수 확인
    print("환경 변수 확인:")
    print(f"NAS_URL: {NAS_URL}")
    print(f"NAS_USERNAME: {NAS_USERNAME}")
    print(f"NAS_FOLDER: {NAS_FOLDER}")
    print(f"BOT_TOKEN: {BOT_TOKEN[:10]}...")  # 토큰은 일부만 표시

    # Application 생성
    application = Application.builder().token(BOT_TOKEN).build()

    # 명령어 핸들러 추가
    application.add_handler(CommandHandler("search", search_command))
    application.add_handler(CommandHandler("share", share_command))
    application.add_handler(CommandHandler("list", list_command))
    application.add_handler(CommandHandler("clear", clear_command))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("start", help_command))

    # 봇 실행
    print("🤖 봇이 시작되었습니다...")
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == '__main__':
    main()

위 스크립트를 scripts 폴더에 search_and_share.py로 저장합니다.

코드의 동작은 다음과 같이 이루어집니다.

1. /search [파일 명] 명령어로, 지정한 공유 폴더에서 재귀적으로 해당 파일을 찾음.
2. 파일을 찾은 경우, 해당 파일 명, 번호, 경로를 리턴 함.
3. /share [번호]로 해당 파일의 공유 링크를 생성함.
4. 생성된 공유 링크의 URL에서 http를 https로, 포트 5000을 5001로 변경함.
5. /clear를 통해 검색 결과를 리셋함.

포트 부분은 모든 사용자의 환경에 맞출 수가 없기에,
필요에 따라 Chat GPT와 같은 AI를 이용하여 도움을 받을 수 있습니다.

간단하게는 위 부분을 수정하는 것으로 가능하며,
포트가 필요 없는 경우에는 AI의 도움을 받으시기 바랍니다.

Docker 실행

SSH에서 sudo -i로 root 권한을 얻은 뒤,
docker-compose.yml이 있는 경로로 진입합니다.

cd /volume1/docker/폴더

이후 docker-compose up -d로 실행할 수 있습니다.

위 이미지는 동작 예시입니다.

---

수고 많으셨습니다!

필자는 아직 코딩에 능숙하지 않지만, 필요한 코드를 작성하고 다듬는 과정에서 시행착오를 겪으며 조금씩 성장하고 있습니다.
이번에 작성한 코드는 간결하고 가볍다고 말하기엔 무리가 있지만
사용하기에 지장은 없을 것으로 생각합니다.

이제 텔레그램 Bot을 통해 언제 어디서나 원하는 파일을 검색하고, 간편하게 공유 링크를 생성할 수 있게 되었습니다.
이를 활용하면 지인들과 손쉽게 파일을 공유할 수 있는 개인 텔레그램 서버를 운영하는 데 유용할 것입니다.

앞으로도 텔레그램 Bot과 시놀로지 Python을 활용한 다양한 주제들을 다룰 예정이니
많은 관심 부탁드립니다.
피드백과 경험 공유도 언제나 환영합니다.

감사합니다!