Python调用Tika实现一站式提取PDF/Word/Excel

# 下载最新版本（以2.9.1为例）
wget https://dlcdn.apache.org/tika/tika-server-2.9.1.jar

java -jar tika-server-2.9.1.jar

java -jar tika-server-2.9.1.jar --port=8080

curl http://localhost:9998/tika

# docker-compose.yml
services:
  tika:
    image: apache/tika:3.2.2.0-full
    container_name: tika-server
    ports:
      - "9998:9998"
    restart: unless-stopped

docker-compose up -d

java -jar tika-app-2.9.1.jar --text myfile.pdf

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.apache.tika</groupId>
            <artifactId>tika-bom</artifactId>
            <version>2.17.0</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <!-- 标准解析器包（必选）-->
    <dependency>
        <groupId>org.apache.tika</groupId>
        <artifactId>tika-parsers-standard-package</artifactId>
    </dependency>
</dependencies>

pip install tika

from tika import parser
# 指定Tika Server地址（如果是本地默认端口，可省略）
server_url = 'http://127.0.0.1:9998'
# 解析文件
parsed = parser.from_file('path/to/document.pdf', server_url)
# 获取文本内容
print("文本内容:\n", parsed["content"])
# 获取元数据
print("\n元数据:", parsed["metadata"])

from tika import parser
parsed = parser.from_file('report.docx')
metadata = parsed["metadata"]
print(f"文档标题: {metadata.get('title', '无')}")
print(f"作者: {metadata.get('Author', '无')}")
print(f"创建时间: {metadata.get('dcterms:created', '无')}")
print(f"修改时间: {metadata.get('Last-Modified', '无')}")
print(f"文件类型: {metadata.get('Content-Type', '无')}")
print(f"页数: {metadata.get('xmpTPg:NPages', '无')}")
print(f"文件大小: {metadata.get('Content-Length', '无')}")

from tika import parser
parsed = parser.from_url('https://example.com/document.pdf')
print(parsed["content"])

from tika import parser
with open('document.pdf', 'rb') as f:
    binary_data = f.read()
parsed = parser.from_buffer(binary_data)
print(parsed["content"])

brew install tesseract
brew install tesseract-lang  # 安装语言包

sudo apt-get install tesseract-ocr
sudo apt-get install tesseract-ocr-chi-sim  # 简体中文语言包

tesseract --version

from tika import parser
# Tika 2.x会自动检测是否为扫描件，并调用Tesseract OCR
parsed = parser.from_file('scanned_document.pdf')
print("OCR识别结果:\n", parsed["content"])

<?xml version="1.0" encoding="UTF-8"?>
<properties>
  <parsers>
    <parser class="org.apache.tika.parser.pdf.PDFParser">
      <params>
        <param name="OCR_STRATEGY" type="string">NO_OCR</param>
      </params>
    </parser>
  </parsers>
</properties>

curl -T scanned.pdf http://localhost:9998/tika \
  -H "X-Tika-PDFOcrStrategy: no_ocr"

from tika import parser
# 通过headers传递密码
headers = {
    "X-Tika-PDFPassword": "your_password"
}
parsed = parser.from_file('encrypted.pdf', headers=headers)
print(parsed["content"])

import requests
def extract_with_attachments(file_path, tika_url='http://localhost:9998'):
    """提取文档内容及其附件"""
    with open(file_path, 'rb') as f:
        files = {'file': f}
        # 使用 /rmeta 端点获取递归元数据
        response = requests.put(
            f'{tika_url}/rmeta/text',
            files=files,
            headers={'Accept': 'application/json'}
        )
    return response.json()
# 使用示例
result = extract_with_attachments('document_with_attachments.pdf')
print(result)

from tika import unpack
# 解析压缩包
parsed = unpack.from_file('archive.zip')
# 遍历压缩包内的所有文件
for file_name, content in parsed.items():
    print(f"文件: {file_name}")
    print(f"内容预览: {content[:200]}...")

