Python基于自然语言处理开发文本摘要系统

def textrank_summarize(text, ratio=0.2):
    """
    使用textrank算法生成文本摘要
    
    参数:
        text (str): 输入文本
        ratio (float): 摘要占原文比例
        
    返回:
        str: 生成的摘要
    """
    # 文本预处理
    sentences = text_to_sentences(text)
    
    # 构建句子相似度矩阵
    similarity_matrix = build_similarity_matrix(sentences)
    
    # 使用networkx库计算textrank得分
    import networkx as nx
    nx_graph = nx.from_numpy_array(similarity_matrix)
    scores = nx.pagerank(nx_graph)
    
    # 根据得分选择重要句子
    ranked_sentences = sorted(((scores[i], s) for i, s in enumerate(sentences)), reverse=true)
    
    # 根据比例选择句子数量
    select_length = int(len(sentences) * ratio)
    
    # 按原文顺序排列选中的句子
    selected_sentences = sorted(
        [ranked_sentences[i][1] for i in range(select_length)],
        key=lambda s: sentences.index(s))
    
    # 生成摘要
    summary = ' '.join(selected_sentences)
    
    return summary

import torch
import torch.nn as nn
import torch.optim as optim

class encoder(nn.module):
    def __init__(self, input_dim, emb_dim, hid_dim, n_layers, dropout):
        super().__init__()
        self.embedding = nn.embedding(input_dim, emb_dim)
        self.rnn = nn.lstm(emb_dim, hid_dim, n_layers, dropout=dropout)
        self.dropout = nn.dropout(dropout)
        
    def forward(self, src):
        # src = [src_len, batch_size]
        embedded = self.dropout(self.embedding(src))
        # embedded = [src_len, batch_size, emb_dim]
        outputs, (hidden, cell) = self.rnn(embedded)
        # outputs = [src_len, batch_size, hid_dim * n_directions]
        # hidden = [n_layers * n_directions, batch_size, hid_dim]
        # cell = [n_layers * n_directions, batch_size, hid_dim]
        return hidden, cell

class decoder(nn.module):
    def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout):
        super().__init__()
        self.output_dim = output_dim
        self.embedding = nn.embedding(output_dim, emb_dim)
        self.rnn = nn.lstm(emb_dim, hid_dim, n_layers, dropout=dropout)
        self.fc_out = nn.linear(hid_dim, output_dim)
        self.dropout = nn.dropout(dropout)
        
    def forward(self, input, hidden, cell):
        # input = [batch_size]
        # hidden = [n_layers * n_directions, batch_size, hid_dim]
        # cell = [n_layers * n_directions, batch_size, hid_dim]
        
        input = input.unsqueeze(0)
        # input = [1, batch_size]
        
        embedded = self.dropout(self.embedding(input))
        # embedded = [1, batch_size, emb_dim]
        
        output, (hidden, cell) = self.rnn(embedded, (hidden, cell))
        # output = [1, batch_size, hid_dim * n_directions]
        # hidden = [n_layers * n_directions, batch_size, hid_dim]
        # cell = [n_layers * n_directions, batch_size, hid_dim]
        
        prediction = self.fc_out(output.squeeze(0))
        # prediction = [batch_size, output_dim]
        
        return prediction, hidden, cell

​​​​​​​class seq2seq(nn.module):
    def __init__(self, encoder, decoder, device):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder
        self.device = device
        
    def forward(self, src, trg, teacher_forcing_ratio=0.5):
        # src = [src_len, batch_size]
        # trg = [trg_len, batch_size]
        
        batch_size = trg.shape[1]
        trg_len = trg.shape[0]
        trg_vocab_size = self.decoder.output_dim
        
        # 存储每一步的预测结果
        outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device)
        
        # 编码器前向传播
        hidden, cell = self.encoder(src)
        
        # 第一个输入是<sos>标记
        input = trg[0,:]
        
        for t in range(1, trg_len):
            # 解码器前向传播
            output, hidden, cell = self.decoder(input, hidden, cell)
            
            # 存储预测结果
            outputs[t] = output
            
            # 决定是否使用teacher forcing
            teacher_force = random.random() < teacher_forcing_ratio
            
            # 获取最可能的词
            top1 = output.argmax(1)
            
            # 如果使用teacher forcing，则下一个输入是真实标签
            # 否则使用模型预测结果
            input = trg[t] if teacher_force else top1
            
        return outputs

from transformers import pipeline

