Python sdk

from docflow import DocflowClient, AuthScope

client = DocflowClient.from_env()

workspace = client.workspace.create(
    name="费用报销处理空间",
    auth_scope=AuthScope.PUBLIC,  # 公开权限
    description="处理公司费用报销单据"
)

print(f"工作空间ID: {workspace.workspace_id}")

# 分页获取
workspaces = client.workspace.list(page=1, page_size=20)

for ws in workspaces.workspaces:
    print(f"{ws.workspace_id}: {ws.name}")

# 使用迭代器自动分页
for workspace in client.workspace.iter():
    print(f"{workspace.workspace_id}: {workspace.name}")

# 绑定工作空间
ws = client.workspace("123")

# 获取工作空间详情
detail = ws.get()

# 更新工作空间
ws.update(
    name="新的工作空间名称",
    auth_scope=AuthScope.PRIVATE
)

# 链式调用：操作类别
cat = ws.category("456")
cat.fields.add(name="发票号码")
cat.tables.add(name="商品明细表")

from docflow import ExtractModel, FieldType

# 定义字段配置
fields = [
    {
        "name": "发票号码",
        "description": "发票唯一标识"
    },
    {
        "name": "开票日期",
        "transform_settings": {
            "type": FieldType.DATETIME.value,
            "datetime_settings": {
                "format": "yyyy-MM-dd"
            }
        }
    },
    {
        "name": "发票类型",
        "transform_settings": {
            "type": FieldType.ENUMERATE.value,
            "enumerate_settings": {
                "items": ["增值税专用发票", "增值税普通发票"]
            }
        }
    }
]

# 创建类别
category = client.category.create(
    workspace_id="123",
    name="增值税发票",
    extract_model=ExtractModel.Model_1,  # 使用模型1
    sample_files=[
        "/path/to/invoice_sample1.pdf",
        "/path/to/invoice_sample2.pdf"
    ],
    fields=fields,
    category_prompt="增值税发票，包含发票信息和商品明细"
)

print(f"类别ID: {category.category_id}")

# 创建类别后，添加表格
cat = client.workspace("123").category(category.category_id)
table = cat.tables.add(
    name="商品明细表",
    prompt="抽取商品名称、规格、数量、单价、金额"
)
print(f"表格ID: {table.table_id}")

# 绑定类别上下文
cat = client.workspace("123").category("456")

# 新增字段
field = cat.fields.add(name="税率", description="税率百分比")

# 获取字段列表
fields = cat.fields.list()
for field in fields.fields:
    print(f"{field.id}: {field.name}")

# 更新字段
cat.fields.update(
    field_id="789",
    name="税率(%)",
    required=True
)

# 删除字段
cat.fields.delete(field_ids=["789"])

# 添加表格
table = cat.tables.add(
    name="商品明细表",
    prompt="抽取商品名称、规格、数量、单价、金额"
)

# 为表格添加列字段
cat.fields.add(
    table_id=table.table_id,
    name="商品名称"
)
cat.fields.add(
    table_id=table.table_id,
    name="数量"
)
cat.fields.add(
    table_id=table.table_id,
    name="单价"
)

# 获取表格列表
tables = cat.tables.list()
for table in tables.tables:
    print(f"{table.table_id}: {table.name}")

# 上传样本文件
sample = cat.samples.upload(file="/path/to/invoice_sample.pdf")

# 获取样本列表
samples = cat.samples.list()

# 下载样本
cat.samples.download(
    sample_id="789",
    save_path="/path/to/save.pdf"
)

# 删除样本
cat.samples.delete(sample_ids=["789"])

# 上传文件（自动开始识别）
response = client.file.upload(
    workspace_id="123",
    category="增值税发票",  # 类别名称
    file_path="/path/to/invoice.pdf"
)

batch_number = response.batch_number
print(f"批次号: {batch_number}")

import time

# 轮询等待识别完成
max_wait = 60
wait_interval = 3
elapsed = 0

