785 lines
30 KiB
Python
785 lines
30 KiB
Python
"""批量 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
|