# 下載最新版本（以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>
    <!-- 標(biāo)準(zhǔn)解析器包（必選）-->
    <dependency>
        <groupId>org.apache.tika</groupId>
        <artifactId>tika-parsers-standard-package</artifactId>
    </dependency>
</dependencies>

pip install tika

from tika import parser
# 指定Tika Server地址（如果是本地默認(rèn)端口，可省略）
server_url = 'http://127.0.0.1:9998'
# 解析文件
parsed = parser.from_file('path/to/document.pdf', server_url)
# 獲取文本內(nèi)容
print("文本內(nèi)容:\n", parsed["content"])
# 獲取元數(shù)據(jù)
print("\n元數(shù)據(jù):", parsed["metadata"])

from tika import parser
parsed = parser.from_file('report.docx')
metadata = parsed["metadata"]
print(f"文檔標(biāo)題: {metadata.get('title', '無(wú)')}")
print(f"作者: {metadata.get('Author', '無(wú)')}")
print(f"創(chuàng)建時(shí)間: {metadata.get('dcterms:created', '無(wú)')}")
print(f"修改時(shí)間: {metadata.get('Last-Modified', '無(wú)')}")
print(f"文件類型: {metadata.get('Content-Type', '無(wú)')}")
print(f"頁(yè)數(shù): {metadata.get('xmpTPg:NPages', '無(wú)')}")
print(f"文件大小: {metadata.get('Content-Length', '無(wú)')}")

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  # 安裝語(yǔ)言包

sudo apt-get install tesseract-ocr
sudo apt-get install tesseract-ocr-chi-sim  # 簡(jiǎn)體中文語(yǔ)言包

tesseract --version

from tika import parser
# Tika 2.x會(huì)自動(dòng)檢測(cè)是否為掃描件，并調(diào)用Tesseract OCR
parsed = parser.from_file('scanned_document.pdf')
print("OCR識(shí)別結(jié)果:\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
# 通過(guò)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'):
    """提取文檔內(nèi)容及其附件"""
    with open(file_path, 'rb') as f:
        files = {'file': f}
        # 使用 /rmeta 端點(diǎn)獲取遞歸元數(shù)據(jù)
        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')
# 遍歷壓縮包內(nèi)的所有文件
for file_name, content in parsed.items():
    print(f"文件: {file_name}")
    print(f"內(nèi)容預(yù)覽: {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']
    }
    # 檢查是否有嵌入的附件（需要進(jìn)一步處理）
    # 實(shí)際實(shí)現(xiàn)中可通過(guò)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:
    """生產(chǎn)級(jí)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:
        """提取文本內(nèi)容"""
        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]:
        """提取元數(shù)據(jù)"""
        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]:
        """同時(shí)提取文本和元數(shù)據(jù)（推薦使用）"""
        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:
        """檢測(cè)文件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"文本長(zhǎng)度: {len(text)} 字符")
    # 提取元數(shù)據(jù)
    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:
    """批量文檔解析器"""
    # 支持的文件擴(kuò)展名
    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]:
        """解析單個(gè)文件"""
        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': '解析返回空結(jié)果',
                    '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}, 錯(cuò)誤: {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"共發(fā)現(xiàn) {len(files_to_parse)} 個(gè)待解析文件")
        # 使用線程池并發(fā)解析
        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):
        """導(dǎo)出結(jié)果為JSON文件"""
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        logger.info(f"結(jié)果已導(dǎo)出到: {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='并發(fā)線程數(shù)')
    parser.add_argument('--tika-server', help='Tika Server地址')
    args = parser.parse_args()
    # 創(chuàng)建解析器
    batch_parser = BatchParser(
        tika_server_url=args.tika_server,
        max_workers=args.workers
    )
    # 判斷輸入是文件還是目錄
    if os.path.isfile(args.input):
        # 單文件解析
        logger.info(f"解析單個(gè)文件: {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)
        # 統(tǒng)計(jì)信息
        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

# 解析單個(gè)文件
python batch_parser.py document.pdf -o result.json

# 指定Tika Server和并發(fā)數(shù)
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中提取的文本內(nèi)容...",
    "metadata": {
      "Author": "張三",
      "Content-Type": "application/pdf",
      "dcterms:created": "2024-01-15T10:30:00Z",
      "xmpTPg:NPages": "25"
    },
    "error": null
  },
  {
    "file": "/documents/data.xlsx",
    "content": "表格數(shù)據(jù)提取結(jié)果...",
    "metadata": {
      "Author": "李四",
      "Content-Type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
      "title": "季度報(bào)表"
    },
    "error": null
  }
]

java -version

from tika import parser
import tika
# 設(shè)置全局超時(shí)（單位：秒）
tika.TikaClientOnly.timeout = 120
parsed = parser.from_file('large_file.pdf')

// Java代碼中限制解析內(nèi)容大小
BodyContentHandler handler = new BodyContentHandler(10 * 1024 * 1024); // 限制10MB

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

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