128K 长文本处理实战：腾讯混元 + 云函数 SCF 构建 PDF 摘要生成器

import fitz  # PyMuPDF 的别名

def extract_text_from_pdf(pdf_path):
    """
    从 PDF 中提取纯文本内容
    :param pdf_path: PDF 文件路径
    :return: 提取的完整文本字符串
    """
    text = ""
    doc = fitz.open(pdf_path)
    for page in doc:
        text += page.get_text()
    return text

def extract_text_with_progress(pdf_path, chunk_size=1024*1024):
    """
    分块提取 PDF 文本，避免内存过载
    :param pdf_path: PDF 文件路径
    :param chunk_size: 每次处理的文本块大小（字节）
    :return: 文本生成器，逐块输出内容
    """
    doc = fitz.open(pdf_path)
    total_pages = len(doc)
    extracted_size = 0
    for page_num, page in enumerate(doc):
        text_block = page.get_text()
        # 按指定块大小分割文本
        while len(text_block) > chunk_size:
            yield text_block[:chunk_size]
            text_block = text_block[chunk_size:]
            extracted_size += chunk_size
            print(f"已提取 {extracted_size / (1024*1024):.2f} MB，进度：{page_num + 1}/{total_pages} 页")
        yield text_block
        extracted_size += len(text_block)
        print(f"已提取 {extracted_size / (1024*1024):.2f} MB，进度：{page_num + 1}/{total_pages} 页")

import re

def preprocess_text(raw_text):
    """
    对提取的文本进行清理与标准化
    :param raw_text: 原始文本字符串
    :return: 清洗后的文本
    """
    # 去除多余空白字符
    cleaned_text = re.sub(r'\s+', ' ', raw_text)
    # 移除常见 PDF 特殊控制字符
    cleaned_text = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F-\x9F]', '', cleaned_text)
    # 纠正常见编码错误（例如将某些误编码的拉丁字符转回正确形式）
    cleaned_text = cleaned_text.encode('latin1', errors='ignore').decode('utf-8', errors='ignore')
    return cleaned_text

import spacy

# 加载小型英文分词模型（可根据实际语言更换）
nlp = spacy.load("en_core_web_sm")

def semantic_chunking(text, max_length):
    """
    基于语义单元进行文本分块
    :param text: 预处理后的完整文本
    :param max_length: 每块最大允许长度（字符数）
    :return: 分块后的文本列表
    """
    chunks = []
    current_chunk = ""
    doc = nlp(text)
    for sent in doc.sents:  # 按句子迭代
        sent_text = sent.text.strip()
        if len(current_chunk) + len(sent_text) + 1 <= max_length:
            # 当前句子可以加入当前块
            if current_chunk:
                current_chunk += " " + sent_text
            else:
                current_chunk = sent_text
        else:
            # 当前块已满，存储并开始新块
            chunks.append(current_chunk)
            current_chunk = sent_text
    if current_chunk:
        chunks.append(current_chunk)
    return chunks

import requests
import json

def call_hunyuan_api(api_key, api_secret, text_chunk, max_summary_length=200):
    """
    调用混元大模型生成摘要
    :param api_key: 模型访问密钥
    :param api_secret: 模型访问密钥
    :param text_chunk: 待摘要文本块
    :param max_summary_length: 摘要最大长度（字符数）
    :return: 生成的摘要文本
    """
    api_url = "https://hunyuan-api.tencent-cloud.com/v1/summarize"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}:{api_secret}"
    }
    payload = {
        "text": text_chunk,
        "max_length": max_summary_length
    }
    response = requests.post(api_url, headers=headers, data=json.dumps(payload))
    if response.status_code == 200:
        result = response.json()
        if result["code"] == 0:
            return result["summary"]
        else:
            print(f"API 调用错误：{result['message']}")
            return None
    else:
        print(f"HTTP 错误：{response.status_code}")
        return None

# main.py - 云函数入口文件

import os
from extraction import extract_text_with_progress, preprocess_text
from semantic_chunking import semantic_chunking
from hunyuan_integration import call_hunyuan_api

def main_handler(event, context):
    """
    云函数入口函数
    :param event: 触发事件数据（包含 PDF 文件存储路径等）
    :param context: 运行时上下文
    :return: 摘要结果或错误信息
    """
    try:
        # 从事件获取 PDF 文件路径
        pdf_path = event["pdf_path"]
        # 步骤 1：文本提取
        raw_text_generator = extract_text_with_progress(pdf_path)
        full_raw_text = ""
        for text_chunk in raw_text_generator:
            full_raw_text += text_chunk
        # 步骤 2：文本预处理
        cleaned_text = preprocess_text(full_raw_text)
        # 步骤 3：语义分块
        text_chunks = semantic_chunking(cleaned_text, max_length=3000)
        # 步骤 4：调用混元生成摘要
        all_summaries = []
        for chunk in text_chunks:
            summary = call_hunyuan_api(
                api_key=os.environ["HUNYUAN_API_KEY"],
                api_secret=os.environ["HUNYUAN_API_SECRET"],
                text_chunk=chunk
            )
            if summary:
                all_summaries.append(summary)
        # 合并所有摘要块
        final_summary = " ".join(all_summaries)
        return {
            "statusCode": 200,
            "summary": final_summary
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "error": str(e)
        }

pymupdf==1.20.0
spacy==3.5.1
requests==2.28.1

POST /generate-summary
{
    "pdf_url": "https://example-bucket.cos.ap-guangzhou.myqcloud.com/sample.pdf"
}

HTTP/1.1 200 OK
{
    "summary": "本文档主要讨论了……",
    "processing_time": "2.45s",
    "confidence_score": 0.93
}