diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f9147ec --- /dev/null +++ b/.gitignore @@ -0,0 +1,47 @@ +# ========================= +# Python 가상환경 +# ========================= +# 각 개발자의 로컬 환경에 따라 내용이 다르며, +# 용량이 매우 크기 때문에 Git으로 관리하지 않습니다. +# 대신 'requirements.txt' 파일을 통해 필요한 라이브러리를 관리합니다. +nllb_env/ + +# ========================= +# AI 모델 파일 +# ========================= +# 수 GB에 달하는 매우 큰 파일들입니다. +# Git은 대용량 파일을 관리하기에 적합하지 않으므로 제외합니다. +# 모델은 별도의 방법(예: 다운로드 스크립트)으로 관리해야 합니다. +models/ + +# ========================= +# 데이터 파일 +# ========================= +# 원본 데이터나 학습 데이터는 용량이 클 수 있으므로 제외합니다. +data/ + +# ========================= +# 실행 결과물 +# ========================= +# 코드를 실행하면 자동으로 생성되는 출력 파일들입니다. +# 소스 코드가 아니므로 버전 관리 대상에서 제외합니다. +output/ + +# ========================= +# Python 캐시 파일 +# ========================= +# Python 인터프리터가 실행 속도 향상을 위해 자동으로 생성하는 파일들입니다. +__pycache__/ +*.pyc + +# ========================= +# macOS 시스템 파일 +# ========================= +# macOS의 Finder가 자동으로 생성하는 시스템 파일입니다. +.DS_Store + +# ========================= +# IDE 및 에디터 설정 +# ========================= +.idea/ +.vscode/ \ No newline at end of file diff --git a/check_installation.py b/check_installation.py new file mode 100644 index 0000000..00f9d92 --- /dev/null +++ b/check_installation.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +""" +환경 설정 확인 스크립트 +""" + +def check_installation(): + print("🔍 설치된 패키지 확인 중...") + + try: + import torch + print(f"✅ PyTorch: {torch.__version__}") + + # Apple Silicon MPS 확인 + if torch.backends.mps.is_available(): + print("✅ Apple Silicon MPS 가속 사용 가능") + else: + print("⚠️ MPS 가속 사용 불가 (CPU 모드)") + + except ImportError: + print("❌ PyTorch 설치 실패") + return False + + try: + import transformers + print(f"✅ Transformers: {transformers.__version__}") + except ImportError: + print("❌ Transformers 설치 실패") + return False + + try: + import sentencepiece + print("✅ SentencePiece 설치 완료") + except ImportError: + print("❌ SentencePiece 설치 실패") + return False + + try: + import accelerate + print("✅ Accelerate 설치 완료") + except ImportError: + print("❌ Accelerate 설치 실패") + return False + + # 문서 처리 라이브러리 확인 + doc_libs = [] + try: + import PyPDF2 + doc_libs.append("PyPDF2") + except ImportError: + pass + + try: + import pdfplumber + doc_libs.append("pdfplumber") + except ImportError: + pass + + try: + from docx import Document + doc_libs.append("python-docx") + except ImportError: + pass + + if doc_libs: + print(f"✅ 문서 처리: {', '.join(doc_libs)}") + else: + print("⚠️ 문서 처리 라이브러리 설치 필요") + + # 시스템 정보 + print(f"\n📊 시스템 정보:") + print(f" Python 버전: {torch.version.python}") + if torch.backends.mps.is_available(): + print(f" 디바이스: Apple Silicon (MPS)") + else: + print(f" 디바이스: CPU") + + return True + +if __name__ == "__main__": + print("🚀 NLLB 번역 시스템 환경 확인") + print("=" * 40) + + if check_installation(): + print("\n🎉 환경 설정 완료!") + print("다음 단계: NLLB 모델 다운로드") + else: + print("\n❌ 환경 설정 실패") + print("패키지 재설치 필요") diff --git a/config/settings.json b/config/settings.json new file mode 100644 index 0000000..c1fc8f7 --- /dev/null +++ b/config/settings.json @@ -0,0 +1,28 @@ +{ + "models": { + "translation": "facebook/nllb-200-3.3B", + "summarization": "ainize/kobart-news", + "embedding": "nomic-embed-text" + }, + "translation": { + "max_length": 512, + "num_beams": 4, + "early_stopping": true, + "batch_size": 4 + }, + "summarization": { + "max_length": 150, + "min_length": 30, + "num_beams": 4 + }, + "processing": { + "chunk_size": 500, + "overlap": 50, + "concurrent_chunks": 3 + }, + "output": { + "html_template": "modern", + "include_toc": true, + "include_summary": true + } +} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..95035c5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,32 @@ +# AI 모델 관련 +torch>=2.0.0 +transformers>=4.30.0 +sentencepiece>=0.1.99 +accelerate>=0.20.0 +datasets>=2.12.0 + +# PDF 처리 +PyPDF2>=3.0.1 +pdfplumber>=0.9.0 + +# 문서 처리 +python-docx>=0.8.11 + +# 언어 감지 +langdetect>=1.0.9 + +# 웹 프레임워크 +fastapi>=0.100.0 +uvicorn[standard]>=0.22.0 +python-multipart>=0.0.6 +jinja2>=3.1.2 +aiofiles>=23.1.0 + +# 유틸리티 +tqdm>=4.65.0 +numpy>=1.24.0 +pandas>=2.0.0 + +# 추가 도구 +requests>=2.31.0 +pathlib2>=2.3.7 diff --git a/src/download_kobart.py b/src/download_kobart.py new file mode 100755 index 0000000..4ac930b --- /dev/null +++ b/src/download_kobart.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +""" +KoBART 한국어 요약 모델 다운로드 및 테스트 +""" + +import torch +import time +from transformers import BartForConditionalGeneration, AutoTokenizer +from pathlib import Path + +def download_kobart_model(): + print("🔄 KoBART 한국어 요약 모델 다운로드 시작...") + print("📊 모델 크기: ~500MB, 다운로드 시간 3-5분 예상") + + model_name = "ainize/kobart-news" + + # 로컬 모델 저장 경로 + model_dir = Path("models/kobart-news") + model_dir.mkdir(parents=True, exist_ok=True) + + try: + # 1. 토크나이저 다운로드 + print("\n📥 1/2: 토크나이저 다운로드 중...") + tokenizer = AutoTokenizer.from_pretrained( + model_name, + cache_dir=str(model_dir) + ) + print("✅ 토크나이저 다운로드 완료") + + # 2. 모델 다운로드 + print("\n📥 2/2: KoBART 모델 다운로드 중...") + start_time = time.time() + + model = BartForConditionalGeneration.from_pretrained( + model_name, + cache_dir=str(model_dir), + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + + download_time = time.time() - start_time + print(f"✅ 모델 다운로드 완료 ({download_time/60:.1f}분 소요)") + + return model, tokenizer + + except Exception as e: + print(f"❌ 다운로드 실패: {e}") + return None, None + +def test_kobart_model(model, tokenizer): + print("\n🧪 KoBART 요약 모델 테스트...") + + # Apple Silicon 최적화 + if torch.backends.mps.is_available(): + device = torch.device("mps") + model = model.to(device) + print("🚀 Apple Silicon MPS 가속 사용") + else: + device = torch.device("cpu") + print("💻 CPU 모드 사용") + + def summarize_text(text, max_length=150, min_length=30): + print(f"\n📝 요약 테스트:") + print(f"원문 ({len(text)}자):") + print(f"{text}") + + start_time = time.time() + + # 텍스트 토큰화 + inputs = tokenizer( + text, + return_tensors="pt", + max_length=1024, + truncation=True, + padding=True + ).to(device) + + # 요약 생성 + with torch.no_grad(): + summary_ids = model.generate( + **inputs, + max_length=max_length, + min_length=min_length, + num_beams=4, + early_stopping=True, + no_repeat_ngram_size=2, + length_penalty=2.0 + ) + + # 결과 디코딩 + summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True) + + process_time = time.time() - start_time + print(f"\n요약 ({len(summary)}자):") + print(f"{summary}") + print(f"소요 시간: {process_time:.2f}초") + print(f"압축률: {len(summary)/len(text)*100:.1f}%") + + return summary, process_time + + try: + # 테스트 케이스 1: 기술 문서 + tech_text = """ + 인공지능 기술이 급속히 발전하면서 우리의 일상생활과 업무 환경에 큰 변화를 가져오고 있습니다. + 특히 자연어 처리 분야에서는 번역, 요약, 질의응답 등의 기술이 크게 향상되어 실용적인 수준에 + 도달했습니다. 기계학습 알고리즘은 대량의 데이터를 학습하여 패턴을 파악하고, 이를 바탕으로 + 새로운 입력에 대해 예측이나 분류를 수행합니다. 딥러닝 기술의 발전으로 이미지 인식, 음성 인식, + 자연어 이해 등의 성능이 인간 수준에 근접하거나 이를 넘어서는 경우도 생겨나고 있습니다. + 이러한 기술들은 의료, 금융, 교육, 엔터테인먼트 등 다양한 분야에서 활용되고 있으며, + 앞으로도 더 많은 혁신을 가져올 것으로 예상됩니다. + """ + + summarize_text(tech_text.strip()) + + # 테스트 케이스 2: 뉴스 스타일 + news_text = """ + 최근 발표된 연구에 따르면 인공지능을 활용한 번역 시스템의 정확도가 크게 향상되었다고 합니다. + 특히 한국어와 영어, 일본어 간의 번역에서 기존 시스템 대비 20% 이상의 성능 개선을 보였습니다. + 연구팀은 대규모 언어 모델과 특화된 번역 모델을 결합하여 문맥을 더 정확히 이해하고 + 자연스러운 번역을 생성할 수 있게 되었다고 설명했습니다. 이번 기술은 개인 사용자뿐만 아니라 + 기업의 글로벌 비즈니스에도 큰 도움이 될 것으로 기대됩니다. + """ + + summarize_text(news_text.strip()) + + print(f"\n✅ KoBART 요약 모델 테스트 완료!") + return True + + except Exception as e: + print(f"❌ 요약 테스트 실패: {e}") + return False + +def check_total_model_size(): + """전체 모델 크기 확인""" + model_dir = Path("models") + if model_dir.exists(): + import subprocess + try: + result = subprocess.run( + ["du", "-sh", str(model_dir)], + capture_output=True, + text=True + ) + if result.returncode == 0: + size = result.stdout.strip().split()[0] + print(f"📊 전체 모델 크기: {size}") + except: + print("📊 모델 크기 확인 불가") + +if __name__ == "__main__": + print("🚀 KoBART 한국어 요약 모델 설치") + print("="*50) + + # 기존 모델 확인 + model_dir = Path("models/kobart-news") + if model_dir.exists() and any(model_dir.iterdir()): + print("✅ 기존 KoBART 모델 발견, 로딩 시도...") + try: + tokenizer = AutoTokenizer.from_pretrained(str(model_dir)) + model = BartForConditionalGeneration.from_pretrained( + str(model_dir), + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + print("✅ 기존 모델 로딩 완료") + except: + print("⚠️ 기존 모델 손상, 재다운로드 필요") + model, tokenizer = download_kobart_model() + else: + model, tokenizer = download_kobart_model() + + if model is not None and tokenizer is not None: + print("\n🧪 요약 모델 테스트 시작...") + if test_kobart_model(model, tokenizer): + check_total_model_size() + print("\n🎉 KoBART 요약 모델 설치 및 테스트 완료!") + print("📝 다음 단계: 통합 번역 시스템 구축") + else: + print("\n❌ 요약 모델 테스트 실패") + else: + print("\n❌ KoBART 모델 다운로드 실패") diff --git a/src/download_korean_summarizer.py b/src/download_korean_summarizer.py new file mode 100755 index 0000000..9d1cb21 --- /dev/null +++ b/src/download_korean_summarizer.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +""" +안정적인 한국어 요약 모델 +""" + +import torch +import time +from transformers import PreTrainedTokenizerFast, BartForConditionalGeneration +from pathlib import Path + +def download_korean_summarizer(): + print("🔄 한국어 요약 모델 다운로드 (대안)") + + # 더 안정적인 모델 사용 + model_name = "gogamza/kobart-summarization" + + model_dir = Path("models/korean-summarizer") + model_dir.mkdir(parents=True, exist_ok=True) + + try: + print("📥 토크나이저 다운로드 중...") + tokenizer = PreTrainedTokenizerFast.from_pretrained( + model_name, + cache_dir=str(model_dir) + ) + print("✅ 토크나이저 다운로드 완료") + + print("📥 모델 다운로드 중...") + start_time = time.time() + + model = BartForConditionalGeneration.from_pretrained( + model_name, + cache_dir=str(model_dir), + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + + download_time = time.time() - start_time + print(f"✅ 모델 다운로드 완료 ({download_time/60:.1f}분 소요)") + + return model, tokenizer + + except Exception as e: + print(f"❌ 다운로드 실패: {e}") + + # 마지막 대안: 간단한 T5 모델 + print("\n🔄 더 간단한 모델로 재시도...") + try: + from transformers import T5Tokenizer, T5ForConditionalGeneration + + alt_model_name = "t5-small" + print(f"📥 {alt_model_name} 다운로드 중...") + + tokenizer = T5Tokenizer.from_pretrained(alt_model_name) + model = T5ForConditionalGeneration.from_pretrained( + alt_model_name, + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + + print("✅ T5 백업 모델 다운로드 완료") + return model, tokenizer + + except Exception as e2: + print(f"❌ 백업 모델도 실패: {e2}") + return None, None + +def test_summarizer(model, tokenizer, model_type="kobart"): + print(f"\n🧪 {model_type} 요약 모델 테스트...") + + # Apple Silicon 최적화 + if torch.backends.mps.is_available(): + device = torch.device("mps") + model = model.to(device) + print("🚀 Apple Silicon MPS 가속 사용") + else: + device = torch.device("cpu") + print("💻 CPU 모드 사용") + + def summarize_korean_text(text): + print(f"\n📝 한국어 요약 테스트:") + print(f"원문 ({len(text)}자):") + print(f"{text[:200]}...") + + start_time = time.time() + + if model_type == "t5": + # T5 모델용 프롬프트 + input_text = f"summarize: {text}" + else: + # KoBART 모델용 + input_text = text + + # 텍스트 토큰화 + inputs = tokenizer( + input_text, + return_tensors="pt", + max_length=1024, + truncation=True, + padding=True + ).to(device) + + # 요약 생성 + with torch.no_grad(): + summary_ids = model.generate( + **inputs, + max_length=150, + min_length=30, + num_beams=4, + early_stopping=True, + no_repeat_ngram_size=2, + length_penalty=1.5 + ) + + # 결과 디코딩 + summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True) + + process_time = time.time() - start_time + print(f"\n📋 요약 결과 ({len(summary)}자):") + print(f"{summary}") + print(f"⏱️ 처리 시간: {process_time:.2f}초") + print(f"📊 압축률: {len(summary)/len(text)*100:.1f}%") + + return summary + + try: + # 한국어 테스트 텍스트 + korean_text = """ + 인공지능과 기계학습 기술이 급속도로 발전하면서 우리의 일상생활과 업무 환경에 + 혁신적인 변화를 가져오고 있습니다. 특히 자연어 처리 분야에서는 번역, 요약, + 대화형 AI 등의 기술이 실용적인 수준에 도달하여 다양한 서비스에 적용되고 있습니다. + 기계학습 알고리즘은 대량의 텍스트 데이터를 학습하여 언어의 패턴과 의미를 이해하고, + 이를 바탕으로 인간과 유사한 수준의 언어 처리 능력을 보여주고 있습니다. + 딥러닝 기술의 발전으로 번역의 정확도가 크게 향상되었으며, 실시간 번역 서비스도 + 일상적으로 사용할 수 있게 되었습니다. 앞으로 이러한 기술들은 교육, 의료, + 비즈니스 등 더 많은 분야에서 활용될 것으로 예상되며, 언어 장벽을 허물어 + 글로벌 소통을 더욱 원활하게 만들 것입니다. + """ + + summarize_korean_text(korean_text.strip()) + + print(f"\n✅ 요약 모델 테스트 완료!") + return True + + except Exception as e: + print(f"❌ 요약 테스트 실패: {e}") + return False + +if __name__ == "__main__": + print("🚀 한국어 요약 모델 설치 (대안)") + print("="*50) + + model, tokenizer = download_korean_summarizer() + + if model is not None and tokenizer is not None: + # 모델 타입 판단 + model_type = "t5" if "t5" in str(type(model)).lower() else "kobart" + + print(f"\n🧪 {model_type} 모델 테스트 시작...") + if test_summarizer(model, tokenizer, model_type): + print("\n🎉 한국어 요약 모델 설치 및 테스트 완료!") + print("📝 다음 단계: 통합 번역 시스템 구축") + else: + print("\n❌ 요약 모델 테스트 실패") + else: + print("\n❌ 모든 요약 모델 다운로드 실패") + print("📝 요약 기능 없이 번역만으로 진행 가능") diff --git a/src/download_nllb.py b/src/download_nllb.py new file mode 100755 index 0000000..d670f95 --- /dev/null +++ b/src/download_nllb.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +""" +NLLB-200-3.3B 모델 다운로드 및 초기 테스트 +""" + +import os +import time +import torch +from transformers import AutoTokenizer, AutoModelForSeq2SeqLM +from pathlib import Path + +def download_nllb_model(): + print("🔄 NLLB-200-3.3B 모델 다운로드 시작...") + print("⚠️ 모델 크기: ~7GB, 다운로드 시간 10-15분 예상") + + model_name = "facebook/nllb-200-3.3B" + + # 로컬 모델 저장 경로 + model_dir = Path("models/nllb-200-3.3B") + model_dir.mkdir(parents=True, exist_ok=True) + + try: + # 1. 토크나이저 다운로드 + print("\n📥 1/2: 토크나이저 다운로드 중...") + tokenizer = AutoTokenizer.from_pretrained( + model_name, + cache_dir=str(model_dir), + local_files_only=False + ) + print("✅ 토크나이저 다운로드 완료") + + # 2. 모델 다운로드 (시간이 오래 걸림) + print("\n📥 2/2: 모델 다운로드 중...") + print("⏳ 진행률을 확인하려면 별도 터미널에서 다음 명령어 실행:") + print(f" du -sh {model_dir}/models--facebook--nllb-200-3.3B") + + start_time = time.time() + + model = AutoModelForSeq2SeqLM.from_pretrained( + model_name, + cache_dir=str(model_dir), + local_files_only=False, + torch_dtype=torch.float16, # 메모리 절약 + low_cpu_mem_usage=True + ) + + download_time = time.time() - start_time + print(f"✅ 모델 다운로드 완료 ({download_time/60:.1f}분 소요)") + + return model, tokenizer + + except Exception as e: + print(f"❌ 다운로드 실패: {e}") + return None, None + +def test_nllb_model(model, tokenizer): + print("\n🧪 NLLB 모델 기본 테스트...") + + # Apple Silicon 최적화 + if torch.backends.mps.is_available(): + device = torch.device("mps") + print("🚀 Apple Silicon MPS 가속 사용") + else: + device = torch.device("cpu") + print("💻 CPU 모드 사용") + + try: + model = model.to(device) + + # NLLB 언어 코드 + lang_codes = { + "eng_Latn": "English", + "jpn_Jpan": "Japanese", + "kor_Hang": "Korean" + } + + def translate_test(text, src_lang, tgt_lang, desc): + print(f"\n📝 {desc} 테스트:") + print(f"원문: {text}") + + tokenizer.src_lang = src_lang + encoded = tokenizer(text, return_tensors="pt", padding=True).to(device) + + start_time = time.time() + + generated_tokens = model.generate( + **encoded, + forced_bos_token_id=tokenizer.lang_code_to_id[tgt_lang], + max_length=200, + num_beams=4, + early_stopping=True, + do_sample=False + ) + + translation_time = time.time() - start_time + result = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0] + + print(f"번역: {result}") + print(f"소요 시간: {translation_time:.2f}초") + + return result, translation_time + + # 테스트 케이스들 + print("\n" + "="*50) + + # 1. 영어 → 한국어 + en_text = "Artificial intelligence is transforming the way we work and live." + translate_test(en_text, "eng_Latn", "kor_Hang", "영어 → 한국어") + + # 2. 일본어 → 한국어 + ja_text = "人工知能は私たちの働き方と生活を変革しています。" + translate_test(ja_text, "jpn_Jpan", "kor_Hang", "일본어 → 한국어") + + # 3. 기술 문서 스타일 + tech_text = "Machine learning algorithms require large datasets for training." + translate_test(tech_text, "eng_Latn", "kor_Hang", "기술 문서") + + print("\n✅ 모든 테스트 완료!") + return True + + except Exception as e: + print(f"❌ 테스트 실패: {e}") + return False + +def check_model_size(): + """다운로드된 모델 크기 확인""" + model_dir = Path("models") + if model_dir.exists(): + import subprocess + try: + result = subprocess.run( + ["du", "-sh", str(model_dir)], + capture_output=True, + text=True + ) + if result.returncode == 0: + size = result.stdout.strip().split()[0] + print(f"📊 다운로드된 모델 크기: {size}") + except: + print("📊 모델 크기 확인 불가") + +if __name__ == "__main__": + print("🚀 NLLB-200-3.3B 모델 설치") + print("="*50) + + # 기존 모델 확인 + model_dir = Path("models/nllb-200-3.3B") + if model_dir.exists() and any(model_dir.iterdir()): + print("✅ 기존 모델 발견, 로딩 시도...") + try: + tokenizer = AutoTokenizer.from_pretrained(str(model_dir)) + model = AutoModelForSeq2SeqLM.from_pretrained( + str(model_dir), + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + print("✅ 기존 모델 로딩 완료") + except: + print("⚠️ 기존 모델 손상, 재다운로드 필요") + model, tokenizer = download_nllb_model() + else: + model, tokenizer = download_nllb_model() + + if model is not None and tokenizer is not None: + print("\n🧪 모델 테스트 시작...") + if test_nllb_model(model, tokenizer): + check_model_size() + print("\n🎉 NLLB 모델 설치 및 테스트 완료!") + print("📝 다음 단계: KoBART 요약 모델 설치") + else: + print("\n❌ 모델 테스트 실패") + else: + print("\n❌ 모델 다운로드 실패") + print("네트워크 연결을 확인하고 다시 시도해주세요.") diff --git a/src/fastapi_final.py b/src/fastapi_final.py new file mode 100644 index 0000000..15ecce1 --- /dev/null +++ b/src/fastapi_final.py @@ -0,0 +1,462 @@ +#!/usr/bin/env python3 +""" +Mac Mini (192.168.1.122) FastAPI + DS1525+ (192.168.1.227) 연동 +최종 운영 버전 +""" + +from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks, Request +from fastapi.templating import Jinja2Templates +from fastapi.staticfiles import StaticFiles +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +import asyncio +import json +import uuid +from pathlib import Path +from typing import Dict, List, Optional +import time +import shutil +from dataclasses import dataclass, asdict +import aiofiles +from datetime import datetime +import subprocess + +# 실제 네트워크 설정 +MAC_MINI_IP = "192.168.1.122" +NAS_IP = "192.168.1.227" + +# NAS 마운트 경로 (Finder에서 연결 시 생성되는 경로) +NAS_MOUNT_POINT = Path("/Volumes/DS1525+") +DOCUMENT_UPLOAD_BASE = NAS_MOUNT_POINT / "Document-upload" + +# 세부 경로들 +NAS_ORIGINALS_PATH = DOCUMENT_UPLOAD_BASE / "originals" +NAS_TRANSLATED_PATH = DOCUMENT_UPLOAD_BASE / "translated" +NAS_STATIC_HOSTING_PATH = DOCUMENT_UPLOAD_BASE / "static-hosting" +NAS_METADATA_PATH = DOCUMENT_UPLOAD_BASE / "metadata" + +# 로컬 작업 디렉토리 +LOCAL_WORK_PATH = Path.home() / "Scripts" / "nllb-translation-system" + +app = FastAPI( + title="AI 번역 시스템", + description=f"Mac Mini ({MAC_MINI_IP}) + DS1525+ ({NAS_IP}) 연동", + version="1.0.0" +) + +# 정적 파일 및 템플릿 +app.mount("/static", StaticFiles(directory=str(LOCAL_WORK_PATH / "static")), name="static") +templates = Jinja2Templates(directory=str(LOCAL_WORK_PATH / "templates")) + +# 작업 상태 관리 +processing_jobs: Dict[str, Dict] = {} + +def check_nas_connection(): + """NAS 연결 상태 실시간 확인""" + try: + # 1. ping 테스트 + ping_result = subprocess.run( + ["ping", "-c", "1", "-W", "3000", NAS_IP], + capture_output=True, + timeout=5 + ) + + if ping_result.returncode != 0: + return {"status": "offline", "error": f"NAS {NAS_IP} ping 실패"} + + # 2. 마운트 상태 확인 + if not NAS_MOUNT_POINT.exists(): + return {"status": "not_mounted", "error": "마운트 포인트 없음"} + + # 3. 실제 마운트 확인 + mount_result = subprocess.run( + ["mount"], + capture_output=True, + text=True + ) + + if str(NAS_MOUNT_POINT) not in mount_result.stdout: + return {"status": "not_mounted", "error": "NAS가 마운트되지 않음"} + + # 4. Document-upload 폴더 확인 + if not DOCUMENT_UPLOAD_BASE.exists(): + return {"status": "folder_missing", "error": "Document-upload 폴더 없음"} + + return {"status": "connected", "mount_point": str(NAS_MOUNT_POINT)} + + except subprocess.TimeoutExpired: + return {"status": "timeout", "error": "연결 타임아웃"} + except Exception as e: + return {"status": "error", "error": str(e)} + +@app.on_event("startup") +async def startup_event(): + """서버 시작시 전체 시스템 상태 확인""" + print(f"🚀 Mac Mini AI 번역 서버 시작") + print(f"📍 Mac Mini IP: {MAC_MINI_IP}") + print(f"📍 NAS IP: {NAS_IP}") + print("-" * 50) + + # NAS 연결 상태 확인 + nas_status = check_nas_connection() + + if nas_status["status"] == "connected": + print(f"✅ NAS 연결 정상: {nas_status['mount_point']}") + + # 필요한 폴더 구조 자동 생성 + try: + current_month = datetime.now().strftime("%Y-%m") + + folders_to_create = [ + NAS_ORIGINALS_PATH / current_month / "pdfs", + NAS_ORIGINALS_PATH / current_month / "docs", + NAS_ORIGINALS_PATH / current_month / "txts", + NAS_TRANSLATED_PATH / current_month / "english-to-korean", + NAS_TRANSLATED_PATH / current_month / "japanese-to-korean", + NAS_TRANSLATED_PATH / current_month / "korean-only", + NAS_STATIC_HOSTING_PATH / "docs", + NAS_STATIC_HOSTING_PATH / "assets", + NAS_METADATA_PATH / "processing-logs" + ] + + for folder in folders_to_create: + folder.mkdir(parents=True, exist_ok=True) + + print(f"✅ 폴더 구조 확인/생성 완료") + + except Exception as e: + print(f"⚠️ 폴더 생성 실패: {e}") + + else: + print(f"❌ NAS 연결 실패: {nas_status['error']}") + print("해결 방법:") + print("1. NAS 전원 및 네트워크 상태 확인") + print("2. Finder에서 DS1525+ 수동 연결:") + print(f" - 이동 → 서버에 연결 → smb://{NAS_IP}") + print("3. 연결 후 서버 재시작") + +@app.get("/", response_class=HTMLResponse) +async def index(request: Request): + """메인 페이지""" + nas_status = check_nas_connection() + return templates.TemplateResponse("index.html", { + "request": request, + "nas_status": nas_status + }) + +@app.get("/system-status") +async def system_status(): + """전체 시스템 상태 확인""" + + # NAS 상태 + nas_status = check_nas_connection() + + # Mac Mini 상태 + mac_status = { + "ip": MAC_MINI_IP, + "hostname": subprocess.getoutput("hostname"), + "uptime": subprocess.getoutput("uptime"), + "disk_usage": {} + } + + # 디스크 사용량 확인 + try: + disk_result = subprocess.run( + ["df", "-h", str(Path.home())], + capture_output=True, + text=True + ) + if disk_result.returncode == 0: + lines = disk_result.stdout.strip().split('\n') + if len(lines) > 1: + parts = lines[1].split() + mac_status["disk_usage"] = { + "total": parts[1], + "used": parts[2], + "available": parts[3], + "percentage": parts[4] + } + except: + pass + + # AI 모델 상태 (간단 체크) + ai_status = { + "models_available": False, + "memory_usage": "Unknown" + } + + try: + # NLLB 모델 폴더 확인 + model_path = LOCAL_WORK_PATH / "models" + if model_path.exists(): + ai_status["models_available"] = True + except: + pass + + # 작업 통계 + job_stats = { + "total_jobs": len(processing_jobs), + "completed": len([j for j in processing_jobs.values() if j["status"] == "completed"]), + "processing": len([j for j in processing_jobs.values() if j["status"] == "processing"]), + "failed": len([j for j in processing_jobs.values() if j["status"] == "error"]) + } + + return { + "timestamp": datetime.now().isoformat(), + "nas": nas_status, + "mac_mini": mac_status, + "ai_models": ai_status, + "jobs": job_stats + } + +@app.post("/upload") +async def upload_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)): + """파일 업로드 (NAS 연결 상태 확인 포함)""" + + # NAS 연결 상태 먼저 확인 + nas_status = check_nas_connection() + if nas_status["status"] != "connected": + raise HTTPException( + status_code=503, + detail=f"NAS 연결 실패: {nas_status['error']}. Finder에서 DS1525+를 연결해주세요." + ) + + # 파일 검증 + if not file.filename: + raise HTTPException(status_code=400, detail="파일이 선택되지 않았습니다.") + + allowed_extensions = {'.pdf', '.txt', '.docx', '.doc'} + file_ext = Path(file.filename).suffix.lower() + + if file_ext not in allowed_extensions: + raise HTTPException( + status_code=400, + detail=f"지원되지 않는 파일 형식입니다. 지원 형식: {', '.join(allowed_extensions)}" + ) + + # 파일 크기 확인 (100MB 제한) + content = await file.read() + if len(content) > 100 * 1024 * 1024: + raise HTTPException(status_code=413, detail="파일 크기는 100MB를 초과할 수 없습니다.") + + # 고유 작업 ID 생성 + job_id = str(uuid.uuid4()) + + # NAS에 체계적으로 저장 + current_month = datetime.now().strftime("%Y-%m") + file_type_folder = { + '.pdf': 'pdfs', + '.doc': 'docs', + '.docx': 'docs', + '.txt': 'txts' + }.get(file_ext, 'others') + + nas_original_dir = NAS_ORIGINALS_PATH / current_month / file_type_folder + nas_original_dir.mkdir(parents=True, exist_ok=True) + + # 안전한 파일명 생성 + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + safe_filename = f"{timestamp}_{job_id[:8]}_{file.filename}".replace(" ", "_") + nas_file_path = nas_original_dir / safe_filename + + # 파일 저장 + try: + async with aiofiles.open(nas_file_path, 'wb') as f: + await f.write(content) + + print(f"📁 파일 저장 완료: {nas_file_path}") + + except Exception as e: + raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") + + # 작업 상태 초기화 + job = { + "id": job_id, + "filename": file.filename, + "file_type": file_ext, + "status": "uploaded", + "progress": 0, + "message": f"파일 업로드 완료, NAS에 저장됨 ({nas_file_path.name})", + "nas_original_path": str(nas_file_path), + "created_at": time.time() + } + + processing_jobs[job_id] = job + + # 백그라운드에서 처리 시작 + background_tasks.add_task(process_document_with_nas_final, job_id, nas_file_path) + + return { + "job_id": job_id, + "message": "파일 업로드 완료, AI 처리를 시작합니다.", + "nas_path": str(nas_file_path) + } + +async def process_document_with_nas_final(job_id: str, nas_file_path: Path): + """최종 문서 처리 파이프라인""" + + try: + processing_jobs[job_id].update({ + "status": "processing", + "progress": 10, + "message": "AI 모델 로딩 및 텍스트 추출 중..." + }) + + # 1. 통합 번역 시스템 로드 + import sys + sys.path.append(str(LOCAL_WORK_PATH / "src")) + + from integrated_translation_system import IntegratedTranslationSystem + + system = IntegratedTranslationSystem() + if not system.load_models(): + raise Exception("AI 모델 로드 실패") + + processing_jobs[job_id].update({ + "progress": 30, + "message": "문서 분석 및 언어 감지 중..." + }) + + # 2. 문서 처리 + result = system.process_document(str(nas_file_path)) + detected_lang = result.metadata["detected_language"] + + processing_jobs[job_id].update({ + "progress": 60, + "message": f"언어 감지: {detected_lang}, 다국어 HTML 생성 중..." + }) + + # 3. HTML 생성 + from html_generator import MultilingualHTMLGenerator + + generator = MultilingualHTMLGenerator() + + # 언어별 콘텐츠 준비 + contents = {} + translation_folder = "" + + if detected_lang == "korean": + contents["korean"] = result.original_text + translation_folder = "korean-only" + elif detected_lang == "english": + contents["english"] = result.original_text + contents["korean"] = result.translated_text + translation_folder = "english-to-korean" + elif detected_lang == "japanese": + contents["japanese"] = result.original_text + contents["korean"] = result.translated_text + translation_folder = "japanese-to-korean" + else: + contents[detected_lang] = result.original_text + contents["korean"] = result.translated_text + translation_folder = "unknown-language" + + # NAS 저장 경로 결정 + current_month = datetime.now().strftime("%Y-%m") + nas_translated_dir = NAS_TRANSLATED_PATH / current_month / translation_folder + nas_translated_dir.mkdir(parents=True, exist_ok=True) + + # 원본 파일명에서 HTML 파일명 생성 + base_name = Path(nas_file_path.stem).stem + # 타임스탬프와 job_id 제거 + if '_' in base_name: + parts = base_name.split('_')[2:] # 처음 2개 부분 제거 + if parts: + base_name = '_'.join(parts) + + html_filename = f"{base_name}.html" + nas_html_path = nas_translated_dir / html_filename + + # 중복 파일명 처리 + counter = 1 + while nas_html_path.exists(): + html_filename = f"{base_name}_{counter}.html" + nas_html_path = nas_translated_dir / html_filename + counter += 1 + + # HTML 생성 + generator.generate_multilingual_html( + title=base_name, + contents=contents, + summary=result.summary, + metadata={ + **result.metadata, + "nas_original_path": str(nas_file_path), + "translation_type": translation_folder, + "nas_ip": NAS_IP, + "mac_mini_ip": MAC_MINI_IP + }, + output_path=str(nas_html_path) + ) + + processing_jobs[job_id].update({ + "progress": 85, + "message": "정적 호스팅 준비 및 인덱스 업데이트 중..." + }) + + # 4. 정적 호스팅 폴더에 복사 + static_hosting_file = NAS_STATIC_HOSTING_PATH / "docs" / html_filename + shutil.copy2(nas_html_path, static_hosting_file) + + # 5. 메타데이터 저장 + await save_processing_metadata(job_id, nas_file_path, nas_html_path, result.metadata) + + processing_jobs[job_id].update({ + "progress": 100, + "status": "completed", + "message": f"처리 완료! ({translation_folder})", + "nas_translated_path": str(nas_html_path), + "static_hosting_path": str(static_hosting_file), + "detected_language": detected_lang, + "translation_type": translation_folder + }) + + print(f"✅ 작업 완료: {job_id} - {html_filename}") + + except Exception as e: + processing_jobs[job_id].update({ + "status": "error", + "progress": 0, + "message": f"처리 실패: {str(e)}" + }) + + print(f"❌ 작업 실패: {job_id} - {str(e)}") + +async def save_processing_metadata(job_id: str, original_path: Path, html_path: Path, metadata: Dict): + """처리 메타데이터 NAS에 저장""" + + log_data = { + "job_id": job_id, + "timestamp": datetime.now().isoformat(), + "original_file": str(original_path), + "html_file": str(html_path), + "metadata": metadata, + "nas_ip": NAS_IP, + "mac_mini_ip": MAC_MINI_IP + } + + # 월별 로그 파일에 추가 + current_month = datetime.now().strftime("%Y-%m") + log_file = NAS_METADATA_PATH / "processing-logs" / f"{current_month}.jsonl" + + try: + async with aiofiles.open(log_file, 'a', encoding='utf-8') as f: + await f.write(json.dumps(log_data, ensure_ascii=False) + '\n') + except Exception as e: + print(f"메타데이터 저장 실패: {e}") + +@app.get("/status/{job_id}") +async def get_job_status(job_id: str): + """작업 상태 조회""" + if job_id not in processing_jobs: + raise HTTPException(status_code=404, detail="작업을 찾을 수 없습니다.") + + return processing_jobs[job_id] + +if __name__ == "__main__": + import uvicorn + + print(f"🚀 Mac Mini AI 번역 서버") + print(f"📡 서버 주소: http://{MAC_MINI_IP}:8080") + print(f"📁 NAS 주소: {NAS_IP}") + + uvicorn.run(app, host="0.0.0.0", port=8080) diff --git a/src/fastapi_media_mount.py b/src/fastapi_media_mount.py new file mode 100644 index 0000000..1be3fe4 --- /dev/null +++ b/src/fastapi_media_mount.py @@ -0,0 +1,575 @@ +#!/usr/bin/env python3 +""" +/Volumes/Media 마운트 기반 FastAPI +최종 운영 버전 +""" + +from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks, Request +from fastapi.templating import Jinja2Templates +from fastapi.staticfiles import StaticFiles +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +import asyncio +import json +import uuid +from pathlib import Path +from typing import Dict, List, Optional +import time +import shutil +from dataclasses import dataclass, asdict +import aiofiles +from datetime import datetime +import subprocess + +# 실제 네트워크 설정 +MAC_MINI_IP = "192.168.1.122" +NAS_IP = "192.168.1.227" + +# 기존 연결된 Media 마운트 사용 +NAS_MOUNT_POINT = Path("/Volumes/Media") +DOCUMENT_UPLOAD_BASE = NAS_MOUNT_POINT / "Document-upload" + +# 세부 경로들 +NAS_ORIGINALS_PATH = DOCUMENT_UPLOAD_BASE / "originals" +NAS_TRANSLATED_PATH = DOCUMENT_UPLOAD_BASE / "translated" +NAS_STATIC_HOSTING_PATH = DOCUMENT_UPLOAD_BASE / "static-hosting" +NAS_METADATA_PATH = DOCUMENT_UPLOAD_BASE / "metadata" + +# 로컬 작업 디렉토리 +LOCAL_WORK_PATH = Path.home() / "Scripts" / "nllb-translation-system" + +app = FastAPI( + title="AI 번역 시스템", + description=f"Mac Mini ({MAC_MINI_IP}) + Media Mount 연동", + version="1.0.0" +) + +# 정적 파일 및 템플릿 (있는 경우에만) +if (LOCAL_WORK_PATH / "static").exists(): + app.mount("/static", StaticFiles(directory=str(LOCAL_WORK_PATH / "static")), name="static") + +if (LOCAL_WORK_PATH / "templates").exists(): + templates = Jinja2Templates(directory=str(LOCAL_WORK_PATH / "templates")) +else: + templates = None + +# 작업 상태 관리 +processing_jobs: Dict[str, Dict] = {} + +def check_nas_connection(): + """NAS 연결 상태 확인""" + try: + # 1. Media 마운트 확인 + if not NAS_MOUNT_POINT.exists(): + return {"status": "not_mounted", "error": "Media 마운트 포인트 없음"} + + # 2. 쓰기 권한 확인 + try: + test_file = NAS_MOUNT_POINT / ".test_write" + test_file.touch() + test_file.unlink() + except: + return {"status": "read_only", "error": "Media 마운트 읽기 전용"} + + # 3. Document-upload 폴더 확인 + if not DOCUMENT_UPLOAD_BASE.exists(): + return {"status": "folder_missing", "error": "Document-upload 폴더 없음"} + + return {"status": "connected", "mount_point": str(NAS_MOUNT_POINT)} + + except Exception as e: + return {"status": "error", "error": str(e)} + +def ensure_nas_directories(): + """NAS 디렉토리 구조 생성""" + try: + # Document-upload 기본 구조 + DOCUMENT_UPLOAD_BASE.mkdir(exist_ok=True) + + current_month = datetime.now().strftime("%Y-%m") + + folders_to_create = [ + NAS_ORIGINALS_PATH / current_month / "pdfs", + NAS_ORIGINALS_PATH / current_month / "docs", + NAS_ORIGINALS_PATH / current_month / "txts", + NAS_TRANSLATED_PATH / current_month / "english-to-korean", + NAS_TRANSLATED_PATH / current_month / "japanese-to-korean", + NAS_TRANSLATED_PATH / current_month / "korean-only", + NAS_STATIC_HOSTING_PATH / "docs", + NAS_STATIC_HOSTING_PATH / "assets", + NAS_STATIC_HOSTING_PATH / "index", + NAS_METADATA_PATH / "processing-logs" + ] + + for folder in folders_to_create: + folder.mkdir(parents=True, exist_ok=True) + + # README 파일 생성 + readme_content = f"""# AI 번역 시스템 문서 저장소 + +자동 생성 시간: {datetime.now().isoformat()} +Mac Mini IP: {MAC_MINI_IP} +NAS IP: {NAS_IP} + +## 폴더 구조 + +- originals/: 업로드된 원본 파일들 +- translated/: 번역된 HTML 파일들 +- static-hosting/: 웹 호스팅용 파일들 +- metadata/: 처리 로그 및 메타데이터 + +## 자동 관리 + +이 폴더는 AI 번역 시스템에서 자동으로 관리됩니다. +수동으로 파일을 수정하지 마세요. +""" + + readme_path = DOCUMENT_UPLOAD_BASE / "README.md" + with open(readme_path, 'w', encoding='utf-8') as f: + f.write(readme_content) + + return True + + except Exception as e: + print(f"❌ 디렉토리 생성 실패: {e}") + return False + +@app.on_event("startup") +async def startup_event(): + """서버 시작시 전체 시스템 상태 확인""" + print(f"🚀 Mac Mini AI 번역 서버 시작") + print(f"📍 Mac Mini IP: {MAC_MINI_IP}") + print(f"📍 NAS Mount: {NAS_MOUNT_POINT}") + print("-" * 50) + + # NAS 연결 상태 확인 + nas_status = check_nas_connection() + + if nas_status["status"] == "connected": + print(f"✅ NAS 연결 정상: {nas_status['mount_point']}") + + if ensure_nas_directories(): + print(f"✅ 폴더 구조 확인/생성 완료") + print(f"📁 문서 저장 경로: {DOCUMENT_UPLOAD_BASE}") + else: + print(f"⚠️ 폴더 생성 실패") + else: + print(f"❌ NAS 연결 실패: {nas_status['error']}") + +@app.get("/") +async def index(request: Request = None): + """메인 페이지""" + nas_status = check_nas_connection() + + if templates: + return templates.TemplateResponse("index.html", { + "request": request, + "nas_status": nas_status + }) + else: + # 템플릿이 없으면 간단한 HTML 반환 + html_content = f""" + + + + + AI 번역 시스템 + + + +
+

