"""
PPT 工具类 - 处理 PPT 到 PDF 和 PDF 到图片的转换。
此工具类专注于文件格式转换功能,只使用 pypdfium2 进行图片转换。
"""
import hashlib
import subprocess
import base64
from pathlib import Path
from typing import List
from io import BytesIO
import logging
# PDF 处理 - 只使用 pypdfium2
try:
import pypdfium2 as pdfium
PDFIUM_AVAILABLE = True
except ImportError:
PDFIUM_AVAILABLE = False
logger = logging.getLogger(__name__)
class PPTUtils:
"""PPT 文件操作的工具类。"""
def __init__(self, cache_dir: str = "./cache"):
"""初始化PPT工具类,设置缓存目录。
Args:
cache_dir: 存储缓存文件的目录
"""
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
# 为不同文件类型创建子目录
self.pdf_cache_dir = self.cache_dir / "pdf"
self.image_cache_dir = self.cache_dir / "images"
self.pdf_cache_dir.mkdir(exist_ok=True)
self.image_cache_dir.mkdir(exist_ok=True)
def _get_file_hash(self, file_path: str) -> str:
"""计算文件的哈希值用于缓存。"""
with open(file_path, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
def ppt_to_pdf(self, ppt_path: str) -> str:
"""将PPT文件转换为PDF。
Args:
ppt_path: PPT文件路径
Returns:
生成的PDF文件路径
Raises:
FileNotFoundError: 如果PPT文件不存在
RuntimeError: 如果转换失败
"""
ppt_path = Path(ppt_path)
if not ppt_path.exists():
raise FileNotFoundError(f"未找到PPT文件: {ppt_path}")
# 计算哈希值用于缓存
file_hash = self._get_file_hash(str(ppt_path))
pdf_filename = f"{ppt_path.stem}_{file_hash}.pdf"
pdf_path = self.pdf_cache_dir / pdf_filename
logger.info(f"转换PPT到PDF: {ppt_path} → {pdf_path}")
# 如果存在缓存的PDF则返回
if pdf_path.exists():
logger.info(f"使用缓存的PDF: {pdf_path}")
return str(pdf_path)
# 使用LibreOffice将PPT转换为PDF
try:
logger.info(f"正在将PPT转换为PDF: {ppt_path}")
# 使用LibreOffice将PPT转换为PDF
cmd = [
"/Applications/LibreOffice.app/Contents/MacOS/soffice",
"--headless",
"--convert-to", "pdf",
"--outdir", str(self.pdf_cache_dir),
str(ppt_path)
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
if result.returncode != 0:
raise RuntimeError(f"LibreOffice转换失败: {result.stderr}")
# LibreOffice创建的PDF使用原始文件名
original_pdf = self.pdf_cache_dir / f"{ppt_path.stem}.pdf"
# 重命名以包含哈希值
if original_pdf.exists():
original_pdf.rename(pdf_path)
logger.info(f"PPT已转换为PDF: {pdf_path}")
return str(pdf_path)
else:
raise RuntimeError("LibreOffice未创建PDF文件")
except subprocess.TimeoutExpired:
raise RuntimeError("PPT转PDF转换超时")
except Exception as e:
raise RuntimeError(f"PPT转PDF转换失败: {str(e)}")
def pdf_to_images(self, pdf_path: str, dpi: int = 200) -> List[str]:
"""使用pypdfium2将PDF转换为图片。
Args:
pdf_path: PDF文件路径
dpi: 图片转换分辨率(转换为scale参数)
Returns:
生成的图片文件路径列表
Raises:
FileNotFoundError: 如果PDF文件不存在
RuntimeError: 如果转换失败或缺少pypdfium2
"""
if not PDFIUM_AVAILABLE:
raise RuntimeError("pypdfium2 未安装。请运行: pip install pypdfium2")
pdf_path = Path(pdf_path)
if not pdf_path.exists():
raise FileNotFoundError(f"未找到PDF文件: {pdf_path}")
# 将DPI转换为scale参数
scale = dpi / 50.0
# 检查图片是否已存在
image_pattern = f"{pdf_path.stem}_page_*.jpg"
existing_images = list(self.image_cache_dir.glob(image_pattern))
if existing_images:
logger.info(f"使用缓存的图片: {len(existing_images)} 页")
return sorted([str(img) for img in existing_images])
try:
logger.info(f"正在使用pypdfium2将PDF转换为图片: {pdf_path}")
# 打开PDF文档
pdf = pdfium.PdfDocument(str(pdf_path))
image_paths = []
# 转换每一页
for i in range(len(pdf)):
page = pdf[i]
image = page.render(scale=scale).to_pil()
# 生成图片文件名
image_filename = f"{pdf_path.stem}_page_{i+1:03d}.jpg"
image_path = self.image_cache_dir / image_filename
# 保存图片
image.save(str(image_path), 'JPEG', quality=95)
image_paths.append(str(image_path))
logger.info(f"PDF已转换为 {len(image_paths)} 张图片")
return image_paths
except Exception as e:
raise RuntimeError(f"PDF转图片转换失败: {str(e)}")
def ppt_to_images(self, ppt_path: str, dpi: int = 200) -> List[str]:
"""直接将PPT转换为图片 (PPT → PDF → 图片)。
Args:
ppt_path: PPT文件路径
dpi: 图片转换分辨率
Returns:
生成的图片文件路径列表
"""
# 首先将PPT转换为PDF
pdf_path = self.ppt_to_pdf(ppt_path)
# 然后将PDF转换为图片
return self.pdf_to_images(pdf_path, dpi)
def get_slide_count_from_pdf(self, pdf_path: str) -> int:
"""从PDF文件中获取页数(替代PPT幻灯片数量)。
Args:
pdf_path: PDF文件路径
Returns:
PDF页数
"""
if not PDFIUM_AVAILABLE:
logger.warning("pypdfium2 未安装,无法获取页数")
return 0
try:
pdf = pdfium.PdfDocument(str(pdf_path))
return len(pdf)
except Exception as e:
logger.error(f"获取PDF页数失败: {e}")
return 0
def cleanup_cache(self, max_age_days: int = 30):
"""清理旧的缓存文件。
Args:
max_age_days: 缓存文件的最大保留天数
"""
import time
cutoff_time = time.time() - (max_age_days * 24 * 60 * 60)
for cache_subdir in [self.pdf_cache_dir, self.image_cache_dir]:
for file_path in cache_subdir.iterdir():
if file_path.is_file() and file_path.stat().st_mtime < cutoff_time:
try:
file_path.unlink()
logger.info(f"已清理旧缓存文件: {file_path}")
except Exception as e:
logger.error(f"清理失败 {file_path}: {e}")
def get_pdf_images_base64(self, pdf_path: str, scale: float = 4.0) -> List[str]:
"""
将PDF转换为base64编码的图片列表(用于文档理解)
Args:
pdf_path: PDF文件路径
scale: 渲染缩放比例,数值越大图片质量越高
Returns:
base64编码的图片字符串列表
Raises:
FileNotFoundError: 如果PDF文件不存在
RuntimeError: 如果转换失败或缺少pypdfium2
"""
if not PDFIUM_AVAILABLE:
raise RuntimeError("pypdfium2 未安装。请运行: pip install pypdfium2")
pdf_path = Path(pdf_path)
if not pdf_path.exists():
raise FileNotFoundError(f"未找到PDF文件: {pdf_path}")
try:
logger.info(f"正在分析PDF文档并转换为base64图片: {pdf_path}")
# 打开PDF文档
pdf = pdfium.PdfDocument(str(pdf_path))
images = []
# 转换每一页为图片并编码为base64
for i in range(len(pdf)):
page = pdf[i]
image = page.render(scale=scale).to_pil()
buffered = BytesIO()
image.save(buffered, format="JPEG")
img_byte = buffered.getvalue()
img_base64 = base64.b64encode(img_byte).decode("utf-8")
images.append(img_base64)
logger.info(f"PDF已转换为 {len(images)} 张base64编码图片")
return images
except Exception as e:
raise RuntimeError(f"PDF文档分析失败: {str(e)}")