1# 1. 下载模型包
2wget -O  db_operator_dev.tar  "上述物料中镜像链接"
3# 2. 进入算子镜像tar包所在目录
4cd /xxx
5# 3. 展开镜像
6docker load < ./db_operator_dev.tar

1docker image ls | grep iregistry.baidu-int.com/doris-rdw/algorithm/operator

1docker run -itd --gpus=all \
2--privileged=true  \
3--shm-size=64g \
4--net=host \
5--name db_op_dev \
6iregistry.baidu-int.com/doris-rdw/algorithm/operator:0.7 \
7/bin/bash

1docker exec -it db_op_dev /bin/bash

1# 1. 进入阈值自定义算子测试目录
2cd /root/databuilder_dev/tests
3# 2. 运行标点替换算子测试脚本
4sh run_punctuation_replace.sh

1# 1. input data
2    samples = [
3        {
4            'images': './images/cat.jpg'
5        },
6        {
7            'images': './images/cat2.jpg'
8        },
9        {
10            'images': './images/lena.jpg'
11        }
12    ]
13
14    input_dataset = RayDataset.from_list(samples)
15
16    # 2. create operator
17    dst_path = './ret'
18    os.makedirs(dst_path, exist_ok=True)
19    op = ImageResizer(width=1024, height=1024, dst_path=dst_path, need_hash_name=False)
20
21    # 3. run
22    output_dataset = input_dataset.run(op)
23
24    # 4. get & check result
25    image_path = output_dataset.get_column(column=op.image_key)
26    
27    for name in image_path:
28        image = Image.open(name)
29        print(f'---result image {name} shape: {image.size}')

1|-- catalog_op
2|   `-- default
3|       |-- image_resizer
4|       |   `-- v1
5|       |       |-- __init__.py
6|       |       `-- image_resizer.py
7|       |-- pdf_parser
8|       |   `-- v1
9|       |       |-- __init__.py
10|       |       `-- pdf_parser.py
11|       |-- punctuation_filter
12|       |   `-- v1
13|       |       |-- __init__.py
14|       |       `-- punctuation_filter.py
15|       |-- punctuation_replacer
16|       |   `-- v1
17|       |       |-- __init__.py
18|       |       `-- punctuation_replacer.py
19|       |-- video_frame_extractor
20|       |   `-- v1
21|       |       |-- __init__.py
22|       |       `-- video_frame_extractor.py
23|       `-- video_frame_split
24|           `-- v1
25|               |-- __init__.py
26|               `-- video_frame_split.py

1from palette.ops.base import Transformer
2from palette.util.file_utils import SmartFile
3import os
4import base64
5import time
6import requests
7
8
9class PdfParser(Transformer):
10    """
11    PDF 解析算子
12    功能：
13    1. 读取 PDF 文件（支持 volume 路径）
14    2. 提交到本地 PDF 解析服务（异步任务模式）
15    3. 轮询查询直到任务完成
16    4. 将解析结果（markdown 文本）写入 text 字段
17    """
18    _op_type = "transform"
19    _batched_op = True
20    _processor = 'cpu'
21    _name = "pdf_parser"
22    _ray_execute_mode = "PIPELINE_TASK"
23    _ray_batch_format = "numpy"
24
25    def __init__(self,
26                 server_url: str = 'http://xxx:8000',
27                 poll_interval: float = 2.0,
28                 timeout: int = 300,
29                 *args, **kwargs):
30        """
31        :param server_url: PDF 解析服务地址，如 http://xxxx:8000
32        :param poll_interval: 轮询间隔（秒）
33        :param timeout: 单个任务最大等待时间（秒）
34        """
35        super().__init__(*args, **kwargs)
36        self.server_url = server_url.rstrip('/')
37        self.push_url = f'{self.server_url}/v2/models/parser_pdf'
38        self.query_url = f'{self.server_url}/v2/models/parser_pdf/query'
39        self.poll_interval = poll_interval
40        self.document_key = 'documents'
41        self.timeout = timeout
42        self.smart_file = SmartFile()
43        self.download_path = os.path.join(
44            os.getcwd(),
45            os.path.splitext(os.path.basename(__file__))[0]
46        )
47
48    def _submit_task(self, session, local_path):
49        """提交 PDF 文件到解析服务，返回 task_id"""
50        file_name = os.path.basename(local_path)
51        with open(local_path, 'rb') as f:
52            file_data = base64.b64encode(f.read()).decode('utf-8')
53
54        response = session.post(self.push_url, json={
55            'file_name': file_name,
56            'file_data': file_data,
57            'preprocess_info': {'angle_adjust': 'false', 'gray': 'no', 'unwarp': 'no'},
58            'mode': 'professional',
59            'detect_threshold': 0,
60            'process_granularity': 'coarse',
61            'submodel_info': {'layout': 'layout_det_v2'}
62        }, timeout=(5, 120))
63        response.raise_for_status()
64        task_id = response.json().get('id')
65        return task_id
66
67    def _wait_for_result(self, session, task_id):
68        """轮询查询任务状态直到完成，返回结果 dict"""
69        deadline = time.time() + self.timeout
70        while time.time() < deadline:
71            resp = session.post(self.query_url, json={'id': task_id}, timeout=(5, 30))
72            resp.raise_for_status()
73            result = resp.json()
74            status = result.get('status')
75            if status == 'succeeded':
76                return result
77            elif status == 'failed':
78                raise RuntimeError(f'task {task_id} failed: {result}')
79            time.sleep(self.poll_interval)
80        raise TimeoutError(f'task {task_id} timed out after {self.timeout}s')
81
82    def transform_batched(self, samples, *args, **kwargs):
83        file_paths = samples[self.document_key]
84        session = requests.Session()
85
86        result_texts = []
87        for file_path in file_paths:
88            try:
89                # volume 路径下载到本地
90                os.makedirs(self.download_path, exist_ok=True)
91                local_path = self.smart_file.volume_2_local(file_path, self.download_path)
92
93                task_id = self._submit_task(session, local_path)
94                result = self._wait_for_result(session, task_id)
95                content = result['result']['result']['content']
96            except Exception as e:
97                content = ''
98                print(f'[PdfParser] error processing {file_path}: {e}')
99
100            result_texts.append(content)
101
102        samples[self.text_key] = result_texts
103        return samples

1cd /root/databuilder_dev
2sh ./build.sh

1cd /root/databuilder_dev/output/dist 
2pip install databuilder_vendor_operators-0.3.0-py3-none-any.whl