본문 바로가기
PC/Programming

[Telegram-bot] NAS_bot(소스 코드)

by dragom 2020. 11. 9.
반응형

2020/10/30 - [PC] - [Telegram-bot] File-management를 위한 NAS_bot

 

[Telegram-bot] File-management를 위한 NAS_bot

이 글 본문은 프로그램의 기능을 설명하는 내용입니다. 프로그램 다운로드는 아래 링크 문서에서 가능합니다. 2020/11/02 - [PC] - [Telegram-bot] File-management를 위한 NAS_bot(실행파일) [Telegram-bot] File..

dragom.tistory.com

2020/11/09 - [PC] - [Telegram-bot] NAS_bot(윈도우OS실행파일)

 

[Telegram-bot] NAS_bot(윈도우OS실행파일)

기능설명은 이전 글 참고 부탁드립니다. 2020/10/30 - [PC] - [Telegram-bot] File-management를 위한 NAS_bot [Telegram-bot] File-management를 위한 NAS_bot 최근 telegram-bot이라는 telegram의 너무나도 고맙..

dragom.tistory.com

 


  • 2020/11/09: File_size_max 함수를 설정 파일에서 읽어온 뒤 숫자로 변환하지 않아 다운로드가 안되는 오류가 있었습니다. 이 부분 수정했습니다.(v1.0.8)

이 블로그 글들과 이 코드를 보시면 아시겠지만, 저는 기본적인 C언어와 리눅스 shell script, matlab m-script, 아주 간단한 python 정도 밖에 할 줄 모릅니다.

따라서 기본기도 제대로 안된 코드이겠지만, 비슷한 기능을 활용하여 무엇인가 만들고자 하는 다른 사람에게 한 번이라도 도움 되면 좋겠다는 생각에 공개합니다.

NAS_bot.py
0.04MB

'''=============Version log=================
2020.10.28 - 2020.10.30    v1.0.0 초안 완성
2020.10.30  v1.0.1 명령어 변경(files->file)
2020.10.30  v1.0.2 logger 추가
2020.10.30  v1.0.3 error handler 추가
2020.10.30  v1.0.4 logger.debug, 주석 추가
2020.10.31  v1.0.5 파일 다운로드/업로드 크기 한도를 환경변수로 변경(file_size_max);
2020.10.31  v1.0.6 일부 에러 수정(except 시 telegram message 보내는 방법 수정)
2020.10.31  v1.0.7 로그 저장 위치를 변수로 설정할 수 있도록 수정
2020.11.09  v1.0.8 File_size_max 변수 받아올 때 int로 변경되도록 수정
========================================='''
import logging
from logging.handlers import TimedRotatingFileHandler
import os
import telegram
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ChatAction, KeyboardButton, ReplyKeyboardMarkup
from telegram.ext import Updater, Filters, CommandHandler, CallbackQueryHandler, MessageHandler
from functools import wraps
from configparser import ConfigParser
import shutil

#=====================log 처리=====================#
try:    # 설정 파일이 있고 그 안에 settings 섹션에 log_file 변수 있으면 불러오기
    parser=ConfigParser()
    parser.read('./NAS_bot.ini')
    log_file=parser.get('settings','log_file')
    logger = logging.getLogger(__name__)
    if not os.path.isdir(os.path.split(log_file)[0]):
        os.mkdir(os.path.split(log_file)[0])  # log data 저장될 폴더 없으면 생성
    formatter = logging.Formatter(u'%(asctime)s [%(levelname)8s] %(message)s')
    logger.setLevel(logging.DEBUG)
    fileHandler = TimedRotatingFileHandler(filename=log_file, when='midnight', interval=1, encoding='utf-8')        
    fileHandler.setFormatter(formatter)
    fileHandler.suffix = '%Y%m%d'
    fileHandler.setLevel(logging.DEBUG)
    logger.addHandler(fileHandler)
except Exception as eee:    # log data 저장될 폴더 없는 경우 로그는 저장되지 않고 그냥 출력 됨.
    logger = logging.getLogger(__name__)
    logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',level=logging.INFO)
    print(eee)
logger.critical('Program started. 도움이 필요한 경우 https://dragom.tistory.com/41에 요청주세요.')  # 프로그램 시작 log

#=====================config 파일 처리=====================#
if not os.path.isfile('NAS_bot.ini') :  # 설정 파일 존재 확인
    print("설정 파일이 없습니다.(./NAS_bot.ini)\n")
    with open('critical.log','w') as f:
            f.write('설정 파일이 없습니다.(./NAS_bot.ini) 도움이 필요한 경우 https://dragom.tistory.com/41에 요청주세요.')
    logger.error('설정 파일을 찾을 수 없어 종료')
    os._exit(0) # 설정 파일 미존재 시 프로그램 종료