🤖 AI 번역 시스템

+ +
+ {'✅ NAS 연결됨: ' + nas_status.get('mount_point', '') if nas_status['status'] == 'connected' else '❌ NAS 연결 실패: ' + nas_status.get('error', '')} +
+ +
+

파일 업로드

+
+ +
+ +
+
+
+ +
+

API 엔드포인트

+ +
+
+ + + + + """ + return HTMLResponse(content=html_content) + +@app.get("/system-status") +async def system_status(): + """전체 시스템 상태""" + nas_status = check_nas_connection() + + # 파일 통계 + file_stats = {"originals": 0, "translated": 0, "total_size": 0} + + try: + if NAS_ORIGINALS_PATH.exists(): + original_files = list(NAS_ORIGINALS_PATH.rglob("*.*")) + file_stats["originals"] = len([f for f in original_files if f.is_file()]) + + if NAS_TRANSLATED_PATH.exists(): + html_files = list(NAS_TRANSLATED_PATH.rglob("*.html")) + file_stats["translated"] = len(html_files) + file_stats["total_size"] = sum(f.stat().st_size for f in html_files if f.exists()) + except: + pass + + return { + "timestamp": datetime.now().isoformat(), + "mac_mini_ip": MAC_MINI_IP, + "nas_mount": str(NAS_MOUNT_POINT), + "nas_status": nas_status, + "file_stats": file_stats, + "active_jobs": len(processing_jobs), + "document_upload_path": str(DOCUMENT_UPLOAD_BASE) + } + +@app.post("/upload") +async def upload_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)): + """파일 업로드""" + + # NAS 연결 상태 먼저 확인 + nas_status = check_nas_connection() + if nas_status["status"] != "connected": + raise HTTPException( + status_code=503, + detail=f"NAS 연결 실패: {nas_status['error']}" + ) + + # 파일 검증 + if not file.filename: + raise HTTPException(status_code=400, detail="파일이 선택되지 않았습니다.") + + allowed_extensions = {'.pdf', '.txt', '.docx', '.doc'} + file_ext = Path(file.filename).suffix.lower() + + if file_ext not in allowed_extensions: + raise HTTPException( + status_code=400, + detail=f"지원되지 않는 파일 형식입니다. 지원: {', '.join(allowed_extensions)}" + ) + + # 파일 크기 확인 (100MB 제한) + content = await file.read() + if len(content) > 100 * 1024 * 1024: + raise HTTPException(status_code=413, detail="파일 크기는 100MB를 초과할 수 없습니다.") + + # 고유 작업 ID 생성 + job_id = str(uuid.uuid4()) + + # NAS에 저장 + current_month = datetime.now().strftime("%Y-%m") + file_type_folder = { + '.pdf': 'pdfs', + '.doc': 'docs', + '.docx': 'docs', + '.txt': 'txts' + }.get(file_ext, 'others') + + nas_original_dir = NAS_ORIGINALS_PATH / current_month / file_type_folder + nas_original_dir.mkdir(parents=True, exist_ok=True) + + # 안전한 파일명 생성 + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + safe_filename = f"{timestamp}_{job_id[:8]}_{file.filename}".replace(" ", "_") + nas_file_path = nas_original_dir / safe_filename + + # 파일 저장 + try: + async with aiofiles.open(nas_file_path, 'wb') as f: + await f.write(content) + + print(f"📁 파일 저장: {nas_file_path}") + + except Exception as e: + raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") + + # 작업 상태 초기화 + job = { + "id": job_id, + "filename": file.filename, + "file_type": file_ext, + "status": "uploaded", + "progress": 0, + "message": f"파일 저장 완료: {safe_filename}", + "nas_original_path": str(nas_file_path), + "created_at": time.time() + } + + processing_jobs[job_id] = job + + # 백그라운드 처리 시작 + background_tasks.add_task(process_document_simple, job_id, nas_file_path) + + return { + "job_id": job_id, + "message": "파일 업로드 완료, 처리를 시작합니다.", + "nas_path": str(nas_file_path) + } + +async def process_document_simple(job_id: str, nas_file_path: Path): + """간단한 문서 처리 (AI 모델 연동 전 테스트용)""" + + try: + processing_jobs[job_id].update({ + "status": "processing", + "progress": 30, + "message": "텍스트 추출 중..." + }) + + await asyncio.sleep(2) # 시뮬레이션 + + processing_jobs[job_id].update({ + "progress": 60, + "message": "언어 감지 및 번역 중..." + }) + + await asyncio.sleep(3) # 시뮬레이션 + + # 간단한 HTML 생성 (테스트용) + current_month = datetime.now().strftime("%Y-%m") + nas_translated_dir = NAS_TRANSLATED_PATH / current_month / "korean-only" + nas_translated_dir.mkdir(parents=True, exist_ok=True) + + base_name = Path(nas_file_path.stem).stem + if '_' in base_name: + parts = base_name.split('_')[2:] # 타임스탬프와 job_id 제거 + if parts: + base_name = '_'.join(parts) + + html_filename = f"{base_name}.html" + nas_html_path = nas_translated_dir / html_filename + + # 테스트용 HTML 생성 + html_content = f""" + + + + {base_name} + + + +
+
+