while elapsed < max_wait:
    result = client.file.fetch(
        workspace_id="123",
        batch_number=batch_number
    )
    
    if result.files and result.files[0].recognition_status == 1:
        file_info = result.files[0]
        print(f"识别完成: {file_info.name}")
        
        # 访问抽取的字段数据
        if file_info.data and 'fields' in file_info.data:
            for field in file_info.data['fields']:
                print(f"{field['name']}: {field['value']}")
        
        # 访问表格数据
        if file_info.data and 'items' in file_info.data:
            for row in file_info.data['items']:
                print(row)
        
        break
    
    time.sleep(wait_interval)
    elapsed += wait_interval

# 1. 创建规则库
repo = client.review.create_repo(
    workspace_id="123",
    name="费用报销审核规则库"
)

# 2. 创建规则组
group = client.review.create_group(
    workspace_id="123",
    repo_id=repo.repo_id,
    name="发票合规性检查"
)

# 3. 创建审核规则
# 获取字段ID映射
fields_response = client.category.fields.list(
    workspace_id="123",
    category_id="456"
)
field_map = {f.name: f.id for f in fields_response.fields}

# 创建规则：必填字段完整性校验
client.review.create_rule(
    workspace_id="123",
    repo_id=int(repo.repo_id),
    group_id=group.group_id,
    name="必填字段完整性校验",
    prompt='检查"发票号码"、"开票日期"、"金额"是否都已填写，任一字段为空则审核不通过',
    category_ids=["456"],
    risk_level=10,  # 高风险
    referenced_fields=[
        {
            "category_id": "456",
            "category_name": "增值税发票",
            "fields": [
                {"field_id": field_map["发票号码"], "field_name": "发票号码"},
                {"field_id": field_map["开票日期"], "field_name": "开票日期"},
                {"field_id": field_map["金额"], "field_name": "金额"}
            ],
            "tables": []
        }
    ]
)

# 创建跨文档规则：验证发票金额与付款金额一致
client.review.create_rule(
    workspace_id="123",
    repo_id=int(repo.repo_id),
    group_id=group.group_id,
    name="跨文档金额匹配",
    prompt="验证发票金额与支付记录交易金额一致，允许±0.1元误差",
    category_ids=["456", "789"],  # 多个类别
    risk_level=10,
    referenced_fields=[
        {
            "category_id": "456",
            "category_name": "增值税发票",
            "fields": [
                {"field_id": invoice_field_map["金额"], "field_name": "金额"}
            ],
            "tables": []
        },
        {
            "category_id": "789",
            "category_name": "支付记录",
            "fields": [
                {"field_id": payment_field_map["交易金额"], "field_name": "交易金额"}
            ],
            "tables": []
        }
    ]
)

# 提交审核任务
review_task = client.review.submit_task(
    workspace_id="123",
    name="2024年3月费用报销审核",
    repo_id=repo.repo_id,
    extract_task_ids=["task_001", "task_002", "task_003"]
)

task_id = review_task['task_id']
print(f"审核任务ID: {task_id}")

import time

# 轮询等待审核完成
max_wait = 120
wait_interval = 5
elapsed = 0

while elapsed < max_wait:
    result = client.review.get_task_result(
        workspace_id="123",
        task_id=task_id
    )
    
    status = result.get('status')
    # 状态：0=待审核, 1=审核通过, 2=审核失败, 4=审核不通过, 7=识别失败
    if status in (1, 2, 4, 7):
        print("审核完成")
        
        # 输出审核结果
        stats = result.get('statistics', {})
        print(f"规则通过数: {stats.get('pass_count', 0)}")
        print(f"规则不通过数: {stats.get('failure_count', 0)}")
        
        # 详细审核结果
        for group in result.get('groups', []):
            print(f"\n【{group['group_name']}】")
            for task in group.get('review_tasks', []):
                result_icon = "✓" if task['review_result'] == 0 else "✗"
                print(f"  {result_icon} {task['rule_name']}")
                print(f"    {task['reasoning']}")
        
        break
    
    time.sleep(wait_interval)
    elapsed += wait_interval