from tika import parser
import json
def recursive_parse(file_path, depth=0):
    """递归解析文档及其附件"""
    parsed = parser.from_file(file_path)
    result = {
        'file': file_path,
        'depth': depth,
        'content': parsed['content'][:500] if parsed['content'] else '',
        'metadata': parsed['metadata']
    }
    # 检查是否有嵌入的附件（需要进一步处理）
    # 实际实现中可通过metadata中的embedded resources判断
    return result

version: '3.8'
services:
  tika:
    image: apache/tika:3.2.2.0-full
    container_name: tika-server
    ports:
      - "9998:9998"
    environment:
      - TIKA_MEMORY_LIMIT=2048m
      - TIKA_TIMEOUT=120
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9998/tika"]
      interval: 30s
      timeout: 10s
      retries: 3

import requests
from typing import Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
class TikaClient:
    """生产级Tika Server客户端"""
    def __init__(self, base_url: str = "http://localhost:9998", timeout: int = 120):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
    def extract_text(self, file_path: str, password: Optional[str] = None) -> str:
        """提取文本内容"""
        with open(file_path, 'rb') as f:
            headers = {}
            if password:
                headers['X-Tika-PDFPassword'] = password
            response = requests.put(
                f"{self.base_url}/tika",
                data=f,
                headers=headers,
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.text
    def extract_metadata(self, file_path: str) -> Dict[str, Any]:
        """提取元数据"""
        with open(file_path, 'rb') as f:
            response = requests.put(
                f"{self.base_url}/meta",
                data=f,
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
    def extract_with_metadata(self, file_path: str) -> Dict[str, Any]:
        """同时提取文本和元数据（推荐使用）"""
        with open(file_path, 'rb') as f:
            response = requests.put(
                f"{self.base_url}/rmeta/text",
                data=f,
                headers={'Accept': 'application/json'},
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
    def detect_type(self, file_path: str) -> str:
        """检测文件MIME类型"""
        with open(file_path, 'rb') as f:
            response = requests.put(
                f"{self.base_url}/detect/stream",
                data=f,
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.text.strip()
    def health_check(self) -> bool:
        """健康检查"""
        try:
            response = requests.get(f"{self.base_url}/tika", timeout=5)
            return response.status_code == 200
        except Exception:
            return False
# 使用示例
client = TikaClient()
if client.health_check():
    # 提取文本
    text = client.extract_text("document.pdf")
    print(f"文本长度: {len(text)} 字符")
    # 提取元数据
    metadata = client.extract_metadata("document.pdf")
    print(f"作者: {metadata.get('Author', '未知')}")

import os
import json
import argparse
from typing import Dict, List, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from tika import parser
import logging
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class BatchParser:
    """批量文档解析器"""
    # 支持的文件扩展名
    SUPPORTED_EXTENSIONS = {
        '.pdf', '.doc', '.docx', '.xls', '.xlsx', 
        '.ppt', '.pptx', '.txt', '.html', '.xml',
        '.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff'
    }
    def __init__(self, tika_server_url: str = None, max_workers: int = 4):
        self.tika_server_url = tika_server_url
        self.max_workers = max_workers
    def parse_single_file(self, file_path: str) -> Dict[str, Any]:
        """解析单个文件"""
        try:
            logger.info(f"解析: {file_path}")
            parsed = parser.from_file(file_path, self.tika_server_url)
            if parsed is None:
                return {
                    'file': file_path,
                    'error': '解析返回空结果',
                    'content': '',
                    'metadata': {}
                }
            return {
                'file': file_path,
                'content': parsed.get('content', ''),
                'metadata': parsed.get('metadata', {}),
                'error': None
            }
        except Exception as e:
            logger.error(f"解析失败: {file_path}, 错误: {str(e)}")
            return {
                'file': file_path,
                'error': str(e),
                'content': '',
                'metadata': {}
            }
    def parse_directory(self, directory: str, recursive: bool = True) -> List[Dict]:
        """批量解析目录下的所有文件"""
        results = []
        files_to_parse = []
        # 收集所有需要解析的文件
        if recursive:
            for root, _, files in os.walk(directory):
                for file in files:
                    ext = os.path.splitext(file)[1].lower()
                    if ext in self.SUPPORTED_EXTENSIONS:
                        files_to_parse.append(os.path.join(root, file))
        else:
            for file in os.listdir(directory):
                file_path = os.path.join(directory, file)
                if os.path.isfile(file_path):
                    ext = os.path.splitext(file)[1].lower()
                    if ext in self.SUPPORTED_EXTENSIONS:
                        files_to_parse.append(file_path)
        logger.info(f"共发现 {len(files_to_parse)} 个待解析文件")
        # 使用线程池并发解析
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_file = {
                executor.submit(self.parse_single_file, file_path): file_path
                for file_path in files_to_parse
            }
            for future in as_completed(future_to_file):
                result = future.result()
                results.append(result)
        return results
    def export_to_json(self, results: List[Dict], output_path: str):
        """导出结果为JSON文件"""
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        logger.info(f"结果已导出到: {output_path}")
def main():
    parser = argparse.ArgumentParser(description='批量文档解析工具')
    parser.add_argument('input', help='输入文件或目录路径')
    parser.add_argument('-o', '--output', default='parsed_results.json', 
                        help='输出JSON文件路径')
    parser.add_argument('-r', '--recursive', action='store_true', 
                        help='递归遍历子目录')
    parser.add_argument('-w', '--workers', type=int, default=4, 
                        help='并发线程数')
    parser.add_argument('--tika-server', help='Tika Server地址')
    args = parser.parse_args()
    # 创建解析器
    batch_parser = BatchParser(
        tika_server_url=args.tika_server,
        max_workers=args.workers
    )
    # 判断输入是文件还是目录
    if os.path.isfile(args.input):
        # 单文件解析
        logger.info(f"解析单个文件: {args.input}")
        result = batch_parser.parse_single_file(args.input)
        batch_parser.export_to_json([result], args.output)
    elif os.path.isdir(args.input):
        # 目录批量解析
        logger.info(f"批量解析目录: {args.input}")
        results = batch_parser.parse_directory(args.input, args.recursive)
        batch_parser.export_to_json(results, args.output)
        # 统计信息
        success_count = sum(1 for r in results if r['error'] is None)
        error_count = len(results) - success_count
        logger.info(f"解析完成: 成功 {success_count}, 失败 {error_count}")
    else:
        logger.error(f"路径不存在: {args.input}")
if __name__ == "__main__":
    main()

# 批量解析目录下所有文件
python batch_parser.py /path/to/documents -o results.json -r

# 解析单个文件
python batch_parser.py document.pdf -o result.json

# 指定Tika Server和并发数
python batch_parser.py /path/to/documents -o results.json -w 8 --tika-server http://192.168.1.100:9998

[
  {
    "file": "/documents/report.pdf",
    "content": "这是从PDF中提取的文本内容...",
    "metadata": {
      "Author": "张三",
      "Content-Type": "application/pdf",
      "dcterms:created": "2024-01-15T10:30:00Z",
      "xmpTPg:NPages": "25"
    },
    "error": null
  },
  {
    "file": "/documents/data.xlsx",
    "content": "表格数据提取结果...",
    "metadata": {
      "Author": "李四",
      "Content-Type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
      "title": "季度报表"
    },
    "error": null
  }
]

java -version

from tika import parser
import tika
# 设置全局超时（单位：秒）
tika.TikaClientOnly.timeout = 120
parsed = parser.from_file('large_file.pdf')

// Java代码中限制解析内容大小
BodyContentHandler handler = new BodyContentHandler(10 * 1024 * 1024); // 限制10MB

curl -T scanned.pdf http://localhost:9998/tika \
  -H "X-Tika-OCRLanguage: chi_sim"

# 通过headers禁用OCR
headers = {"X-Tika-PDFOcrStrategy": "no_ocr"}
parsed = parser.from_file('dual_layer.pdf', headers=headers)