Update src/main-backup.py

src/main-backup.py

@@ -7,99 +7,134 @@ import cv2
 import numpy as np
 import os
 import requests
+from concurrent.futures import ThreadPoolExecutor
 from urllib.parse import quote, unquote
 import tempfile
 import re
+from functools import lru_cache
+from typing import List, Dict, Any
+import logging
+from contextlib import contextmanager
+
+# 로깅 설정
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 app = Flask(__name__, static_folder='static')
 app.config['TITLE'] = 'Sign Language Translate'
 
+# 전역 변수를 초기화하고 캐싱
 nlp, dict_docs_spacy = sp.load_spacy_values()
 dataset, list_2000_tokens = dg.load_data()
 
-def clean_quotes(text):
+# 스레드 풀 생성
+executor = ThreadPoolExecutor(max_workers=4)
+
+# 메모리 캐시 데코레이터
+@lru_cache(maxsize=1000)
+def clean_quotes(text: str) -> str:
     """따옴표 정리 함수"""
     text = re.sub(r"'+", "'", text)
     text = re.sub(r'\s+', ' ', text).strip()
     return text
 
-def is_korean(text):
+@lru_cache(maxsize=1000)
+def is_korean(text: str) -> bool:
     """한글이 포함되어 있는지 확인"""
     return bool(re.search('[가-힣]', text))
 
-def is_english(text):
+@lru_cache(maxsize=1000)
+def is_english(text: str) -> bool:
     """텍스트가 영어인지 확인하는 함수"""
     text_without_quotes = re.sub(r"'[^']*'|\s", "", text)
     return bool(re.match(r'^[A-Za-z.,!?-]*$', text_without_quotes))
 
-def normalize_quotes(text):
+@lru_cache(maxsize=1000)
+def normalize_quotes(text: str) -> str:
     """따옴표 형식을 정규화하는 함수"""
     text = re.sub(r"'+", "'", text)
     text = re.sub(r'\s+', ' ', text).strip()
     
-    # 이미 따옴표로 묶인 단어가 있으면 그대로 반환
     if re.search(r"'[^']*'", text):
         return text
-    
     return text
 
-def find_quoted_words(text):
+@lru_cache(maxsize=1000)
+def find_quoted_words(text: str) -> List[str]:
     """작은따옴표로 묶인 단어들을 찾는 함수"""
     return re.findall(r"'([^']*)'", text)
 
-def spell_out_word(word):
+@lru_cache(maxsize=1000)
+def spell_out_word(word: str) -> str:
     """단어를 개별 알파벳으로 분리하는 함수"""
     return ' '.join(list(word.lower()))
 
-def translate_korean_text(text):
-    """한글 전용 번역 함수"""
+def translate_text_chunk(text: str, source_lang: str, target_lang: str) -> str:
+    """텍스트 번역 함수"""
     try:
-        quoted_match = re.search(r"'([^']*)'", text)
-        if not quoted_match:
-            return text
-
-        quoted_word = quoted_match.group(1)
-        
         url = "https://translate.googleapis.com/translate_a/single"
         params = {
             "client": "gtx",
-            "sl": "ko",
-            "tl": "en",
+            "sl": source_lang,
+            "tl": target_lang,
             "dt": "t",
-            "q": text.replace(f"'{quoted_word}'", "XXXXX")
+            "q": text
         }
+        
         response = requests.get(url, params=params)
         if response.status_code != 200:
+            logger.error(f"Translation API error: {response.status_code}")
             return text
+            
+        data = response.json()
+        return ' '.join(item[0] for item in data[0] if item[0])
+            
+    except Exception as e:
+        logger.error(f"Translation error: {e}")
+        return text
+
+def translate_korean_text(text: str) -> str:
+    """한글 전용 번역 함수"""
+    try:
+        quoted_match = re.search(r"'([^']*)'", text)
+        if not quoted_match:
+            # 인용부호가 없는 경우 전체 텍스트를 번역하도록 함
+            translated_text = translate_text_chunk(text, "ko", "en")
+            # 번역 후 추가 전처리 (필요한 경우)
+            translated_text = re.sub(r'\bNAME\b', 'name', translated_text)
+            translated_text = translated_text.replace(" .", ".")
+            return translated_text
 
-        translated_text = ' '.join(item[0] for item in response.json()[0] if item[0])
+        quoted_word = quoted_match.group(1)
+        
+        # 본문 번역: 인용부호 단어를 placeholder로 대체
+        main_text = text.replace(f"'{quoted_word}'", "XXXXX")
+        translated_main = translate_text_chunk(main_text, "ko", "en")
         
+        # 인용된 단어 처리
         if re.match(r'^[A-Za-z]+$', quoted_word):
             proper_noun = quoted_word.upper()
         else:
-            params["q"] = quoted_word
-            response = requests.get(url, params=params)
-            if response.status_code == 200:
-                proper_noun = response.json()[0][0][0].upper()
-            else:
-                proper_noun = quoted_word.upper()
+            proper_noun = translate_text_chunk(quoted_word, "ko", "en").upper()
 
-        final_text = translated_text.replace("XXXXX", f"'{proper_noun}'")
+        final_text = translated_main.replace("XXXXX", f"'{proper_noun}'")
         final_text = re.sub(r'\bNAME\b', 'name', final_text)
         final_text = final_text.replace(" .", ".")
         
         return final_text
 
     except Exception as e:
-        print(f"Korean translation error: {e}")
+        logger.error(f"Korean translation error: {e}")
         return text
 
-def translate_korean_to_english(text):
+
+def translate_korean_to_english(text: str) -> str:
     """전체 텍스트 번역 함수"""
     try:
         text = normalize_quotes(text)
         
         if is_english(text):
+            # 영어 문장 안의 작은따옴표 단어를 대문자로 바꿔주는 처리
             quoted_match = re.search(r"'([^']*)'", text)
             if quoted_match:
                 quoted_word = quoted_match.group(1).upper()
@@ -111,9 +146,73 @@ def translate_korean_to_english(text):
 
         return text
     except Exception as e:
-        print(f"Translation error: {e}")
+        logger.error(f"Translation error: {e}")
         return text
 
+def process_frame(frame_data: bytes) -> np.ndarray:
+    """프레임 처리 함수"""
+    try:
+        frame_content = frame_data.split(b'\r\n\r\n')[1]
+        nparr = np.frombuffer(frame_content, np.uint8)
+        return cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+    except Exception as e:
+        logger.error(f"Frame processing error: {e}")
+        raise
+
+@contextmanager
+def video_writer(path: str, frame_size: tuple, fps: int = 25):
+    """비디오 작성을 위한 컨텍스트 매니저"""
+    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+    writer = cv2.VideoWriter(path, fourcc, fps, frame_size)
+    try:
+        yield writer
+    finally:
+        writer.release()
+
+def generate_complete_video(gloss_list: List[str], dataset: Dict[str, Any], list_2000_tokens: List[str]) -> bytes:
+    """최적화된 비디오 생성 함수"""
+    try:
+        frames = []
+        is_spelling = False
+        
+        # 프레임 생성을 병렬로 처리
+        with ThreadPoolExecutor() as executor:
+            for gloss in gloss_list:
+                if gloss == 'FINGERSPELL-START':
+                    is_spelling = True
+                    continue
+                elif gloss == 'FINGERSPELL-END':
+                    is_spelling = False
+                    continue
+                
+                frame_futures = [
+                    executor.submit(process_frame, frame)
+                    for frame in dg.generate_video([gloss], dataset, list_2000_tokens)
+                ]
+                frames.extend([future.result() for future in frame_futures])
+        
+        if not frames:
+            raise Exception("No frames generated")
+
+        height, width = frames[0].shape[:2]
+        
+        with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as temp_file:
+            temp_path = temp_file.name
+            
+            with video_writer(temp_path, (width, height)) as out:
+                for frame in frames:
+                    out.write(frame)
+            
+            with open(temp_path, 'rb') as f:
+                video_bytes = f.read()
+            
+        os.remove(temp_path)
+        return video_bytes
+        
+    except Exception as e:
+        logger.error(f"Video generation error: {str(e)}")
+        raise
+
 @app.route('/')
 def index():
     return render_template('index.html', title=app.config['TITLE'])
@@ -123,6 +222,7 @@ def result():
     if request.method == 'POST':
         input_text = request.form['inputSentence'].strip()
         if not input_text:
+            # 에러 페이지로 이동
             return render_template('error.html', error="Please enter text to translate")
         
         try:
@@ -133,10 +233,15 @@ def result():
             
             quoted_words = find_quoted_words(english_text)
             
-            clean_english = re.sub(r"'([^']*)'", r"\1", english_text)
-            eng_to_asl_translator = NlpSpacyBaseTranslator(sentence=clean_english)
-            generated_gloss = eng_to_asl_translator.translate_to_gloss()
+            # NLP 처리를 스레드 풀에서 실행
+            def process_nlp():
+                clean_english = re.sub(r"'([^']*)'", r"\1", english_text)
+                eng_to_asl_translator = NlpSpacyBaseTranslator(sentence=clean_english)
+                return eng_to_asl_translator.translate_to_gloss()
+                
+            generated_gloss = executor.submit(process_nlp).result()
             
+            # 작은따옴표 처리(지정된 영단어는 fingerspell로 변환)
             processed_gloss = []
             words = generated_gloss.split()
             
@@ -150,86 +255,59 @@ def result():
             
             gloss_sentence_before_synonym = " ".join(processed_gloss)
             
+            # 동의어 처리
             final_gloss = []
             i = 0
             while i < len(processed_gloss):
                 if processed_gloss[i] == 'FINGERSPELL-START':