📄 {base_name}

+

처리 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

+

원본 파일: {nas_file_path.name}

+

Job ID: {job_id}

+
+ +
+

테스트 문서

+

이것은 AI 번역 시스템의 테스트 출력입니다.

+

실제 AI 모델이 연동되면 이 부분에 번역된 내용이 표시됩니다.

+ +

시스템 정보

+ +
+
+ +""" + + # HTML 파일 저장 + async with aiofiles.open(nas_html_path, 'w', encoding='utf-8') as f: + await f.write(html_content) + + processing_jobs[job_id].update({ + "progress": 100, + "status": "completed", + "message": "처리 완료!", + "nas_translated_path": str(nas_html_path) + }) + + print(f"✅ 테스트 처리 완료: {job_id} - {html_filename}") + + except Exception as e: + processing_jobs[job_id].update({ + "status": "error", + "progress": 0, + "message": f"처리 실패: {str(e)}" + }) + + print(f"❌ 처리 실패: {job_id} - {str(e)}") + +@app.get("/status/{job_id}") +async def get_job_status(job_id: str): + """작업 상태 조회""" + if job_id not in processing_jobs: + raise HTTPException(status_code=404, detail="작업을 찾을 수 없습니다.") + + return processing_jobs[job_id] + +@app.get("/download/{job_id}") +async def download_result(job_id: str): + """결과 파일 다운로드""" + if job_id not in processing_jobs: + raise HTTPException(status_code=404, detail="작업을 찾을 수 없습니다.") + + job = processing_jobs[job_id] + + if job["status"] != "completed" or not job.get("nas_translated_path"): + raise HTTPException(status_code=400, detail="처리가 완료되지 않았습니다.") + + result_path = Path(job["nas_translated_path"]) + if not result_path.exists(): + raise HTTPException(status_code=404, detail="결과 파일을 찾을 수 없습니다.") + + return FileResponse( + path=result_path, + filename=f"{Path(job['filename']).stem}.html", + media_type="text/html" + ) + +@app.get("/nas-info") +async def nas_info(): + """NAS 정보 및 통계""" + + nas_status = check_nas_connection() + + info = { + "nas_status": nas_status, + "mount_point": str(NAS_MOUNT_POINT), + "document_upload_base": str(DOCUMENT_UPLOAD_BASE), + "folders": {}, + "statistics": { + "total_jobs": len(processing_jobs), + "completed_jobs": len([j for j in processing_jobs.values() if j["status"] == "completed"]), + "failed_jobs": len([j for j in processing_jobs.values() if j["status"] == "error"]) + } + } + + # 폴더 정보 수집 + try: + for folder_name, folder_path in [ + ("originals", NAS_ORIGINALS_PATH), + ("translated", NAS_TRANSLATED_PATH), + ("static-hosting", NAS_STATIC_HOSTING_PATH), + ("metadata", NAS_METADATA_PATH) + ]: + if folder_path.exists(): + files = list(folder_path.rglob("*.*")) + file_count = len([f for f in files if f.is_file()]) + total_size = sum(f.stat().st_size for f in files if f.is_file()) + + info["folders"][folder_name] = { + "exists": True, + "file_count": file_count, + "total_size_mb": round(total_size / (1024 * 1024), 2) + } + else: + info["folders"][folder_name] = {"exists": False} + except Exception as e: + info["error"] = str(e) + + return info + +if __name__ == "__main__": + import uvicorn + + print(f"🚀 Mac Mini AI 번역 서버") + print(f"📡 서버 주소: http://{MAC_MINI_IP}:8080") + print(f"📁 NAS Mount: {NAS_MOUNT_POINT}") + + # 시작 전 연결 확인 + nas_status = check_nas_connection() + if nas_status["status"] == "connected": + print(f"✅ NAS 연결 확인됨") + ensure_nas_directories() + else: + print(f"❌ NAS 연결 문제: {nas_status['error']}") + + uvicorn.run(app, host="0.0.0.0", port=8080) diff --git a/src/fastapi_port_20080.py b/src/fastapi_port_20080.py new file mode 100644 index 0000000..6a0b618 --- /dev/null +++ b/src/fastapi_port_20080.py @@ -0,0 +1,848 @@ +#!/usr/bin/env python3 +""" +포트 20080으로 실행하는 AI 번역 시스템 +Mac Mini (192.168.1.122:20080) + Media Mount 연동 +""" + +from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks, Request +from fastapi.templating import Jinja2Templates +from fastapi.staticfiles import StaticFiles +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse +import asyncio +import json +import uuid +from pathlib import Path +from typing import Dict, List, Optional +import time +import shutil +from dataclasses import dataclass, asdict +import aiofiles +from datetime import datetime +import subprocess + +# 실제 네트워크 설정 (포트 20080) +MAC_MINI_IP = "192.168.1.122" +MAC_MINI_PORT = 20080 +NAS_IP = "192.168.1.227" + +# 기존 연결된 Media 마운트 사용 +NAS_MOUNT_POINT = Path("/Volumes/Media") +DOCUMENT_UPLOAD_BASE = NAS_MOUNT_POINT / "Document-upload" + +# 세부 경로들 +NAS_ORIGINALS_PATH = DOCUMENT_UPLOAD_BASE / "originals" +NAS_TRANSLATED_PATH = DOCUMENT_UPLOAD_BASE / "translated" +NAS_STATIC_HOSTING_PATH = DOCUMENT_UPLOAD_BASE / "static-hosting" +NAS_METADATA_PATH = DOCUMENT_UPLOAD_BASE / "metadata" + +# 로컬 작업 디렉토리 +LOCAL_WORK_PATH = Path.home() / "Scripts" / "nllb-translation-system" + +app = FastAPI( + title="AI 번역 시스템", + description=f"Mac Mini ({MAC_MINI_IP}:{MAC_MINI_PORT}) + Media Mount 연동", + version="1.0.0" +) + +# 정적 파일 및 템플릿 (있는 경우에만) +if (LOCAL_WORK_PATH / "static").exists(): + app.mount("/static", StaticFiles(directory=str(LOCAL_WORK_PATH / "static")), name="static") + +if (LOCAL_WORK_PATH / "templates").exists(): + templates = Jinja2Templates(directory=str(LOCAL_WORK_PATH / "templates")) +else: + templates = None + +# 작업 상태 관리 +processing_jobs: Dict[str, Dict] = {} + +def check_nas_connection(): + """NAS 연결 상태 확인""" + try: + # 1. Media 마운트 확인 + if not NAS_MOUNT_POINT.exists(): + return {"status": "not_mounted", "error": "Media 마운트 포인트 없음"} + + # 2. 쓰기 권한 확인 + try: + test_file = NAS_MOUNT_POINT / ".test_write" + test_file.touch() + test_file.unlink() + except: + return {"status": "read_only", "error": "Media 마운트 읽기 전용"} + + # 3. Document-upload 폴더 확인 + if not DOCUMENT_UPLOAD_BASE.exists(): + return {"status": "folder_missing", "error": "Document-upload 폴더 없음"} + + return {"status": "connected", "mount_point": str(NAS_MOUNT_POINT)} + + except Exception as e: + return {"status": "error", "error": str(e)} + +def ensure_nas_directories(): + """NAS 디렉토리 구조 생성""" + try: + # Document-upload 기본 구조 + DOCUMENT_UPLOAD_BASE.mkdir(exist_ok=True) + + current_month = datetime.now().strftime("%Y-%m") + + folders_to_create = [ + NAS_ORIGINALS_PATH / current_month / "pdfs", + NAS_ORIGINALS_PATH / current_month / "docs", + NAS_ORIGINALS_PATH / current_month / "txts", + NAS_TRANSLATED_PATH / current_month / "english-to-korean", + NAS_TRANSLATED_PATH / current_month / "japanese-to-korean", + NAS_TRANSLATED_PATH / current_month / "korean-only", + NAS_STATIC_HOSTING_PATH / "docs", + NAS_STATIC_HOSTING_PATH / "assets", + NAS_STATIC_HOSTING_PATH / "index", + NAS_METADATA_PATH / "processing-logs" + ] + + for folder in folders_to_create: + folder.mkdir(parents=True, exist_ok=True) + + # README 파일 생성 + readme_content = f"""# AI 번역 시스템 문서 저장소 + +자동 생성 시간: {datetime.now().isoformat()} +Mac Mini: {MAC_MINI_IP}:{MAC_MINI_PORT} +NAS IP: {NAS_IP} + +## 접속 정보 + +- 웹 인터페이스: http://{MAC_MINI_IP}:{MAC_MINI_PORT} +- 시스템 상태: http://{MAC_MINI_IP}:{MAC_MINI_PORT}/system-status +- NAS 정보: http://{MAC_MINI_IP}:{MAC_MINI_PORT}/nas-info +- API 문서: http://{MAC_MINI_IP}:{MAC_MINI_PORT}/docs + +## 폴더 구조 + +- originals/: 업로드된 원본 파일들 +- translated/: 번역된 HTML 파일들 +- static-hosting/: 웹 호스팅용 파일들 +- metadata/: 처리 로그 및 메타데이터 + +## VPN 접속 + +내부 네트워크에서만 접근 가능합니다. +외부에서 접속시 VPN 연결이 필요합니다. +""" + + readme_path = DOCUMENT_UPLOAD_BASE / "README.md" + with open(readme_path, 'w', encoding='utf-8') as f: + f.write(readme_content) + + return True + + except Exception as e: + print(f"❌ 디렉토리 생성 실패: {e}") + return False + +@app.on_event("startup") +async def startup_event(): + """서버 시작시 전체 시스템 상태 확인""" + print(f"🚀 Mac Mini AI 번역 서버 시작") + print(f"📍 접속 주소: http://{MAC_MINI_IP}:{MAC_MINI_PORT}") + print(f"📍 NAS Mount: {NAS_MOUNT_POINT}") + print("-" * 60) + + # NAS 연결 상태 확인 + nas_status = check_nas_connection() + + if nas_status["status"] == "connected": + print(f"✅ NAS 연결 정상: {nas_status['mount_point']}") + + if ensure_nas_directories(): + print(f"✅ 폴더 구조 확인/생성 완료") + print(f"📁 문서 저장 경로: {DOCUMENT_UPLOAD_BASE}") + else: + print(f"⚠️ 폴더 생성 실패") + else: + print(f"❌ NAS 연결 실패: {nas_status['error']}") + +@app.get("/") +async def index(request: Request = None): + """메인 페이지""" + nas_status = check_nas_connection() + + if templates: + return templates.TemplateResponse("index.html", { + "request": request, + "nas_status": nas_status + }) + else: + # 포트 20080 반영된 기본 HTML + html_content = f""" + + + + + + 🤖 AI 번역 시스템 + + + +
+
+