else :
    parser=ConfigParser()
    parser.read('./NAS_bot.ini')    # config 파일 로드
    logger.debug('설정 파일 load')
    if not 'settings' in parser.sections() :    # 설정 파일 settings section 존재 여부 확인
        logger.error("==설정 파일(NAS_bot.ini)을 찾았으나 [settings] 섹션이 없습니다.")
        os._exit(0)
    try:    # 설정 파일 settings section 내 bot_token 항목 확인
        my_token=parser.get('settings','bot_token')
        logger.debug('==bot_token 설정')
    except Exception as eee:
        logger.error("==설정 파일(NAS_bot.ini)을 찾았으나 bot_token 항목이 없습니다.\t에러:{}".format(eee))
        os._exit(0)
    try:    # 설정 파일 settings section 내 administrator 항목 확인
        admin_id = parser.get('settings','administrator')
        logger.debug('==admin_id 설정')
    except Exception as eee:
        logger.error("==설정 파일(NAS_bot.ini)을 찾았으나 adminitrator 항목이 없습니다.\t에러:{}".format(eee))
        os._exit(0)
    try:    # 2GB 이상 크기를 가지는 파일 처리를 위한 변수
        file_size_max = int(parser.get('settings','file_size_max'))
        uri_path = parser.get('settings','uri_path')
        http_uri = parser.get('settings','http_uri')
        logger.debug('==uri_path,http_uri 설정')
    except Exception as eee:
        file_size_max = 2000000000  #2GB
        uri_path = None
        http_uri = None
        logger.warning("==uri_path/http_uri가 설정되어 있지 않습니다. 2GB이상 파일 다운로드 불가능 합니다.\t에러:{}".format(eee))
    try:    # 승인된 유저 리스트가 있는지 확인
        users_id=parser.get('settings','users').split()
        logger.debug('==users 설정')
    except Exception as eee:
        users_id=[admin_id]
        logger.warning("==urers가 설정되어 있지 않습니다. admin만 허용 해 놓습니다.\t에러:{}".format(eee))
    try:    # 파일 매니저 기준 경로 설정
        base_path = format(parser.get('settings','base_path'))
        logger.debug('==base_path 설정')
    except Exception as eee :
        base_path = os.getcwd()
        logger.warning("==base_path가 설정되어 있지 않습니다. 현위치를 base_path로 설정합니다.\t에러:{}".format(eee))

#=====================사용할 변수 준비=====================#
try:
    help_text = '<b>==NAS helper bot==</b>\
        \n<code>/help</code>\t: 도움말\
        \n<code>/file</code>\t: 파일 매니저 시작\
        \n\n <b>업로드:</b> <code>/file</code> 이후 파일 첨부 시 현재 위치에 업로드 진행\
        \n └ 용량 제한: 2GB이하\
        \n └ 사진/음성/영상 등 묻는 경우 꼭 <code>파일</code>로 선택해야 함.\
        \n\n<b>주의사항</b>: <u>00시~00시5분 서비스 중지됨</u>\
        \n<b>주의사항</b>: <code>[삭제]</code>의 경우 별도의 확인 없이 삭제되므로 <u>주의</u>'
    # 아래 변수들은 유저 수 만큼의 False 값을 가지는 list로 생성 됨.
    current_path=[False for i in range(users_id.__len__())]
    current_path_p=[False for i in range(users_id.__len__())]
    files_list=[False for i in range(users_id.__len__())]
    target_file=[False for i in range(users_id.__len__())]
    files_enable=[False for i in range(users_id.__len__())]
    files_copy=[False for i in range(users_id.__len__())]
    files_move=[False for i in range(users_id.__len__())]
    logger.debug("초기 변수 생성 완료")
except Exception as eee:
    logger.critical("초기 변수 생성 실패")
    logger.critical(eee)
    exit(0)