​​​​​​​def transformer_summarize(text, max_length=150, min_length=30):
    """
    使用预训练的transformer模型生成摘要
    
    参数:
        text (str): 输入文本
        max_length (int): 摘要最大长度
        min_length (int): 摘要最小长度
        
    返回:
        str: 生成的摘要
    """
    # 初始化摘要pipeline
    summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
    
    # 生成摘要
    summary = summarizer(text, max_length=max_length, min_length=min_length, do_sample=false)
    
    return summary[0]['summary_text']

from flask import flask, render_template, request, jsonify
from werkzeug.utils import secure_filename
import os
from summarizer import textranksummarizer, seq2seqsummarizer, transformersummarizer
from file_processor import process_file

app = flask(__name__)
app.config['upload_folder'] = 'uploads/'
app.config['max_content_length'] = 16 * 1024 * 1024  # 限制上传文件大小为16mb

# 确保上传目录存在
os.makedirs(app.config['upload_folder'], exist_ok=true)

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/summarize', methods=['post'])
def summarize():
    # 获取参数
    text = request.form.get('text', '')
    file = request.files.get('file')
    method = request.form.get('method', 'textrank')
    ratio = float(request.form.get('ratio', 0.2))
    max_length = int(request.form.get('max_length', 150))
    min_length = int(request.form.get('min_length', 30))
    
    # 如果上传了文件，处理文件内容
    if file and file.filename != '':
        filename = secure_filename(file.filename)
        file_path = os.path.join(app.config['upload_folder'], filename)
        file.save(file_path)
        text = process_file(file_path)
        os.remove(file_path)  # 处理完成后删除文件
    
    # 检查文本是否为空
    if not text:
        return jsonify({'error': '请提供文本内容或上传文件'}), 400
    
    # 根据选择的方法生成摘要
    if method == 'textrank':
        summarizer = textranksummarizer()
        summary = summarizer.summarize(text, ratio=ratio)
    elif method == 'seq2seq':
        summarizer = seq2seqsummarizer()
        summary = summarizer.summarize(text, max_length=max_length)
    elif method == 'transformer':
        summarizer = transformersummarizer()
        summary = summarizer.summarize(text, max_length=max_length, min_length=min_length)
    else:
        return jsonify({'error': '不支持的摘要方法'}), 400
    
    return jsonify({'summary': summary})

​​​​​​​if __name__ == '__main__':
    app.run(debug=true)

import os
import pypdf2
import docx
from bs4 import beautifulsoup

def process_file(file_path):
    """
    根据文件类型处理文件，提取文本内容
    
    参数:
        file_path (str): 文件路径
        
    返回:
        str: 提取的文本内容
    """
    file_ext = os.path.splitext(file_path)[1].lower()
    
    if file_ext == '.txt':
        return process_txt(file_path)
    elif file_ext == '.pdf':
        return process_pdf(file_path)
    elif file_ext == '.docx':
        return process_docx(file_path)
    elif file_ext in ['.html', '.htm']:
        return process_html(file_path)
    else:
        raise valueerror(f"不支持的文件类型: {file_ext}")

def process_txt(file_path):
    """处理txt文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return f.read()

def process_pdf(file_path):
    """处理pdf文件"""
    text = ""
    with open(file_path, 'rb') as f:
        pdf_reader = pypdf2.pdfreader(f)
        for page_num in range(len(pdf_reader.pages)):
            page = pdf_reader.pages[page_num]
            text += page.extract_text()
    return text

def process_docx(file_path):
    """处理docx文件"""
    doc = docx.document(file_path)
    text = ""
    for para in doc.paragraphs:
        text += para.text + "\n"
    return text

​​​​​​​def process_html(file_path):
    """处理html文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        soup = beautifulsoup(f.read(), 'html.parser')
        # 去除script和style元素
        for script in soup(["script", "style"]):
            script.extract()
        # 获取文本
        text = soup.get_text()
        # 处理多余的空白字符
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
        text = '\n'.join(chunk for chunk in chunks if chunk)
    return text

git clone https://github.com/username/text-summarization-system.git
cd text-summarization-system

python -m venv venv
source venv/bin/activate  # linux/macos
venv\scripts\activate  # windows

pip install -r requirements.txt

python download_models.py

python app.py

python summarize.py --input input.txt --method transformer --output summary.txt

import requests

url = "http://localhost:5000/summarize"
data = {
    "text": "这是一段需要摘要的长文本...",
    "method": "transformer",
    "max_length": 150,
    "min_length": 30
}

response = requests.post(url, data=data)
summary = response.json()["summary"]
print(summary)