🤖 AI 번역 시스템

+

PDF/문서를 다국어 HTML로 자동 변환

+
+ +
+
+ 🖥️ + Mac Mini: {MAC_MINI_IP}:{MAC_MINI_PORT} +
+
+ 💾 + NAS: {NAS_IP} (DS1525+) +
+
+ +
+ {'✅ NAS 연결됨: ' + nas_status.get('mount_point', '') if nas_status['status'] == 'connected' else '❌ NAS 연결 실패: ' + nas_status.get('error', '')} +
+ +
+
+

📁 파일 업로드

+

PDF, TXT, DOCX 파일 지원 (최대 100MB)

+
+ +
+ +
+ +
+
처리 중...
+
+
+
+
+ +
+

🎉 변환 완료!

+
+ 📥 HTML 다운로드 + 📊 NAS 정보 +
+
+
+ +
+

🔗 시스템 링크

+ +
+
+
+ + + + + """ + return HTMLResponse(content=html_content) + +@app.get("/system-status") +async def system_status(): + """전체 시스템 상태""" + nas_status = check_nas_connection() + + # 파일 통계 + file_stats = {"originals": 0, "translated": 0, "total_size": 0} + + try: + if NAS_ORIGINALS_PATH.exists(): + original_files = list(NAS_ORIGINALS_PATH.rglob("*.*")) + file_stats["originals"] = len([f for f in original_files if f.is_file()]) + + if NAS_TRANSLATED_PATH.exists(): + html_files = list(NAS_TRANSLATED_PATH.rglob("*.html")) + file_stats["translated"] = len(html_files) + file_stats["total_size"] = sum(f.stat().st_size for f in html_files if f.exists()) + except: + pass + + return { + "timestamp": datetime.now().isoformat(), + "server": f"{MAC_MINI_IP}:{MAC_MINI_PORT}", + "nas_ip": NAS_IP, + "nas_mount": str(NAS_MOUNT_POINT), + "nas_status": nas_status, + "file_stats": file_stats, + "active_jobs": len(processing_jobs), + "document_upload_path": str(DOCUMENT_UPLOAD_BASE), + "access_urls": { + "main": f"http://{MAC_MINI_IP}:{MAC_MINI_PORT}", + "system_status": f"http://{MAC_MINI_IP}:{MAC_MINI_PORT}/system-status", + "nas_info": f"http://{MAC_MINI_IP}:{MAC_MINI_PORT}/nas-info", + "api_docs": f"http://{MAC_MINI_IP}:{MAC_MINI_PORT}/docs" + } + } + +@app.post("/upload") +async def upload_file(background_tasks: BackgroundTasks, file: UploadFile = File(...)): + """파일 업로드""" + + # NAS 연결 상태 먼저 확인 + nas_status = check_nas_connection() + if nas_status["status"] != "connected": + raise HTTPException( + status_code=503, + detail=f"NAS 연결 실패: {nas_status['error']}" + ) + + # 파일 검증 + if not file.filename: + raise HTTPException(status_code=400, detail="파일이 선택되지 않았습니다.") + + allowed_extensions = {'.pdf', '.txt', '.docx', '.doc'} + file_ext = Path(file.filename).suffix.lower() + + if file_ext not in allowed_extensions: + raise HTTPException( + status_code=400, + detail=f"지원되지 않는 파일 형식입니다. 지원: {', '.join(allowed_extensions)}" + ) + + # 파일 크기 확인 (100MB 제한) + content = await file.read() + if len(content) > 100 * 1024 * 1024: + raise HTTPException(status_code=413, detail="파일 크기는 100MB를 초과할 수 없습니다.") + + # 고유 작업 ID 생성 + job_id = str(uuid.uuid4()) + + # NAS에 저장 + current_month = datetime.now().strftime("%Y-%m") + file_type_folder = { + '.pdf': 'pdfs', + '.doc': 'docs', + '.docx': 'docs', + '.txt': 'txts' + }.get(file_ext, 'others') + + nas_original_dir = NAS_ORIGINALS_PATH / current_month / file_type_folder + nas_original_dir.mkdir(parents=True, exist_ok=True) + + # 안전한 파일명 생성 + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + safe_filename = f"{timestamp}_{job_id[:8]}_{file.filename}".replace(" ", "_") + nas_file_path = nas_original_dir / safe_filename + + # 파일 저장 + try: + async with aiofiles.open(nas_file_path, 'wb') as f: + await f.write(content) + + print(f"📁 파일 저장: {nas_file_path}") + + except Exception as e: + raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") + + # 작업 상태 초기화 + job = { + "id": job_id, + "filename": file.filename, + "file_type": file_ext, + "status": "uploaded", + "progress": 20, + "message": f"파일 저장 완료: {safe_filename}", + "nas_original_path": str(nas_file_path), + "created_at": time.time(), + "server": f"{MAC_MINI_IP}:{MAC_MINI_PORT}" + } + + processing_jobs[job_id] = job + + # 백그라운드 처리 시작 + background_tasks.add_task(process_document_simple, job_id, nas_file_path) + + return { + "job_id": job_id, + "message": "파일 업로드 완료, 처리를 시작합니다.", + "nas_path": str(nas_file_path), + "server": f"{MAC_MINI_IP}:{MAC_MINI_PORT}" + } + +async def process_document_simple(job_id: str, nas_file_path: Path): + """간단한 문서 처리 (AI 모델 연동 전 테스트용)""" + + try: + processing_jobs[job_id].update({ + "status": "processing", + "progress": 40, + "message": "텍스트 추출 중..." + }) + + await asyncio.sleep(2) # 시뮬레이션 + + processing_jobs[job_id].update({ + "progress": 70, + "message": "언어 감지 및 번역 중..." + }) + + await asyncio.sleep(3) # 시뮬레이션 + + # 간단한 HTML 생성 (테스트용) + current_month = datetime.now().strftime("%Y-%m") + nas_translated_dir = NAS_TRANSLATED_PATH / current_month / "korean-only" + nas_translated_dir.mkdir(parents=True, exist_ok=True) + + base_name = Path(nas_file_path.stem).stem + if '_' in base_name: + parts = base_name.split('_')[2:] # 타임스탬프와 job_id 제거 + if parts: + base_name = '_'.join(parts) + + html_filename = f"{base_name}.html" + nas_html_path = nas_translated_dir / html_filename + + # 테스트용 HTML 생성 + html_content = f""" + + + + + {base_name} - AI 번역 결과 + + + +
+
+