#=====================유저 승인 처리=====================#
def update_user(chat_ID,phone_num):
    logger.debug('새 유저 추가.\t요청ID={}'.format(chat_ID))
    global parser
    global users_id
    global current_path
    global current_path_p
    global files_list
    global target_file
    global files_enable
    global files_copy
    global files_move
    try:    # 새 유저 추가에 따른 list 변수 원소 수 추가
        current_path.append(False)
        current_path_p.append(False)
        files_list.append(False)
        target_file.append(False)
        files_enable.append(False)
        files_copy.append(False)
        files_move.append(False)
    except Exception as eee:
        logger.critical('배열 변수 배열 증가 실패. 재시작 필요 {}'.format(eee))
        telegram.Bot(token=my_token).sendMessage(chat_id=admin_id,text='배열 변수 배열 증가 실패. 재시작 필요')
    users_id=users_id+[format(chat_ID)]
    logger.debug('users_id 변수 변경: {}'.format(users_id))
    try:    # 설정 파일에 추가된 user 적용
        parser.set('settings','users',' '.join(users_id))
        if not parser.has_section('Users'):
            parser.add_section('Users')
        parser.set('Users',format(chat_ID),format(phone_num))   # 추후 Chat_ID - 신원 확인을 위해 전화번호도 함께 저장
        with open('NAS_bot.ini','w') as f:
            parser.write(f)
        logger.debug('NAS_bot.ini에 users 변경 사항 저장')
    except Exception as eee:
        logger.critical('NAS_bot.ini에 변경사항 저장 실패')
        telegram.Bot(token=my_token).send_message(chat_id=admin_id,text='NAS_bot.ini에 변경사항 저장 실패')

#=====================상위 폴더 선택 시 처리=====================#
def files_upperdir(update):
    global current_path 
    global current_path_p
    try:
        current_path_p[users_id.index(format(update.effective_user.id))].append(current_path[users_id.index(format(update.effective_user.id))])
        current_path[users_id.index(format(update.effective_user.id))]= os.sep.join(current_path[users_id.index(format(update.effective_user.id))].split(os.path.sep)[0:-1])
    except Exception as eee:
        logger.critical('upperdir 작업 실패.\t요청자:{}\tcurrent_path:{}\tError 내용: {}'.format(update.effective_user.id,current_path_p[users_id.index(format(update.effective_user.id))],eee))

#=====================이전 폴더 선택 시 처리=====================#
def files_backdir(update):
    global current_path
    global current_path_p
    try:
        current_path[users_id.index(format(update.effective_user.id))]=current_path_p[users_id.index(format(update.effective_user.id))][-1]
        del current_path_p[users_id.index(format(update.effective_user.id))][-1]
    except Exception as eee:
        logger.critical('Backdir 실패.\t요청자:{}\tcurrent_path:{}\tError 내용: {}'.format(update.effective_user.id,current_path_p[users_id.index(format(update.effective_user.id))],eee))

#=====================메뉴 생성을 위한 함수=====================#
def build_menu(buttons, n_cols, header_buttons=None, footer_buttons=None):
    menu = [buttons[i:i + n_cols] for i in range(0, len(buttons), n_cols)]
    if header_buttons:
        menu.insert(0, header_buttons)
    if footer_buttons:
        menu.append(footer_buttons)
    return menu

#=====================자주 사용되는 메뉴 생성=====================#
buttons_footer=[]
buttons_footer.append(InlineKeyboardButton("이전 위치", callback_data="back_dir")) # add cancel button
buttons_footer.append(InlineKeyboardButton("종료", callback_data="exit")) # add cancel button
buttons_footer1=[InlineKeyboardButton("종료", callback_data="exit")] # add cancel button

#=====================/help 처리=====================#
def command_help(update,context) :
    context.bot.send_message(chat_id=update.effective_user.id,text=help_text,parse_mode=telegram.ParseMode.HTML)

