Xberg Python 教程：批量文档抽取、OCR 与结果校验

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'xberg==1.3.9'
python -c 'from importlib.metadata import version; print(version("xberg"))'
mkdir input

"""Fixed Xberg 1.3.9 local-byte demo; no remote URLs or optional LLMs."""
import argparse
import asyncio
import hashlib
import json
from importlib.metadata import version
from pathlib import Path
import time

from xberg import ExtractInput,ExtractionConfig,OcrConfig,extract

MIMES={'.txt':'text/plain','.html':'text/html','.csv':'text/csv',
       '.docx':'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
       '.pdf':'application/pdf','.png':'image/png'}

async def process(path,root):
    record={'source':path.name,'status':'failed','xberg_version':version('xberg')}
    try:
        if path.is_symlink() or not path.resolve().is_relative_to(root.resolve()):
            raise ValueError('outside input directory or symlink')
        mime=MIMES.get(path.suffix.lower())
        if not mime:raise ValueError('unsupported extension in this demo')
        if path.stat().st_size>2*1024*1024:raise ValueError('file exceeds demo 2 MiB limit')
        data=path.read_bytes()
        if not data:raise ValueError('empty file')
        record['sha256']=hashlib.sha256(data).hexdigest()
        record['source_bytes']=len(data)
        is_image=mime.startswith('image/')
        config=ExtractionConfig(use_cache=False,disable_ocr=not is_image,
            ocr=OcrConfig(backend='tesseract',language=['eng']) if is_image else None,
            extraction_timeout_secs=15,output_format='plain')
        started=time.perf_counter()
        output=await extract(ExtractInput(kind='bytes',bytes=data,mime_type=mime,filename=path.name),config)
        record['extract_ms']=round((time.perf_counter()-started)*1000,2)
        if output.errors:raise RuntimeError('Xberg returned extraction errors: '+str(output.errors)[:400])
        if len(output.results)!=1:raise RuntimeError('expected exactly one extracted document')
        doc=output.results[0]
        if not doc.content.strip():raise RuntimeError('empty extracted content')
        record.update(status='ok',content=doc.content,mime_type=doc.mime_type,
                      table_count=len(doc.tables),ocr_requested=is_image)
    except Exception as exc:
        record.update(error_type=type(exc).__name__,error=str(exc)[:600])
    return record

async def main():
    parser=argparse.ArgumentParser()
    parser.add_argument('input_directory',type=Path)
    args=parser.parse_args()
    root=args.input_directory.resolve()
    if not root.is_dir():raise SystemExit('input must be an existing directory')
    files=sorted(p for p in root.iterdir() if p.is_file())
    if not files or len(files)>20:raise SystemExit('demo expects 1 to 20 files')
    # Sequential execution makes errors and resource use easier to inspect.
    records=[await process(p,root) for p in files]
    print(json.dumps({'records':records,'summary':{'attempted':len(records),
        'ok':sum(r['status']=='ok' for r in records),
        'failed':sum(r['status']!='ok' for r in records)}},ensure_ascii=False,indent=2))

if __name__=='__main__':asyncio.run(main())

python extract_batch.py input > extraction-output.json

import json
from pathlib import Path

report = json.loads(Path('extraction-output.json').read_text())
records = report['records']
assert records, 'no extraction records'
assert report['summary']['attempted'] == len(records)
bad = [row['source'] for row in records if row['status'] != 'ok']
if bad:
    raise SystemExit('review failed files: ' + ', '.join(bad))
print('All files passed the basic extraction gate.')