📄 {base_name}

+

AI 번역 시스템 처리 결과

+
+ +
+
+
+ ⏰ + {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} +
+
+ 📁 + {nas_file_path.name} +
+
+ 🆔 + {job_id[:8]} +
+
+ 🖥️ + {MAC_MINI_IP}:{MAC_MINI_PORT} +
+
+
+ +
+
+

📋 문서 정보

+
+

원본 파일: {nas_file_path.name}

+

저장 위치: {nas_html_path}

+

처리 서버: Mac Mini ({MAC_MINI_IP}:{MAC_MINI_PORT})

+

NAS 저장소: DS1525+ ({NAS_IP})

+
+
+ +
+

🤖 AI 처리 결과

+
+

테스트 모드

+

현재 AI 번역 시스템이 테스트 모드로 실행 중입니다.

+

실제 AI 모델(NLLB, KoBART)이 연동되면 이 부분에 다음 내용이 표시됩니다:

+
    +
  • 자동 언어 감지 결과
    • +
  • 한국어 번역 텍스트
    • +
  • 문서 요약
    • +
  • 다국어 지원 인터페이스
    • +
+
+
+ +
+

🌐 시스템 구성

+
+
+

Mac Mini M4 Pro

+

IP: {MAC_MINI_IP}:{MAC_MINI_PORT}