#=====================폴더/파일 리스트 보여주는 함수=====================#
# user 권한 확인을 위한 command_files_init을 제외하면 이 함수가 파일 매니저의 몸통 #
def command_files(update,context,silent=False,msg_id=None) :
    global files_list
    try:
        local_path=current_path[users_id.index(format(update.effective_user.id))]
        local_path_p=current_path_p[users_id.index(format(update.effective_user.id))]
        files_list[users_id.index(format(update.effective_user.id))]=os.listdir(local_path)
        local_files_list=files_list[users_id.index(format(update.effective_user.id))]
        local_copy=files_copy[users_id.index(format(update.effective_user.id))]
        local_move=files_move[users_id.index(format(update.effective_user.id))]
        temp_num=0  # 폴더/파일 선택 index
        if local_path==base_path:   # 최상위 위치인 경우 상위폴더로 가지 못하도록 별도 header 버튼 사용
            header_buttons=[]
        else:   # 최상위가 아닌 경우 상위 폴더로 갈 수 있는 버튼 나오도록 header 생성
            header_buttons=[InlineKeyboardButton("/ (최상위)", callback_data="root_dir"),InlineKeyboardButton("../ (위로)", callback_data="upper_dir")]
        show_list = []
        for target in local_files_list :
            if (local_copy!=False)|(local_move!=False): # copy/move 이용 중인 경우 폴더만 표시되도록
                if os.path.isdir(os.path.join(local_path,target)) :
                    show_list.append(InlineKeyboardButton('[폴더] '+target, callback_data=temp_num))
            else :  # copy/move 중이 아닌 경우 폴더와 파일 모두 표시
                show_list.append(InlineKeyboardButton('[폴더] '+target if os.path.isdir(os.path.join(local_path,target)) else target, callback_data=temp_num))
            temp_num+=1
        if (local_copy!=False)|(local_move!=False): # copy/move 중인 경우 하단 메뉴
            temp_foot=[InlineKeyboardButton("여기",callback_data="here_dir"),InlineKeyboardButton("취소",callback_data="cancel_dir")]
            show_markup = InlineKeyboardMarkup(build_menu(show_list,2,header_buttons=header_buttons,footer_buttons=temp_foot))
        else:   # copy/move 중이 아닌 경우 하단 메뉴
            if not local_path_p :   # 이전 경로가 없는 경우
                show_markup = InlineKeyboardMarkup(build_menu(show_list, 2,header_buttons=header_buttons,footer_buttons=buttons_footer1)) # [종료]
            else :
                show_markup = InlineKeyboardMarkup(build_menu(show_list, 2,header_buttons=header_buttons,footer_buttons=buttons_footer)) # [이전]+[종료]
        if not silent : # 평소의 경우 이전 폴더 이동 history를 남기기 위해 새 메시지로 발송
            context.bot.send_message(chat_id=update.effective_user.id,text=local_path, reply_markup=show_markup)
        else :  # silent argument를 입력 받은 경우 이전 message를 수정하는 방식으로 적용
            context.bot.edit_message_text(chat_id=update.effective_user.id,text=local_path, reply_markup=show_markup,message_id=msg_id)
    except Exception as eee:
        logger.critical('command_file 작업 실패.\t요청자:{}\t위치:{}\t에러:{}'.format(update.effective_user.id,local_path,eee))
        context.bot.send_message(chat_id=update.effective_user.id,text='알 수 없는 에러 발생')

#=====================/start 처리=====================#
def command_start(update,context):
    command_help(update,context)    # 처음 접속하는 초보 유저를 위해 /help 자동 실행
    if not format(update.effective_user.id) in users_id:    # 권한 있는 user인지 확인
        reply_markup = telegram.ReplyKeyboardMarkup([[telegram.KeyboardButton('권한 요청 (전화번호 전송)', request_contact=True)]],one_time_keyboard=True)  # 권한 요청을 위해 연락처 전송.(연락처 전송은 Inline버튼 불가)
        context.bot.send_message(chat_id=update.effective_user.id,text='수행할 권한이 없습니다.\n↓ 권한 요청 클릭 ↓',reply_markup=reply_markup)

#=====================/file 처리=====================#
def command_files_init(update,context):
    global current_path
    global current_path_p
    global files_enable
    try:
        if format(update.effective_user.id) in users_id:    # user 권한 확인
            files_enable[users_id.index(format(update.effective_user.id))]=True # 업로드 가능하도록 설정
            temp=telegram.ReplyKeyboardRemove() # 연락처 전송 시 사용된 버튼이 계속 남는 경우 보기 싫기에 삭제되도록. (제대로 진행되었다면 안떠있을테지만 혹시 몰라서...)
            context.bot.send_message(chat_id=update.effective_user.id,text='<code>==File manager==</code>',reply_markup=temp,parse_mode=telegram.ParseMode.HTML)
            current_path[users_id.index(format(update.effective_user.id))]=base_path    # current_path를 기본 경로로 초기화
            current_path_p[users_id.index(format(update.effective_user.id))]=[] # 이전 경로 초기화
            command_files(update,context)   # 기본 경로에 대한 폴더/파일 리스트 표시
        else:   # user 권하이 없는데도 /file 수행 시 다시 권한 요청 띄우기.
            files_enable[users_id.index(format(update.effective_user.id))]=False
            reply_markup = telegram.ReplyKeyboardMarkup([[telegram.KeyboardButton('권한 요청(전화번호 전송)', request_contact=True)]],one_time_keyboard=True)
            context.bot.send_message(chat_id=update.effective_user.id,text='수행할 권한이 없습니다.\n↓ 권한 요청 클릭 ↓',reply_markup=reply_markup)
    except Exception as eee:
        logger.critical('command_files_init 작업 실패.\t요청자:{}\t에러:{}'.format(update.effective_user.id,eee))