-                    final_gloss.extend(processed_gloss[i:i+2])
+                    # fingerspell 구간은 그대로 넣는다
+                    final_gloss.extend(processed_gloss[i:i+2])  # FINGERSPELL-START + 첫 글자
                     i += 2
                     while i < len(processed_gloss) and processed_gloss[i] != 'FINGERSPELL-END':
                         final_gloss.append(processed_gloss[i])
                         i += 1
                     if i < len(processed_gloss):
-                        final_gloss.append(processed_gloss[i])
+                        final_gloss.append(processed_gloss[i])  # FINGERSPELL-END
                         i += 1
                 else:
                     word = processed_gloss[i]
-                    final_gloss.append(sp.find_synonyms(word, nlp, dict_docs_spacy, list_2000_tokens))
+                    # 동의어 찾기를 스레드 풀에서 비동기로 실행하여 결과 얻기
+                    synonym = executor.submit(
+                        sp.find_synonyms,
+                        word,
+                        nlp,
+                        dict_docs_spacy,
+                        list_2000_tokens
+                    ).result()
+                    final_gloss.append(synonym)
                     i += 1
             
             gloss_sentence_after_synonym = " ".join(final_gloss)
             
-            return render_template('result.html',
-                                title=app.config['TITLE'],
-                                original_sentence=input_text,
-                                english_translation=english_text,
-                                gloss_sentence_before_synonym=gloss_sentence_before_synonym,
-                                gloss_sentence_after_synonym=gloss_sentence_after_synonym)
+            return render_template(
+                'result.html',
+                title=app.config['TITLE'],
+                original_sentence=input_text,
+                english_translation=english_text,
+                gloss_sentence_before_synonym=gloss_sentence_before_synonym,
+                gloss_sentence_after_synonym=gloss_sentence_after_synonym
+            )
+                                
         except Exception as e:
+            logger.error(f"Translation processing error: {str(e)}")
             return render_template('error.html', error=f"Translation error: {str(e)}")
 
-def generate_complete_video(gloss_list, dataset, list_2000_tokens):
-    try:
-        frames = []
-        is_spelling = False
-        
-        for gloss in gloss_list:
-            if gloss == 'FINGERSPELL-START':
-                is_spelling = True
-                continue
-            elif gloss == 'FINGERSPELL-END':
-                is_spelling = False
-                continue
-                
-            for frame in dg.generate_video([gloss], dataset, list_2000_tokens):
-                frame_data = frame.split(b'\r\n\r\n')[1]
-                nparr = np.frombuffer(frame_data, np.uint8)
-                img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
-                frames.append(img)
-        
-        if not frames:
-            raise Exception("No frames generated")
-
-        height, width = frames[0].shape[:2]
-        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-        
-        with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as temp_file:
-            temp_path = temp_file.name
-            
-        out = cv2.VideoWriter(temp_path, fourcc, 25, (width, height))
-        
-        for frame in frames:
-            out.write(frame)
-        out.release()
-        
-        with open(temp_path, 'rb') as f:
-            video_bytes = f.read()
-        
-        os.remove(temp_path)
-        return video_bytes
-    except Exception as e:
-        print(f"Error generating video: {str(e)}")
-        raise
-
 @app.route('/video_feed')
 def video_feed():
     sentence = request.args.get('gloss_sentence_to_display', '')
     gloss_list = sentence.split()
-    return Response(dg.generate_video(gloss_list, dataset, list_2000_tokens), 
-                   mimetype='multipart/x-mixed-replace; boundary=frame')
+    return Response(
+        dg.generate_video(gloss_list, dataset, list_2000_tokens), 
+        mimetype='multipart/x-mixed-replace; boundary=frame'
+    )
 
 @app.route('/download_video/<path:gloss_sentence>')
-def download_video(gloss_sentence):
+def download_video(gloss_sentence: str):
     try:
         decoded_sentence = unquote(gloss_sentence)
         gloss_list = decoded_sentence.split()
@@ -249,8 +327,16 @@ def download_video(gloss_sentence):
             download_name='sign_language.mp4'
         )
     except Exception as e:
-        print(f"Download error: {str(e)}")
+        logger.error(f"Video download error: {str(e)}")
         return f"Error downloading video: {str(e)}", 500
 
+# 만약 main.py를 직접 실행하는 경우(개발 환경)
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=7860, debug=True)
+    app.run(host="0.0.0.0", port=7860, debug=True)
+
+# ---------------------------------------------------------------------
+# Gunicorn 설정을 main.py 파일 내에 포함 (Gunicorn 실행 시 -c 옵션으로 사용)
+# 아래 변수들은 Gunicorn이 main.py를 설정 파일로 사용할 경우 적용됩니다.
+bind = "0.0.0.0:7860"
+workers = 4
+timeout = 120  # worker timeout을 120초로 설정