+

역할: AI 처리 서버

+

모델: NLLB + KoBART

+
+
+

Synology DS1525+

+

IP: {NAS_IP}

+

역할: 파일 저장소

+

마운트: /Volumes/Media

+
+
+

네트워크

+

연결: 2.5GbE

+

접근: VPN 필요

+

프로토콜: SMB

+
+
+
+ +
+

🔗 관련 링크

+
+

🏠 메인 페이지

+

📊 시스템 상태

+

💾 NAS 정보

+

📖 API 문서

+
+
+
+ +
+

AI 번역 시스템 v1.0.0 | Mac Mini M4 Pro + Synology DS1525+ | 자동 생성 문서

+
+
+ +""" + + # HTML 파일 저장 + async with aiofiles.open(nas_html_path, 'w', encoding='utf-8') as f: + await f.write(html_content) + + processing_jobs[job_id].update({ + "progress": 100, + "status": "completed", + "message": "처리 완료!", + "nas_translated_path": str(nas_html_path) + }) + + print(f"✅ 테스트 처리 완료: {job_id} - {html_filename}") + + except Exception as e: + processing_jobs[job_id].update({ + "status": "error", + "progress": 0, + "message": f"처리 실패: {str(e)}" + }) + + print(f"❌ 처리 실패: {job_id} - {str(e)}") + +@app.get("/status/{job_id}") +async def get_job_status(job_id: str): + """작업 상태 조회""" + if job_id not in processing_jobs: + raise HTTPException(status_code=404, detail="작업을 찾을 수 없습니다.") + + return processing_jobs[job_id] + +@app.get("/download/{job_id}") +async def download_result(job_id: str): + """결과 파일 다운로드""" + if job_id not in processing_jobs: + raise HTTPException(status_code=404, detail="작업을 찾을 수 없습니다.") + + job = processing_jobs[job_id] + + if job["status"] != "completed" or not job.get("nas_translated_path"): + raise HTTPException(status_code=400, detail="처리가 완료되지 않았습니다.") + + result_path = Path(job["nas_translated_path"]) + if not result_path.exists(): + raise HTTPException(status_code=404, detail="결과 파일을 찾을 수 없습니다.") + + return FileResponse( + path=result_path, + filename=f"{Path(job['filename']).stem}.html", + media_type="text/html" + ) + +@app.get("/nas-info") +async def nas_info(): + """NAS 정보 및 통계""" + + nas_status = check_nas_connection() + + info = { + "server": f"{MAC_MINI_IP}:{MAC_MINI_PORT}", + "nas_ip": NAS_IP, + "nas_status": nas_status, + "mount_point": str(NAS_MOUNT_POINT), + "document_upload_base": str(DOCUMENT_UPLOAD_BASE), + "folders": {}, + "statistics": { + "total_jobs": len(processing_jobs), + "completed_jobs": len([j for j in processing_jobs.values() if j["status"] == "completed"]), + "processing_jobs": len([j for j in processing_jobs.values() if j["status"] == "processing"]), + "failed_jobs": len([j for j in processing_jobs.values() if j["status"] == "error"]) + }, + "recent_jobs": list(processing_jobs.values())[-5:] if processing_jobs else [] + } + + # 폴더 정보 수집 + try: + for folder_name, folder_path in [ + ("originals", NAS_ORIGINALS_PATH), + ("translated", NAS_TRANSLATED_PATH), + ("static-hosting", NAS_STATIC_HOSTING_PATH), + ("metadata", NAS_METADATA_PATH) + ]: + if folder_path.exists(): + files = list(folder_path.rglob("*.*")) + file_count = len([f for f in files if f.is_file()]) + total_size = sum(f.stat().st_size for f in files if f.is_file()) + + info["folders"][folder_name] = { + "exists": True, + "path": str(folder_path), + "file_count": file_count, + "total_size_mb": round(total_size / (1024 * 1024), 2) + } + else: + info["folders"][folder_name] = { + "exists": False, + "path": str(folder_path) + } + except Exception as e: + info["error"] = str(e) + + return info + +if __name__ == "__main__": + import uvicorn + + print(f"🚀 Mac Mini AI 번역 서버") + print(f"📡 접속 주소: http://{MAC_MINI_IP}:{MAC_MINI_PORT}") + print(f"📁 NAS Mount: {NAS_MOUNT_POINT}") + print(f"🔗 VPN 접속 전용 (내부 네트워크)") + + # 시작 전 연결 확인 + nas_status = check_nas_connection() + if nas_status["status"] == "connected": + print(f"✅ NAS 연결 확인됨") + ensure_nas_directories() + else: + print(f"❌ NAS 연결 문제: {nas_status['error']}") + + print("-" * 60) + uvicorn.run(app, host="0.0.0.0", port=MAC_MINI_PORT) diff --git a/src/html_generator.py b/src/html_generator.py new file mode 100755 index 0000000..bd8de21 --- /dev/null +++ b/src/html_generator.py @@ -0,0 +1,597 @@ +#!/usr/bin/env python3 +""" +다국어 HTML 생성기 +원문 + 번역문을 언어 전환 가능한 HTML로 생성 +""" + +from pathlib import Path +from typing import Dict, List +import json +import re +from datetime import datetime + +class MultilingualHTMLGenerator: + def __init__(self): + self.templates = self._load_templates() + + def _load_templates(self) -> Dict: + """HTML 템플릿들""" + return { + "base": """ + + + + + {title} + + + +
+
+
+ +
+
+ {language_buttons} +
+
+ +
+
+

{title}

