final

app.py

@@ -0,0 +1,407 @@
+import os
+import zipfile
+import requests
+import gradio as gr
+import whisper
+import subprocess
+import uuid
+import torch
+import re
+import matplotlib.pyplot as plt
+import language_tool_python
+import difflib
+from transformers import (
+    AutoTokenizer,
+    AutoModelForSeq2SeqLM,
+    pipeline as hf_pipeline,
+)
+
+# ────────────────────────────────────────────────────────────────
+# Optional evaluation libraries
+try:
+    from rouge_score import rouge_scorer
+except ImportError:
+    rouge_scorer = None
+    print("[Warning] rouge_score 패키지가 없습니다. pip install rouge-score")
+
+try:
+    from bert_score import score as bert_score_func
+except ImportError:
+    bert_score_func = None
+    print("[Warning] bert-score 패키지가 없습니다. pip install bert-score")
+
+# ────────────────────────────────────────────────────────────────
+# 한글 맞춤법 검사(py‑hanspell)
+try:
+    from hanspell import spell_checker
+except ImportError:
+    spell_checker = None
+
+# ────────────────────────────────────────────────────────────────
+# LanguageTool 룰 기반 교정 (영어 전용)
+try:
+    lt_tool = language_tool_python.LanguageTool('en-US')
+except Exception as e:
+    lt_tool = None
+    print(f"[Warning] LanguageTool 초기화 실패: {e}")
+
+# ────────────────────────────────────────────────────────────────
+# FFmpeg
+yt_dlp_path = "C:/Windows/System32/yt-dlp.exe"  
+ffmpeg_path = "C:/ProgramData/chocolatey/bin" 
+def download_ffmpeg(dest_bin):
+    if os.path.isdir(dest_bin) and os.path.isfile(os.path.join(dest_bin, "ffmpeg.exe")):
+        return dest_bin
+    url = "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip"
+    zip_path = os.path.join(os.getcwd(), "ffmpeg.zip")
+    extract_root = os.path.dirname(dest_bin)
+    os.makedirs(extract_root, exist_ok=True)
+    resp = requests.get(url, stream=True); resp.raise_for_status()
+    with open(zip_path, "wb") as f:
+        for chunk in resp.iter_content(8192): f.write(chunk)
+    with zipfile.ZipFile(zip_path, "r") as zf: zf.extractall(extract_root)
+    os.remove(zip_path)
+    for root, _, files in os.walk(extract_root):
+        if "ffmpeg.exe" in files:
+            os.makedirs(dest_bin, exist_ok=True)
+            for fn in ("ffmpeg.exe","ffprobe.exe","ffplay.exe"):
+                src, dst = os.path.join(root,fn), os.path.join(dest_bin,fn)
+                if os.path.isfile(src): os.replace(src, dst)
+            return dest_bin
+    raise RuntimeError("FFmpeg 설치 실패")
+
+download_ffmpeg(ffmpeg_path)
+os.environ["PATH"] = ffmpeg_path + os.pathsep + os.environ.get("PATH","")
+
+# ────────────────────────────────────────────────────────────────
+# Whisper
+asr_model = whisper.load_model("medium")
+
+# ────────────────────────────────────────────────────────────────
+# 요약 모델(모델/토크나이저 직접 사용, pipeline X)
+SUMMARY_MODELS = {
+    "mT5_multilingual_XLSum": "csebuetnlp/mT5_multilingual_XLSum",
+    "Pegasus XSum":           "google/pegasus-xsum",
+    "BART-large CNN":         "facebook/bart-large-cnn",
+    "DistilBART CNN":         "sshleifer/distilbart-cnn-12-6"
+}
+tokenizers, models = {}, {}
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+def load_summarizer(label: str):
+    if label in models:
+        return
+    repo = SUMMARY_MODELS[label]
+    tok = AutoTokenizer.from_pretrained(repo, use_fast=False)
+    model = AutoModelForSeq2SeqLM.from_pretrained(repo).to(device)
+    model.eval()
+    tokenizers[label] = tok
+    models[label]     = model
+
+if rouge_scorer:
+    scorer = rouge_scorer.RougeScorer(["rouge1","rouge2","rougeL"], use_stemmer=True)
+
+# ────────────────────────────────────────────────────────────────
+# 문법 교정
+GRAMMAR_MODELS = {
+    "LanguageTool-en": None,
+    "py-hanspell":     None,
+    "GEC-한국어":      "Soyoung97/gec_kr"
+}
+grammar_pipes = {}
+
+def load_grammar_pipe(name: str):
+    repo = GRAMMAR_MODELS[name]
+    grammar_pipes[name] = hf_pipeline(
+        "text2text-generation",
+        model=repo,
+        tokenizer=AutoTokenizer.from_pretrained(repo),
+        device=0 if torch.cuda.is_available() else -1
+    )
+
+def correct_spelling(text, max_chunk=500):
+    if not spell_checker: return text
+    parts, curr = re.split(r'([.?!]\s*)', text), ""
+    segs, out = [], []
+    for p in parts:
+        if len(curr)+len(p) <= max_chunk: curr += p
+        else: segs.append(curr); curr = p
+    if curr: segs.append(curr)
+    for s in segs:
+        try: out.append(spell_checker.check(s).checked)
+        except: out.append(s)
+    return " ".join(o.strip() for o in out)
+
+def correct_text(text, method="GEC-한국어"):
+    if method=="py-hanspell":
+        return correct_spelling(text)
+    if method=="LanguageTool-en" and lt_tool:
+        matches = lt_tool.check(text)
+        return language_tool_python.utils.correct(text, matches)
+    if method=="GEC-한국어":
+        if method not in grammar_pipes:
+            load_grammar_pipe(method)
+        pipe = grammar_pipes[method]
+        sents = re.split(r'(?<=[.?!])\s+', text)
+        corrected=[]
+        for sent in sents:
+            gen = pipe(sent, max_length=256, min_length=1, do_sample=False)[0]["generated_text"]
+            corrected.append(gen.strip())
+        return " ".join(corrected)
+    return text
+
+# ────────────────────────────────────────────────────────────────
+# 교정률 + Diff
+def calculate_correction_rate(original, corrected):
+    orig_tokens = original.split()
+    corr_tokens = corrected.split()
+    sm = difflib.SequenceMatcher(None, orig_tokens, corr_tokens)
+    diff_count = sum((i2 - i1) for tag, i1, i2, j1, j2 in sm.get_opcodes() if tag != 'equal')
+    total = max(len(orig_tokens), 1)
+    return round(100 * diff_count / total, 2)
+
+def highlight_diff(original: str, corrected: str) -> str:
+    diff = difflib.ndiff(original.split(), corrected.split())
+    html_parts = []
+    for token in diff:
+        if token.startswith("+ "):
+            html_parts.append(f"<span style='color:red;'>{token[2:]}</span>")
+        elif token.startswith("- "):
+            continue
+        else:
+            html_parts.append(token[2:])
+    return " ".join(html_parts)
+
+# ────────────────────────────────────────────────────────────────
+# YouTube
+def download_audio(url):
+    fname = f"yt_{uuid.uuid4().hex[:8]}.mp3"
+    cmd = [yt_dlp_path,"-f","bestaudio","--extract-audio","--audio-format","mp3","-o",fname,url]
+    res = subprocess.run(cmd, capture_output=True, text=True)
+    if res.returncode!=0: raise RuntimeError(res.stderr)
+    return fname
+
+def get_transcript(url, state):
+    if state and state.get("url")==url:
+        return state["orig"], state
+    audio = download_audio(url)
+    res   = asr_model.transcribe(audio)
+    orig  = res.get("text","")
+    os.remove(audio)
+    return orig, {"url":url, "orig":orig}
+
+# ────────────────────────────────────────────────────────────────
+# 안전한 청크 요약 (model.generate 직접 호출)
+def summarize_long_text(text: str, label: str, chunk_size: int = 512) -> str:
+    load_summarizer(label)
+    tok  = tokenizers[label]
+    model= models[label]
+
+    enc = tok(text, return_tensors="pt", truncation=False)
+    ids = enc.input_ids[0]
+    summaries = []
+
+    max_ctx = getattr(model.config, "max_position_embeddings", 1024) - 4
+    chunk_size = min(chunk_size, max_ctx)
+
+    for i in range(0, len(ids), chunk_size):
+        chunk_ids = ids[i:i+chunk_size].unsqueeze(0).to(device)
+        out_ids = model.generate(
+            chunk_ids,
+            max_new_tokens=128,
+            num_beams=4,
+            do_sample=False
+        )
+        summ = tok.decode(out_ids[0], skip_special_tokens=True)
+        summaries.append(summ)
+
+    combined = " ".join(summaries)
+    enc2 = tok(combined, return_tensors="pt", truncation=True, max_length=max_ctx).to(device)
+    out_ids = model.generate(
+        **enc2,
+        max_new_tokens=128,
+        num_beams=4,
+        do_sample=False
+    )
+    final = tok.decode(out_ids[0], skip_special_tokens=True)
+    return final
+
+# ────────────────────────────────────────────────────────────────
+def summarize_single(url, label, grammar_method, transcript_state):
+    orig, new_state = get_transcript(url, transcript_state)
+    corr = correct_text(orig, method=grammar_method)
+    corr_rate = calculate_correction_rate(orig, corr)
+    corr_html = f"<div><b>교정률:</b> {corr_rate}%</div><hr/>{highlight_diff(orig, corr)}"
+
+    summary = summarize_long_text(corr, label) if len(corr) > 100 else "⚠️ 요약 불가"
+
+    rouge_vals=[0,0,0]
+    if rouge_scorer and summary.strip():
+        sc = scorer.score(orig, summary)
+        rouge_vals=[sc["rouge1"].fmeasure, sc["rouge2"].fmeasure, sc["rougeL"].fmeasure]
+
+    bert_f1=0
+    if bert_score_func and summary.strip():
+        try:
+            _,_,F = bert_score_func([summary],[orig],lang="ko")
+        except Exception:
+            _,_,F = bert_score_func([summary],[orig],lang="en")
+        bert_f1=float(F.mean())
+
+    fig,ax=plt.subplots()
+    ax.bar(["R1","R2","RL","BERT-F1"], rouge_vals+[bert_f1])
+    ax.set_ylim(0,1); ax.set_ylabel("Score"); ax.set_title("Summary Fidelity")
+    plt.tight_layout()
+
+    return orig, corr_html, summary, fig, new_state
+
+# ────────────────────────────────────────────────────────────────
+def summarize_all(url, grammar_method, transcript_state):
+    orig, new_state = get_transcript(url, transcript_state)
+    corr = correct_text(orig, method=grammar_method)
+    corr_rate = calculate_correction_rate(orig, corr)
+    corr_html = f"<div><b>교정률:</b> {corr_rate}%</div><hr/>{highlight_diff(orig, corr)}"
+
+    figs, interps, rv_list, bf_list = [], [], [], []
+    summaries_plain = []
+    labels = list(SUMMARY_MODELS.keys())
+
+    for label in labels:
+        summ = summarize_long_text(corr, label)
+        summaries_plain.append(summ)
+
+        rv=[0,0,0]; bf=0
+        if rouge_scorer:
+            sc = scorer.score(orig, summ)
+            rv=[sc["rouge1"].fmeasure, sc["rouge2"].fmeasure, sc["rougeL"].fmeasure]
+        if bert_score_func:
+            try:
+                _,_,F = bert_score_func([summ],[orig],lang="ko")
+            except Exception:
+                _,_,F = bert_score_func([summ],[orig],lang="en")
+            bf=float(F.mean())
+        rv_list.append(rv); bf_list.append(bf)
+
+        fig,ax=plt.subplots()
+        ax.bar(["R1","R2","RL","BERT-F1"], rv+[bf])
+        ax.set_ylim(0,1); ax.set_title(label)
+        plt.tight_layout(); figs.append(fig)
+
+        note="정보 손실 많음"
+        if bf>0.8: note="핵심 정보 잘 반영"
+        elif bf>0.5: note="주요 내용 포함"
+        interps.append(f"{label}: {note} (F1={bf:.2f})")
+
+    html = "<h3>모델별 요약 & Fidelity Metrics</h3>"
+    html+= f"<p><b>교정률:</b> {corr_rate}%</p>"
+    html+= "<table border='1' style='border-collapse:collapse; width:100%; table-layout:fixed;'>"
+    html+= "<tr><th style='width:12%'>모델</th><th style='width:58%'>요약문</th><th style='width:5%'>R1</th><th style='width:5%'>R2</th><th style='width:5%'>RL</th><th style='width:7%'>BERT-F1</th><th style='width:8%'>해석</th></tr>"
+
+    for i,label in enumerate(labels):
+        r1,r2,rl = rv_list[i]
+        bf       = bf_list[i]
+        note     = "정보 손실 많음"
+        if bf>0.8: note="핵심 정보 잘 반영"
+        elif bf>0.5: note="주요 내용 포함"
+
+        summ_html = summaries_plain[i].replace("<", "&lt;")
+        html+= (
+            f"<tr>"
+            f"<td>{label}</td>"
+            f"<td style='white-space:pre-wrap; word-break:break-word'>{summ_html}</td>"
+            f"<td>{r1:.2f}</td><td>{r2:.2f}</td><td>{rl:.2f}</td>"
+            f"<td>{bf:.2f}</td><td>{note}</td>"
+            f"</tr>"
+        )
+    html+="</table>"
+
+    return [orig, corr_html] + figs + interps + [html, new_state]
+
+# ────────────────────────────────────────────────────────────────
+def save_summary(url, label):
+    orig, _ = get_transcript(url, None)
+    corr = correct_text(orig, "GEC-한국어")
+    summary = summarize_long_text(corr, label)
+    path = os.path.join(os.getcwd(), f"summary_{label}.txt")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(summary)
+    return path
+
+# ────────────────────────────────────────────────────────────────
+# CSS (교정 자막을 박스처럼 보이게)
+CUSTOM_CSS = """
+#corr_box, #corr_box_all {
+  border: 1px solid #ccc;
+  padding: 10px;
+  border-radius: 6px;
+  background-color: #fafafa;
+  max-height: 300px;
+  overflow-y: auto;
+  white-space: pre-wrap;
+}
+"""
+
+# Gradio
+with gr.Blocks(css=CUSTOM_CSS) as demo:
+    gr.Markdown("## 🎬 YouTube 요약 서비스 (교정 + 교정률 + Diff 강조, 안전 청크요약)")
+
+    with gr.Tabs():
+        with gr.TabItem("단일 모델 요약"):
+            url_input        = gr.Textbox(label="YouTube URL")
+            model_sel        = gr.Dropdown(list(SUMMARY_MODELS.keys()), label="요약 모델")
+            grammar_sel      = gr.Dropdown(list(GRAMMAR_MODELS.keys()), label="교정 모델", value="GEC-한국어")
+            transcript_state = gr.State(None)
+            btn_single       = gr.Button("요약 실행")
+
+            orig_tb       = gr.Textbox(label="원문 자막", lines=10)
+            corr_tb       = gr.HTML(label="교정 자막 (변경점 강조)", elem_id="corr_box")
+            sum_tb        = gr.Textbox(label="요약 결과", lines=8)
+            fidelity_plot = gr.Plot(label="Fidelity Metrics")
+            save_btn      = gr.Button("요약 저장")
+            download_single = gr.File(label="다운로드 파일")
+
+            btn_single.click(
+                fn=summarize_single,
+                inputs=[url_input, model_sel, grammar_sel, transcript_state],
+                outputs=[orig_tb, corr_tb, sum_tb, fidelity_plot, transcript_state]
+            )
+            save_btn.click(
+                fn=save_summary,
+                inputs=[url_input, model_sel],
+                outputs=[download_single]
+            )
+
+        with gr.TabItem("전체 모델 비교"):
+            url_all              = gr.Textbox(label="YouTube URL")
+            grammar_sel_all      = gr.Dropdown(list(GRAMMAR_MODELS.keys()), label="교정 모델", value="GEC-한국어")
+            transcript_state_all = gr.State(None)
+            btn_all              = gr.Button("모두 실행")
+
+            orig_all    = gr.Textbox(label="원문 자막", lines=10)
+            corr_all    = gr.HTML(label="교정 자막 (변경점 강조)", elem_id="corr_box_all")
+
+            plot_components, interp_components = [], []
+            for label in SUMMARY_MODELS:
+                plot_components.append(gr.Plot(label=f"{label} Metrics"))
+                interp_components.append(gr.HTML(label=f"{label} 해석"))
+
+            agg_table    = gr.HTML(label="모델별 요약 & Fidelity Metrics")
+            save_all_sel = gr.Radio(list(SUMMARY_MODELS.keys()), label="저장 모델 지정")
+            save_all_btn = gr.Button("선택 요약 저장")
+            download_all = gr.File(label="다운로드 파일")
+
+            btn_all.click(
+                fn=summarize_all,
+                inputs=[url_all, grammar_sel_all, transcript_state_all],
+                outputs=[orig_all, corr_all] + plot_components + interp_components + [agg_table, transcript_state_all]
+            )
+            save_all_btn.click(
+                fn=save_summary,
+                inputs=[url_all, save_all_sel],
+                outputs=[download_all]
+            )
+
+if __name__ == '__main__':
+    # 자동 포트 할당
+    demo.launch(server_name="127.0.0.1")
+    # 혹은 완전 자동: demo.launch()

requirements.txt

@@ -0,0 +1,12 @@
+torch
+transformers
+sentencepiece
+gradio
+git+https://github.com/openai/whisper.git
+matplotlib
+requests
+uuid
+language-tool-python
+rouge-score
+bert-score
+yt-dlp