first commit

lib/batch_processor.py

@@ -0,0 +1,784 @@
+"""批量 Word 转换与自动签名核心模块。
+
+实现 SRS 第 11、14 章：
+- 加载/保存 rules.config（JSON 格式），损坏文件不自动覆盖。
+- 递归扫描 Word 文件夹，只处理 .docx。
+- 多条正则规则：捕获组序号 >=1，cell_mode (single/horizontal/vertical)，cell_match (partial/whole)。
+- 表格逻辑网格：识别 gridSpan 横向合并 / vMerge 纵向合并。
+- 命中后将可见姓名替换为隐藏的 ``SET USER_<name> \\* MERGEFORMAT`` 域代码 + 补偿空格。
+- 调用 word_signer.sign_word 以 auto 模式完成自动签名。
+- 输出 _signed.docx；冲突时自动生成 _signed_001、_signed_002 等不冲突文件名。
+- 单文件失败跳过 + WARN，批量任务继续。
+"""
+import io
+import os
+import re
+import json
+import logging
+import shutil
+import tempfile
+from copy import deepcopy
+from datetime import datetime
+
+from docx import Document
+from docx.oxml.ns import qn
+from docx.oxml import OxmlElement
+
+from lib import field_codes
+from lib.validators import validate_rule
+from lib.word_signer import sign_word
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_RULES = [
+    {
+        'name': '编制：姓名/标识',
+        'pattern': r'编制[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'capture_group': 1,
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+        'enabled': True,
+    },
+    {
+        'name': '审核：姓名/标识',
+        'pattern': r'审核[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'capture_group': 1,
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+        'enabled': True,
+    },
+    {
+        'name': '签字人：姓名/标识',
+        'pattern': r'签字人[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'capture_group': 1,
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+        'enabled': True,
+    },
+    {
+        'name': '表格值整格',
+        'pattern': r'^([一-龥A-Za-z0-9._-]{1,20})$',
+        'capture_group': 1,
+        'cell_mode': 'single',
+        'cell_match': 'whole',
+        'enabled': False,
+    },
+]
+
+REGEX_SNIPPETS = [
+    {
+        'id': 'single_compiled_name',
+        'label': '编制：姓名/标识',
+        'help': '适用于正文中"编制：张三"或"编制: zhang_san1"的标签加姓名组合；表格方式选 single，部分匹配。',
+        'pattern': r'编制[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+    },
+    {
+        'id': 'audit_compiled_name',
+        'label': '审核：姓名/标识',
+        'help': '适用于"审核：李四"组合；表格方式选 single，部分匹配。',
+        'pattern': r'审核[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+    },
+    {
+        'id': 'signer_compiled_name',
+        'label': '签字人：姓名/标识',
+        'help': '适用于"签字人：王五"组合；表格方式选 single，部分匹配。',
+        'pattern': r'签字人[:：]\s*([一-龥A-Za-z0-9._-]{1,20})',
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+    },
+    {
+        'id': 'horizontal_label_value',
+        'label': '表格横向相邻两格',
+        'help': '同一行：左格"编制："，右格"张三"。规则匹配标签，捕获组指向右值；表格方式选 horizontal，整格匹配。',
+        'pattern': r'编制[:：]',
+        'cell_mode': 'horizontal',
+        'cell_match': 'whole',
+        'capture_group_hint': '在表格中通过相邻右格文本另用通用片段提取',
+    },
+    {
+        'id': 'vertical_label_value',
+        'label': '表格纵向相邻两格',
+        'help': '同一列：上格"审核："，下格"李四"。规则匹配标签，从下格提取；表格方式选 vertical，整格匹配。',
+        'pattern': r'审核[:：]',
+        'cell_mode': 'vertical',
+        'cell_match': 'whole',
+    },
+    {
+        'id': 'whole_name_in_cell',
+        'label': '表格值整格',
+        'help': '姓名/标识单独占据整个单元格时使用；表格方式选 single，整格匹配。',
+        'pattern': r'^([一-龥A-Za-z0-9._-]{1,20})$',
+        'cell_mode': 'single',
+        'cell_match': 'whole',
+    },
+    {
+        'id': 'partial_name_in_cell',
+        'label': '表格值部分',
+        'help': '姓名/标识只是单元格部分文本时使用；表格方式选 single，部分匹配。',
+        'pattern': r'([一-龥A-Za-z0-9._-]{1,20})',
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+    },
+    {
+        'id': 'chinese_only_2_4',
+        'label': '纯中文姓名 2-4 字',
+        'help': '仅匹配 2-4 字中文姓名；表格方式选 single，整格或部分均可。',
+        'pattern': r'([一-龥]{2,4})',
+        'cell_mode': 'single',
+        'cell_match': 'partial',
+    },
+]
+
+
+def get_default_rules_path():
+    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    return os.path.join(project_root, 'rules.config')
+
+
+def resolve_rules_path(path=None):
+    if not path:
+        return get_default_rules_path()
+    if not os.path.isabs(path):
+        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        return os.path.join(project_root, path)
+    return path
+
+
+def load_rules(path=None):
+    """加载规则文件；损坏或不存在时返回默认规则并提示，不自动覆盖原文件。"""
+    target = resolve_rules_path(path)
+    if not os.path.exists(target):
+        logger.warning("规则文件不存在，加载内置默认规则: %s", target, extra={'log_source': '批量'})
+        return {'rules': deepcopy(DEFAULT_RULES), 'warning': f'规则文件不存在，已加载默认规则：{target}'}
+    try:
+        with open(target, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        rules = data.get('rules') if isinstance(data, dict) else data
+        if not isinstance(rules, list):
+            raise ValueError("规则文件格式非法：缺少 rules 列表")
+        normalized, errors = _normalize_rules(rules)
+        warning = '; '.join(errors) if errors else None
+        return {'rules': normalized, 'warning': warning}
+    except Exception as e:
+        logger.warning("规则文件损坏或非法 (%s): %s", target, str(e), extra={'log_source': '批量'})
+        return {'rules': deepcopy(DEFAULT_RULES),
+                'warning': f'规则文件损坏或非法：{e}。已加载内置默认规则，请修复或重新选择。'}
+
+
+def save_rules(rules, path=None):
+    """保存规则文件；若文件已存在且解析失败则不覆盖。"""
+    target = resolve_rules_path(path)
+    normalized, errors = _normalize_rules(rules)
+
+    if os.path.exists(target):
+        try:
+            with open(target, 'r', encoding='utf-8') as f:
+                json.load(f)
+        except Exception:
+            logger.error("规则文件已损坏，拒绝覆盖: %s", target, extra={'log_source': '批量'})
+            raise RuntimeError(f'规则文件已损坏，未覆盖：{target}')
+
+    os.makedirs(os.path.dirname(target), exist_ok=True)
+    with open(target, 'w', encoding='utf-8') as f:
+        json.dump({'rules': normalized}, f, ensure_ascii=False, indent=2)
+    logger.info("规则已保存到 %s", target, extra={'log_source': '批量'})
+    return {'rules': normalized, 'path': target, 'warnings': errors}
+
+
+def _normalize_rules(rules):
+    normalized = []
+    errors = []
+    for idx, rule in enumerate(rules):
+        if not isinstance(rule, dict):
+            errors.append(f"规则{idx + 1}: 非对象，跳过")
+            continue
+        norm_rule = {
+            'name': str(rule.get('name') or '').strip(),
+            'pattern': str(rule.get('pattern') or '').strip(),
+            'capture_group': rule.get('capture_group', 1),
+            'cell_mode': rule.get('cell_mode', 'single'),
+            'cell_match': rule.get('cell_match', 'partial'),
+            'enabled': bool(rule.get('enabled', True)),
+        }
+        rule_errors = validate_rule(norm_rule)
+        if rule_errors:
+            errors.append(f"规则{idx + 1} ({norm_rule['name'] or '未命名'}): {'; '.join(rule_errors)}")
+            norm_rule['enabled'] = False
+        normalized.append(norm_rule)
+    return normalized, errors
+
+
+def get_regex_snippets():
+    return {'snippets': REGEX_SNIPPETS}
+
+
+def test_rules(rules, text):
+    """在纯文本上测试规则，不修改 Word 文件。"""
+    normalized, errors = _normalize_rules(rules)
+    matches = []
+    matched_spans = []
+    for idx, rule in enumerate(normalized):
+        if not rule.get('enabled'):
+            continue
+        try:
+            pattern = re.compile(rule['pattern'])
+        except re.error as e:
+            errors.append(f"规则{idx + 1}: 正则错误 {e}")
+            continue
+        for m in pattern.finditer(text or ''):
+            try:
+                captured = m.group(rule['capture_group'])
+            except (IndexError, error_group_error_class()):
+                errors.append(f"规则{idx + 1}: 捕获组 {rule['capture_group']} 越界")
+                continue
+            if captured is None:
+                continue
+            name = captured.strip()
+            if not name:
+                continue
+            span = (m.start(rule['capture_group']), m.end(rule['capture_group']))
+            if any(_spans_overlap(span, existing) for existing in matched_spans):
+                errors.append(f"规则{idx + 1} 命中 '{name}' 与已有命中重叠，跳过")
+                continue
+            matched_spans.append(span)
+            field_code = f'{{{{ SET USER_{name} \\* MERGEFORMAT }}}}'
+            preview_head = text[max(0, m.start(rule['capture_group']) - 6):m.start(rule['capture_group'])]
+            preview_tail = text[m.end(rule['capture_group']):m.end(rule['capture_group']) + 6]
+            matches.append({
+                'rule_index': idx,
+                'rule_name': rule['name'],
+                'match_text': m.group(0),
+                'captured': name,
+                'user_marker': f'USER_{name}',
+                'field_code': field_code,
+                'cell_mode': rule['cell_mode'],
+                'cell_match': rule['cell_match'],
+                'preview': f"{preview_head}<{name}>{preview_tail}",
+            })
+    return {'matches': matches, 'match_count': len(matches), 'errors': errors}
+
+
+def error_group_error_class():
+    return IndexError
+
+
+def _spans_overlap(a, b):
+    return not (a[1] <= b[0] or b[1] <= a[0])
+
+
+def scan_docx_files(folder, recursive=True):
+    """递归扫描 .docx 文件，跳过 .doc 与其他格式。"""
+    folder = os.path.abspath(folder)
+    if not os.path.isdir(folder):
+        return [], [f"文件夹不存在: {folder}"]
+    files = []
+    skipped = []
+    if recursive:
+        for root, dirs, names in os.walk(folder):
+            for name in names:
+                full = os.path.join(root, name)
+                if name.lower().endswith('.docx'):
+                    files.append(full)
+                elif name.lower().endswith('.doc'):
+                    skipped.append(full)
+                elif name.lower().endswith(('.wps', '.pdf', '.rtf')):
+                    skipped.append(full)
+    else:
+        for name in os.listdir(folder):
+            full = os.path.join(folder, name)
+            if os.path.isfile(full):
+                if name.lower().endswith('.docx'):
+                    files.append(full)
+                elif name.lower().endswith(('.doc', '.wps', '.pdf', '.rtf')):
+                    skipped.append(full)
+    return files, skipped
+
+
+def _build_table_grid(table):
+    """基于 gridSpan 和 vMerge 构建"逻辑网格"。
+
+    返回 ``(rows, grid)``，grid[row][col] = {
+        'tc': <w:tc element>, 'text': str, 'is_origin': bool (vMerge origin),
+        'is_continue': bool (vMerge continue)
+    }
+    """
+    tbl = table if hasattr(table, '_element') else table
+    if hasattr(tbl, '_element'):
+        tbl = tbl._element
+
+    rows = tbl.findall(qn('w:tr'))
+    grid = []
+    pending_vmerge = {}
+
+    for tr in rows:
+        cells_in_row = tr.findall(qn('w:tc'))
+        row_cells = []
+        col_index = 0
+        cell_iter = iter(cells_in_row)
+        col_fill = {}
+
+        while True:
+            if col_index in col_fill:
+                row_cells.append(col_fill[col_index])
+                col_index += 1
+                continue
+            try:
+                tc = next(cell_iter)
+            except StopIteration:
+                break
+
+            tcPr = tc.find(qn('w:tcPr'))
+            grid_span = 1
+            vmerge_kind = None
+            if tcPr is not None:
+                gs = tcPr.find(qn('w:gridSpan'))
+                if gs is not None:
+                    try:
+                        grid_span = int(gs.get(qn('w:val')) or '1')
+                    except ValueError:
+                        grid_span = 1
+                vm = tcPr.find(qn('w:vMerge'))
+                if vm is not None:
+                    val = vm.get(qn('w:val'))
+                    vmerge_kind = 'restart' if val == 'restart' else 'continue'
+
+            text = _extract_cell_text(tc)
+            cell_info = {
+                'tc': tc,
+                'text': text,
+                'grid_span': grid_span,
+                'vmerge_kind': vmerge_kind,
+                'row': len(grid),
+                'col': col_index,
+            }
+
+            if vmerge_kind == 'restart':
+                pending_vmerge[col_index] = {
+                    'origin_tc': tc,
+                    'origin_text': text,
+                    'origin_row': len(grid),
+                    'rows': [cell_info],
+                }
+                cell_info['is_origin'] = True
+                cell_info['is_continue'] = False
+            elif vmerge_kind == 'continue':
+                if col_index in pending_vmerge:
+                    origin = pending_vmerge[col_index]
+                    origin['rows'].append(cell_info)
+                    cell_info['merged_origin'] = origin['origin_tc']
+                    cell_info['merged_origin_text'] = origin['origin_text']
+                cell_info['is_origin'] = False
+                cell_info['is_continue'] = True
+            else:
+                if col_index in pending_vmerge:
+                    pending_vmerge.pop(col_index, None)
+                cell_info['is_origin'] = True
+                cell_info['is_continue'] = False
+
+            row_cells.append(cell_info)
+            for k in range(1, grid_span):
+                col_fill[col_index + k] = cell_info
+                row_cells.append(cell_info)
+            col_index += grid_span
+
+        grid.append(row_cells)
+
+    return rows, grid
+
+
+def _extract_cell_text(tc):
+    parts = []
+    for p in tc.findall(qn('w:p')):
+        buf = []
+        for t in p.iter(qn('w:t')):
+            if t.text:
+                buf.append(t.text)
+        parts.append(''.join(buf))
+    return '\n'.join(parts).strip()
+
+
+def _iter_paragraphs_in_doc(doc):
+    yield from field_codes._iter_all_paragraph_elements(doc)
+
+
+def _iter_tables_in_doc(doc):
+    yield from field_codes._iter_all_tables(doc)
+
+
+def process_document(doc, rules, log_source='批量', trace_id=None):
+    """对单个 Document 应用规则，返回 (替换条数, 告警列表)。"""
+    normalized, errors = _normalize_rules(rules)
+    warnings = list(errors)
+    replaced = 0
+
+    consumed_paragraph_runs = set()
+    consumed_table_cells = set()
+
+    for rule in normalized:
+        if not rule.get('enabled'):
+            continue
+        try:
+            pattern = re.compile(rule['pattern'])
+        except re.error as e:
+            warnings.append(f"规则 '{rule['name']}' 正则错误: {e}")
+            continue
+
+        if rule['cell_mode'] == 'single':
+            replaced += _apply_rule_to_paragraphs(doc, pattern, rule, consumed_paragraph_runs,
+                                                  log_source=log_source, trace_id=trace_id, warnings=warnings)
+            replaced += _apply_rule_single_cell_tables(doc, pattern, rule, consumed_table_cells,
+                                                       log_source=log_source, trace_id=trace_id, warnings=warnings)
+        elif rule['cell_mode'] == 'horizontal':
+            replaced += _apply_rule_horizontal_cells(doc, pattern, rule, consumed_table_cells,
+                                                     log_source=log_source, trace_id=trace_id, warnings=warnings)
+        elif rule['cell_mode'] == 'vertical':
+            replaced += _apply_rule_vertical_cells(doc, pattern, rule, consumed_table_cells,
+                                                   log_source=log_source, trace_id=trace_id, warnings=warnings)
+
+    return replaced, warnings
+
+
+def _apply_rule_to_paragraphs(doc, pattern, rule, consumed, log_source, trace_id, warnings):
+    """对正文段落与文本框/页眉页脚段落应用 single 模式规则。"""
+    count = 0
+    capture_group = int(rule.get('capture_group', 1))
+    cell_match = rule.get('cell_match', 'partial')
+
+    for para in _iter_paragraphs_in_doc(doc):
+        runs = para.findall(qn('w:r'))
+        run_texts = []
+        for r in runs:
+            t = r.find(qn('w:t'))
+            run_texts.append(t.text if t is not None and t.text else '')
+        full_text = ''.join(run_texts)
+
+        if not full_text:
+            continue
+
+        if cell_match == 'whole':
+            m = pattern.fullmatch(full_text)
+            matches = [m] if m else []
+        else:
+            matches = list(pattern.finditer(full_text))
+
+        for m in matches:
+            try:
+                captured = m.group(capture_group)
+            except IndexError:
+                warnings.append(f"规则 '{rule['name']}' 捕获组 {capture_group} 越界")
+                continue
+            if not captured:
+                continue
+            name = captured.strip()
+            if not name:
+                continue
+
+            start_offset = m.start(capture_group)
+            end_offset = m.end(capture_group)
+
+            start_run_idx, start_off, end_run_idx, end_off = _map_text_to_runs(run_texts, start_offset, end_offset)
+            if start_run_idx is None:
+                continue
+            key = (id(para), start_run_idx, start_off, end_run_idx, end_off)
+            if key in consumed:
+                continue
+            consumed.add(key)
+
+            ok = field_codes.replace_text_with_set_field(
+                para, start_run_idx, start_off, end_run_idx, end_off, captured, name,
+                compensate_spaces=True
+            )
+            if ok:
+                count += 1
+                logger.info("[批量] 段落匹配替换为 USER_%s (规则=%s)",
+                            name, rule['name'], extra={'log_source': log_source, 'trace_id': trace_id})
+    return count
+
+
+def _map_text_to_runs(run_texts, start_offset, end_offset):
+    """将整段字符串中的字符偏移映射回 (run_index, offset_in_run) 对。"""
+    cumulative = 0
+    start_run_idx = None
+    start_off = None
+    end_run_idx = None
+    end_off = None
+    for idx, text in enumerate(run_texts):
+        new_cum = cumulative + len(text)
+        if start_run_idx is None and start_offset < new_cum:
+            start_run_idx = idx
+            start_off = start_offset - cumulative
+        if end_offset <= new_cum and end_run_idx is None:
+            end_run_idx = idx
+            end_off = end_offset - cumulative
+            break
+        cumulative = new_cum
+    if start_run_idx is None or end_run_idx is None:
+        return None, None, None, None
+    return start_run_idx, start_off, end_run_idx, end_off
+
+
+def _apply_rule_single_cell_tables(doc, pattern, rule, consumed, log_source, trace_id, warnings):
+    """single 模式下补充：表格单元格内/整格匹配（基于段落处理已能覆盖单格，这里仅对合并单元格特例）。"""
+    return 0
+
+
+def _apply_rule_horizontal_cells(doc, pattern, rule, consumed, log_source, trace_id, warnings):
+    """横向相邻两格：左格匹配标签 + 右格作为值。
+
+    假设规则 pattern 仅匹配标签（如 ``编制[:：]``），值在右相邻格中。
+    捕获组在右格文本上重新应用通用片段 ``([一-龥A-Za-z0-9._-]{1,20})`` 提取。
+    """
+    count = 0
+    value_pattern = re.compile(r'([一-龥A-Za-z0-9._-]{1,20})')
+    cell_match = rule.get('cell_match', 'whole')
+
+    for tbl in _iter_tables_in_doc(doc):
+        try:
+            _rows, grid = _build_table_grid(tbl)
+        except Exception as e:
+            warnings.append(f"表格逻辑网格构建失败: {e}")
+            continue
+        for r in grid:
+            for c in range(len(r) - 1):
+                left = r[c]
+                right = r[c + 1]
+                if left is right or right is None:
+                    continue
+                left_text = left.get('text', '') or ''
+                if not pattern.search(left_text):
+                    continue
+                right_text = right.get('text', '') or ''
+                if not right_text:
+                    continue
+                m_val = value_pattern.fullmatch(right_text) if cell_match == 'whole' else value_pattern.search(right_text)
+                if not m_val:
+                    continue
+                name = m_val.group(1).strip()
+                if not name:
+                    continue
+                key = ('h', id(right['tc']))
+                if key in consumed:
+                    continue
+                consumed.add(key)
+                ok = _replace_cell_text_with_set_field(right['tc'], m_val.group(1), name,
+                                                       partial=(cell_match == 'partial'),
+                                                       match_text=m_val.group(1))
+                if ok:
+                    count += 1
+                    logger.info("[批量] 横向相邻匹配替换为 USER_%s (规则=%s)",
+                                name, rule['name'],
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+    return count
+
+
+def _apply_rule_vertical_cells(doc, pattern, rule, consumed, log_source, trace_id, warnings):
+    """纵向相邻两格：上格匹配标签 + 下格作为值。"""
+    count = 0
+    value_pattern = re.compile(r'([一-龥A-Za-z0-9._-]{1,20})')
+    cell_match = rule.get('cell_match', 'whole')
+
+    for tbl in _iter_tables_in_doc(doc):
+        try:
+            _rows, grid = _build_table_grid(tbl)
+        except Exception as e:
+            warnings.append(f"表格逻辑网格构建失败: {e}")
+            continue
+        max_cols = max((len(r) for r in grid), default=0)
+        for c in range(max_cols):
+            for r in range(len(grid) - 1):
+                if c >= len(grid[r]) or c >= len(grid[r + 1]):
+                    continue
+                top = grid[r][c]
+                bottom = grid[r + 1][c]
+                if top is bottom or bottom is None:
+                    continue
+                top_text = top.get('text', '') or ''
+                if not pattern.search(top_text):
+                    continue
+                bottom_text = bottom.get('text', '') or ''
+                if not bottom_text:
+                    continue
+                m_val = value_pattern.fullmatch(bottom_text) if cell_match == 'whole' else value_pattern.search(bottom_text)
+                if not m_val:
+                    continue
+                name = m_val.group(1).strip()
+                if not name:
+                    continue
+                key = ('v', id(bottom['tc']))
+                if key in consumed:
+                    continue
+                consumed.add(key)
+                ok = _replace_cell_text_with_set_field(bottom['tc'], m_val.group(1), name,
+                                                       partial=(cell_match == 'partial'),
+                                                       match_text=m_val.group(1))
+                if ok:
+                    count += 1
+                    logger.info("[批量] 纵向相邻匹配替换为 USER_%s (规则=%s)",
+                                name, rule['name'],
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+    return count
+
+
+def _replace_cell_text_with_set_field(tc, captured, name, partial, match_text):
+    """将单元格中的 match_text 替换为 SET USER_<name> 域 + 补偿空格。
+
+    简化策略：定位单元格中第一个包含 match_text 的段落与 run。
+    """
+    for p in tc.findall(qn('w:p')):
+        runs = p.findall(qn('w:r'))
+        for r_idx, r in enumerate(runs):
+            t = r.find(qn('w:t'))
+            if t is None or not t.text:
+                continue
+            if match_text not in t.text:
+                continue
+            text = t.text
+            pos = text.find(match_text)
+            head = text[:pos]
+            tail = text[pos + len(match_text):]
+            field_runs = field_codes.build_set_field_runs(name)
+            compensate_run = field_codes._make_text_run(' ' * len(captured))
+
+            parent = p
+            anchor = r
+            idx = list(parent).index(anchor)
+            parent.remove(anchor)
+
+            head_run = field_codes._make_text_run(head)
+            tail_run = field_codes._make_text_run(tail)
+
+            insert_idx = idx
+            for elem in [head_run] + field_runs + [compensate_run, tail_run]:
+                if elem is None:
+                    continue
+                parent.insert(insert_idx, elem)
+                insert_idx += 1
+            return True
+    return False
+
+
+def _next_available_path(base_path):
+    """生成不冲突的输出文件路径：a/_signed.docx → a/_signed_001.docx → ..."""
+    if not os.path.exists(base_path):
+        return base_path
+    directory = os.path.dirname(base_path)
+    name, ext = os.path.splitext(os.path.basename(base_path))
+    if not ext:
+        ext = '.docx'
+    counter = 1
+    while True:
+        candidate = os.path.join(directory, f"{name}_{counter:03d}{ext}")
+        if not os.path.exists(candidate):
+            return candidate
+        counter += 1
+        if counter > 999:
+            return os.path.join(directory, f"{name}_{datetime.now().strftime('%Y%m%d%H%M%S')}{ext}")
+
+
+def process_folder(folder, recursive=True, auto_sign=True, rules=None, request_options=None,
+                   log_source='批量', trace_id=None):
+    """批量处理整个文件夹。"""
+    files, skipped = scan_docx_files(folder, recursive=recursive)
+    summary = {
+        'scanned': len(files),
+        'processed': 0,
+        'skipped': len(skipped),
+        'failed': 0,
+        'files': [],
+    }
+    logger.info("[批量] 开始处理文件夹 %s (扫描 %d 个 docx, 跳过 %d 个非 docx)",
+                folder, len(files), len(skipped),
+                extra={'log_source': log_source, 'trace_id': trace_id})
+
+    for path in skipped:
+        summary['files'].append({
+            'path': path, 'status': 'skipped', 'markers': 0, 'output': '',
+            'warnings': ['类型不支持，仅处理 .docx']
+        })
+
+    for docx_path in files:
+        file_entry = {
+            'path': docx_path, 'status': 'running', 'markers': 0,
+            'output': '', 'warnings': []
+        }
+        try:
+            with open(docx_path, 'rb') as f:
+                docx_bytes = f.read()
+            doc = Document(io.BytesIO(docx_bytes))
+            replaced, warnings = process_document(
+                doc, rules or [], log_source=log_source, trace_id=trace_id
+            )
+            file_entry['markers'] = replaced
+            file_entry['warnings'].extend(warnings)
+
+            converted_buf = io.BytesIO()
+            doc.save(converted_buf)
+            converted_bytes = converted_buf.getvalue()
+
+            if auto_sign and replaced > 0:
+                base_name = os.path.splitext(os.path.basename(docx_path))[0]
+                docx_name = f"{base_name}_signed.docx"
+                sign_params = {
+                    'docx_name': docx_name,
+                    'match_mode': 'auto',
+                    'stamps': [],
+                }
+                sign_params.update(request_options or {})
+                result, error = sign_word(
+                    converted_bytes, sign_params,
+                    log_source=log_source, trace_id=trace_id
+                )
+                if error:
+                    file_entry['status'] = 'partial'
+                    file_entry['warnings'].append(f"自动签名失败: {error.get('message')}")
+                    final_bytes = converted_bytes
+                    output_name = f"{base_name}_signed.docx"
+                else:
+                    final_bytes = result['data']
+                    output_name = result['filename']
+                    file_entry['warnings'].extend(result.get('warnings', []))
+                    file_entry['status'] = 'success' if not result.get('warnings') else 'partial'
+            else:
+                final_bytes = converted_bytes
+                output_name = f"{os.path.splitext(os.path.basename(docx_path))[0]}_signed.docx"
+                file_entry['status'] = 'converted'
+
+            output_path = os.path.join(os.path.dirname(docx_path), output_name)
+            output_path = _next_available_path(output_path)
+            try:
+                with open(output_path, 'wb') as f:
+                    f.write(final_bytes)
+                file_entry['output'] = output_path
+                summary['processed'] += 1
+                logger.info("[批量] 文件处理完成: %s -> %s (标记 %d)",
+                            docx_path, output_path, replaced,
+                            extra={'log_source': log_source, 'trace_id': trace_id})
+            except PermissionError:
+                file_entry['status'] = 'skipped'
+                file_entry['warnings'].append('输出文件被锁定，跳过')
+                summary['skipped'] += 1
+                logger.warning("[批量] 输出文件锁定: %s", output_path,
+                               extra={'log_source': log_source, 'trace_id': trace_id})
+        except PermissionError:
+            file_entry['status'] = 'skipped'
+            file_entry['warnings'].append('源文件被锁定，跳过')
+            summary['skipped'] += 1
+            logger.warning("[批量] 源文件锁定: %s", docx_path,
+                           extra={'log_source': log_source, 'trace_id': trace_id})
+        except Exception as e:
+            file_entry['status'] = 'failed'
+            file_entry['warnings'].append(f"处理异常: {e}")
+            summary['failed'] += 1
+            logger.error("[批量] 文件处理失败 %s: %s", docx_path, str(e),
+                         extra={'log_source': log_source, 'trace_id': trace_id})
+        summary['files'].append(file_entry)
+
+    summary['success'] = summary['failed'] == 0
+    logger.info("[批量] 文件夹处理完成: 扫描 %d, 处理 %d, 跳过 %d, 失败 %d",
+                summary['scanned'], summary['processed'], summary['skipped'], summary['failed'],
+                extra={'log_source': log_source, 'trace_id': trace_id})
+    return summary

lib/config.py

@@ -0,0 +1,86 @@
+import configparser
+import os
+import logging
+
+logger = logging.getLogger(__name__)
+
+_config = None
+_config_path = None
+
+
+def load_config(path=None):
+    global _config, _config_path
+    if path is None:
+        path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'config.ini')
+    _config_path = path
+    _config = configparser.ConfigParser()
+    read_files = _config.read(path, encoding='utf-8')
+    if not read_files:
+        logger.warning("配置文件不存在，使用默认值: %s", path)
+    else:
+        logger.info("配置文件加载完成: %s", path)
+    return _config
+
+
+def get_config():
+    global _config
+    if _config is None:
+        load_config()
+    return _config
+
+
+def get_db_type():
+    cfg = get_config()
+    return cfg.get('database', 'type', fallback='mysql')
+
+
+def get_mysql_config():
+    cfg = get_config()
+    return {
+        'host': cfg.get('mysql', 'host', fallback='127.0.0.1'),
+        'port': cfg.getint('mysql', 'port', fallback=33366),
+        'user': cfg.get('mysql', 'user', fallback='admin'),
+        'password': cfg.get('mysql', 'password', fallback='admin'),
+        'database': cfg.get('mysql', 'database', fallback='word_sign'),
+    }
+
+
+def get_sqlite_config():
+    cfg = get_config()
+    return {
+        'path': cfg.get('sqlite', 'path', fallback='./data/word_sign.db'),
+    }
+
+
+def get_web_config():
+    cfg = get_config()
+    return {
+        'host': cfg.get('web', 'host', fallback='0.0.0.0'),
+        'port': cfg.getint('web', 'port', fallback=5001),
+        'debug': cfg.getboolean('web', 'debug', fallback=False),
+    }
+
+
+def get_log_config():
+    cfg = get_config()
+    return {
+        'level': cfg.get('log', 'level', fallback='INFO'),
+        'file_path': cfg.get('log', 'file_path', fallback='./logs/'),
+    }
+
+
+def get_limits_config():
+    cfg = get_config()
+    return {
+        'docx_max_bytes': cfg.getint('limits', 'docx_max_bytes', fallback=1024 * 1024),
+        'image_max_bytes': cfg.getint('limits', 'image_max_bytes', fallback=256 * 1024),
+        'stamp_max_count': cfg.getint('limits', 'stamp_max_count', fallback=100),
+        'log_max_entries': cfg.getint('limits', 'log_max_entries', fallback=1000),
+    }
+
+
+def get_oa_config():
+    cfg = get_config()
+    return {
+        'action_timeout_seconds': cfg.getint('oa', 'action_timeout_seconds', fallback=30),
+    }

lib/db.py

@@ -0,0 +1,570 @@
+import sqlite3
+import os
+import logging
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+_active_db = None
+_mysql_pool = None
+
+
+def _get_project_root():
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _get_sqlite_db_path():
+    from lib.config import get_sqlite_config
+    cfg = get_sqlite_config()
+    path = cfg['path']
+    if not os.path.isabs(path):
+        path = os.path.join(_get_project_root(), path)
+    return path
+
+
+def _try_mysql_connect():
+    try:
+        import pymysql
+    except ImportError:
+        logger.warning("pymysql 未安装, 无法使用 MySQL")
+        return None
+
+    from lib.config import get_mysql_config
+    cfg = get_mysql_config()
+
+    try:
+        conn = pymysql.connect(
+            host=cfg['host'],
+            port=cfg['port'],
+            user=cfg['user'],
+            password=cfg['password'],
+            database=cfg['database'],
+            charset='utf8mb4',
+            cursorclass=pymysql.cursors.DictCursor,
+            connect_timeout=5,
+        )
+        return conn
+    except Exception as e:
+        logger.warning("MySQL 连接失败: %s", str(e))
+        return None
+
+
+def _ensure_mysql_database():
+    try:
+        import pymysql
+    except ImportError:
+        return False
+
+    from lib.config import get_mysql_config
+    cfg = get_mysql_config()
+
+    try:
+        conn = pymysql.connect(
+            host=cfg['host'],
+            port=cfg['port'],
+            user=cfg['user'],
+            password=cfg['password'],
+            charset='utf8mb4',
+            cursorclass=pymysql.cursors.DictCursor,
+            connect_timeout=5,
+        )
+        with conn.cursor() as cursor:
+            cursor.execute(
+                "SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = %s",
+                (cfg['database'],)
+            )
+            if not cursor.fetchone():
+                cursor.execute(
+                    "CREATE DATABASE `%s` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci" % cfg['database']
+                )
+                conn.commit()
+                logger.info("MySQL 数据库 '%s' 创建成功", cfg['database'])
+        conn.close()
+        return True
+    except Exception as e:
+        logger.warning("MySQL 确保数据库存在失败: %s", str(e))
+        return False
+
+
+def _get_mysql_connection():
+    global _mysql_pool
+    if _mysql_pool is not None:
+        try:
+            _mysql_pool.ping(reconnect=True)
+            return _mysql_pool
+        except Exception:
+            _mysql_pool = None
+
+    conn = _try_mysql_connect()
+    if conn:
+        _mysql_pool = conn
+    return conn
+
+
+def _get_sqlite_connection():
+    db_path = _get_sqlite_db_path()
+    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA foreign_keys=ON")
+    return conn
+
+
+def get_connection():
+    if _active_db == 'mysql':
+        conn = _get_mysql_connection()
+        if conn:
+            return conn
+    if _active_db == 'sqlite':
+        return _get_sqlite_connection()
+    return None
+
+
+def init_database(strict_mysql=False):
+    """初始化数据库。
+
+    参数:
+    - ``strict_mysql``: True 表示严格按 SRS 15.3.4 — 配置了 mysql 但不可用时返回失败，
+      不静默回退 SQLite。False（启动期默认）允许回退以便开发模式可用。
+    """
+    global _active_db
+
+    from lib.config import get_db_type
+    db_type = get_db_type()
+
+    if db_type == 'mysql':
+        ok = _ensure_mysql_database()
+        conn = _try_mysql_connect() if ok else None
+        if conn:
+            _active_db = 'mysql'
+            _init_mysql_tables(conn)
+            conn.close()
+            logger.info("数据库初始化完成 (MySQL)")
+            return True
+        else:
+            if strict_mysql:
+                logger.error("MySQL 不可用且 strict_mysql=True，拒绝服务")
+                return False
+            logger.warning("MySQL 不可用, 回退到 SQLite")
+            _active_db = 'sqlite'
+    elif db_type == 'sqlite':
+        _active_db = 'sqlite'
+    else:
+        conn = _try_mysql_connect()
+        if conn:
+            _active_db = 'mysql'
+            _ensure_mysql_database()
+            _init_mysql_tables(conn)
+            conn.close()
+            logger.info("数据库初始化完成 (MySQL, 自动检测)")
+            return True
+        else:
+            if strict_mysql:
+                logger.error("MySQL 不可用且 strict_mysql=True，拒绝服务")
+                return False
+            logger.warning("MySQL 不可用, 回退到 SQLite")
+            _active_db = 'sqlite'
+
+    if _active_db == 'sqlite':
+        _init_sqlite_tables()
+        logger.info("数据库初始化完成 (SQLite): %s", _get_sqlite_db_path())
+    return True
+
+
+def ensure_mysql_available():
+    """SRS 15.3.4：API 路径在配置 type=mysql 时必须保证 MySQL 可用。
+
+    返回 True 表示可用；False 表示应返回 5xx。
+    """
+    from lib.config import get_db_type
+    db_type = get_db_type()
+    if db_type == 'sqlite':
+        return True
+    global _active_db
+    if _active_db == 'mysql':
+        conn = _get_mysql_connection()
+        return conn is not None
+    return False
+
+
+def _init_mysql_tables(conn):
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS user_sign (
+                    user_id VARCHAR(50) PRIMARY KEY,
+                    user_name VARCHAR(100) NOT NULL,
+                    sign_image LONGBLOB NOT NULL,
+                    is_signature TINYINT(1) NOT NULL DEFAULT 0,
+                    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
+                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+            """)
+        conn.commit()
+        logger.info("MySQL 数据表初始化完成")
+    except Exception as e:
+        logger.error("MySQL 数据表初始化失败: %s", str(e))
+        raise
+
+
+def _init_sqlite_tables():
+    conn = _get_sqlite_connection()
+    try:
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS user_sign (
+                user_id TEXT PRIMARY KEY,
+                user_name TEXT NOT NULL,
+                sign_image BLOB NOT NULL,
+                is_signature INTEGER NOT NULL DEFAULT 0,
+                create_time TEXT DEFAULT (datetime('now','localtime')),
+                update_time TEXT DEFAULT (datetime('now','localtime'))
+            )
+        """)
+        conn.commit()
+    except Exception as e:
+        conn.rollback()
+        logger.error("SQLite 数据表初始化失败: %s", str(e))
+        raise
+    finally:
+        conn.close()
+
+
+def query_by_user_id(user_id):
+    if _active_db == 'mysql':
+        return _mysql_query_by_user_id(user_id)
+    return _sqlite_query_by_user_id(user_id)
+
+
+def _mysql_query_by_user_id(user_id):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_query_by_user_id(user_id)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("SELECT * FROM user_sign WHERE user_id = %s", (user_id,))
+            row = cursor.fetchone()
+            return row if row else None
+    except Exception as e:
+        logger.error("MySQL 查询失败: %s", str(e))
+        return _sqlite_query_by_user_id(user_id)
+
+
+def _sqlite_query_by_user_id(user_id):
+    conn = _get_sqlite_connection()
+    try:
+        cursor = conn.execute("SELECT * FROM user_sign WHERE user_id = ?", (user_id,))
+        row = cursor.fetchone()
+        return dict(row) if row else None
+    finally:
+        conn.close()
+
+
+def query_by_user_name(user_name):
+    if _active_db == 'mysql':
+        return _mysql_query_by_user_name(user_name)
+    return _sqlite_query_by_user_name(user_name)
+
+
+def _mysql_query_by_user_name(user_name):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_query_by_user_name(user_name)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("SELECT * FROM user_sign WHERE user_name = %s", (user_name,))
+            return cursor.fetchall()
+    except Exception as e:
+        logger.error("MySQL 查询失败: %s", str(e))
+        return _sqlite_query_by_user_name(user_name)
+
+
+def _sqlite_query_by_user_name(user_name):
+    conn = _get_sqlite_connection()
+    try:
+        cursor = conn.execute("SELECT * FROM user_sign WHERE user_name = ?", (user_name,))
+        return [dict(r) for r in cursor.fetchall()]
+    finally:
+        conn.close()
+
+
+def query_all(page=1, page_size=20, keyword=None):
+    if _active_db == 'mysql':
+        return _mysql_query_all(page, page_size, keyword)
+    return _sqlite_query_all(page, page_size, keyword)
+
+
+def _mysql_query_all(page=1, page_size=20, keyword=None):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_query_all(page, page_size, keyword)
+    try:
+        offset = (page - 1) * page_size
+        with conn.cursor() as cursor:
+            if keyword:
+                cursor.execute(
+                    "SELECT COUNT(*) as total FROM user_sign WHERE user_id LIKE %s OR user_name LIKE %s",
+                    (f'%{keyword}%', f'%{keyword}%')
+                )
+                total = cursor.fetchone()['total']
+                cursor.execute(
+                    "SELECT user_id, user_name, is_signature, create_time, update_time FROM user_sign WHERE user_id LIKE %s OR user_name LIKE %s ORDER BY create_time DESC LIMIT %s OFFSET %s",
+                    (f'%{keyword}%', f'%{keyword}%', page_size, offset)
+                )
+            else:
+                cursor.execute("SELECT COUNT(*) as total FROM user_sign")
+                total = cursor.fetchone()['total']
+                cursor.execute(
+                    "SELECT user_id, user_name, is_signature, create_time, update_time FROM user_sign ORDER BY create_time DESC LIMIT %s OFFSET %s",
+                    (page_size, offset)
+                )
+            rows = cursor.fetchall()
+        return {'total': total, 'page': page, 'page_size': page_size, 'data': rows}
+    except Exception as e:
+        logger.error("MySQL 查询失败: %s", str(e))
+        return _sqlite_query_all(page, page_size, keyword)
+
+
+def _sqlite_query_all(page=1, page_size=20, keyword=None):
+    conn = _get_sqlite_connection()
+    try:
+        offset = (page - 1) * page_size
+        if keyword:
+            count_row = conn.execute(
+                "SELECT COUNT(*) as total FROM user_sign WHERE user_id LIKE ? OR user_name LIKE ?",
+                (f'%{keyword}%', f'%{keyword}%')
+            ).fetchone()
+            total = count_row['total']
+            cursor = conn.execute(
+                "SELECT user_id, user_name, is_signature, create_time, update_time FROM user_sign WHERE user_id LIKE ? OR user_name LIKE ? ORDER BY create_time DESC LIMIT ? OFFSET ?",
+                (f'%{keyword}%', f'%{keyword}%', page_size, offset)
+            )
+        else:
+            count_row = conn.execute("SELECT COUNT(*) as total FROM user_sign").fetchone()
+            total = count_row['total']
+            cursor = conn.execute(
+                "SELECT user_id, user_name, is_signature, create_time, update_time FROM user_sign ORDER BY create_time DESC LIMIT ? OFFSET ?",
+                (page_size, offset)
+            )
+        rows = [dict(r) for r in cursor.fetchall()]
+        return {'total': total, 'page': page, 'page_size': page_size, 'data': rows}
+    finally:
+        conn.close()
+
+
+def insert_record(user_id, user_name, sign_image, is_signature=0):
+    if _active_db == 'mysql':
+        return _mysql_insert_record(user_id, user_name, sign_image, is_signature)
+    return _sqlite_insert_record(user_id, user_name, sign_image, is_signature)
+
+
+def _mysql_insert_record(user_id, user_name, sign_image, is_signature=0):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_insert_record(user_id, user_name, sign_image, is_signature)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute(
+                "INSERT INTO user_sign (user_id, user_name, sign_image, is_signature) VALUES (%s, %s, %s, %s)",
+                (user_id, user_name, sign_image, is_signature)
+            )
+        conn.commit()
+        logger.info("新增记录(MySQL): user_id=%s, user_name=%s", user_id, user_name)
+        return True
+    except Exception as e:
+        conn.rollback()
+        if 'Duplicate entry' in str(e) or 'PRIMARY' in str(e):
+            logger.error("新增记录失败(主键冲突): user_id=%s, %s", user_id, str(e))
+            raise ValueError(f"人员ID '{user_id}' 已存在")
+        logger.error("MySQL 新增记录失败: %s", str(e))
+        raise
+    except:
+        conn.rollback()
+        raise
+
+
+def _sqlite_insert_record(user_id, user_name, sign_image, is_signature=0):
+    conn = _get_sqlite_connection()
+    try:
+        conn.execute(
+            "INSERT INTO user_sign (user_id, user_name, sign_image, is_signature) VALUES (?, ?, ?, ?)",
+            (user_id, user_name, sign_image, is_signature)
+        )
+        conn.commit()
+        logger.info("新增记录(SQLite): user_id=%s, user_name=%s", user_id, user_name)
+        return True
+    except sqlite3.IntegrityError as e:
+        conn.rollback()
+        logger.error("新增记录失败(主键冲突): user_id=%s, %s", user_id, str(e))
+        raise ValueError(f"人员ID '{user_id}' 已存在")
+    except Exception as e:
+        conn.rollback()
+        logger.error("新增记录失败: %s", str(e))
+        raise
+    finally:
+        conn.close()
+
+
+def update_record(user_id, user_name=None, sign_image=None, is_signature=None):
+    if _active_db == 'mysql':
+        return _mysql_update_record(user_id, user_name, sign_image, is_signature)
+    return _sqlite_update_record(user_id, user_name, sign_image, is_signature)
+
+
+def _mysql_update_record(user_id, user_name=None, sign_image=None, is_signature=None):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_update_record(user_id, user_name, sign_image, is_signature)
+    try:
+        sets = []
+        params = []
+        if user_name is not None:
+            sets.append("user_name = %s")
+            params.append(user_name)
+        if sign_image is not None:
+            sets.append("sign_image = %s")
+            params.append(sign_image)
+        if is_signature is not None:
+            sets.append("is_signature = %s")
+            params.append(is_signature)
+        if not sets:
+            return True
+        params.append(user_id)
+        with conn.cursor() as cursor:
+            cursor.execute(f"UPDATE user_sign SET {', '.join(sets)} WHERE user_id = %s", params)
+        conn.commit()
+        logger.info("更新记录(MySQL): user_id=%s", user_id)
+        return True
+    except Exception as e:
+        conn.rollback()
+        logger.error("MySQL 更新记录失败: %s", str(e))
+        raise
+
+
+def _sqlite_update_record(user_id, user_name=None, sign_image=None, is_signature=None):
+    conn = _get_sqlite_connection()
+    try:
+        sets = []
+        params = []
+        if user_name is not None:
+            sets.append("user_name = ?")
+            params.append(user_name)
+        if sign_image is not None:
+            sets.append("sign_image = ?")
+            params.append(sign_image)
+        if is_signature is not None:
+            sets.append("is_signature = ?")
+            params.append(is_signature)
+        if not sets:
+            return True
+        sets.append("update_time = datetime('now','localtime')")
+        params.append(user_id)
+        conn.execute(f"UPDATE user_sign SET {', '.join(sets)} WHERE user_id = ?", params)
+        conn.commit()
+        logger.info("更新记录(SQLite): user_id=%s", user_id)
+        return True
+    except Exception as e:
+        conn.rollback()
+        logger.error("更新记录失败: %s", str(e))
+        raise
+    finally:
+        conn.close()
+
+
+def delete_record(user_id):
+    if _active_db == 'mysql':
+        return _mysql_delete_record(user_id)
+    return _sqlite_delete_record(user_id)
+
+
+def _mysql_delete_record(user_id):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_delete_record(user_id)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("DELETE FROM user_sign WHERE user_id = %s", (user_id,))
+        conn.commit()
+        logger.info("删除记录(MySQL): user_id=%s", user_id)
+        return True
+    except Exception as e:
+        conn.rollback()
+        logger.error("MySQL 删除记录失败: %s", str(e))
+        raise
+
+
+def _sqlite_delete_record(user_id):
+    conn = _get_sqlite_connection()
+    try:
+        conn.execute("DELETE FROM user_sign WHERE user_id = ?", (user_id,))
+        conn.commit()
+        logger.info("删除记录(SQLite): user_id=%s", user_id)
+        return True
+    except Exception as e:
+        conn.rollback()
+        logger.error("删除记录失败: %s", str(e))
+        raise
+    finally:
+        conn.close()
+
+
+def get_image_by_user_id(user_id):
+    if _active_db == 'mysql':
+        return _mysql_get_image_by_user_id(user_id)
+    return _sqlite_get_image_by_user_id(user_id)
+
+
+def _mysql_get_image_by_user_id(user_id):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_get_image_by_user_id(user_id)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("SELECT sign_image, is_signature FROM user_sign WHERE user_id = %s", (user_id,))
+            row = cursor.fetchone()
+            return row if row else None
+    except Exception as e:
+        logger.error("MySQL 查询图片失败: %s", str(e))
+        return _sqlite_get_image_by_user_id(user_id)
+
+
+def _sqlite_get_image_by_user_id(user_id):
+    conn = _get_sqlite_connection()
+    try:
+        cursor = conn.execute("SELECT sign_image, is_signature FROM user_sign WHERE user_id = ?", (user_id,))
+        row = cursor.fetchone()
+        return dict(row) if row else None
+    finally:
+        conn.close()
+
+
+def get_image_by_user_name(user_name):
+    if _active_db == 'mysql':
+        return _mysql_get_image_by_user_name(user_name)
+    return _sqlite_get_image_by_user_name(user_name)
+
+
+def _mysql_get_image_by_user_name(user_name):
+    conn = _get_mysql_connection()
+    if not conn:
+        return _sqlite_get_image_by_user_name(user_name)
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute("SELECT sign_image, is_signature, user_id, user_name FROM user_sign WHERE user_name = %s", (user_name,))
+            return cursor.fetchall()
+    except Exception as e:
+        logger.error("MySQL 查询图片失败: %s", str(e))
+        return _sqlite_get_image_by_user_name(user_name)
+
+
+def _sqlite_get_image_by_user_name(user_name):
+    conn = _get_sqlite_connection()
+    try:
+        cursor = conn.execute("SELECT sign_image, is_signature, user_id, user_name FROM user_sign WHERE user_name = ?", (user_name,))
+        return [dict(r) for r in cursor.fetchall()]
+    finally:
+        conn.close()
+
+
+def get_active_db_type():
+    return _active_db or 'unknown'

lib/field_codes.py

@@ -0,0 +1,837 @@
+"""Word SET 域代码处理工具。
+
+支持两类域代码定位：
+1. ``w:fldSimple`` 简单域，``w:instr`` 属性内含 ``SET USER_xxx``。
+2. ``w:instrText`` + ``w:fldChar`` 组合的复杂域。
+
+提供:
+- :func:`scan_user_markers` 扫描文档中所有 ``SET USER_<name>`` 标记。
+- :func:`insert_image_after_field` 在指定标记的域之后插入签名图片。
+- :func:`build_set_field_xml` 构造一段 ``SET USER_<name> \\* MERGEFORMAT`` 域代码 XML。
+- :func:`replace_text_with_set_field` 将段落中命中的可见文本替换为 SET 域 + 宽度补偿空格。
+"""
+import re
+import logging
+from copy import deepcopy
+
+from docx.oxml.ns import qn
+from docx.oxml import OxmlElement
+from docx.shared import Cm, Emu
+
+logger = logging.getLogger(__name__)
+
+USER_MARKER_RE = re.compile(r'\bSET\s+USER_(\S+)', re.IGNORECASE)
+USER_MARKER_RE_CLEAN = re.compile(r'^USER_(\S+?)\s*(?:\\.*)?$')
+
+W_NS = 'http://schemas.openxmlformats.org/wordprocessingml/2006/main'
+WP_NS = 'http://schemas.openxmlformats.org/drawingml/2006/wordprocessingDrawing'
+A_NS = 'http://schemas.openxmlformats.org/drawingml/2006/main'
+PIC_NS = 'http://schemas.openxmlformats.org/drawingml/2006/picture'
+R_NS = 'http://schemas.openxmlformats.org/officeDocument/2006/relationships'
+
+
+def _strip_switches(name):
+    """去除 ``\\* MERGEFORMAT`` 等 Word 开关，并 trim 末尾空白。
+
+    输入示例:
+    - ``USER_zhangsan`` -> ``zhangsan``
+    - ``USER_zhangsan \\* MERGEFORMAT`` -> ``zhangsan``
+    """
+    if name is None:
+        return ''
+    txt = name.strip()
+    if txt.startswith('USER_'):
+        txt = txt[5:]
+    txt = txt.split('\\', 1)[0]
+    return txt.strip()
+
+
+def _iter_all_paragraphs(doc):
+    """递归遍历文档中所有段落：正文、表格（含嵌套）、文本框、页眉页脚。
+
+    返回 ``(paragraph_element, part)`` 元组，``part`` 用于 ``add_picture`` 关系绑定。
+    """
+    body = doc.element.body
+    doc_part = doc.part
+    for para in body.iter(qn('w:p')):
+        yield para, doc_part
+
+    for section in doc.sections:
+        for hdr in (section.header, section.first_page_header, section.even_page_header):
+            if hdr is not None:
+                hdr_part = hdr.part
+                for para in hdr._element.iter(qn('w:p')):
+                    yield para, hdr_part
+        for ftr in (section.footer, section.first_page_footer, section.even_page_footer):
+            if ftr is not None:
+                ftr_part = ftr.part
+                for para in ftr._element.iter(qn('w:p')):
+                    yield para, ftr_part
+
+
+def _iter_all_paragraph_elements(doc):
+    """仅迭代段落元素（不带 part）。"""
+    for para, _part in _iter_all_paragraphs(doc):
+        yield para
+
+
+def _iter_all_tables(doc):
+    """递归遍历所有表格，包括正文、文本框、页眉页脚中的嵌套表格。"""
+    body = doc.element.body
+    for tbl in body.iter(qn('w:tbl')):
+        yield tbl
+
+    for section in doc.sections:
+        for hdr in (section.header, section.first_page_header, section.even_page_header):
+            if hdr is not None:
+                for tbl in hdr._element.iter(qn('w:tbl')):
+                    yield tbl
+        for ftr in (section.footer, section.first_page_footer, section.even_page_footer):
+            if ftr is not None:
+                for tbl in ftr._element.iter(qn('w:tbl')):
+                    yield tbl
+
+
+def scan_user_markers(doc, log_source='前端操作'):
+    """扫描文档中所有 ``SET USER_<name>`` 标记，返回去重后的 ``[{marker, name}]`` 列表。"""
+    found = []
+    seen_keys = set()
+
+    for para, _part in _iter_all_paragraphs(doc):
+        for elem in para.iter():
+            tag = elem.tag
+            if tag == qn('w:fldSimple'):
+                instr = elem.get(qn('w:instr')) or ''
+                m = USER_MARKER_RE.search(instr)
+                if m:
+                    name = _strip_switches(m.group(1))
+                    if not name:
+                        continue
+                    key = name.lower()
+                    if key in seen_keys:
+                        continue
+                    seen_keys.add(key)
+                    found.append({'marker': f'USER_{name}', 'name': name, 'kind': 'simple'})
+            elif tag == qn('w:instrText'):
+                text = ''.join(elem.itertext()) if elem.text is None else elem.text
+                if not text:
+                    continue
+                m = USER_MARKER_RE.search(text)
+                if m:
+                    name = _strip_switches(m.group(1))
+                    if not name:
+                        continue
+                    key = name.lower()
+                    if key in seen_keys:
+                        continue
+                    seen_keys.add(key)
+                    found.append({'marker': f'USER_{name}', 'name': name, 'kind': 'complex'})
+
+    logger.info("[auto扫描] 发现 %d 个 SET USER_<name> 标记", len(found),
+                extra={'log_source': log_source})
+    return found
+
+
+def _find_field_elements_for_marker(doc, marker_name):
+    """根据 marker 名称（如 ``USER_zhangsan`` 或 ``zhangsan``）查找域代码宿主元素。
+
+    返回 ``(kind, parent_paragraph, fld_element_or_run, part)`` 列表，供插入图片使用。
+    """
+    target = marker_name
+    if not target.startswith('USER_'):
+        target_full = f'USER_{target}'
+    else:
+        target_full = target
+    target_name = target_full[5:] if target_full.startswith('USER_') else target_full
+    target_lc = target_name.lower()
+
+    results = []
+
+    for para, part in _iter_all_paragraphs(doc):
+        for elem in para.iter():
+            tag = elem.tag
+            if tag == qn('w:fldSimple'):
+                instr = elem.get(qn('w:instr')) or ''
+                m = USER_MARKER_RE.search(instr)
+                if m:
+                    name = _strip_switches(m.group(1))
+                    if name.lower() == target_lc:
+                        results.append(('simple', para, elem, part))
+            elif tag == qn('w:instrText'):
+                text = elem.text or ''
+                m = USER_MARKER_RE.search(text)
+                if m:
+                    name = _strip_switches(m.group(1))
+                    if name.lower() == target_lc:
+                        results.append(('complex', para, elem, part))
+
+    return results
+
+
+def _build_run_with_drawing(image_path, width_cm, height_cm):
+    """构造一个包含图片 drawing 的 ``<w:r>`` 元素。
+
+    保留作为低层 helper；正式插入使用 :func:`_insert_picture_into_paragraph_after_anchor`，
+    通过 python-docx 的 ``Run.add_picture`` 完成关系绑定与 drawing XML。
+    """
+    run = OxmlElement('w:r')
+    return run
+
+
+def _add_picture_run_to_paragraph(paragraph_element, doc_part, image_path, width_cm, height_cm):
+    """在指定段落元素的末尾添加一个图片 run。
+
+    使用 python-docx 的 Paragraph/Run 高层 API，避免手动维护 rels。
+    """
+    from docx.text.paragraph import Paragraph
+    from docx.shared import Cm as _Cm
+    parent = paragraph_element.getparent()
+    para_obj = Paragraph(paragraph_element, parent)
+    run_obj = para_obj.add_run()
+    run_obj.add_picture(image_path, width=_Cm(width_cm), height=_Cm(height_cm))
+    return run_obj
+
+
+def insert_image_after_field(doc, marker, image_path, width_cm, height_cm,
+                             log_source='前端操作', relax_layout=True):
+    """在指定 ``USER_<name>`` 域之后插入签名图片。
+
+    返回插入的图片数量。
+
+    ``relax_layout=True``（默认）时，若标记段落或所在表格行有固定高度约束
+    （``lineRule="exact"`` / ``hRule="exact"``）小于签名需要的高度，
+    自动把约束放宽为 ``atLeast`` 并上调数值，避免 Word 硬裁剪图片。
+    设为 ``False`` 时严格保留原文档布局，仅靠等比例缩放把图片压进约束内。
+    """
+    target_name = marker
+    if target_name.startswith('USER_'):
+        target_name = target_name[5:]
+    target_name = _strip_switches(target_name)
+
+    matches = _find_field_elements_for_marker(doc, target_name)
+    if not matches:
+        logger.warning("[Word处理] 未找到标记 %s 的域代码", marker, extra={'log_source': log_source})
+        return 0
+
+    # 提前读取文档级默认字号（styles.xml docDefaults），用于段落无 sz 时的 fallback
+    doc_default_pt = _get_doc_default_font_size_pt(doc)
+
+    inserted = 0
+    from docx.text.paragraph import Paragraph
+    from docx.shared import Cm as _Cm
+
+    for kind, para_xml, host_elem, part in matches:
+        try:
+            if kind == 'simple':
+                anchor_elem = host_elem
+            else:
+                end_elem = _find_complex_field_end(para_xml, host_elem)
+                anchor_elem = end_elem if end_elem is not None else host_elem
+
+            # 1. 计算目标高度（用户指定或 auto 推算）
+            cur_h = height_cm
+            cur_w = width_cm
+            if not cur_h or cur_h <= 0:
+                max_h_cm_initial = _detect_paragraph_max_height_cm(para_xml)
+                font_pt = _detect_paragraph_font_size_pt(para_xml)
+                if font_pt is None:
+                    font_pt = doc_default_pt
+                    logger.info("[Word处理] 段落无显式字号，使用文档默认字号 %spt",
+                                f"{font_pt}" if font_pt else "无",
+                                extra={'log_source': log_source})
+                cur_h = _auto_size_height_cm(max_h_cm_initial, font_pt)
+                # 按 image 实际宽高比反推 width_cm，保持比例
+                try:
+                    from PIL import Image as _PILImage
+                    with _PILImage.open(image_path) as _im:
+                        _iw, _ih = _im.size
+                    if _ih and cur_h:
+                        cur_w = cur_h * (_iw / _ih)
+                except Exception:
+                    cur_w = None
+                logger.info("[Word处理] 自动尺寸: 字号=%spt → 推算 height=%.2f cm",
+                            f"{font_pt}" if font_pt else "无",
+                            cur_h, extra={'log_source': log_source})
+
+            # 2. 放宽约束（默认开启）：把小于签名高度的 exact 约束改成 atLeast
+            if relax_layout and cur_h and cur_h > 0:
+                _relax_paragraph_for_image(para_xml, cur_h, log_source)
+                _relax_table_row_for_image(para_xml, cur_h, log_source)
+
+            # 3. 重新检测约束（放宽后 max_h_cm 可能变成 None）
+            max_h_cm = _detect_paragraph_max_height_cm(para_xml)
+            max_w_cm = _detect_table_cell_max_width_cm(para_xml)
+            left_cm, right_cm = _detect_paragraph_indent_cm(para_xml)
+            if max_w_cm and (left_cm or right_cm):
+                max_w_cm = max(0.5, max_w_cm - left_cm - right_cm)
+
+            # 4. 兜底缩放（处理 atLeast/缩进等软约束；或 relax_layout=False 时走原逻辑）
+            final_w, final_h = _scale_to_fit(cur_w, cur_h, max_w_cm, max_h_cm)
+            if (final_w, final_h) != (cur_w, cur_h):
+                logger.info("[Word处理] 图片缩放以适配插入位置: %.2fx%.2f cm -> %.2fx%.2f cm "
+                            "(段落行高上限=%s, 单元格宽上限=%s, 左右缩进=%.2f+%.2fcm)",
+                            cur_w or 0, cur_h or 0, final_w or 0, final_h or 0,
+                            f"{max_h_cm:.2f}cm" if max_h_cm else "无",
+                            f"{max_w_cm:.2f}cm" if max_w_cm else "无",
+                            left_cm, right_cm,
+                            extra={'log_source': log_source})
+
+            parent_xml = para_xml.getparent()
+            para_obj = Paragraph(para_xml, part)
+            run_obj = para_obj.add_run()
+            run_obj.add_picture(image_path, width=_Cm(final_w), height=_Cm(final_h))
+
+            new_run_xml = run_obj._element
+            anchor_elem.addnext(new_run_xml)
+            inserted += 1
+            logger.info("[Word处理] 在 %s 域后插入图片成功", marker, extra={'log_source': log_source})
+        except Exception as e:
+            logger.error("[Word处理] 插入图片失败 %s: %s", marker, str(e), extra={'log_source': log_source})
+
+    return inserted
+
+
+_TWIPS_PER_CM = 1440.0 / 2.54  # 1 cm ≈ 566.93 twips
+
+
+def _detect_paragraph_max_height_cm(para_xml):
+    """从段落属性读取固定行高（lineRule="exact"），返回 cm 或 None。
+
+    Word 段落 ``<w:pPr><w:spacing w:line="..." w:lineRule="exact"/></w:pPr>`` 中，
+    line 字段在 exact 模式下单位为 1/20 pt (twips)。其他模式（auto/atLeast）不视为硬约束。
+    """
+    pPr = para_xml.find(qn('w:pPr'))
+    if pPr is None:
+        return None
+    spacing = pPr.find(qn('w:spacing'))
+    if spacing is None:
+        return None
+    line = spacing.get(qn('w:line'))
+    line_rule = spacing.get(qn('w:lineRule'))
+    if not line or line_rule != 'exact':
+        return None
+    try:
+        return int(line) / _TWIPS_PER_CM
+    except (ValueError, TypeError):
+        return None
+
+
+def _detect_table_row_height_rule(para_xml):
+    """读取段落所在表格行的固定高度约束。
+
+    向上查找最近的 ``<w:tr>``，读 ``<w:trPr><w:trHeight w:val w:hRule/></w:trPr>``。
+    返回 ``(val_cm, rule)`` —— ``rule ∈ {'exact', 'atLeast', None}``。
+    按 OOXML 规范，``hRule`` 缺省为 ``atLeast``（不裁剪），仅 ``exact`` 视为硬约束。
+    """
+    parent = para_xml.getparent()
+    while parent is not None and parent.tag != qn('w:tr'):
+        parent = parent.getparent()
+    if parent is None or parent.tag != qn('w:tr'):
+        return None, None
+    trPr = parent.find(qn('w:trPr'))
+    if trPr is None:
+        return None, None
+    trHeight = trPr.find(qn('w:trHeight'))
+    if trHeight is None:
+        return None, None
+    val = trHeight.get(qn('w:val'))
+    rule = trHeight.get(qn('w:hRule')) or 'atLeast'
+    if not val:
+        return None, rule
+    try:
+        val_cm = int(val) / _TWIPS_PER_CM
+    except (ValueError, TypeError):
+        return None, rule
+    return val_cm, rule
+
+
+def _detect_paragraph_indent_cm(para_xml):
+    """读段落左右缩进 ``<w:ind w:left w:right>``，返回 ``(left_cm, right_cm)``。"""
+    pPr = para_xml.find(qn('w:pPr'))
+    if pPr is None:
+        return 0.0, 0.0
+    ind = pPr.find(qn('w:ind'))
+    if ind is None:
+        return 0.0, 0.0
+    left = ind.get(qn('w:left')) or ind.get(qn('w:start')) or '0'
+    right = ind.get(qn('w:right')) or ind.get(qn('w:end')) or '0'
+    try:
+        return int(left) / _TWIPS_PER_CM, int(right) / _TWIPS_PER_CM
+    except (ValueError, TypeError):
+        return 0.0, 0.0
+
+
+def _relax_paragraph_for_image(para_xml, needed_h_cm, log_source):
+    """确保段落行高至少能容纳签名高度，避免 inline 图片被行盒子裁切。
+
+    inline 图片默认骑在文字 baseline 上方，所需高度 = 图片高度。
+    若段落行高（由 spacing/@line + lineRule 决定，或基于字号默认推算）小于该值，
+    baseline 上方会溢出并被 Word 裁掉（典型表现：图片上方被切）。
+
+    处理所有 spacing 状态：
+    - 无 pPr / 无 spacing → 主动创建 ``spacing line=needed atLeast``
+    - ``lineRule='exact'`` 且 line < needed → 改 ``atLeast`` + 上调 line
+    - ``lineRule='atLeast'`` 且 line < needed → 上调 line
+    - ``lineRule='auto'`` / 缺省 → 改 ``atLeast`` + 设 line = needed
+      （默认行高基于字号，对 inline 图片通常不够）
+    - 已 ≥ needed → no-op
+
+    返回是否发生过改动。
+    """
+    import math
+    if not needed_h_cm or needed_h_cm <= 0:
+        return False
+    needed_twips = int(math.ceil(needed_h_cm * _TWIPS_PER_CM))
+
+    pPr = para_xml.find(qn('w:pPr'))
+    if pPr is None:
+        pPr = OxmlElement('w:pPr')
+        para_xml.insert(0, pPr)
+
+    spacing = pPr.find(qn('w:spacing'))
+    if spacing is None:
+        spacing = OxmlElement('w:spacing')
+        spacing.set(qn('w:line'), str(needed_twips))
+        spacing.set(qn('w:lineRule'), 'atLeast')
+        pPr.append(spacing)
+        logger.info("[Word处理] 段落无显式行高，新增 spacing atLeast/%d twips (%.2fcm) "
+                    "以容纳 %.2fcm 高签名",
+                    needed_twips, needed_twips / _TWIPS_PER_CM, needed_h_cm,
+                    extra={'log_source': log_source})
+        return True
+
+    line = spacing.get(qn('w:line'))
+    line_rule = spacing.get(qn('w:lineRule')) or 'auto'
+
+    if line:
+        try:
+            line_twips = int(line)
+            if line_rule in ('exact', 'atLeast') and line_twips >= needed_twips:
+                return False
+        except (ValueError, TypeError):
+            pass
+
+    old_rule = line_rule
+    old_line = line or 'auto'
+    spacing.set(qn('w:lineRule'), 'atLeast')
+    spacing.set(qn('w:line'), str(needed_twips))
+    logger.info("[Word处理] 放宽段落行高: lineRule %s→atLeast, line %s→%d twips "
+                "(%.2fcm) 以容纳 %.2fcm 高签名",
+                old_rule, old_line, needed_twips,
+                needed_twips / _TWIPS_PER_CM, needed_h_cm,
+                extra={'log_source': log_source})
+    return True
+
+
+def _relax_table_row_for_image(para_xml, needed_h_cm, log_source):
+    """放宽表格行**硬高度约束**，避免图片被表格行盒子裁切。
+
+    OOXML 规范下表格行高规则：
+    - ``hRule='exact'``：行高**精确**为 val，超过部分被裁切 ← 硬约束
+    - ``hRule='atLeast'`` 或缺省：行高**至少**为 val，可自动撑高 ← 软约束，不裁切
+
+    因此本函数仅在 ``hRule='exact'`` 且 ``val < needed_h_cm`` 时介入：
+    - 改 ``hRule`` 为 ``atLeast`` + 上调 ``val`` 到 needed_h_cm
+
+    其他情况（无 trHeight、atLeast、已 ≥ needed、auto） → no-op。
+    """
+    import math
+    if not needed_h_cm or needed_h_cm <= 0:
+        return False
+
+    parent = para_xml.getparent()
+    while parent is not None and parent.tag != qn('w:tr'):
+        parent = parent.getparent()
+    if parent is None or parent.tag != qn('w:tr'):
+        return False
+
+    trPr = parent.find(qn('w:trPr'))
+    if trPr is None:
+        return False
+    trHeight = trPr.find(qn('w:trHeight'))
+    if trHeight is None:
+        return False
+
+    val = trHeight.get(qn('w:val'))
+    rule = trHeight.get(qn('w:hRule')) or 'atLeast'
+    if rule != 'exact' or not val:
+        return False
+    try:
+        val_cm = int(val) / _TWIPS_PER_CM
+    except (ValueError, TypeError):
+        return False
+    if val_cm >= needed_h_cm:
+        return False
+
+    needed_twips = int(math.ceil(needed_h_cm * _TWIPS_PER_CM))
+    trHeight.set(qn('w:hRule'), 'atLeast')
+    trHeight.set(qn('w:val'), str(needed_twips))
+    logger.info("[Word处理] 放宽表格行高: hRule exact→atLeast, val %s→%d twips "
+                "(%.2fcm → %.2fcm) 以容纳 %.2fcm 高签名",
+                val, needed_twips, val_cm, needed_twips / _TWIPS_PER_CM, needed_h_cm,
+                extra={'log_source': log_source})
+    return True
+
+
+def _detect_table_cell_max_width_cm(para_xml):
+    """若段落位于表格单元格内，读取单元格宽度，返回 cm 或 None。"""
+    parent = para_xml.getparent()
+    while parent is not None and parent.tag != qn('w:tc'):
+        parent = parent.getparent()
+    if parent is None or parent.tag != qn('w:tc'):
+        return None
+    tcPr = parent.find(qn('w:tcPr'))
+    if tcPr is None:
+        return None
+    tcW = tcPr.find(qn('w:tcW'))
+    if tcW is None:
+        return None
+    w_val = tcW.get(qn('w:w'))
+    w_type = tcW.get(qn('w:type'))
+    if not w_val or w_type == 'pct':
+        return None
+    try:
+        return int(w_val) / _TWIPS_PER_CM
+    except (ValueError, TypeError):
+        return None
+
+
+def _scale_to_fit(width_cm, height_cm, max_w_cm, max_h_cm):
+    """等比例缩放图片，使其宽高都不超过约束。返回 (final_w, final_h)。"""
+    if not width_cm or not height_cm:
+        return width_cm, height_cm
+    ratio = 1.0
+    if max_w_cm and width_cm > max_w_cm:
+        ratio = min(ratio, max_w_cm / width_cm)
+    if max_h_cm and height_cm > max_h_cm:
+        ratio = min(ratio, max_h_cm / height_cm)
+    return width_cm * ratio, height_cm * ratio
+
+
+def _detect_paragraph_font_size_pt(para_xml):
+    """读取段落字体大小（point）。
+
+    优先级：
+    1. 段落第一个有 ``<w:rPr><w:sz>`` 的 run
+    2. 段落标记的 rPr（``<w:pPr><w:rPr><w:sz>``）
+
+    ``w:sz`` 的 val 是 half-points（如 21 = 10.5pt）。
+    找不到返回 None（由调用方决定 fallback）。
+    """
+    for r in para_xml.iter(qn('w:r')):
+        rPr = r.find(qn('w:rPr'))
+        if rPr is None:
+            continue
+        sz = rPr.find(qn('w:sz'))
+        if sz is not None:
+            val = sz.get(qn('w:val'))
+            if val:
+                try:
+                    pt = int(val) / 2.0
+                    if 4.0 <= pt <= 96.0:
+                        return pt
+                except (ValueError, TypeError):
+                    pass
+            break
+
+    pPr = para_xml.find(qn('w:pPr'))
+    if pPr is not None:
+        rPr = pPr.find(qn('w:rPr'))
+        if rPr is not None:
+            sz = rPr.find(qn('w:sz'))
+            if sz is not None:
+                val = sz.get(qn('w:val'))
+                if val:
+                    try:
+                        pt = int(val) / 2.0
+                        if 4.0 <= pt <= 96.0:
+                            return pt
+                    except (ValueError, TypeError):
+                        pass
+    return None
+
+
+def _get_doc_default_font_size_pt(doc):
+    """读文档级默认字号，按 Word 实际渲染优先级。
+
+    优先级：
+    1. ``Normal`` 样式（默认段落样式）的 ``<w:sz>`` —— Word 渲染段落时实际使用
+    2. ``<w:docDefaults><w:rPrDefault><w:rPr><w:sz>`` —— 全局兜底
+
+    找不到返回 None（由调用方做最终 fallback）。
+    """
+    try:
+        styles_element = doc.styles.element
+    except Exception:
+        return None
+
+    # 1. 先读 Normal 样式
+    for style in styles_element.iter(qn('w:style')):
+        if style.get(qn('w:styleId')) != 'Normal':
+            continue
+        rPr = style.find(qn('w:rPr'))
+        if rPr is not None:
+            sz = rPr.find(qn('w:sz'))
+            if sz is not None:
+                val = sz.get(qn('w:val'))
+                if val:
+                    try:
+                        pt = int(val) / 2.0
+                        if 4.0 <= pt <= 96.0:
+                            return pt
+                    except (ValueError, TypeError):
+                        pass
+        break
+
+    # 2. 再读 docDefaults
+    doc_defaults = styles_element.find(qn('w:docDefaults'))
+    if doc_defaults is None:
+        return None
+    rPr_default = doc_defaults.find(qn('w:rPrDefault'))
+    if rPr_default is None:
+        return None
+    rPr = rPr_default.find(qn('w:rPr'))
+    if rPr is None:
+        return None
+    sz = rPr.find(qn('w:sz'))
+    if sz is None:
+        return None
+    val = sz.get(qn('w:val'))
+    if not val:
+        return None
+    try:
+        pt = int(val) / 2.0
+        if 4.0 <= pt <= 96.0:
+            return pt
+    except (ValueError, TypeError):
+        return None
+    return None
+
+
+_CM_PER_PT = 2.54 / 72.0
+
+
+def _auto_size_height_cm(line_height_cm, font_pt, default_cm=2.0):
+    """根据段落行高和字号自动推算签名高度（cm）。
+
+    优先级：
+    1. 段落 exact line height → 用其 95%（留 5% 安全裕量，避免顶满行）
+    2. 字号 → 用 2 倍字号高度（中文文档常见签名比例：四号 14pt → 0.99cm，
+       视觉自然不抢戏，又能看清笔画）
+    3. 字号缺失 → 用 10.5pt（中文 Word 通用默认）按 2 倍推算兜底，
+       避免 fallback 到 default_cm 导致过大
+    4. 默认 default_cm
+
+    范围约束：返回值 clamp 到 [0.5, 5.0] cm，避免极端值。
+    """
+    if line_height_cm and 0.3 <= line_height_cm <= 10.0:
+        return round(min(max(line_height_cm * 0.95, 0.5), 5.0), 2)
+    effective_pt = font_pt if (font_pt and 6.0 <= font_pt <= 72.0) else 10.5
+    cm = effective_pt * 2 * _CM_PER_PT
+    return round(min(max(cm, 0.5), 5.0), 2)
+
+
+def _find_complex_field_end(paragraph, instr_text_elem):
+    """从 instrText 开始向后查找同段落的 fldChar end 元素。"""
+    current = instr_text_elem
+    found_sep = False
+    for elem in paragraph.iter():
+        if elem is current:
+            continue
+        if elem.tag == qn('w:fldChar'):
+            ftype = elem.get(qn('w:fldCharType'))
+            if ftype == 'separate':
+                found_sep = True
+            elif ftype == 'end' and found_sep:
+                return elem
+            elif ftype == 'end':
+                return elem
+    return None
+
+
+def _attach_image_relationship(doc, run_elem, image_path):
+    """通过 python-docx 的 part 接口建立图片关系并补充 blip。"""
+    try:
+        import os
+        from docx.image.image import Image
+        from docx.parts.image import ImagePart
+        from docx.opc.constants import RELATIONSHIP_TYPE as RT
+
+        with open(image_path, 'rb') as f:
+            image_data = f.read()
+        image = Image.from_blob(image_data)
+        image_part = doc.part.related_part_lookup_by_image(image) if hasattr(doc.part, 'related_part_lookup_by_image') else None
+        if image_part is None:
+            image_part = ImagePart.from_image(image, doc.part.package)
+            rId = doc.part.relate_to(image_part, RT.IMAGE)
+        else:
+            rId = None
+            for rel in doc.part.rels.values():
+                if rel.target_part is image_part:
+                    rId = rel.rId
+                    break
+
+        drawing = run_elem.find(qn('w:drawing'))
+        if drawing is None:
+            return
+        inline = drawing.find(qn('wp:inline'))
+        if inline is None:
+            return
+        graphic = inline.find(qn('a:graphic'))
+        if graphic is None:
+            return
+        graphic_data = graphic.find(qn('a:graphicData'))
+        if graphic_data is None:
+            return
+        pic = graphic_data.find(qn('pic:pic'))
+        if pic is None:
+            return
+        blip_fill = OxmlElement('pic:blipFill')
+        blip = OxmlElement('a:blip')
+        blip.set(qn('r:embed'), rId)
+        blip_fill.append(blip)
+        pic.insert(0, blip_fill)
+    except Exception as e:
+        logger.error("[Word处理] 绑定图片关系失败: %s", str(e))
+
+
+def build_set_field_runs(name):
+    """构造一段复杂域 XML，表示 ``SET USER_<name> \\* MERGEFORMAT``。
+
+    返回三个 ``<w:r>`` 元素：begin / instrText / separate / 空白 result / end。
+    合并为一个 list 返回，调用方依次追加到段落。
+    """
+    runs = []
+
+    r1 = OxmlElement('w:r')
+    fld_begin = OxmlElement('w:fldChar')
+    fld_begin.set(qn('w:fldCharType'), 'begin')
+    r1.append(fld_begin)
+    runs.append(r1)
+
+    r2 = OxmlElement('w:r')
+    instr = OxmlElement('w:instrText')
+    instr.set(qn('xml:space'), 'preserve')
+    instr.text = f' SET USER_{name} \\* MERGEFORMAT '
+    r2.append(instr)
+    runs.append(r2)
+
+    r3 = OxmlElement('w:r')
+    fld_sep = OxmlElement('w:fldChar')
+    fld_sep.set(qn('w:fldCharType'), 'separate')
+    r3.append(fld_sep)
+    runs.append(r3)
+
+    r4 = OxmlElement('w:r')
+    t4 = OxmlElement('w:t')
+    t4.set(qn('xml:space'), 'preserve')
+    t4.text = ''
+    r4.append(t4)
+    runs.append(r4)
+
+    r5 = OxmlElement('w:r')
+    fld_end = OxmlElement('w:fldChar')
+    fld_end.set(qn('w:fldCharType'), 'end')
+    r5.append(fld_end)
+    runs.append(r5)
+
+    return runs
+
+
+def _approx_char_width_emu(font_size_pt=None):
+    """估算单字符宽度（EMU），用于补偿空格宽度。
+
+    Word 默认正文字号 10.5pt（小四），半角空格宽度约为字号的 0.25 倍。
+    """
+    pt = font_size_pt or 10.5
+    return int(pt * 0.25 * 12700)
+
+
+def replace_text_with_set_field(paragraph_element, start_run_idx, start_offset, end_run_idx, end_offset,
+                                captured_text, name, compensate_spaces=True):
+    """在段落 XML 中将指定 Run 区间内的字符替换为 SET 域 + 可选补偿空格。
+
+    参数:
+    - ``paragraph_element``: ``<w:p>`` lxml 元素。
+    - ``start_run_idx``/``end_run_idx``: 起止 ``<w:r>`` 在段落中的索引。
+    - ``start_offset``/``end_offset``: 起止 ``<w:t>`` 内字符偏移。
+    - ``captured_text``: 被替换的原始可见文本（用于估算补偿宽度）。
+    - ``name``: 用于 ``USER_<name>`` 的标识。
+    - ``compensate_spaces``: 是否插入普通空格补偿原宽度。
+    """
+    runs = paragraph_element.findall(qn('w:r'))
+    if start_run_idx >= len(runs):
+        return False
+
+    start_run = runs[start_run_idx]
+    start_t = start_run.find(qn('w:t'))
+    if start_t is None:
+        return False
+
+    if start_run_idx == end_run_idx:
+        original_text = start_t.text or ''
+        head = original_text[:start_offset]
+        tail = original_text[end_offset:]
+        head_run = _make_text_run(head)
+        tail_run = _make_text_run(tail)
+        field_runs = build_set_field_runs(name)
+        compensate_run = _make_text_run(' ' * len(captured_text)) if compensate_spaces else None
+
+        parent = start_run.getparent()
+        idx = list(parent).index(start_run)
+        parent.remove(start_run)
+        insert_idx = idx
+        for elem in [head_run] + field_runs + ([compensate_run] if compensate_spaces else []) + [tail_run]:
+            if elem is None:
+                continue
+            parent.insert(insert_idx, elem)
+            insert_idx += 1
+        return True
+
+    # 跨多个 run：简化处理，清掉中间 run，仅在首 run 处插入域
+    original_text_parts = []
+    for ridx in range(start_run_idx, end_run_idx + 1):
+        r = runs[ridx]
+        t = r.find(qn('w:t'))
+        original_text_parts.append(t.text if t is not None and t.text else '')
+
+    first_text = original_text_parts[0]
+    head = first_text[:start_offset]
+    last_text = original_text_parts[-1]
+    tail = last_text[end_offset:]
+
+    head_run = _make_text_run(head)
+    tail_run = _make_text_run(tail)
+    field_runs = build_set_field_runs(name)
+    full_text = ''.join(original_text_parts)
+    captured = full_text[start_offset:(
+        sum(len(p) for p in original_text_parts[:-1]) + end_offset
+    )]
+    compensate_run = _make_text_run(' ' * len(captured)) if compensate_spaces else None
+
+    parent = paragraph_element
+    to_remove = runs[start_run_idx:end_run_idx + 1]
+    anchor = to_remove[0]
+    idx = list(parent).index(anchor)
+    for r in to_remove:
+        parent.remove(r)
+    insert_idx = idx
+    for elem in [head_run] + field_runs + ([compensate_run] if compensate_spaces else []) + [tail_run]:
+        if elem is None:
+            continue
+        parent.insert(insert_idx, elem)
+        insert_idx += 1
+    return True
+
+
+def _make_text_run(text):
+    if text is None:
+        return None
+    if text == '':
+        return None
+    r = OxmlElement('w:r')
+    t = OxmlElement('w:t')
+    t.set(qn('xml:space'), 'preserve')
+    t.text = text
+    r.append(t)
+    return r

lib/image_processor.py

@@ -0,0 +1,251 @@
+import io
+import random
+import math
+import logging
+import numpy as np
+from scipy import ndimage
+from PIL import Image, ImageDraw, ImageChops
+
+logger = logging.getLogger(__name__)
+
+
+def process_image(image_bytes, is_signature=False, noise_level=0,
+                  rotate_min=0, rotate_max=0, log_source='前端操作'):
+    """处理签名/印章图片，返回紧致裁剪后的 PNG 字节。
+
+    ``height_cm`` **不再在此函数处理**：图片处理始终在原始 DPI 下进行，
+    保证笔画细节不丢失；最终在 Word 里的显示尺寸由 ``field_codes.insert_image_after_field``
+    的 ``height_cm`` 参数控制（Word 渲染时按文档 DPI 高质量地下采样）。
+    """
+    img = Image.open(io.BytesIO(image_bytes))
+    logger.info("[图片处理] 原始图片: 模式=%s, 尺寸=%s", img.mode, str(img.size), extra={'log_source': log_source})
+
+    if is_signature:
+        img = _binarize(img, log_source)
+
+    img = _remove_white_background(img, log_source)
+
+    img = _crop_transparent_border(img, log_source)
+
+    if noise_level > 0:
+        img = _add_edge_noise(img, noise_level, log_source)
+
+    if rotate_min != 0 or rotate_max != 0:
+        img = _rotate_image(img, rotate_min, rotate_max, log_source)
+
+    # 剔除零散噪点簇（连通域面积过小）：二值化可能产生离主体很远的孤立黑像素，
+    # 它们会撑大后续紧致裁剪的 bbox，导致签名在图片里只占一角、四周大片空白。
+    # 必须在 _crop_to_content 之前执行，否则裁剪已经被噪点污染。
+    img = _remove_small_clusters(img, log_source)
+
+    # 流水线最后再做一次紧致裁剪：噪点和旋转会在签名外围产生低 alpha 像素，
+    # 撑大图片 bbox，让签名不紧贴图片边缘；Word 插入时按图片整体定位，
+    # 签名会被顶出段落可见区域。用 alpha 阈值裁剪，确保最终图片紧贴签名笔画。
+    img = _crop_to_content(img, log_source)
+
+    buf = io.BytesIO()
+    img.save(buf, format='PNG')
+    buf.seek(0)
+    result = buf.read()
+    logger.info("[图片处理] 处理完成: 最终尺寸=%s, 大小=%d bytes", str(img.size), len(result), extra={'log_source': log_source})
+    return result
+
+
+def _remove_small_clusters(img, log_source, alpha_threshold=30, brightness_threshold=210,
+                           min_area_ratio=0.0001, min_area_abs=50):
+    """剔除零散噪点簇：用连通域分析找出面积过小的内容块并清除。
+
+    判定"内容像素"：``(alpha > alpha_threshold) AND (亮度 < brightness_threshold)``
+    （与 :func:`_crop_to_content` 一致）。8-连通域分组后，面积低于
+    ``max(min_area_ratio × 图像总像素, min_area_abs)`` 的簇被视为噪点，
+    对应像素的 alpha 置 0。
+
+    双阈值设计：
+    - ``min_area_ratio=0.0001``（0.01%）：高分辨率图（如 2025×921）的相对阈值
+      ≈ 186 像素，足以剔除孤立黑像素（实测噪点簇 ≤ 13 px），又保留真实笔画
+      （实测最小真实笔画 285 px）
+    - ``min_area_abs=50``：低分辨率图的兜底，防止相对阈值过小失效
+
+    若需更激进清洗（如打印文档无小笔画），可把 ``min_area_ratio`` 调到 0.001。
+    """
+    if img.mode != 'RGBA':
+        img = img.convert('RGBA')
+
+    arr = np.array(img)
+    alpha = arr[:, :, 3]
+    rgb_max = arr[:, :, :3].max(axis=2).astype(np.int16)
+
+    content_mask = (alpha > alpha_threshold) & (rgb_max < brightness_threshold)
+    if not content_mask.any():
+        return img
+
+    structure = np.ones((3, 3), dtype=int)  # 8-连通
+    labeled, num_features = ndimage.label(content_mask, structure=structure)
+    if num_features == 0:
+        return img
+
+    total_pixels = content_mask.size
+    area_threshold = max(min_area_ratio * total_pixels, min_area_abs)
+
+    # sizes[0] 是背景，从 1 开始才是真实簇
+    sizes = ndimage.sum(content_mask, labeled, range(num_features + 1)).astype(np.int64)
+    small_labels = np.where(sizes < area_threshold)[0]
+    small_labels = small_labels[small_labels > 0]  # 排除背景
+    if len(small_labels) == 0:
+        logger.info("[图片处理] 剔除噪点簇: 共 %d 簇，全部 >= 阈值 %.0f px，无需清理",
+                    num_features, area_threshold, extra={'log_source': log_source})
+        return img
+
+    small_mask = np.isin(labeled, small_labels)
+    removed_pixels = int(small_mask.sum())
+    removed_content = int(sizes[small_labels].sum())
+
+    arr[small_mask] = (0, 0, 0, 0)
+    cleaned = Image.fromarray(arr, 'RGBA')
+
+    logger.info("[图片处理] 剔除噪点簇: 共 %d 簇，移除 %d 簇 (%d 像素，占内容 %.2f%%)；"
+                "阈值=%.0f px (相对 %.3f%% + 绝对 %d px)",
+                num_features, len(small_labels), removed_pixels,
+                100.0 * removed_content / max(int(content_mask.sum()), 1),
+                area_threshold, min_area_ratio * 100, min_area_abs,
+                extra={'log_source': log_source})
+    return cleaned
+
+
+def _crop_to_content(img, log_source, alpha_threshold=30, brightness_threshold=210):
+    """末尾紧致裁剪：联合 alpha 和 RGB 亮度判断内容区域。
+
+    内容判据：``(alpha > 30) AND (亮度 < 210)``
+    """
+    if img.mode != 'RGBA':
+        img = img.convert('RGBA')
+
+    r, g, b, a = img.split()
+    luma = Image.merge('RGB', (r, g, b)).convert('L')
+
+    alpha_mask = a.point(lambda p: 255 if p > alpha_threshold else 0)
+    luma_mask = luma.point(lambda p: 255 if p < brightness_threshold else 0)
+    combined = ImageChops.darker(alpha_mask, luma_mask)  # 二值 mask 逐像素 min = AND
+
+    bbox = combined.getbbox()
+    total_px = img.size[0] * img.size[1] or 1
+    alpha_cnt = sum(alpha_mask.histogram()[1:])
+    dark_cnt = sum(luma_mask.histogram()[1:])
+    content_cnt = sum(combined.histogram()[1:])
+    logger.info("[图片处理] 紧致裁剪诊断: 尺寸=%s, bbox=%s, "
+                "高alpha像素=%d/%d (%.1f%%), 暗像素=%d/%d (%.1f%%), 内容像素=%d/%d (%.1f%%)",
+                str(img.size), str(bbox),
+                alpha_cnt, total_px, 100.0 * alpha_cnt / total_px,
+                dark_cnt, total_px, 100.0 * dark_cnt / total_px,
+                content_cnt, total_px, 100.0 * content_cnt / total_px,
+                extra={'log_source': log_source})
+
+    if bbox and bbox != (0, 0, img.size[0], img.size[1]):
+        before = img.size
+        img = img.crop(bbox)
+        logger.info("[图片处理] 紧致裁剪: %s -> %s (bbox=%s)",
+                    str(before), str(img.size), str(bbox), extra={'log_source': log_source})
+    return img
+
+
+def _binarize(img, log_source, threshold=128):
+    logger.info("[图片处理] 执行手写签字黑白二值化", extra={'log_source': log_source})
+    if img.mode != 'L':
+        img = img.convert('L')
+    # 阈值化二值化（保持 L 模式，避免 PIL point + mode='1' 的兼容性问题）
+    img = img.point(lambda x: 255 if x > threshold else 0)
+    img = img.convert('RGBA')
+    return img
+
+
+def _remove_white_background(img, log_source):
+    logger.info("[图片处理] 执行白底转透明抠图", extra={'log_source': log_source})
+    if img.mode != 'RGBA':
+        img = img.convert('RGBA')
+    datas = img.getdata()
+    new_data = []
+    for item in datas:
+        r, g, b, a = item
+        if r > 240 and g > 240 and b > 240:
+            new_data.append((255, 255, 255, 0))
+        else:
+            new_data.append((r, g, b, a))
+    img.putdata(new_data)
+    return img
+
+
+def _crop_transparent_border(img, log_source):
+    logger.info("[图片处理] 执行透明边框裁剪", extra={'log_source': log_source})
+    if img.mode != 'RGBA':
+        img = img.convert('RGBA')
+    bbox = img.getbbox()
+    if bbox:
+        img = img.crop(bbox)
+    return img
+
+
+def _add_edge_noise(img, noise_level, log_source):
+    logger.info("[图片处理] 添加边缘噪声: level=%d", noise_level, extra={'log_source': log_source})
+    if img.mode != 'RGBA':
+        img = img.convert('RGBA')
+    w, h = img.size
+    pixels = img.load()
+    intensity = noise_level * 2
+    for x in range(w):
+        for y in range(h):
+            r, g, b, a = pixels[x, y]
+            if a == 0:
+                continue
+            is_edge = False
+            for dx in range(-2, 3):
+                for dy in range(-2, 3):
+                    nx, ny = x + dx, y + dy
+                    if 0 <= nx < w and 0 <= ny < h:
+                        if pixels[nx, ny][3] == 0:
+                            is_edge = True
+                            break
+                if is_edge:
+                    break
+            if is_edge and random.random() < noise_level / 10.0:
+                offset_x = random.randint(-intensity, intensity)
+                offset_y = random.randint(-intensity, intensity)
+                nx = x + offset_x
+                ny = y + offset_y
+                if 0 <= nx < w and 0 <= ny < h:
+                    nr = min(255, max(0, r + random.randint(-20, 20)))
+                    ng = min(255, max(0, g + random.randint(-20, 20)))
+                    nb = min(255, max(0, b + random.randint(-20, 20)))
+                    na = min(255, max(0, a + random.randint(-30, 10)))
+                    if pixels[nx, ny][3] == 0:
+                        pixels[nx, ny] = (nr, ng, nb, na)
+    return img
+
+
+def _rotate_image(img, rotate_min, rotate_max, log_source):
+    if rotate_min > rotate_max:
+        rotate_min, rotate_max = rotate_max, rotate_min
+    angle = random.randint(rotate_min, rotate_max)
+    logger.info("[图片处理] 旋转: %d度 (范围 %d~%d)", angle, rotate_min, rotate_max, extra={'log_source': log_source})
+    img = img.rotate(angle, resample=Image.BICUBIC, expand=True, fillcolor=(0, 0, 0, 0))
+    return img
+
+
+def preview_remove_white(image_bytes):
+    img = Image.open(io.BytesIO(image_bytes))
+    img = _remove_white_background(img, '前端操作')
+    img = _crop_transparent_border(img, '前端操作')
+    buf = io.BytesIO()
+    img.save(buf, format='PNG')
+    buf.seek(0)
+    return buf.read()
+
+
+def preview_binarize(image_bytes):
+    img = Image.open(io.BytesIO(image_bytes))
+    img = _binarize(img, '前端操作')
+    img = _remove_white_background(img, '前端操作')
+    img = _crop_transparent_border(img, '前端操作')
+    buf = io.BytesIO()
+    img.save(buf, format='PNG')
+    buf.seek(0)
+    return buf.read()

lib/log_handler.py

@@ -0,0 +1,164 @@
+import logging
+import threading
+import time
+import json
+from collections import deque
+from datetime import datetime
+
+_log_deque = deque(maxlen=2000)
+_subscribers = {}
+_sub_lock = threading.Lock()
+_sub_id_counter = 0
+_log_id_lock = threading.Lock()
+_log_id_counter = 0
+
+SOURCE_ALIASES = {
+    'api': '外部API',
+    'frontend': '前端操作',
+    'db': '数据库',
+    'batch': '批量',
+    'system': '系统',
+    '外部API': '外部API',
+    '前端操作': '前端操作',
+    '数据库': '数据库',
+    '批量': '批量',
+    '系统': '系统',
+}
+
+
+def _next_log_id():
+    global _log_id_counter
+    with _log_id_lock:
+        _log_id_counter += 1
+        return _log_id_counter
+
+
+def _resolve_source(token):
+    if not token:
+        return None
+    return SOURCE_ALIASES.get(token, token)
+
+
+class SSELogHandler(logging.Handler):
+    def emit(self, record):
+        try:
+            msg = self.format(record)
+            level = record.levelname
+            if level == 'WARNING':
+                level = 'WARN'
+            source = getattr(record, 'log_source', '系统')
+            trace_id = getattr(record, 'trace_id', None) or ''
+            entry = {
+                'id': _next_log_id(),
+                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
+                'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
+                'level': level,
+                'source': source,
+                'message': msg,
+                'trace_id': trace_id,
+            }
+            _log_deque.append(entry)
+            _notify_subscribers(entry)
+        except Exception:
+            pass
+
+
+def _notify_subscribers(entry):
+    with _sub_lock:
+        for q in _subscribers.values():
+            q.append(entry)
+
+
+def subscribe():
+    global _sub_id_counter
+    with _sub_lock:
+        _sub_id_counter += 1
+        sub_id = _sub_id_counter
+        q = deque(maxlen=500)
+        _subscribers[sub_id] = q
+    return sub_id, q
+
+
+def unsubscribe(sub_id):
+    with _sub_lock:
+        _subscribers.pop(sub_id, None)
+
+
+def get_all_logs():
+    return list(_log_deque)
+
+
+def query_logs(after_id=0, level=None, source=None, keyword=None, limit=200):
+    after_id = int(after_id or 0)
+    limit = max(1, min(int(limit or 200), 1000))
+    norm_source = _resolve_source(source) if source else None
+    keyword_lc = (keyword or '').lower()
+
+    matched = []
+    for entry in _log_deque:
+        if entry['id'] <= after_id:
+            continue
+        if level and entry.get('level') != level:
+            continue
+        if norm_source and entry.get('source') != norm_source:
+            continue
+        if keyword_lc and keyword_lc not in entry.get('message', '').lower():
+            continue
+        matched.append(entry)
+    matched = matched[-limit:] if len(matched) > limit else matched
+    last_id = matched[-1]['id'] if matched else after_id
+    return {'logs': matched, 'last_id': last_id}
+
+
+def setup_logging():
+    from lib.config import get_log_config
+    log_cfg = get_log_config()
+    log_dir = log_cfg['file_path']
+    import os
+    if not os.path.isabs(log_dir):
+        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        log_dir = os.path.join(project_root, log_dir)
+    os.makedirs(log_dir, exist_ok=True)
+
+    log_level = getattr(logging, log_cfg['level'].upper(), logging.INFO)
+
+    root_logger = logging.getLogger()
+    root_logger.setLevel(log_level)
+
+    fmt = logging.Formatter('%(message)s')
+
+    sse_handler = SSELogHandler()
+    sse_handler.setLevel(log_level)
+    sse_handler.setFormatter(fmt)
+    root_logger.addHandler(sse_handler)
+
+    file_handler = logging.FileHandler(
+        os.path.join(log_dir, f"seal_{datetime.now().strftime('%Y%m%d')}.log"),
+        encoding='utf-8',
+    )
+    file_handler.setLevel(log_level)
+    file_handler.setFormatter(logging.Formatter(
+        '[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S'
+    ))
+    root_logger.addHandler(file_handler)
+
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(log_level)
+    console_handler.setFormatter(logging.Formatter(
+        '[%(asctime)s] [%(levelname)s] %(message)s',
+        datefmt='%H:%M:%S'
+    ))
+    root_logger.addHandler(console_handler)
+
+
+def log_info(msg, source='前端操作', trace_id=None):
+    logging.info(msg, extra={'log_source': source, 'trace_id': trace_id or ''})
+
+
+def log_warn(msg, source='前端操作', trace_id=None):
+    logging.warning(msg, extra={'log_source': source, 'trace_id': trace_id or ''})
+
+
+def log_error(msg, source='前端操作', trace_id=None):
+    logging.error(msg, extra={'log_source': source, 'trace_id': trace_id or ''})

lib/monitor.py

@@ -0,0 +1,75 @@
+import threading
+from collections import deque
+from datetime import datetime
+
+_requests_lock = threading.Lock()
+_requests = deque(maxlen=500)
+
+
+def register_request(trace_id, client_ip, endpoint, request_body_summary=None):
+    entry = {
+        'trace_id': trace_id,
+        'request_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+        'client_ip': client_ip or '-',
+        'endpoint': endpoint,
+        'status': 'running',
+        'response_status': None,
+        'elapsed_ms': None,
+        'request_summary': request_body_summary or {},
+        'response_summary': None,
+        'warnings': [],
+        'error_message': None,
+    }
+    with _requests_lock:
+        _requests.append(entry)
+    return entry
+
+
+def update_request(trace_id, **fields):
+    with _requests_lock:
+        for entry in _requests:
+            if entry['trace_id'] == trace_id:
+                entry.update(fields)
+                return entry
+    return None
+
+
+def finish_request(trace_id, status, response_status=None, elapsed_ms=None,
+                   response_summary=None, warnings=None, error_message=None):
+    return update_request(
+        trace_id,
+        status=status,
+        response_status=response_status,
+        elapsed_ms=elapsed_ms,
+        response_summary=response_summary,
+        warnings=warnings or [],
+        error_message=error_message,
+    )
+
+
+def query_requests(status=None, keyword=None, limit=200):
+    limit = max(1, min(int(limit or 200), 500))
+    keyword_lc = (keyword or '').lower().strip()
+    items = []
+    with _requests_lock:
+        snapshot = list(_requests)
+    snapshot.reverse()
+    for entry in snapshot:
+        if status and entry.get('status') != status:
+            continue
+        if keyword_lc:
+            hay = ' '.join(str(entry.get(k) or '') for k in ('trace_id', 'client_ip', 'endpoint')).lower()
+            if keyword_lc not in hay:
+                continue
+        items.append(entry)
+        if len(items) >= limit:
+            break
+    return items
+
+
+def get_request(trace_id):
+    with _requests_lock:
+        for entry in reversed(_requests):
+            if entry['trace_id'] == trace_id:
+                return entry
+    return None

lib/validators.py

@@ -0,0 +1,159 @@
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+
+USER_ID_PATTERN = re.compile(r'^[a-zA-Z0-9]+$')
+
+
+def validate_user_id(user_id):
+    if not user_id:
+        return False, "人员ID不能为空"
+    if not USER_ID_PATTERN.match(user_id):
+        return False, "人员ID仅支持大小写英文字母和数字"
+    if len(user_id) > 50:
+        return False, "人员ID长度不能超过50"
+    return True, ""
+
+
+def validate_user_name(user_name):
+    if not user_name:
+        return False, "人员姓名不能为空"
+    if len(user_name) > 100:
+        return False, "人员姓名长度不能超过100"
+    return True, ""
+
+
+def validate_height(height):
+    if height is None or height == "":
+        return True, None
+    try:
+        h = float(height)
+        if 0.1 <= h <= 10.0:
+            return True, round(h, 1)
+        return False, f"高度 {height} 超出有效范围 0.1~10.0"
+    except (ValueError, TypeError):
+        return False, f"高度 {height} 不是有效数值"
+
+
+def validate_noise_level(level):
+    if level is None or level == "":
+        return True, 0
+    try:
+        n = int(level)
+        if 0 <= n <= 10:
+            return True, n
+        return False, f"噪声强度 {level} 超出有效范围 0~10"
+    except (ValueError, TypeError):
+        return False, f"噪声强度 {level} 不是有效整数"
+
+
+def validate_rotate(min_val, max_val):
+    def _clamp(v):
+        try:
+            v = int(v)
+            return max(-180, min(180, v))
+        except (ValueError, TypeError):
+            return 0
+    r_min = _clamp(min_val) if min_val not in (None, "") else 0
+    r_max = _clamp(max_val) if max_val not in (None, "") else 0
+    return True, (r_min, r_max)
+
+
+def validate_match_mode(mode):
+    if mode in ('upload', 'id', 'name', 'auto'):
+        return True, ""
+    return False, f"匹配模式 '{mode}' 无效，支持: upload/id/name/auto"
+
+
+def validate_docx_name(name):
+    if not name:
+        return False, "文件名不能为空"
+    if not name.lower().endswith('.docx'):
+        return False, "仅支持.docx格式"
+    return True, ""
+
+
+def validate_docx_size(data_bytes, max_bytes):
+    if not isinstance(data_bytes, (bytes, bytearray)):
+        return False, "文档数据无效"
+    if len(data_bytes) > max_bytes:
+        return False, f"文档大小 {len(data_bytes)} 字节超过限制 {max_bytes} 字节"
+    return True, ""
+
+
+def validate_image_size(data_bytes, max_bytes):
+    if not isinstance(data_bytes, (bytes, bytearray)):
+        return False, "图片数据无效"
+    if len(data_bytes) > max_bytes:
+        return False, f"图片大小 {len(data_bytes)} 字节超过限制 {max_bytes} 字节"
+    return True, ""
+
+
+def validate_capture_group(group):
+    if group is None or group == "":
+        return False, "捕获组序号不能为空"
+    try:
+        g = int(group)
+    except (ValueError, TypeError):
+        return False, f"捕获组序号 '{group}' 不是整数"
+    if g < 1:
+        return False, "捕获组序号禁止为 0 或负数，必须 >=1"
+    return True, g
+
+
+def validate_rule(rule):
+    errors = []
+    if not rule.get('name'):
+        errors.append("规则名称不能为空")
+    pattern = rule.get('pattern')
+    if not pattern:
+        errors.append("规则正则表达式不能为空")
+    else:
+        try:
+            re.compile(pattern)
+        except re.error as e:
+            errors.append(f"正则表达式非法: {e}")
+    valid, msg_or_val = validate_capture_group(rule.get('capture_group'))
+    if not valid:
+        errors.append(msg_or_val)
+    cell_mode = rule.get('cell_mode', 'single')
+    if cell_mode not in ('single', 'horizontal', 'vertical'):
+        errors.append(f"表格方式 '{cell_mode}' 无效")
+    cell_match = rule.get('cell_match', 'partial')
+    if cell_match not in ('partial', 'whole'):
+        errors.append(f"匹配方式 '{cell_match}' 无效")
+    return errors
+
+
+def validate_sign_request(params):
+    errors = []
+
+    valid, msg = validate_docx_name(params.get('docx_name', ''))
+    if not valid:
+        errors.append(msg)
+
+    valid, msg = validate_match_mode(params.get('match_mode', ''))
+    if not valid:
+        errors.append(msg)
+
+    stamps = params.get('stamps', [])
+    if params.get('match_mode') != 'auto' and not stamps:
+        errors.append("非auto模式必须提供盖章项列表")
+
+    if len(stamps) > 100:
+        errors.append("盖章项不能超过100个")
+
+    height = params.get('height')
+    if height is not None and height != "":
+        valid, _ = validate_height(height)
+        if not valid:
+            errors.append(f"全局高度无效: {validate_height(height)[1]}")
+
+    noise = params.get('noise_level')
+    if noise is not None and noise != "":
+        valid, _ = validate_noise_level(noise)
+        if not valid:
+            errors.append(f"全局噪声无效: {validate_noise_level(noise)[1]}")
+
+    return errors

lib/word_signer.py

@@ -0,0 +1,381 @@
+import io
+import os
+import re
+import base64
+import logging
+import tempfile
+from datetime import datetime
+from docx import Document
+from docx.shared import Cm
+import lib.db as db
+from lib.image_processor import process_image
+from lib import field_codes
+
+logger = logging.getLogger(__name__)
+
+
+def _new_trace_id():
+    return datetime.now().strftime('%Y%m%d%H%M%S%f') + f'_{os.getpid()}'
+
+
+def sign_word(docx_data, params, log_source='前端操作', trace_id=None, client_ip=None):
+    if trace_id is None:
+        trace_id = _new_trace_id()
+    logger.info("[签章流程] ====== 开始处理 ====== trace_id=%s", trace_id,
+                extra={'log_source': log_source, 'trace_id': trace_id})
+
+    try:
+        docx_name = params.get('docx_name', '')
+        logger.info("[签章流程] 文件名: %s", docx_name,
+                    extra={'log_source': log_source, 'trace_id': trace_id})
+
+        if not docx_name.lower().endswith('.docx'):
+            logger.error("[参数校验] 文件类型错误: %s, 仅支持.docx", docx_name,
+                         extra={'log_source': log_source, 'trace_id': trace_id})
+            return None, {'success': False, 'code': 'INVALID_FILE_TYPE',
+                          'message': f'仅支持.docx格式，当前: {docx_name}',
+                          'trace_id': trace_id}
+
+        match_mode = params.get('match_mode', '')
+        if match_mode not in ('upload', 'id', 'name', 'auto'):
+            logger.error("[参数校验] 匹配模式无效: %s", match_mode,
+                         extra={'log_source': log_source, 'trace_id': trace_id})
+            return None, {'success': False, 'code': 'INVALID_MATCH_MODE',
+                          'message': f'匹配模式无效: {match_mode}', 'trace_id': trace_id}
+
+        logger.info("[签章流程] 匹配模式: %s", match_mode,
+                    extra={'log_source': log_source, 'trace_id': trace_id})
+
+        try:
+            doc = Document(io.BytesIO(docx_data))
+        except Exception as e:
+            logger.error("[参数校验] Word文档解析失败: %s", str(e),
+                         extra={'log_source': log_source, 'trace_id': trace_id})
+            return None, {'success': False, 'code': 'INVALID_DOCX',
+                          'message': f'Word文档解析失败或已损坏: {e}', 'trace_id': trace_id}
+        logger.info("[签章流程] Word文档加载成功",
+                    extra={'log_source': log_source, 'trace_id': trace_id})
+
+        stamps = params.get('stamps', [])
+        use_global_align = params.get('use_global_align', True)
+        align_h = params.get('align_h', 'left')
+        align_v = params.get('align_v', 'center')
+        use_global_height = params.get('use_global_height', True)
+        global_height = params.get('height')
+        use_global_noise = params.get('use_global_noise', True)
+        global_noise = params.get('noise_level', 0)
+        use_global_rotate = params.get('use_global_rotate', True)
+        global_rotate_min = params.get('rotate_min', 0)
+        global_rotate_max = params.get('rotate_max', 0)
+        relax_layout = params.get('relax_layout', True)
+
+        if match_mode == 'auto':
+            stamps = [s for s in stamps if (s.get('marker') or '').strip()]
+            if not stamps:
+                scanned = field_codes.scan_user_markers(doc, log_source=log_source)
+                stamps = [{'marker': item['marker']} for item in scanned]
+                logger.info("[签章流程] auto模式扫描到 %d 个标记", len(stamps),
+                            extra={'log_source': log_source, 'trace_id': trace_id})
+
+        if len(stamps) > 100:
+            logger.error("[参数校验] 盖章项超过100个限制: %d", len(stamps),
+                         extra={'log_source': log_source, 'trace_id': trace_id})
+            return None, {'success': False, 'code': 'TOO_MANY_STAMPS',
+                          'message': '盖章项超过100个限制', 'trace_id': trace_id}
+
+        warnings = []
+        success_count = 0
+
+        for i, stamp in enumerate(stamps):
+            logger.info("[签章流程] ---- 处理盖章项 %d/%d ----", i + 1, len(stamps),
+                        extra={'log_source': log_source, 'trace_id': trace_id})
+            try:
+                marker = stamp.get('marker', '')
+                image_data = None
+                stamp_is_signature = False
+
+                if match_mode == 'upload':
+                    image_b64 = stamp.get('image_base64', '')
+                    if image_b64:
+                        image_data = base64.b64decode(image_b64)
+                        logger.info("[图片匹配] 使用上传图片: %s", stamp.get('image_filename', ''),
+                                    extra={'log_source': log_source, 'trace_id': trace_id})
+                    else:
+                        warnings.append(f"盖章项{i + 1}: 缺少图片数据")
+                        logger.warning("[图片匹配] 盖章项%d缺少图片数据", i + 1,
+                                       extra={'log_source': log_source, 'trace_id': trace_id})
+                        continue
+
+                elif match_mode == 'id':
+                    image_id = stamp.get('image_id', '')
+                    if not image_id:
+                        # SRS 3.1.3：空字符串改用 image_name 匹配
+                        image_name = stamp.get('image_name', '')
+                        if image_name:
+                            logger.info("[图片匹配] image_id 为空，改用姓名查询: %s", image_name,
+                                        extra={'log_source': log_source, 'trace_id': trace_id})
+                            records = db.get_image_by_user_name(image_name)
+                            if records:
+                                if len(records) > 1:
+                                    warnings.append(f"盖章项{i + 1}: 姓名={image_name} 匹配到 {len(records)} 条记录，使用第一条")
+                                    logger.warning("[图片匹配] 姓名匹配多条: %s -> %d 条，使用第一条",
+                                                   image_name, len(records),
+                                                   extra={'log_source': log_source, 'trace_id': trace_id})
+                                image_data = records[0]['sign_image']
+                                stamp_is_signature = bool(records[0]['is_signature'])
+                            else:
+                                warnings.append(f"盖章项{i + 1}: 未找到姓名={image_name} 的记录")
+                                logger.warning("[图片匹配] 未找到姓名=%s", image_name,
+                                               extra={'log_source': log_source, 'trace_id': trace_id})
+                                continue
+                        else:
+                            warnings.append(f"盖章项{i + 1}: 缺少人员ID和姓名")
+                            logger.warning("[图片匹配] 盖章项%d缺少人员ID和姓名", i + 1,
+                                           extra={'log_source': log_source, 'trace_id': trace_id})
+                            continue
+                    else:
+                        logger.info("[图片匹配] 按ID查询: %s", image_id,
+                                    extra={'log_source': log_source, 'trace_id': trace_id})
+                        record = db.get_image_by_user_id(image_id)
+                        if record:
+                            image_data = record['sign_image']
+                            stamp_is_signature = bool(record['is_signature'])
+                            logger.info("[图片匹配] ID匹配成功: %s", image_id,
+                                        extra={'log_source': log_source, 'trace_id': trace_id})
+                        else:
+                            warnings.append(f"盖章项{i + 1}: 未找到ID={image_id} 的记录")
+                            logger.warning("[图片匹配] 未找到ID=%s", image_id,
+                                           extra={'log_source': log_source, 'trace_id': trace_id})
+                            continue
+
+                elif match_mode == 'name':
+                    image_name = stamp.get('image_name', '')
+                    if not image_name:
+                        warnings.append(f"盖章项{i + 1}: 缺少人员姓名")
+                        logger.warning("[图片匹配] 盖章项%d缺少人员姓名", i + 1,
+                                       extra={'log_source': log_source, 'trace_id': trace_id})
+                        continue
+                    logger.info("[图片匹配] 按姓名查询: %s", image_name,
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+                    records = db.get_image_by_user_name(image_name)
+                    if records:
+                        if len(records) > 1:
+                            warnings.append(f"盖章项{i + 1}: 姓名={image_name} 匹配到 {len(records)} 条记录，使用第一条")
+                            logger.warning("[图片匹配] 姓名匹配多条: %s -> %d 条，使用第一条",
+                                           image_name, len(records),
+                                           extra={'log_source': log_source, 'trace_id': trace_id})
+                        image_data = records[0]['sign_image']
+                        stamp_is_signature = bool(records[0]['is_signature'])
+                        logger.info("[图片匹配] 姓名匹配成功: %s", image_name,
+                                    extra={'log_source': log_source, 'trace_id': trace_id})
+                    else:
+                        warnings.append(f"盖章项{i + 1}: 未找到姓名={image_name} 的记录")
+                        logger.warning("[图片匹配] 未找到姓名=%s", image_name,
+                                       extra={'log_source': log_source, 'trace_id': trace_id})
+                        continue
+
+                elif match_mode == 'auto':
+                    name_from_marker = marker
+                    if name_from_marker.startswith('USER_'):
+                        name_from_marker = name_from_marker[5:]
+                    name_from_marker = field_codes._strip_switches(name_from_marker)
+                    logger.info("[图片匹配] auto模式, 标记=%s, 查询姓名=%s", marker, name_from_marker,
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+                    records = db.get_image_by_user_name(name_from_marker)
+                    if records:
+                        if len(records) > 1:
+                            warnings.append(f"盖章项{i + 1}: 姓名={name_from_marker} 匹配多条，使用第一条")
+                            logger.warning("[图片匹配] auto 姓名匹配多条: %s -> %d", name_from_marker, len(records),
+                                           extra={'log_source': log_source, 'trace_id': trace_id})
+                        image_data = records[0]['sign_image']
+                        stamp_is_signature = bool(records[0]['is_signature'])
+                        logger.info("[图片匹配] auto匹配成功: %s", name_from_marker,
+                                    extra={'log_source': log_source, 'trace_id': trace_id})
+                    else:
+                        warnings.append(f"盖章项{i + 1}: auto模式未找到姓名={name_from_marker} 的记录")
+                        logger.warning("[图片匹配] auto未找到姓名=%s", name_from_marker,
+                                       extra={'log_source': log_source, 'trace_id': trace_id})
+                        continue
+
+                if not image_data:
+                    warnings.append(f"盖章项{i + 1}: 无有效图片数据")
+                    continue
+
+                is_sig = stamp.get('is_signature', stamp_is_signature)
+                h = _resolve_height(stamp, use_global_height, global_height)
+                noise = _resolve_noise(stamp, use_global_noise, global_noise)
+                rot_min, rot_max = _resolve_rotate(stamp, use_global_rotate, global_rotate_min, global_rotate_max)
+                s_align_h = stamp.get('align_h') if not use_global_align else align_h
+                s_align_v = stamp.get('align_v') if not use_global_align else align_v
+
+                processed = process_image(
+                    image_data,
+                    is_signature=is_sig,
+                    noise_level=noise,
+                    rotate_min=rot_min,
+                    rotate_max=rot_max,
+                    log_source=log_source,
+                )
+
+                # ============= 调试：把"即将插入文档的图"落到磁盘 =============
+                # 目的：直接对比项目实际产出的图 vs 测试脚本（image_test/process.py）的输出
+                # 落盘位置: image_test/output/project_run/<trace_id>_<stamp_idx>_*.png
+                try:
+                    import os as _os
+                    _debug_dir = _os.path.join(
+                        _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))),
+                        'image_test', 'output', 'project_run'
+                    )
+                    _os.makedirs(_debug_dir, exist_ok=True)
+                    _tag = f"{trace_id or 'no_trace'}_stamp{i + 1}"
+                    with open(_os.path.join(_debug_dir, f"{_tag}_1_input.png"), 'wb') as _f:
+                        _f.write(image_data)
+                    with open(_os.path.join(_debug_dir, f"{_tag}_2_processed.png"), 'wb') as _f:
+                        _f.write(processed)
+                    logger.info("[签章流程][DEBUG] 已保存处理前后对比图到 %s (tag=%s)",
+                                _debug_dir, _tag,
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+                except Exception as _e:
+                    logger.warning("[签章流程][DEBUG] 保存调试图失败: %s", str(_e),
+                                   extra={'log_source': log_source, 'trace_id': trace_id})
+                # ============= 调试结束 =============
+
+                inserted = _insert_signature_image(doc, marker, processed, s_align_h, s_align_v,
+                                                    height_cm=h,
+                                                    relax_layout=relax_layout,
+                                                    log_source=log_source, trace_id=trace_id)
+                if inserted > 0:
+                    success_count += inserted
+                    logger.info("[签章流程] 盖章项%d签章成功: marker=%s, 插入%d处",
+                                i + 1, marker, inserted,
+                                extra={'log_source': log_source, 'trace_id': trace_id})
+                else:
+                    warnings.append(f"盖章项{i + 1}: 未找到标记 '{marker}'")
+                    logger.warning("[签章流程] 未找到标记: %s", marker,
+                                   extra={'log_source': log_source, 'trace_id': trace_id})
+
+            except Exception as e:
+                warnings.append(f"盖章项{i + 1}: 处理异常 - {str(e)}")
+                logger.error("[签章流程] 盖章项%d处理异常: %s", i + 1, str(e),
+                             extra={'log_source': log_source, 'trace_id': trace_id})
+
+        output_buf = io.BytesIO()
+        doc.save(output_buf)
+        output_buf.seek(0)
+        result_data = output_buf.read()
+
+        base_name = os.path.splitext(docx_name)[0]
+        output_name = f"{base_name}_signed.docx"
+
+        logger.info("[签章流程] ====== 处理完成 ====== 成功%d项, 告警%d条, trace_id=%s",
+                    success_count, len(warnings), trace_id,
+                    extra={'log_source': log_source, 'trace_id': trace_id})
+
+        return {'data': result_data, 'filename': output_name,
+                'warnings': warnings, 'trace_id': trace_id,
+                'success_count': success_count}, None
+
+    except Exception as e:
+        logger.error("[签章流程] 处理失败: %s", str(e),
+                     extra={'log_source': log_source, 'trace_id': trace_id})
+        return None, {'success': False, 'code': 'INTERNAL_ERROR',
+                      'message': str(e), 'trace_id': trace_id}
+
+
+def _insert_signature_image(doc, marker, image_data, align_h, align_v, height_cm=None,
+                            relax_layout=True,
+                            log_source='前端操作', trace_id=None):
+    """插入签名图片到 marker 对应的 SET 域后。
+
+    ``height_cm`` 为用户配置的目标高度（厘米）。若为 None，则**透传 None 给
+    insert_image_after_field**，由后者根据插入位置的段落行高 + 字号自动推算
+    （`_auto_size_height_cm`）。
+
+    ``relax_layout`` 透传给 :func:`field_codes.insert_image_after_field`，
+    控制是否自动放宽段落 / 表格行的固定高度约束以避免签名被裁剪。
+    """
+    with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp:
+        tmp.write(image_data)
+        tmp_path = tmp.name
+
+    try:
+        width_cm = None
+        if height_cm and height_cm > 0:
+            from PIL import Image as PILImage
+            img = PILImage.open(io.BytesIO(image_data))
+            img_w, img_h = img.size
+            width_cm = img_w / img_h * height_cm if img_h else height_cm
+
+        count = field_codes.insert_image_after_field(
+            doc, marker, tmp_path, width_cm, height_cm,
+            log_source=log_source, relax_layout=relax_layout,
+        )
+        return count
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except Exception:
+            pass
+
+
+def _resolve_height(stamp, use_global, global_val):
+    if use_global:
+        if global_val in (None, ""):
+            return None
+        try:
+            h = float(global_val)
+            if 0.1 <= h <= 10.0:
+                return round(h, 1)
+        except (ValueError, TypeError):
+            return None
+        return None
+    h = stamp.get('height')
+    if h not in (None, ""):
+        try:
+            h = float(h)
+            if 0.1 <= h <= 10.0:
+                return round(h, 1)
+        except (ValueError, TypeError):
+            pass
+    return None
+
+
+def _resolve_noise(stamp, use_global, global_val):
+    if use_global:
+        try:
+            return int(global_val) if global_val not in (None, "") else 0
+        except (ValueError, TypeError):
+            return 0
+    n = stamp.get('noise_level')
+    if n not in (None, ""):
+        try:
+            n = int(n)
+            if 0 <= n <= 10:
+                return n
+        except (ValueError, TypeError):
+            pass
+    return 0
+
+
+def _resolve_rotate(stamp, use_global, global_min, global_max):
+    if use_global:
+        try:
+            rmin = int(global_min) if global_min not in (None, "") else 0
+        except (ValueError, TypeError):
+            rmin = 0
+        try:
+            rmax = int(global_max) if global_max not in (None, "") else 0
+        except (ValueError, TypeError):
+            rmax = 0
+        return max(-180, min(180, rmin)), max(-180, min(180, rmax))
+    r_min = stamp.get('rotate_min', 0)
+    r_max = stamp.get('rotate_max', 0)
+    try:
+        r_min = max(-180, min(180, int(r_min)))
+    except (ValueError, TypeError):
+        r_min = 0
+    try:
+        r_max = max(-180, min(180, int(r_max)))
+    except (ValueError, TypeError):
+        r_max = 0
+    return r_min, r_max