+
+ {metadata} +
+
+ + {toc_section} + + {summary_section} + +
+
+ 📖 본문 내용 +
+
+ {content_sections} +
+
+
+ + + +""", + + "language_button": """""", + + "metadata_item": """
+ {icon} + {label}: {value} +
""", + + "content_section": """
+
{content}
+
""", + + "summary_section": """
+
+ 📋 요약 +
+
+
+
문서 요약
+ {summary_content} +
+
+
""", + + "toc_section": """
+
📑 목차
+ +
""" + } + + def _generate_toc(self, content: str) -> List[Dict]: + """목차 자동 생성""" + # 간단한 헤딩 감지 (대문자로 시작하는 짧은 줄들) + lines = content.split('\n') + toc_items = [] + + for i, line in enumerate(lines): + line = line.strip() + # 헤딩 감지 조건 + if (len(line) < 100 and + len(line) > 5 and + (line.isupper() or + line.startswith(('Chapter', '장', '제', '1.', '2.', '3.', '4.', '5.')))): + + toc_items.append({ + "title": line, + "anchor": f"section-{i}", + "line_number": i + }) + + return toc_items[:10] # 최대 10개만 + + def generate_multilingual_html(self, + title: str, + contents: Dict[str, str], # {lang_code: content} + summary: str = "", + metadata: Dict = None, + output_path: str = "output.html") -> str: + """다국어 HTML 생성""" + + # 언어 정보 매핑 + lang_info = { + "korean": {"name": "한국어", "code": "ko", "icon": "🇰🇷"}, + "english": {"name": "English", "code": "en", "icon": "🇺🇸"}, + "japanese": {"name": "日本語", "code": "ja", "icon": "🇯🇵"} + } + + # 기본 메타데이터 + if metadata is None: + metadata = {} + + default_metadata = { + "처리_시간": datetime.now().strftime("%Y-%m-%d %H:%M"), + "언어_수": len(contents), + "총_문자수": sum(len(content) for content in contents.values()), + "생성_도구": "NLLB 번역 시스템" + } + default_metadata.update(metadata) + + # 언어 버튼 생성 + language_buttons = [] + for lang_key, content in contents.items(): + lang_data = lang_info.get(lang_key, {"name": lang_key.title(), "code": lang_key[:2], "icon": "🌐"}) + button_html = self.templates["language_button"].format( + lang_code=lang_data["code"], + lang_name=f"{lang_data['icon']} {lang_data['name']}" + ) + language_buttons.append(button_html) + + # 메타데이터 생성 + metadata_items = [] + metadata_icons = { + "처리_시간": "⏰", + "언어_수": "🌍", + "총_문자수": "📝", + "생성_도구": "⚙️", + "원본_언어": "🔤", + "파일_크기": "📊" + } + + for key, value in default_metadata.items(): + icon = metadata_icons.get(key, "📄") + label = key.replace("_", " ").title() + + metadata_html = self.templates["metadata_item"].format( + icon=icon, + label=label, + value=value + ) + metadata_items.append(metadata_html) + + # 콘텐츠 섹션 생성 + content_sections = [] + default_lang = list(contents.keys())[0] + + for lang_key, content in contents.items(): + lang_data = lang_info.get(lang_key, {"code": lang_key[:2]}) + + # 내용을 문단별로 정리 + formatted_content = self._format_content(content) + + section_html = self.templates["content_section"].format( + lang_code=lang_data["code"], + content=formatted_content + ) + content_sections.append(section_html) + + # 목차 생성 (첫 번째 언어 기준) + first_content = list(contents.values())[0] + toc_items = self._generate_toc(first_content) + + toc_html = "" + if toc_items: + toc_item_htmls = [] + for item in toc_items: + toc_item_htmls.append(f'
  • {item["title"]}
    • ') + + toc_html = self.templates["toc_section"].format( + toc_items="\n".join(toc_item_htmls) + ) + + # 요약 섹션 + summary_html = "" + if summary: + summary_content_sections = [] + for lang_key in contents.keys(): + lang_data = lang_info.get(lang_key, {"code": lang_key[:2]}) + if lang_key == "korean" or "korean" not in contents: + summary_content = f'
    {summary}
    ' + else: + summary_content = f'
    {summary}
    ' + summary_content_sections.append(summary_content) + + summary_html = self.templates["summary_section"].format( + summary_content="\n".join(summary_content_sections) + ) + + # 최종 HTML 생성 + default_lang_code = lang_info.get(default_lang, {"code": default_lang[:2]})["code"] + + html_content = self.templates["base"].format( + title=title, + language_buttons="\n".join(language_buttons), + metadata="\n".join(metadata_items), + toc_section=toc_html, + summary_section=summary_html, + content_sections="\n".join(content_sections), + default_lang=default_lang_code + ) + + # 파일 저장 + output_file = Path(output_path) + output_file.parent.mkdir(parents=True, exist_ok=True) + + with open(output_file, 'w', encoding='utf-8') as f: + f.write(html_content) + + print(f"다국어 HTML 생성 완료: {output_path}") + return str(output_file) + + def _format_content(self, content: str) -> str: + """내용 포맷팅""" + # 기본 텍스트 정리 + content = content.strip() + + # 문단 구분 개선 + content = re.sub(r'\n\s*\n\s*\n+', '\n\n', content) + + # 특수 문자 이스케이프 + content = content.replace('<', '<').replace('>', '>') + + return content + +def main(): + """HTML 생성기 테스트""" + generator = MultilingualHTMLGenerator() + + # 테스트 데이터 + test_contents = { + "english": """Chapter 1: Introduction to Machine Learning + +Machine learning represents one of the most transformative technologies of our time. This comprehensive guide explores the core concepts, methodologies, and applications that define this rapidly evolving field. + +The power of machine learning lies in its ability to handle complex problems that would be difficult or impossible to solve using conventional programming methods.""", + + "korean": """제1장: 기계학습 소개 + +기계학습은 우리 시대의 가장 혁신적인 기술 중 하나를 나타냅니다. 이 포괄적인 가이드는 빠르게 발전하는 이 분야를 정의하는 핵심 개념, 방법론 및 응용 분야를 탐구합니다. + +기계학습의 힘은 기존 프로그래밍 방법으로는 해결하기 어렵거나 불가능한 복잡한 문제를 처리할 수 있는 능력에 있습니다.""" + } + + test_summary = "이 문서는 기계학습의 기본 개념과 응용 분야에 대해 설명합니다. 기계학습이 현대 기술에서 차지하는 중요성과 복잡한 문제 해결 능력을 강조합니다." + + test_metadata = { + "원본_언어": "English", + "파일_크기": "2.3 MB" + } + + # HTML 생성 + output_path = generator.generate_multilingual_html( + title="기계학습 완전 가이드", + contents=test_contents, + summary=test_summary, + metadata=test_metadata, + output_path="output/test_multilingual.html" + ) + + print(f"테스트 HTML 생성됨: {output_path}") + +if __name__ == "__main__": + main() diff --git a/src/integrated_translation_system.py b/src/integrated_translation_system.py new file mode 100755 index 0000000..db78f22 --- /dev/null +++ b/src/integrated_translation_system.py @@ -0,0 +1,364 @@ +#!/usr/bin/env python3 +""" +완전한 PDF -> HTML 번역 시스템 +PDF OCR -> NLLB 번역 -> KoBART 요약 -> HTML 생성 +""" + +import torch +import time +import json +from pathlib import Path +from typing import Dict, List, Optional, Tuple +import re +from dataclasses import dataclass + +# 문서 처리 +try: + import PyPDF2 + import pdfplumber + from docx import Document +except ImportError: + print("문서 처리 라이브러리 설치 필요: pip install PyPDF2 pdfplumber python-docx") + +# 번역 및 요약 모델 +from transformers import ( + AutoTokenizer, AutoModelForSeq2SeqLM, + PreTrainedTokenizerFast, BartForConditionalGeneration +) + +@dataclass +class TranslationResult: + original_text: str + translated_text: str + summary: str + processing_time: float + metadata: Dict + +class IntegratedTranslationSystem: + def __init__(self): + self.device = self._setup_device() + self.models = {} + self.tokenizers = {} + self.config = self._load_config() + + print(f"번역 시스템 초기화 (디바이스: {self.device})") + + def _setup_device(self) -> torch.device: + """최적 디바이스 설정""" + if torch.backends.mps.is_available(): + return torch.device("mps") + elif torch.cuda.is_available(): + return torch.device("cuda") + else: + return torch.device("cpu") + + def _load_config(self) -> Dict: + """설정 파일 로드""" + config_path = Path("config/settings.json") + if config_path.exists(): + with open(config_path, 'r', encoding='utf-8') as f: + return json.load(f) + + # 기본 설정 + return { + "translation": { + "chunk_size": 500, + "max_length": 512, + "num_beams": 4, + "batch_size": 4 + }, + "summarization": { + "max_length": 150, + "min_length": 30, + "num_beams": 4 + } + } + + def load_models(self): + """모든 모델 로드""" + print("모델 로딩 중...") + + # 1. NLLB 번역 모델 + print(" NLLB 번역 모델...") + try: + self.tokenizers['nllb'] = AutoTokenizer.from_pretrained("facebook/nllb-200-3.3B") + self.models['nllb'] = AutoModelForSeq2SeqLM.from_pretrained( + "facebook/nllb-200-3.3B", + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ).to(self.device) + print(" NLLB 모델 로드 완료") + except Exception as e: + print(f" NLLB 모델 로드 실패: {e}") + return False + + # 2. KoBART 요약 모델 + print(" KoBART 요약 모델...") + try: + self.tokenizers['kobart'] = PreTrainedTokenizerFast.from_pretrained("gogamza/kobart-summarization") + self.models['kobart'] = BartForConditionalGeneration.from_pretrained( + "gogamza/kobart-summarization", + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ).to(self.device) + print(" KoBART 모델 로드 완료") + except Exception as e: + print(f" KoBART 모델 로드 실패: {e}") + print(" 요약 없이 번역만 진행") + self.models['kobart'] = None + + print("모델 로딩 완료!") + return True + + def detect_language(self, text: str) -> str: + """언어 자동 감지""" + # 간단한 휴리스틱 언어 감지 + korean_chars = len(re.findall(r'[가-힣]', text)) + japanese_chars = len(re.findall(r'[ひらがなカタカナ一-龯]', text)) + english_chars = len(re.findall(r'[a-zA-Z]', text)) + + total_chars = len(text.replace(' ', '')) + + if total_chars == 0: + return "unknown" + + if korean_chars / total_chars > 0.3: + return "korean" + elif japanese_chars / total_chars > 0.1: + return "japanese" + elif english_chars / total_chars > 0.5: + return "english" + else: + return "unknown" + + def extract_text_from_pdf(self, pdf_path: str) -> str: + """PDF에서 텍스트 추출""" + print(f"PDF 텍스트 추출: {pdf_path}") + + text = "" + try: + # pdfplumber 우선 시도 + import pdfplumber + with pdfplumber.open(pdf_path) as pdf: + for page_num, page in enumerate(pdf.pages, 1): + page_text = page.extract_text() + if page_text: + text += f"\n\n{page_text}" + + print(f"PDF 텍스트 추출 완료: {len(text)}자") + + except Exception as e: + print(f"pdfplumber 실패: {e}") + + # PyPDF2 백업 + try: + import PyPDF2 + with open(pdf_path, 'rb') as file: + pdf_reader = PyPDF2.PdfReader(file) + for page in pdf_reader.pages: + page_text = page.extract_text() + if page_text: + text += f"\n\n{page_text}" + + print(f"PyPDF2로 텍스트 추출 완료: {len(text)}자") + + except Exception as e2: + print(f"PDF 텍스트 추출 완전 실패: {e2}") + return "" + + return self._clean_text(text) + + def _clean_text(self, text: str) -> str: + """추출된 텍스트 정리""" + # 과도한 공백 정리 + text = re.sub(r'\n\s*\n\s*\n', '\n\n', text) + text = re.sub(r'[ \t]+', ' ', text) + + # 페이지 번호 제거 + text = re.sub(r'\n\d+\n', '\n', text) + + return text.strip() + + def split_text_into_chunks(self, text: str, chunk_size: int = 500) -> List[str]: + """텍스트를 번역 가능한 청크로 분할""" + sentences = re.split(r'[.!?]\s+', text) + chunks = [] + current_chunk = "" + + for sentence in sentences: + test_chunk = current_chunk + " " + sentence if current_chunk else sentence + + if len(test_chunk) > chunk_size and current_chunk: + chunks.append(current_chunk.strip()) + current_chunk = sentence + else: + current_chunk = test_chunk + + if current_chunk: + chunks.append(current_chunk.strip()) + + print(f"텍스트 분할: {len(chunks)}개 청크") + return chunks + + def translate_text(self, text: str, src_lang: str = "english") -> str: + """NLLB로 텍스트 번역""" + if src_lang == "korean": + return text # 한국어는 번역하지 않음 + + # 언어 코드 매핑 + lang_map = { + "english": "eng_Latn", + "japanese": "jpn_Jpan", + "korean": "kor_Hang" + } + + src_code = lang_map.get(src_lang, "eng_Latn") + tgt_code = "kor_Hang" + + tokenizer = self.tokenizers['nllb'] + model = self.models['nllb'] + + # 청크별 번역 + chunks = self.split_text_into_chunks(text, self.config["translation"]["chunk_size"]) + translated_chunks = [] + + print(f"번역 시작: {src_lang} -> 한국어") + + for i, chunk in enumerate(chunks): + print(f" 청크 {i+1}/{len(chunks)} 번역 중...") + + inputs = tokenizer(chunk, return_tensors="pt", padding=True, truncation=True).to(self.device) + + with torch.no_grad(): + translated_tokens = model.generate( + **inputs, + forced_bos_token_id=tokenizer.convert_tokens_to_ids(tgt_code), + max_length=self.config["translation"]["max_length"], + num_beams=self.config["translation"]["num_beams"], + early_stopping=True + ) + + translated_chunk = tokenizer.decode(translated_tokens[0], skip_special_tokens=True) + translated_chunks.append(translated_chunk) + + result = "\n\n".join(translated_chunks) + print(f"번역 완료: {len(result)}자") + return result + + def summarize_text(self, text: str) -> str: + """KoBART로 한국어 텍스트 요약""" + if self.models['kobart'] is None: + print("요약 모델 없음, 첫 300자 반환") + return text[:300] + "..." if len(text) > 300 else text + + print("텍스트 요약 중...") + + tokenizer = self.tokenizers['kobart'] + model = self.models['kobart'] + + inputs = tokenizer( + text, + return_tensors="pt", + max_length=1024, + truncation=True, + padding=True, + return_token_type_ids=False + ).to(self.device) + + with torch.no_grad(): + summary_ids = model.generate( + input_ids=inputs['input_ids'], + attention_mask=inputs['attention_mask'], + max_length=self.config["summarization"]["max_length"], + min_length=self.config["summarization"]["min_length"], + num_beams=self.config["summarization"]["num_beams"], + early_stopping=True, + no_repeat_ngram_size=2, + length_penalty=1.2 + ) + + summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True) + print(f"요약 완료: {len(summary)}자") + return summary + + def process_document(self, input_path: str, output_dir: str = "output") -> TranslationResult: + """전체 문서 처리 파이프라인""" + start_time = time.time() + + print(f"문서 처리 시작: {input_path}") + + # 1. 텍스트 추출 + if input_path.lower().endswith('.pdf'): + original_text = self.extract_text_from_pdf(input_path) + else: + with open(input_path, 'r', encoding='utf-8') as f: + original_text = f.read() + + if not original_text: + raise ValueError("텍스트 추출 실패") + + # 2. 언어 감지 + detected_lang = self.detect_language(original_text) + print(f"감지된 언어: {detected_lang}") + + # 3. 번역 + if detected_lang == "korean": + translated_text = original_text + print("한국어 문서, 번역 생략") + else: + translated_text = self.translate_text(original_text, detected_lang) + + # 4. 요약 + summary = self.summarize_text(translated_text) + + # 5. 결과 저장 + output_path = Path(output_dir) + output_path.mkdir(exist_ok=True) + + base_name = Path(input_path).stem + + # 텍스트 파일 저장 + with open(output_path / f"{base_name}_translated.txt", 'w', encoding='utf-8') as f: + f.write(translated_text) + + with open(output_path / f"{base_name}_summary.txt", 'w', encoding='utf-8') as f: + f.write(summary) + + processing_time = time.time() - start_time + + result = TranslationResult( + original_text=original_text, + translated_text=translated_text, + summary=summary, + processing_time=processing_time, + metadata={ + "input_file": input_path, + "detected_language": detected_lang, + "original_chars": len(original_text), + "translated_chars": len(translated_text), + "summary_chars": len(summary), + "compression_ratio": len(summary) / len(translated_text) * 100 if translated_text else 0 + } + ) + + print(f"문서 처리 완료! ({processing_time/60:.1f}분 소요)") + return result + +def main(): + """메인 실행 함수""" + system = IntegratedTranslationSystem() + + if not system.load_models(): + print("모델 로딩 실패") + return None + + print("\n" + "="*60) + print("통합 번역 시스템 준비 완료!") + print("사용법:") + print(" result = system.process_document('input.pdf')") + print("="*60) + + return system + +if __name__ == "__main__": + system = main() diff --git a/src/nas_mount_setup.py b/src/nas_mount_setup.py new file mode 100755 index 0000000..d377169 --- /dev/null +++ b/src/nas_mount_setup.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +""" +NAS 자동 마운트 및 연결 설정 +DS1525+ (192.168.1.227) → Mac Mini (192.168.1.122) +""" + +import subprocess +import os +from pathlib import Path +import time + +class NASMountManager: + def __init__(self): + self.nas_ip = "192.168.1.227" + self.nas_share = "Media" # 또는 실제 공유 폴더명 + self.mount_point = Path("/Volumes/DS1525+") + self.smb_url = f"smb://{self.nas_ip}/{self.nas_share}" + + def check_nas_connection(self): + """NAS 연결 상태 확인""" + try: + # ping으로 NAS 접근 가능한지 확인 + result = subprocess.run( + ["ping", "-c", "3", self.nas_ip], + capture_output=True, + timeout=10 + ) + + if result.returncode == 0: + print(f"✅ NAS 연결 가능: {self.nas_ip}") + return True + else: + print(f"❌ NAS 연결 불가: {self.nas_ip}") + return False + + except subprocess.TimeoutExpired: + print(f"⏰ NAS 연결 타임아웃: {self.nas_ip}") + return False + except Exception as e: + print(f"❌ 연결 확인 오류: {e}") + return False + + def check_mount_status(self): + """마운트 상태 확인""" + if self.mount_point.exists(): + # 실제 마운트되어 있는지 확인 + try: + result = subprocess.run( + ["mount"], + capture_output=True, + text=True + ) + + if str(self.mount_point) in result.stdout: + print(f"✅ NAS 이미 마운트됨: {self.mount_point}") + return True + else: + print(f"📁 마운트 포인트 존재하지만 연결 안됨: {self.mount_point}") + return False + + except Exception as e: + print(f"❌ 마운트 상태 확인 실패: {e}") + return False + else: + print(f"📁 마운트 포인트 없음: {self.mount_point}") + return False + + def mount_nas(self, username=None, password=None): + """NAS 마운트""" + + if self.check_mount_status(): + return True + + if not self.check_nas_connection(): + return False + + try: + # 마운트 포인트 생성 + self.mount_point.mkdir(parents=True, exist_ok=True) + + # SMB 마운트 시도 + if username and password: + # 인증 정보가 있는 경우 + mount_cmd = [ + "mount", "-t", "smbfs", + f"//{username}:{password}@{self.nas_ip}/{self.nas_share}", + str(self.mount_point) + ] + else: + # 게스트 접근 시도 + mount_cmd = [ + "mount", "-t", "smbfs", + f"//{self.nas_ip}/{self.nas_share}", + str(self.mount_point) + ] + + print(f"🔄 NAS 마운트 시도: {self.smb_url}") + result = subprocess.run(mount_cmd, capture_output=True, text=True) + + if result.returncode == 0: + print(f"✅ NAS 마운트 성공: {self.mount_point}") + return True + else: + print(f"❌ NAS 마운트 실패: {result.stderr}") + return False + + except Exception as e: + print(f"❌ 마운트 프로세스 오류: {e}") + return False + + def unmount_nas(self): + """NAS 언마운트""" + try: + if self.check_mount_status(): + result = subprocess.run( + ["umount", str(self.mount_point)], + capture_output=True, + text=True + ) + + if result.returncode == 0: + print(f"✅ NAS 언마운트 완료: {self.mount_point}") + return True + else: + print(f"❌ 언마운트 실패: {result.stderr}") + return False + else: + print("ℹ️ NAS가 마운트되어 있지 않음") + return True + + except Exception as e: + print(f"❌ 언마운트 오류: {e}") + return False + + def create_document_upload_structure(self): + """Document-upload 폴더 구조 생성""" + if not self.check_mount_status(): + print("❌ NAS가 마운트되지 않음") + return False + + base_path = self.mount_point / "Document-upload" + + try: + # 기본 구조 생성 + folders = [ + "originals", + "originals/2025-01/pdfs", + "originals/2025-01/docs", + "originals/2025-01/txts", + "translated", + "translated/2025-01/english-to-korean", + "translated/2025-01/japanese-to-korean", + "translated/2025-01/korean-only", + "static-hosting/docs", + "static-hosting/assets", + "static-hosting/index", + "metadata/processing-logs" + ] + + for folder in folders: + folder_path = base_path / folder + folder_path.mkdir(parents=True, exist_ok=True) + print(f"📁 폴더 생성: {folder}") + + # 기본 README 파일 생성 + readme_content = """# Document Upload System + +이 폴더는 AI 번역 시스템에서 자동으로 관리됩니다. + +## 폴더 구조 + +- originals/: 업로드된 원본 파일들 (월별/타입별 정리) +- translated/: 번역된 HTML 파일들 (월별/언어별 정리) +- static-hosting/: 웹 호스팅용 파일들 +- metadata/: 처리 로그 및 인덱스 정보 + +## 자동 관리 + +- 파일은 월별로 자동 정리됩니다 +- 언어 감지에 따라 적절한 폴더에 저장됩니다 +- 메타데이터는 자동으로 기록됩니다 +""" + + readme_path = base_path / "README.md" + with open(readme_path, 'w', encoding='utf-8') as f: + f.write(readme_content) + + print(f"✅ Document-upload 구조 생성 완료: {base_path}") + return True + + except Exception as e: + print(f"❌ 폴더 구조 생성 실패: {e}") + return False + +def main(): + """NAS 설정 메인 함수""" + print("🚀 NAS 마운트 및 설정 시작") + print("=" * 50) + + mount_manager = NASMountManager() + + # 1. NAS 연결 확인 + if not mount_manager.check_nas_connection(): + print("\n❌ NAS에 연결할 수 없습니다.") + print("확인사항:") + print("- NAS 전원이 켜져 있는지") + print("- 네트워크 연결 상태") + print("- IP 주소: 192.168.1.227") + return False + + # 2. 마운트 시도 + if not mount_manager.check_mount_status(): + print("\n🔄 NAS 마운트 시도...") + + # 우선 게스트 접근 시도 + if not mount_manager.mount_nas(): + print("\n🔐 인증이 필요할 수 있습니다.") + print("Finder에서 수동으로 연결하거나:") + print("1. Finder → 이동 → 서버에 연결") + print("2. smb://192.168.1.227 입력") + print("3. 인증 정보 입력") + + return False + + # 3. 폴더 구조 생성 + print("\n📁 Document-upload 폴더 구조 생성...") + if mount_manager.create_document_upload_structure(): + print("\n🎉 NAS 설정 완료!") + print(f"📁 마운트 위치: {mount_manager.mount_point}") + print(f"📁 문서 저장 위치: {mount_manager.mount_point}/Document-upload") + return True + else: + print("\n❌ 폴더 구조 생성 실패") + return False + +if __name__ == "__main__": + main() diff --git a/src/test_nllb_fixed.py b/src/test_nllb_fixed.py new file mode 100755 index 0000000..f0f8a88 --- /dev/null +++ b/src/test_nllb_fixed.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +""" +NLLB 모델 테스트 (수정된 버전) +""" + +import torch +import time +from transformers import AutoTokenizer, AutoModelForSeq2SeqLM + +def test_nllb_fixed(): + print("🧪 NLLB 모델 테스트 (수정된 버전)") + + model_name = "facebook/nllb-200-3.3B" + + # 모델 로드 + print("📥 모델 로딩 중...") + tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir="models/nllb-200-3.3B") + model = AutoModelForSeq2SeqLM.from_pretrained( + model_name, + cache_dir="models/nllb-200-3.3B", + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + + # Apple Silicon 최적화 + if torch.backends.mps.is_available(): + device = torch.device("mps") + model = model.to(device) + print("🚀 Apple Silicon MPS 가속 사용") + else: + device = torch.device("cpu") + print("💻 CPU 모드 사용") + + # NLLB 언어 코드 (수정된 방식) + def get_lang_id(tokenizer, lang_code): + """언어 코드를 토큰 ID로 변환""" + return tokenizer.convert_tokens_to_ids(lang_code) + + def translate_text(text, src_lang, tgt_lang, description): + print(f"\n📝 {description}:") + print(f"원문: {text}") + + start_time = time.time() + + # 입력 텍스트 토큰화 + inputs = tokenizer(text, return_tensors="pt", padding=True).to(device) + + # 번역 생성 (수정된 방식) + with torch.no_grad(): + generated_tokens = model.generate( + **inputs, + forced_bos_token_id=get_lang_id(tokenizer, tgt_lang), + max_length=200, + num_beams=4, + early_stopping=True, + do_sample=False, + pad_token_id=tokenizer.pad_token_id + ) + + # 결과 디코딩 + result = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0] + + translation_time = time.time() - start_time + print(f"번역: {result}") + print(f"소요 시간: {translation_time:.2f}초") + + return result, translation_time + + print("\n" + "="*60) + + try: + # 1. 영어 → 한국어 + en_text = "Artificial intelligence is transforming the way we work and live." + translate_text(en_text, "eng_Latn", "kor_Hang", "영어 → 한국어 번역") + + # 2. 일본어 → 한국어 + ja_text = "人工知能は私たちの働き方と生活を変革しています。" + translate_text(ja_text, "jpn_Jpan", "kor_Hang", "일본어 → 한국어 번역") + + # 3. 기술 문서 테스트 + tech_text = "Machine learning algorithms require large datasets for training and validation." + translate_text(tech_text, "eng_Latn", "kor_Hang", "기술 문서 번역") + + print(f"\n✅ 모든 테스트 성공!") + return True + + except Exception as e: + print(f"❌ 테스트 중 오류: {e}") + return False + +if __name__ == "__main__": + if test_nllb_fixed(): + print("\n🎉 NLLB 모델 테스트 완료!") + print("📝 다음 단계: KoBART 요약 모델 설치") + else: + print("\n❌ 테스트 실패") diff --git a/src/test_summarizer_fixed.py b/src/test_summarizer_fixed.py new file mode 100755 index 0000000..bfccc55 --- /dev/null +++ b/src/test_summarizer_fixed.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +요약 모델 테스트 (토큰 오류 수정) +""" + +import torch +import time +from transformers import PreTrainedTokenizerFast, BartForConditionalGeneration + +def test_summarizer_fixed(): + print("🧪 한국어 요약 모델 테스트 (수정된 버전)") + + model_name = "gogamza/kobart-summarization" + + try: + # 모델 로드 + print("📥 모델 로딩 중...") + tokenizer = PreTrainedTokenizerFast.from_pretrained(model_name) + model = BartForConditionalGeneration.from_pretrained( + model_name, + torch_dtype=torch.float16, + low_cpu_mem_usage=True + ) + + # Apple Silicon 최적화 + if torch.backends.mps.is_available(): + device = torch.device("mps") + model = model.to(device) + print("🚀 Apple Silicon MPS 가속 사용") + else: + device = torch.device("cpu") + print("💻 CPU 모드 사용") + + def summarize_text_fixed(text): + print(f"\n📝 요약 테스트:") + print(f"원문 ({len(text)}자):") + print(f"{text[:150]}...") + + start_time = time.time() + + # 토큰화 (token_type_ids 제거) + inputs = tokenizer( + text, + return_tensors="pt", + max_length=1024, + truncation=True, + padding=True, + return_token_type_ids=False # 이 부분이 핵심! + ).to(device) + + # 요약 생성 + with torch.no_grad(): + summary_ids = model.generate( + input_ids=inputs['input_ids'], + attention_mask=inputs['attention_mask'], + max_length=150, + min_length=30, + num_beams=4, + early_stopping=True, + no_repeat_ngram_size=2, + length_penalty=1.2, + do_sample=False + ) + + # 결과 디코딩 + summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True) + + process_time = time.time() - start_time + print(f"\n📋 요약 결과 ({len(summary)}자):") + print(f"{summary}") + print(f"⏱️ 처리 시간: {process_time:.2f}초") + print(f"📊 압축률: {len(summary)/len(text)*100:.1f}%") + + return summary + + # 테스트 실행 + test_text = """ + 인공지능과 기계학습 기술이 급속도로 발전하면서 우리의 일상생활과 업무 환경에 혁신적인 변화를 가져오고 있습니다. + 특히 자연어 처리 분야에서는 번역, 요약, 대화형 AI 등의 기술이 실용적인 수준에 도달하여 다양한 서비스에 적용되고 있습니다. + 기계학습 알고리즘은 대량의 텍스트 데이터를 학습하여 언어의 패턴과 의미를 이해하고, 이를 바탕으로 인간과 유사한 수준의 + 언어 처리 능력을 보여주고 있습니다. 딥러닝 기술의 발전으로 번역의 정확도가 크게 향상되었으며, 실시간 번역 서비스도 + 일상적으로 사용할 수 있게 되었습니다. + """ + + summarize_text_fixed(test_text.strip()) + + print(f"\n✅ 요약 모델 테스트 성공!") + return True + + except Exception as e: + print(f"❌ 테스트 실패: {e}") + return False + +if __name__ == "__main__": + print("🚀 한국어 요약 모델 테스트 (수정)") + print("="*50) + + if test_summarizer_fixed(): + print("\n🎉 요약 모델 정상 작동!") + print("📝 다음 단계: 통합 번역 시스템 구축") + else: + print("\n❌ 여전히 문제 있음") + print("📝 요약 없이 번역만으로 진행 고려") diff --git a/templates/index.html b/templates/index.html new file mode 100644 index 0000000..ab11568 --- /dev/null +++ b/templates/index.html @@ -0,0 +1,468 @@ + + + + + + AI 번역 시스템 + + + +
    +
    +

    🤖 AI 번역 시스템

    +

    PDF/문서를 다국어 HTML로 자동 변환

    +
    + +
    +
    + 📤 업로드 + 📚 라이브러리 + 📊 시스템 +
    +
    + +
    +
    + {% if nas_status.status == "connected" %} + ✅ NAS 연결됨 (DS1525+) + {% else %} + ❌ NAS 연결 실패: {{ nas_status.error }} + {% endif %} +
    +
    + 🚀 Mac Mini Ready +
    +
    + +
    + {% if nas_status.status != "connected" %} +
    +

    NAS 연결 필요

    +

    {{ nas_status.error }}

    +

    해결 방법:

    +
      +
    • Finder → 이동 → 서버에 연결
      • +
    • smb://192.168.1.227 입력
      • +
    • DS1525+ 연결 후 페이지 새로고침
      • +
    +
    + {% endif %} + +
    +
    +
    📁
    +
    파일을 드래그하거나 클릭하여 업로드
    +
    PDF, TXT, DOCX 파일 지원 (최대 100MB)
    + + +
    +
    + +
    +
    처리 진행 상황
    +
    +
    +
    +
    준비 중...
    +
    + + +
    +
    + + + + diff --git a/templates/library_organized.html b/templates/library_organized.html new file mode 100644 index 0000000..4a59b23 --- /dev/null +++ b/templates/library_organized.html @@ -0,0 +1,373 @@ + + + + + + 문서 라이브러리 - AI 번역 시스템 + + + +
    +
    +

    📚 문서 라이브러리

    +

    체계적으로 분류된 다국어 번역 문서

    +
    + +
    +
    + 📤 업로드 + 📚 라이브러리 + 📊 시스템 +
    +
    + +
    +
    + 📄 + 총 {{ total_documents }}개 문서 +
    +
    + 🗂️ + {{ categories|length }}개 카테고리 +
    +
    + 🌍 + 다국어 지원 +
    +
    + +
    +
    + + +
    + + {% if categories %} + {% for category_name, documents in categories.items() %} +
    +
    +
    + {% if category_name == "English To Korean" %} + 🇺🇸→🇰🇷 English to Korean + {% elif category_name == "Japanese To Korean" %} + 🇯🇵→🇰🇷 Japanese to Korean + {% elif category_name == "Korean Only" %} + 🇰🇷 Korean Documents + {% else %} + {{ category_name }} + {% endif %} +
    +
    {{ documents|length }}개
    +
    + +
    + {% for doc in documents %} +
    +
    {{ doc.language_display }}
    +
    {{ doc.name }}
    +
    +
    + 📅 + {{ doc.created }} +
    +
    + 📊 + {{ doc.size }} +
    +
    + 📁 + {{ doc.month }} +
    +
    + 🌐 + {{ doc.language_type.replace("-", " ")|title }} +
    +
    +
    + 👁️ 보기 + 📥 다운로드 +
    +
    + {% endfor %} +
    +
    + {% endfor %} + {% else %} +
    +
    📚
    +
    아직 변환된 문서가 없습니다
    +
    PDF나 텍스트 파일을 업로드하여 다국어 문서를 만들어보세요!
    + 📤 첫 번째 문서 업로드 +
    + {% endif %} +
    +
    + + + +