from docflow import (
    ExtractModel,      # 提取模型类型
    EnabledStatus,     # 启用状态（查询用）
    EnabledFlag,       # 启用标志（更新用）
    AuthScope,         # 权限范围
    FieldType,         # 字段类型
    MismatchAction,    # 不匹配处理模式
    RecognitionStatus, # 识别状态
)

# ExtractModel - 提取模型
ExtractModel.Model_1  # 速度快，抽取结果稳定
ExtractModel.Model_2  # 适用复杂文档理解
ExtractModel.Model_3  # 多模态，适用简单抽取

# AuthScope - 权限范围
AuthScope.PRIVATE  # 0 - 私有权限
AuthScope.PUBLIC   # 1 - 公共权限

# EnabledStatus - 启用状态（查询用）
EnabledStatus.ALL       # "all" - 全部
EnabledStatus.DISABLED  # "0"   - 未启用
EnabledStatus.ENABLED   # "1"   - 已启用

# EnabledFlag - 启用标志（更新用）
EnabledFlag.DISABLED  # 0 - 未启用
EnabledFlag.ENABLED   # 1 - 已启用

# FieldType - 字段转换类型
FieldType.DATETIME   # "datetime"   - 日期时间
FieldType.ENUMERATE  # "enumerate"  - 枚举
FieldType.REGEX      # "regex"      - 正则表达式

# RecognitionStatus - 识别状态
RecognitionStatus.PENDING    # 0 - 待识别
RecognitionStatus.SUCCESS    # 1 - 识别成功
RecognitionStatus.FAILED     # 2 - 识别失败

# 工作空间迭代器
for workspace in client.workspace.iter():
    print(f"{workspace.workspace_id}: {workspace.name}")
    if some_condition:
        break  # 可以随时中断

# 类别迭代器
for category in client.category.iter(workspace_id="123"):
    print(f"{category.category_id}: {category.name}")

# 限制最大页数
for category in client.category.iter(workspace_id="123", max_pages=5):
    print(category.name)

# 转换为列表（获取所有数据）
all_workspaces = list(client.workspace.iter())

from docflow.exceptions import (
    DocflowException,
    AuthenticationError,
    ValidationError,
    ResourceNotFoundError,
    APIError,
)

try:
    workspace = client.workspace.get(workspace_id="123")
except AuthenticationError as e:
    print(f"认证失败: {e.message}")
except ResourceNotFoundError as e:
    print(f"工作空间不存在: {e.message}")
except ValidationError as e:
    print(f"参数校验失败: {e.message}")
except APIError as e:
    print(f"API错误 [HTTP {e.status_code}]: {e.message}")
except DocflowException as e:
    print(f"SDK错误: {e.message}")

from docflow import DocflowClient, set_language

# 使用英文
set_language('en_US')
client = DocflowClient.from_env()

# 使用中文（默认）
set_language('zh_CN')

# 动态切换语言
client.set_language('en_US')

# 获取当前语言
current_lang = client.get_language()  # 'en_US' 或 'zh_CN'

client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    timeout=60,           # 请求超时（秒），默认 30
    max_retries=5,        # 最大重试次数，默认 3
    retry_backoff_factor=1.0,  # 退避因子，默认 1.0
)

# 自定义重试状态码
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    retry_status_codes=[429, 503],  # 只重试 429 和 503
)

# 自定义重试方法
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    retry_methods=["GET"],  # 只允许 GET 请求重试
)

# 禁用重试
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    max_retries=0  # 禁用重试
)

client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    base_url="https://custom-api.example.com"
)

with DocflowClient.from_env() as client:
    workspaces = client.workspace.list()
    # 退出时自动关闭连接

import logging
logging.getLogger("docflow").setLevel(logging.DEBUG)