#=====================업로드 처리=====================#
def command_gotfile(update,context):
    try:
        if files_enable[users_id.index(format(update.effective_user.id))]:  # /file 수행 중에만 current_path가 있기에 업로드 수행될 수 있다. 
            # 해당 여부 확인은 current_path 가 설정되어있는지로 확인할 수 있지만, 혹시 추후 업로드만 따로 어찌 해야할지도 몰라서 일단 분리함.
            temp=os.path.join(current_path[users_id.index(format(update.effective_user.id))],update.message.document.file_name) # 이 부분의 document 부분 때문에 "파일"로 받은 파일만 처리 가능. 
            while os.path.isfile(temp): # 현재 위치에 업로드 된 파일과 동일한 이름이 이미 있는 경우 _(1)을 계속 붙여 덮어쓰기 회피
                temp1,temp2=os.path.splitext(temp)
                temp=temp1+'_(1)'+temp2
            with open(temp,'wb') as f:
                context.bot.get_file(update.message.document).download(out=f)
            logger.debug('업로드 작업 성공.\t요청자:{}\t대상 파일:{}'.format(update.effective_user.id,temp))
    except Exception as eee:
        logger.critical('GotFile 작업 실패.\t요청자:{}\t대상 파일:{}\t에러:{}'.format(update.effective_user.id,temp,eee))

#=====================권한요청을 위한 연락처 수신 처리=====================#
def command_gotcontact(update,context):
    try:
        # 관리자에게 보낼 메시지 작성
        temp_text='==권한 요청=='\
            +'\nChat_ID: <code>'+format(update.effective_user.id)\
            +'</code>\n전화번호: <code>'+format(update.message.contact.phone_number)\
            +'</code>\nID: <code>'+format(update.effective_user.username)\
            +'</code>\n이름: <code>'+format(update.effective_user.last_name)+' '+format(update.effective_user.first_name)+'</code>'\
            +'\n<a href="tg://user?id={}">대화하기</a>'.format(update.effective_user.id)+''
        show_list=[]
        show_list.append(InlineKeyboardButton("허용", callback_data="accept_user"))
        show_list.append(InlineKeyboardButton("거부", callback_data="deny_user"))
        show_markup = InlineKeyboardMarkup(build_menu(show_list, 2))
        context.bot.send_message(chat_id=admin_id,text=temp_text,reply_markup=show_markup,parse_mode=telegram.ParseMode.HTML)
        logger.debug('권한 요청 메시지 발송.\t요청자:{}'.format(update.effective_user.id))
    except Exception as eee:
        logger.critical('연락처 처리 실패.\t요청자:{}\t에러:{}'.format(update.effective_user.id,eee))

#=====================폴더/파일 선택 처리=====================#
def selected_files(update,context,sel_num,query):
    global target_file
    try:
        local_path=current_path[users_id.index(format(update.effective_user.id))]
        local_files_list=files_list[users_id.index(format(update.effective_user.id))]
        context.bot.edit_message_text(message_id=query.message.message_id,chat_id=query.message.chat_id,text=local_path)    # 폴더/파일 리스트 버튼 없애기 위한 메시지 수정
        target_file[users_id.index(format(update.effective_user.id))]=os.path.join(local_path,local_files_list[sel_num])    # 파일 선택 시 처리를 위한 변수 설정
        header_buttons=[InlineKeyboardButton(" 다운로드", callback_data="download_file")]
        show_list = []
        show_list.append(InlineKeyboardButton("복사", callback_data="copy_file"))
        show_list.append(InlineKeyboardButton("이동", callback_data="move_file"))
        show_list.append(InlineKeyboardButton("삭제", callback_data="delete_file"))
        buttons_footer2=[InlineKeyboardButton("선택 취소", callback_data="cancel_file")]
        show_markup = InlineKeyboardMarkup(build_menu(show_list, 3,header_buttons=header_buttons, footer_buttons=buttons_footer2))
        context.bot.send_message(text=local_files_list[sel_num],chat_id=query.message.chat_id, reply_markup=show_markup)    # 파일 선택 메뉴 메시지
    except Exception as eee:
        logger.critical('Selected_files 작업 실패.\t요청자:{}\t에러:{}'.format(query.message.chat_id,eee))

