paperless_api.py
3.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
#!/usr/bin/env python3
"""
清关文件上传 API
按照 form-data 方式提交两个字段：
- params: JSON 字符串（包含 code、slipId、uid 等）
- fileUpload: PDF 文件
"""
import os
import time
import json
import requests
from typing import Optional
from .api_config import API_CONFIG
def upload_clearance_file(
code: str,
slip_id: int,
pdf_path: str,
file_index: int = 0,
uid: Optional[int] = None,
Authorization: Optional[str] = None,
) -> str:
"""根据运单ID上传清关文件（PDF）。
Args:
code: 运单编号（必填）
slip_id: 创建运单返回的ID（必填）
pdf_path: 本地 PDF 文件路径（必填）
file_index: 文件索引，默认 0
uid: 文件 uid，可不传，默认使用时间戳生成
Authorization: 授权令牌，如果提供则使用该值，否则使用 API_CONFIG 中的默认值
Returns:
文本：成功/失败信息。
"""
if not code:
return "参数错误: code 不能为空"
if not slip_id and slip_id != 0:
return "参数错误: slip_id 不能为空"
if not os.path.isfile(pdf_path):
return f"参数错误: 文件不存在 - {pdf_path}"
url = f"{API_CONFIG['base_url']}/Exp/manager-server/attachmentNew/upload/paperless/clean"
# 生成 uid
real_uid = uid if isinstance(uid, int) else int(time.time() * 1000)
params_payload = {
"uploadType": "other",
"code": code,
"fileIndex": file_index,
"fileType": "picType_1",
"inputType": 0,
"slipId": slip_id,
"slipType": "slipType_chm_pdf_od",
"fileUpload": [{"uid": real_uid}],
"fileSizeTotal": 18425,
"splitSuccess": 0,
"needBackSplit": 0,
"splitPicTotal": 0,
"item": [],
}
# form-data: params 是 JSON 字符串，fileUpload 是文件
data = {
"params": json.dumps(params_payload, ensure_ascii=False),
}
files = {
"fileUpload": (os.path.basename(pdf_path), open(pdf_path, "rb"), "application/pdf"),
}
# 构建 headers，优先使用传入的 Authorization
headers = dict(API_CONFIG.get("headers", {}))
if Authorization:
headers['Authorization'] = Authorization
# 移除 Content-Type（由 requests 根据 multipart 自动设置）
headers.pop("Content-Type", None)
try:
resp = requests.post(url, headers=headers, data=data, files=files, timeout=60)
# 确保文件句柄尽快关闭
try:
files["fileUpload"][1].close()
except Exception:
pass
if resp.status_code == 200:
return (
f"上传清关文件成功\n"
f"运单号：{code}\n"
f"slipId：{slip_id}\n"
f"uid：{real_uid}"
)
return f"上传失败: HTTP {resp.status_code}, 错误信息: {resp.text}"
except requests.exceptions.RequestException as e:
return f"网络请求失败: {str(e)}"
except Exception as e:
return f"上传清关文件失败: {str(e)}"