434 lines
18 KiB
Python
434 lines
18 KiB
Python
import base64
|
||
import io
|
||
import logging
|
||
import time
|
||
from urllib.parse import quote
|
||
from flask import Blueprint, request, jsonify, Response
|
||
from PIL import Image as PILImage
|
||
from docx import Document
|
||
from lib.word_signer import sign_word, _new_trace_id
|
||
from lib.image_processor import process_image
|
||
from lib import field_codes
|
||
from lib.validators import validate_sign_request, validate_docx_size, validate_image_size
|
||
from lib.config import get_limits_config, get_oa_config
|
||
from lib import monitor
|
||
import lib.db as db
|
||
|
||
api_bp = Blueprint('api', __name__)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _error_body(code, message, trace_id, **extra):
|
||
body = {'success': False, 'code': code, 'message': message, 'trace_id': trace_id}
|
||
if extra:
|
||
body['data'] = extra
|
||
return body
|
||
|
||
|
||
@api_bp.route('/api/word/sign', methods=['POST'])
|
||
def word_sign():
|
||
log_source = '外部API'
|
||
trace_id = _new_trace_id()
|
||
client_ip = request.remote_addr
|
||
start_ts = time.time()
|
||
limits = get_limits_config()
|
||
|
||
monitor.register_request(
|
||
trace_id, client_ip, '/api/word/sign',
|
||
request_body_summary={'content_type': request.content_type}
|
||
)
|
||
logger.info("[请求接入] 客户端IP=%s, 接口=/api/word/sign, trace_id=%s",
|
||
client_ip, trace_id, extra={'log_source': log_source, 'trace_id': trace_id})
|
||
|
||
content_type = request.content_type or ''
|
||
|
||
if 'application/json' in content_type:
|
||
json_data = request.get_json(silent=True)
|
||
if not json_data:
|
||
logger.error("[参数校验] 请求体JSON解析失败",
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='INVALID_JSON')
|
||
return jsonify(_error_body('INVALID_JSON', '请求体JSON解析失败', trace_id)), 400
|
||
|
||
params = json_data
|
||
docx_base64 = params.get('docx_base64', '')
|
||
if not docx_base64:
|
||
logger.error("[参数校验] 缺少docx_base64字段",
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='MISSING_DOCX')
|
||
return jsonify(_error_body('MISSING_DOCX', '缺少docx_base64字段', trace_id)), 400
|
||
|
||
try:
|
||
docx_data = base64.b64decode(docx_base64)
|
||
except Exception as e:
|
||
logger.error("[参数校验] docx_base64解码失败: %s", str(e),
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='INVALID_BASE64')
|
||
return jsonify(_error_body('INVALID_BASE64', f'docx_base64解码失败: {e}', trace_id)), 400
|
||
|
||
elif 'multipart/form-data' in content_type:
|
||
docx_file = request.files.get('docx')
|
||
if not docx_file:
|
||
logger.error("[参数校验] multipart请求缺少docx文件",
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='MISSING_DOCX')
|
||
return jsonify(_error_body('MISSING_DOCX', '缺少docx文件', trace_id)), 400
|
||
docx_data = docx_file.read()
|
||
params = {
|
||
'docx_name': docx_file.filename or 'upload.docx',
|
||
'match_mode': request.form.get('match_mode', 'upload'),
|
||
'stamps': [],
|
||
}
|
||
import json
|
||
try:
|
||
params['stamps'] = json.loads(request.form.get('stamps', '[]'))
|
||
except Exception:
|
||
params['stamps'] = []
|
||
|
||
image_file = request.files.get('image')
|
||
if image_file:
|
||
img_bytes = image_file.read()
|
||
valid, msg = validate_image_size(img_bytes, limits['image_max_bytes'])
|
||
if not valid:
|
||
logger.error("[参数校验] 上传图片过大: %s", msg,
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='IMAGE_TOO_LARGE')
|
||
return jsonify(_error_body('IMAGE_TOO_LARGE', msg, trace_id)), 400
|
||
img_b64 = base64.b64encode(img_bytes).decode('utf-8')
|
||
for s in params['stamps']:
|
||
s['image_base64'] = img_b64
|
||
s['image_filename'] = image_file.filename or 'image.png'
|
||
|
||
for key in ['use_global_align', 'align_h', 'align_v', 'use_global_height',
|
||
'height', 'use_global_noise', 'noise_level', 'use_global_rotate',
|
||
'rotate_min', 'rotate_max']:
|
||
if key in request.form:
|
||
val = request.form[key]
|
||
if val.lower() in ('true', 'false'):
|
||
params[key] = val.lower() == 'true'
|
||
else:
|
||
try:
|
||
if '.' in val:
|
||
params[key] = float(val)
|
||
else:
|
||
params[key] = int(val)
|
||
except ValueError:
|
||
params[key] = val
|
||
else:
|
||
logger.error("[参数校验] 不支持的Content-Type: %s", content_type,
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='INVALID_CONTENT_TYPE')
|
||
return jsonify(_error_body('INVALID_CONTENT_TYPE',
|
||
f'不支持的Content-Type: {content_type}', trace_id)), 400
|
||
|
||
# SRS 9.1/15.3.3：docx 文档 ≤ 1 MB
|
||
valid, msg = validate_docx_size(docx_data, limits['docx_max_bytes'])
|
||
if not valid:
|
||
logger.error("[参数校验] %s", msg, extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='DOCX_TOO_LARGE')
|
||
return jsonify(_error_body('DOCX_TOO_LARGE', msg, trace_id)), 400
|
||
|
||
# SRS 15.3.3：upload 模式下图片大小校验
|
||
if params.get('match_mode') == 'upload':
|
||
for idx, s in enumerate(params.get('stamps', [])):
|
||
b64 = s.get('image_base64') or ''
|
||
if b64:
|
||
try:
|
||
img_bytes = base64.b64decode(b64)
|
||
valid, msg = validate_image_size(img_bytes, limits['image_max_bytes'])
|
||
if not valid:
|
||
logger.error("[参数校验] 盖章项%d图片过大: %s", idx + 1, msg,
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='IMAGE_TOO_LARGE')
|
||
return jsonify(_error_body('IMAGE_TOO_LARGE', f"盖章项{idx + 1}: {msg}", trace_id)), 400
|
||
except Exception as e:
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='INVALID_BASE64')
|
||
return jsonify(_error_body('INVALID_BASE64',
|
||
f"盖章项{idx + 1}: 图片Base64解码失败", trace_id)), 400
|
||
|
||
errors = validate_sign_request(params)
|
||
if errors:
|
||
logger.error("[参数校验] 校验失败: %s", '; '.join(errors),
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=400,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='VALIDATION_ERROR')
|
||
return jsonify(_error_body('VALIDATION_ERROR', '; '.join(errors), trace_id)), 400
|
||
|
||
logger.info("[参数校验] 校验通过", extra={'log_source': log_source, 'trace_id': trace_id})
|
||
|
||
# SRS 15.3.4：需要数据库的模式下 MySQL 未启动或连接失败，返回 5xx
|
||
if params.get('match_mode') in ('id', 'name', 'auto'):
|
||
if not db.ensure_mysql_available():
|
||
logger.error("[数据库] MySQL 不可用，按 SRS 15.3.4 返回 5xx",
|
||
extra={'log_source': log_source, 'trace_id': trace_id})
|
||
monitor.finish_request(trace_id, 'error', response_status=503,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message='MYSQL_UNAVAILABLE')
|
||
return jsonify(_error_body('MYSQL_UNAVAILABLE',
|
||
'MySQL 不可用，请检查数据库配置', trace_id)), 503
|
||
|
||
result, error = sign_word(docx_data, params, log_source=log_source, trace_id=trace_id, client_ip=client_ip)
|
||
|
||
if error:
|
||
code = error.get('code', 'INTERNAL_ERROR')
|
||
status_code = 400 if code in ('INVALID_FILE_TYPE', 'INVALID_MATCH_MODE',
|
||
'TOO_MANY_STAMPS', 'INVALID_DOCX',
|
||
'DOCX_TOO_LARGE', 'IMAGE_TOO_LARGE',
|
||
'VALIDATION_ERROR') else 500
|
||
monitor.finish_request(trace_id, 'error', response_status=status_code,
|
||
elapsed_ms=int((time.time() - start_ts) * 1000),
|
||
error_message=code)
|
||
return jsonify(error), status_code
|
||
|
||
filename = result['filename']
|
||
response = Response(
|
||
result['data'],
|
||
mimetype='application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
||
)
|
||
ascii_fallback = filename.encode('ascii', errors='replace').decode('ascii')
|
||
response.headers['Content-Disposition'] = (
|
||
f"attachment; filename=\"{ascii_fallback}\"; "
|
||
f"filename*=UTF-8''{quote(filename, safe='')}"
|
||
)
|
||
response.headers['X-SEAL-Trace-Id'] = result.get('trace_id', trace_id)
|
||
response.headers['X-SEAL-Warnings'] = str(len(result.get('warnings', [])))
|
||
if result.get('warnings'):
|
||
joined = '; '.join(result['warnings'])
|
||
encoded = quote(joined, safe='')
|
||
encoded_bytes = encoded.encode('latin-1', errors='replace')
|
||
while len(encoded_bytes) > 1024:
|
||
encoded = encoded[:-1]
|
||
encoded_bytes = encoded.encode('latin-1', errors='replace')
|
||
response.headers['X-SEAL-Warning-Detail'] = encoded
|
||
|
||
elapsed_ms = int((time.time() - start_ts) * 1000)
|
||
monitor.finish_request(
|
||
trace_id, 'success', response_status=200, elapsed_ms=elapsed_ms,
|
||
response_summary={'filename': filename, 'success_count': result.get('success_count', 0)},
|
||
warnings=result.get('warnings', [])
|
||
)
|
||
return response
|
||
|
||
|
||
def _image_data_url(data):
|
||
"""根据魔数判断图片格式并构造 data URL。"""
|
||
if data[:8] == b'\x89PNG\r\n\x1a\n':
|
||
mime = 'image/png'
|
||
elif data[:3] == b'\xff\xd8\xff':
|
||
mime = 'image/jpeg'
|
||
elif data[:6] in (b'GIF87a', b'GIF89a'):
|
||
mime = 'image/gif'
|
||
else:
|
||
mime = 'image/png'
|
||
return f'data:{mime};base64,' + base64.b64encode(data).decode('ascii')
|
||
|
||
|
||
def _resolve_preview_image(payload):
|
||
"""根据 payload 解析出 (image_bytes, is_signature) 或 (None, error_msg)。"""
|
||
match_mode = payload.get('match_mode', 'upload')
|
||
|
||
if match_mode == 'upload':
|
||
b64 = payload.get('image_base64') or ''
|
||
if not b64:
|
||
return None, 'upload 模式缺少 image_base64'
|
||
try:
|
||
return base64.b64decode(b64), None
|
||
except Exception as e:
|
||
return None, f'Base64 解码失败: {e}'
|
||
|
||
if not db.ensure_mysql_available():
|
||
return None, 'MySQL 不可用，无法按人员查询'
|
||
|
||
is_sig_from_db = None
|
||
if match_mode == 'id':
|
||
image_id = (payload.get('image_id') or '').strip()
|
||
if not image_id:
|
||
return None, 'id 模式缺少 image_id'
|
||
record = db.get_image_by_user_id(image_id)
|
||
if not record:
|
||
return None, f'未找到 image_id={image_id}'
|
||
return record['sign_image'], None
|
||
|
||
if match_mode == 'name':
|
||
name = (payload.get('image_name') or '').strip()
|
||
if not name:
|
||
return None, 'name 模式缺少 image_name'
|
||
records = db.get_image_by_user_name(name)
|
||
if not records:
|
||
return None, f'未找到 image_name={name}'
|
||
return records[0]['sign_image'], None
|
||
|
||
if match_mode == 'auto':
|
||
marker = (payload.get('marker') or '').strip()
|
||
if not marker:
|
||
return None, 'auto 模式缺少 marker'
|
||
name = marker[5:] if marker.startswith('USER_') else marker
|
||
name = field_codes._strip_switches(name)
|
||
if not name:
|
||
return None, f'marker={marker} 无法解析出姓名'
|
||
records = db.get_image_by_user_name(name)
|
||
if not records:
|
||
return None, f'未找到姓名={name}'
|
||
return records[0]['sign_image'], None
|
||
|
||
return None, f'不支持的 match_mode: {match_mode}'
|
||
|
||
|
||
def _parse_float(value, default=None, lo=None, hi=None):
|
||
if value in (None, ''):
|
||
return default
|
||
try:
|
||
v = float(value)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
if lo is not None and v < lo:
|
||
return default
|
||
if hi is not None and v > hi:
|
||
return default
|
||
return v
|
||
|
||
|
||
def _parse_int(value, default=0, lo=None, hi=None):
|
||
if value in (None, ''):
|
||
return default
|
||
try:
|
||
v = int(value)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
if lo is not None and v < lo:
|
||
return default
|
||
if hi is not None and v > hi:
|
||
return default
|
||
return v
|
||
|
||
|
||
@api_bp.route('/api/preview/sign-image', methods=['POST'])
|
||
def preview_sign_image():
|
||
"""预览：原签名图 vs 处理后效果（走完整 process_image 流水线）。"""
|
||
log_source = '前端操作'
|
||
payload = request.get_json(silent=True) or {}
|
||
|
||
image_data, err = _resolve_preview_image(payload)
|
||
if err:
|
||
logger.warning("[图片预览] 解析图片失败: %s", err, extra={'log_source': log_source})
|
||
return jsonify({'success': False, 'message': err}), 400
|
||
|
||
is_sig = bool(payload.get('is_signature', False))
|
||
|
||
height_cm = _parse_float(payload.get('height_cm'), default=None, lo=0.1, hi=10.0)
|
||
noise_level = _parse_int(payload.get('noise_level'), default=0, lo=0, hi=10)
|
||
rotate_min = _parse_int(payload.get('rotate_min'), default=0, lo=-180, hi=180)
|
||
rotate_max = _parse_int(payload.get('rotate_max'), default=0, lo=-180, hi=180)
|
||
|
||
try:
|
||
processed = process_image(
|
||
image_data,
|
||
is_signature=is_sig,
|
||
noise_level=noise_level,
|
||
rotate_min=rotate_min,
|
||
rotate_max=rotate_max,
|
||
log_source=log_source,
|
||
)
|
||
except Exception as e:
|
||
logger.error("[图片预览] 处理失败: %s", str(e), extra={'log_source': log_source})
|
||
return jsonify({'success': False, 'message': f'图片处理失败: {e}'}), 500
|
||
|
||
try:
|
||
pimg = PILImage.open(io.BytesIO(processed))
|
||
p_w, p_h = pimg.size
|
||
except Exception:
|
||
p_w, p_h = 0, 0
|
||
|
||
try:
|
||
oimg = PILImage.open(io.BytesIO(image_data))
|
||
o_w, o_h = oimg.size
|
||
except Exception:
|
||
o_w, o_h = 0, 0
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'original': _image_data_url(image_data),
|
||
'processed': _image_data_url(processed),
|
||
'original_size': len(image_data),
|
||
'processed_size': len(processed),
|
||
'original_dimensions': [o_w, o_h],
|
||
'processed_dimensions': [p_w, p_h],
|
||
'is_signature': is_sig,
|
||
'params': {
|
||
'height_cm': height_cm, 'noise_level': noise_level,
|
||
'rotate_min': rotate_min, 'rotate_max': rotate_max,
|
||
},
|
||
})
|
||
|
||
|
||
@api_bp.route('/api/doc/scan-markers', methods=['POST'])
|
||
def doc_scan_markers():
|
||
"""扫描 docx 中所有 SET USER_<name> 标记，用于 auto 模式预览。"""
|
||
log_source = '前端操作'
|
||
payload = request.get_json(silent=True) or {}
|
||
docx_b64 = payload.get('docx_base64') or ''
|
||
if not docx_b64:
|
||
return jsonify({'success': False, 'message': '缺少 docx_base64'}), 400
|
||
try:
|
||
docx_data = base64.b64decode(docx_b64)
|
||
except Exception as e:
|
||
return jsonify({'success': False, 'message': f'docx Base64 解码失败: {e}'}), 400
|
||
|
||
try:
|
||
doc = Document(io.BytesIO(docx_data))
|
||
except Exception as e:
|
||
return jsonify({'success': False, 'message': f'Word 解析失败: {e}'}), 400
|
||
|
||
markers = field_codes.scan_user_markers(doc, log_source=log_source)
|
||
return jsonify({
|
||
'success': True,
|
||
'markers': [{'marker': m['marker'], 'name': m['name']} for m in markers],
|
||
})
|
||
|
||
|
||
@api_bp.route('/api/health', methods=['GET'])
|
||
def health():
|
||
limits = get_limits_config()
|
||
oa = get_oa_config()
|
||
mysql_enabled = (db.get_active_db_type() == 'mysql')
|
||
return jsonify({
|
||
'success': True,
|
||
'mysql_enabled': mysql_enabled,
|
||
'docx_max_bytes': limits['docx_max_bytes'],
|
||
'image_max_bytes': limits['image_max_bytes'],
|
||
'stamp_max_count': limits['stamp_max_count'],
|
||
'action_timeout_seconds': oa['action_timeout_seconds'],
|
||
'db_type': db.get_active_db_type(),
|
||
})
|
||
|
||
|
||
@api_bp.route('/api/db/init', methods=['POST'])
|
||
def init_db_endpoint():
|
||
try:
|
||
ok = db.init_database(strict_mysql=False)
|
||
if not ok:
|
||
return jsonify({'success': False, 'message': 'MySQL 不可用'}), 500
|
||
return jsonify({'success': True, 'message': f"数据库已初始化: {db.get_active_db_type()}"})
|
||
except Exception as e:
|
||
logger.error("数据库初始化失败: %s", str(e))
|
||
return jsonify({'success': False, 'message': str(e)}), 500
|