#=====================callback 처리=====================#
def cd_button(update, context) :
    global current_path
    global current_path_p
    global files_enable
    global files_copy
    global files_move
    try:
        query = update.callback_query
        data = query.data
        local_path=current_path[users_id.index(format(update.effective_user.id))]
        local_files_list=files_list[users_id.index(format(update.effective_user.id))]
        local_target_file=target_file[users_id.index(format(update.effective_user.id))]
        #=====================상위 폴더 선택 처리=====================#
        if 'upper_dir' in data:
            context.bot.send_chat_action(chat_id=update.effective_user.id,action=ChatAction.TYPING) # 입력 중 표시
            context.bot.edit_message_text(text=local_path,chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
            files_upperdir(update)  # 상위폴더 이동 함수 호출(current_path 수정)
            command_files(update,context)   # current_path에 대한 폴더/파일 표시 처리
        #=====================이전 폴더 이동 처리=====================#
        elif 'back_dir' in data:
            context.bot.send_chat_action(chat_id=update.effective_user.id,action=ChatAction.TYPING) # 입력 중 표시
            context.bot.edit_message_text(text=local_path,chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
            files_backdir(update)   # 이전폴더 이동 함수 호출(current_path 수정)
            command_files(update,context)   # current_path에 대한 폴더/파일 표시 처리
        #=====================종료 처리=====================#
        elif 'exit' in data:
            files_enable[users_id.index(format(update.effective_user.id))]=False    # 파일 업로드 불가하도록 변수 수정
            context.bot.edit_message_text(text=local_path,chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
            context.bot.send_message(text='접속 종료',chat_id=query.message.chat_id)
            return
        #=====================파일 선택 후 다운로드 선택 처리=====================#
        elif 'download_file' in data:
            if os.path.getsize(local_target_file) <= file_size_max : # 파일 용량 확인(2GB 보다 조금 작게 잡음)
                # 2GB 이하 파일은 메시지로 보냄.
                try:
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)  # 폴더/파일 리스트 버튼 삭제
                    context.bot.send_document(chat_id=query.message.chat_id,document=open(local_target_file,'rb'))  # 파일 전송
                    logger.debug('파일 다운로드.\t요청자:{}\t대상 파일:{}'.format(query.message.chat_id,local_target_file))
                except Exception as e:
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)  # 폴더/파일 리스트 버튼 삭제
                    context.bot.send_message(chat_id=query.message.chat_id,text='파일 다운로드 실패')
                    logger.critical('파일 전송 실패.\t요청자:{}\t대상 파일:{}\t에러:{}'.format(query.message.chat_id,local_target_file,e))
            else:   # 2GB보다 큰 파일 처리
                context.bot.edit_message_text(text=os.path.split(local_target_file)[-1]+' 처리 중',chat_id=query.message.chat_id,message_id=query.message.message_id)   # 용량 큰 파일은 이동에 시간이 소요되기에 처리 중 표시
                temp=os.path.join(uri_path,os.path.split(local_target_file)[-1])
                while os.path.isfile(temp): # url 만들 위치에 파일 중복 여부 확인 후 이름 변경
                    temp1,temp2=os.path.splitext(temp)
                    temp=temp1+'_(1)'+temp2
                try:
                    shutil.copy(local_target_file,temp)
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)
                    context.bot.send_message(chat_id=query.message.chat_id,text='파일 크기가 2GB를 넘어 링크를 제공합니다.(00시 자동 삭제)\n<a href="{}/{}">다운로드</a>'.format(http_uri,os.path.split(temp)[-1]),parse_mode=telegram.ParseMode.HTML)
                    logger.debug('파일 링크 생성.\t요청자:{}\t대상 파일:{}'.format(query.message.chat_id,local_target_file))
                except OSError as e:
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)
                    context.bot.send_message(text=os.path.split(local_target_file)[-1]+' 다운로드 링크 생성 실패\n에러: '+e,chat_id=query.message.chat_id,message_id=query.message.message_id)
                    logger.critical('파일 링크 생성 실패.\t요청자:{}\t대상 파일:{}\t에러:{}'.format(query.message.chat_id,local_target_file,e))
            command_files(update,context)   
        #=====================파일 선택 후 선택 취소 선택 처리=====================#
        elif 'cancel' in data:
            context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
            context.bot.send_chat_action(chat_id=update.effective_user.id,action=ChatAction.TYPING)
            command_files(update,context)
        #=====================최상위 위치 선택 처리=====================#
        elif 'root_dir' in data:
            context.bot.edit_message_text(text=local_path,chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
            current_path[users_id.index(format(update.effective_user.id))]=base_path
            command_files(update,context)
        #=====================권한 요청 허가 선택 처리=====================#
        elif 'accept_user' in data:
            temp=query.message.text.split()[3]  # 이전 메시지에서 요청자 chat_ID 추출
            context.bot.send_message(chat_id=update.effective_user.id,text=format(temp)+'의 권한 요청 승인')    # 관리자에게
            update_user(temp,query.message.text.split()[5]) # 신규 유저 처리 함수에 chat_ID와 전화번호 전달
            context.bot.send_message(chat_id=temp,text='권한 요청이 승인 되었습니다.',reply_markup=telegram.ReplyKeyboardRemove())  # 요청자에게
            logger.debug('관리자가 권한 요청에 대한 승인 처리함.\t요청자:{}'.format(temp))
        #=====================권한 요청 거부 선택 처리=====================#
        elif 'deny_user' in data:
            temp=query.message.text.split()[3]  # 이전 메시지에서 요청자 chat_ID 추출
            context.bot.send_message(chat_id=update.effective_user.id,text=format(temp)+'의 권한 요청 거부')    # 관리자에게
            context.bot.send_message(chat_id=temp,text='권한 요청이 거부 되었습니다.\n재요청: 연락처 재전송')    # 요청자에게
            logger.debug('관리자가 권한 요청에 대한 승인 거부함.\t요청자:{}'.format(temp))
        #=====================파일 선택 후 복사 선택 처리=====================#
        elif 'copy_file' in data:
            context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)  # 파일 선택 버튼 삭제
            files_copy[users_id.index(format(update.effective_user.id))]=local_target_file  # 폴더 탐색 중 copy 중이라는 것을 알려주는 변수
            command_files(update,context)
        #=====================파일 선택 후 이동 선택 처리=====================#
        elif 'move_file' in data:
            context.bot.edit_message_text(text=os.path.split(local_target_file)[-1],chat_id=query.message.chat_id,message_id=query.message.message_id)  # 파일 선택 버튼 삭제
            files_move[users_id.index(format(update.effective_user.id))]=local_target_file  # 폴더 탐색 중 move 중이라는 것을 알려주는 변수
            command_files(update,context)
        #=====================파일 선택 후 복사/이동 선택 후 폴더 탐색 중 여기 선택 처리=====================#
        elif 'here_dir' in data:
            local_copy=files_copy[users_id.index(format(update.effective_user.id))]
            local_move=files_move[users_id.index(format(update.effective_user.id))]
            if local_copy!=False:
                temp=os.path.join(current_path[users_id.index(format(update.effective_user.id))],os.path.split(local_copy)[-1])
            elif local_move!=False:
                temp=os.path.join(current_path[users_id.index(format(update.effective_user.id))],os.path.split(local_move)[-1])
            while os.path.isfile(temp): # 중복 파일 확인 후 이름변경
                temp1,temp2=os.path.splitext(temp)
                temp=temp1+'_(1)'+temp2
            if local_copy!=False:
                try:
                    shutil.copy(local_copy,temp)    # 복사 진행
                    logger.debug('파일 복사.\t요청자:{}\t원본 파일:{}\t대상 파일:{}'.format(update.effective_user.id,local_copy,temp))
                    context.bot.edit_message_text(text=local_path+'로 복사 완료',chat_id=query.message.chat_id,message_id=query.message.message_id)
                except OSError as e:
                    logger.critical('파일 복사.\t요청자:{}\t원본 파일:{}\t대상 파일:{}\t에러:{}'.format(update.effective_user.id,local_copy,temp,e))
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1]+' 복사 실패\n에러: '+e,chat_id=query.message.chat_id,message_id=query.message.message_id)
            elif local_move!=False:
                try:
                    shutil.move(local_move,temp)    # 이동 진행
                    logger.debug('파일 이동.\t요청자:{}\t원본 파일:{}\t대상 파일:{}'.format(update.effective_user.id,local_copy,temp))
                    context.bot.edit_message_text(text=local_path+'로 이동 완료',chat_id=query.message.chat_id,message_id=query.message.message_id)
                except OSError as e:
                    logger.critical('파일 이동.\t요청자:{}\t원본 파일:{}\t대상 파일:{}\t에러:{}'.format(update.effective_user.id,local_copy,temp,e))
                    context.bot.edit_message_text(text=os.path.split(local_target_file)[-1]+' 이동 실패\n에러: '+e,chat_id=query.message.chat_id,message_id=query.message.message_id)
            files_copy[users_id.index(format(update.effective_user.id))]=False  # 파일 복사 여부 변수 초기화
            files_move[users_id.index(format(update.effective_user.id))]=False  # 파일 이동 여부 변수 초기화
            command_files(update,context)
        #=====================파일 선택 후 복사/이동 선택 후 폴더 탐색 중 취소 선택 처리=====================#
        elif 'cancel_dir' in data:
            files_copy[users_id.index(format(update.effective_user.id))]=False  # 파일 복사 여부 변수 초기화
            files_move[users_id.index(format(update.effective_user.id))]=False  # 파일 이동 여부 변수 초기화
            command_files(update,context)
        #=====================파일 선택 후 삭제 선택 처리=====================#
        elif 'delete_file' in data:
            try:
                os.remove(local_target_file)    # 삭제 진행
                logger.debug('파일 삭제.\t요청자:{}\t대상 파일:{}'.format(query.message.chat_id,local_target_file))
                context.bot.edit_message_text(text=os.path.split(local_target_file)[-1]+' 삭제',chat_id=query.message.chat_id,message_id=query.message.message_id)
            except OSError as e:
                logger.critical('파일 삭제.\t요청자:{}\t대상 파일:{}\t에러:{}'.format(query.message.chat_id,local_target_file,e))
                context.bot.edit_message_text(text=os.path.split(local_target_file)[-1]+' 삭제 실패\n에러: '+e,chat_id=query.message.chat_id,message_id=query.message.message_id)
            command_files(update,context)
        #=====================폴더/파일 선택 처리=====================#
        elif data.isdigit():
            if os.path.isfile(os.path.join(local_path,local_files_list[int(data)])):    # 파일 여부 확인
                selected_files(update,context,int(data),query)
            elif os.path.isdir(os.path.join(local_path,local_files_list[int(data)])):   # 폴더 여부 확인
                context.bot.edit_message_text(text=local_path,chat_id=query.message.chat_id,message_id=query.message.message_id)    # 폴더/파일 리스트 버튼 삭제
                current_path_p[users_id.index(format(update.effective_user.id))].append(local_path) # 이전 경로 추가
                current_path[users_id.index(format(update.effective_user.id))]=os.path.join(local_path,local_files_list[int(data)]) # 선택한 폴더로 현재 경로 변경
                command_files(update,context)   # 폴더/파일 리스트 업데이트
            else:   # 폴더도 파일도 아닌 경우? 권한이 없는 경우?
                logger.critical("폴더/파일 선택 에러.\t요청자:{}\t현재 경로:{}\t선택 사항:{}".format(query.message.chat_id,local_path,local_files_list[int(data)]))
                context.bot.edit_message_text(text='에러 발생',chat_id=query.message.chat_id,message_id=query.message.message_id)
                command_files(update,context)
        else :  # query 결과에 해당하는 선택이 없는 경우
            logger.critical("폴더/파일 선택 에러.\t요청자:{}\t현재 경로:{}\t선택 사항:{}".format(query.message.chat_id,local_path,data))
            context.bot.edit_message_text(text='[{}] : 알 수 없는 작업.'.format(data),chat_id=query.message.chat_id,message_id=query.message.message_id)
    except Exception as eee:
        logger.critical('cd_buttons 작업 실패.\t요청자:{}\t에러:{}'.format(query.message.chat_id,eee))

#=====================예제에 있던 에러처리 함수=====================#
def error(update,context) :
    logger.warning('Update "%s" caused error "%s"',context,update.error)

#=====================Thread 함수(bot 구동)=====================#
updater = Updater(my_token,use_context=True)
# 명령어 처리 #
updater.dispatcher.add_handler(CommandHandler('help', command_help))
updater.dispatcher.add_handler(CommandHandler('start', command_start))
updater.dispatcher.add_handler(CommandHandler('file', command_files_init))
updater.dispatcher.add_handler(MessageHandler(Filters.contact,command_gotcontact))  # 연락처 수신 판단
updater.dispatcher.add_handler(MessageHandler(Filters.document,command_gotfile))    # 파일 수신 판단
updater.dispatcher.add_handler(CallbackQueryHandler(cd_button)) # 버튼 선택 판단(callback)
updater.dispatcher.add_error_handler(error) # 에러 발생 처리 함수 (예제에 있던)
updater.start_polling(timeout=5,clean=True) # bot이 서버에 5번 연결 시도, 읽어온 메시지는 서버에서 삭제
updater.idle()
logger.critical('Program Terminated')   # 프로그램 종료 로그

 

평소 저는 코드 작성하면서 주석을 잘 안다는 편인데, 이번에는 코드 공개를 위해 최대한 주석을 유지하도록 노력하였습니다.

앞서 작성했던 다른 글에 나와있듯 이 코드를 돌리실 때에는 NAS_bot.ini 파일이 함께 있지 않으면 초기에 구동되지 않고 종료됩니다.

python module은 python-telegram-bot 만 설치해도 동작합니다.

반응형

댓글