import re import subprocess wanted_variations = [] skip_variations = ["horizontal_group_bar_chart_13"] def parse_translate(transform_str): """解析SVG transform属性中的translate值""" if not transform_str: return 0.0, 0.0 total_x, total_y = 0.0, 0.0 # 查找所有translate(x, y)或translate(x)模式 translate_matches = re.finditer(r'translate\(\s*([-\d.]+)(?:\s*,\s*([-\d.]+))?\s*\)', transform_str) for match in translate_matches: x = float(match.group(1)) y = float(match.group(2)) if match.group(2) else 0.0 total_x += x total_y += y return total_x, total_y def parse_percentage(value_str, reference): """解析SVG中的百分比值""" if not value_str: return 0.0 value_str = str(value_str).strip() # 处理百分比 if '%' in value_str: percentage = float(value_str.replace('%', '')) return reference * percentage / 100.0 # 处理像素值 if 'px' in value_str: return float(value_str.replace('px', '')) # 处理纯数字 try: return float(value_str) except ValueError: return 0.0 def get_accumulated_transform(elem, root): # 获取从根元素到当前元素的所有transform累加 total_dx, total_dy = 0.0, 0.0 current = elem # 存储父元素路径以便从外向内累积变换 path = [] while current is not None: path.append(current) if current == root: break current = current.getparent() # 从外向内累积变换（从根元素到目标元素） for current in reversed(path): transform = current.get("transform") if transform: dx, dy = parse_translate(transform) total_dx += dx total_dy += dy return total_dx, total_dy def get_text_bounding_box(text_elem, svg_width, svg_height): """获取文本元素的精确边界框，考虑各种对齐方式和基线属性""" # 获取SVG的根元素 root = text_elem.getroottree().getroot() # 获取累积的变换信息 dx, dy = get_accumulated_transform(text_elem, root) # 基本位置信息 x = parse_percentage(text_elem.get('x', '0'), svg_width) + dx y = parse_percentage(text_elem.get('y', '0'), svg_height) + dy # 获取字体大小 font_size_str = text_elem.get('font-size', '16') font_size = float(font_size_str.replace('px', '')) if 'px' in font_size_str else float(font_size_str) # 获取文本内容及长度估算 text_content = text_elem.text or "" # 更精确的宽度估算 - 实际应用中可能需要根据字体特性调整 text_width = font_size * 0.6 * len(text_content) text_height = font_size * 1.2 # 考虑行高 # 处理文本对齐方式 text_anchor = text_elem.get('text-anchor', 'start') if text_anchor == 'start': text_left = x text_right = x + text_width elif text_anchor == 'middle': text_left = x - text_width / 2 text_right = x + text_width / 2 elif text_anchor == 'end': text_left = x - text_width text_right = x else: # 默认为start text_left = x text_right = x + text_width # 处理基线属性 dominant_baseline = text_elem.get('dominant-baseline', 'auto') if dominant_baseline == 'hanging': # 悬挂基线，文本从y坐标开始向下展示 text_top = y text_bottom = y + text_height elif dominant_baseline == 'middle': # 中间基线，文本在y坐标上下对称分布 text_top = y - text_height / 2 text_bottom = y + text_height / 2 else: # 默认为alphabetic或auto # 字母基线，文本基线在y坐标上 text_top = y - 0.8 * text_height text_bottom = y + 0.2 * text_height return { 'x1': text_left, 'y1': text_top, 'x2': text_right, 'y2': text_bottom, 'width': text_right - text_left, 'height': text_bottom - text_top, 'content': text_content } def check_text_overlap(tree): """检查SVG中所有文本元素是否存在重叠""" root = tree.getroot() svg_width = float(root.get('width', '0').replace('px', '')) svg_height = float(root.get('height', '0').replace('px', '')) # 收集所有文本元素的边界框 text_boxes = [] text_elems = [] # 存储对应的文本元素 for text_elem in root.xpath("//text | //*[local-name()='text']"): # 检查文本元素或其父元素是否存在rotate变换 if has_rotate_transform(text_elem, root): continue # 跳过有rotate变换的文本元素 bbox = get_text_bounding_box(text_elem, svg_width, svg_height) text_boxes.append(bbox) text_elems.append(text_elem) # 检查任意两个文本框是否重叠 overlapping_pairs = [] for i, box1 in enumerate(text_boxes): for j, box2 in enumerate(text_boxes[i+1:], i+1): # 检查两个盒子是否重叠 if (box1['x1'] < box2['x2'] and box1['x2'] > box2['x1'] and box1['y1'] < box2['y2'] and box1['y2'] > box2['y1']): # 计算重叠区域的大小 overlap_width = min(box1['x2'], box2['x2']) - max(box1['x1'], box2['x1']) overlap_height = min(box1['y2'], box2['y2']) - max(box1['y1'], box2['y1']) overlap_area = overlap_width * overlap_height # 计算重叠比例 min_area = min(box1['width'] * box1['height'], box2['width'] * box2['height']) overlap_ratio = overlap_area / min_area if min_area > 0 else 0 # 如果重叠比例超过阈值，记录这对重叠文本 if overlap_ratio > 0.25: # 可以根据需要调整阈值 overlapping_pairs.append({ 'text1': box1['content'], 'text2': box2['content'], 'overlap_ratio': overlap_ratio, 'box1': box1, 'box2': box2, 'elem1': text_elems[i], # 保存对应的文本元素 'elem2': text_elems[j] }) return overlapping_pairs import os import json import shutil import argparse from multiprocessing import Pool, Manager, Lock from tqdm import tqdm from lxml import etree def has_image_tag(svg_content): """检查SVG内容是否包含 1: # 保留第一个，删除其他的 first_found = False def replace_xmlns(match): nonlocal first_found if not first_found: first_found = True return match.group(0) # 保留第一个 else: return '' # 删除其他的 content = re.sub(r'\s+xmlns\s*=\s*["\']http://www\.w3\.org/2000/svg["\']', replace_xmlns, content) changed = True if changed: # 清理多余的空格 content = re.sub(r'\s+', ' ', content) content = re.sub(r'\s+>', '>', content) # 写回文件 with open(svg_path, 'w', encoding='utf-8') as f: f.write(content) return True # 表示文件被修改了 return False # 表示文件没有被修改 except Exception as e: print(f"修复SVG命名空间失败 {svg_path}: {e}") return False def process_single_folder(args): """处理单个文件夹""" # 解包参数，兼容keep_all if len(args) == 5: folder_path, output_dir, counter, lock, keep_all = args else: folder_path, output_dir, counter, lock = args keep_all = False # 先检查SVG是否存在 svg_path = os.path.join(folder_path, "chart.svg") png_path = os.path.join(folder_path, "chart.png") if not os.path.exists(svg_path): print(f"跳过文件夹 {folder_path}，缺少文件 chart.svg") return None # 检查并补全SVG根节点属性，并修正image标签的xlink:href try: # 先修复可能的重复命名空间声明 fix_duplicate_xmlns(svg_path) tree = etree.parse(svg_path) root = tree.getroot() changed = False # 补全根节点命名空间 if root.get("xmlns") != "http://www.w3.org/2000/svg": root.set("xmlns", "http://www.w3.org/2000/svg") changed = True # 检查xlink命名空间声明（使用正确的命名空间格式） xlink_ns = "{http://www.w3.org/2000/xmlns/}xlink" if root.get(xlink_ns) != "http://www.w3.org/1999/xlink": root.set(xlink_ns, "http://www.w3.org/1999/xlink") changed = True # 修正所有image标签的xlink:href和href nsmap = root.nsmap.copy() if root.nsmap else {} if None in nsmap: nsmap["svg"] = nsmap.pop(None) for image in root.xpath('.//svg:image | .//image', namespaces=nsmap): href = image.get('href') xlink_href = image.get('{http://www.w3.org/1999/xlink}href') # 如果只有href没有xlink:href，则补充 if href and not xlink_href: image.set('{http://www.w3.org/1999/xlink}href', href) changed = True # 如果只有xlink:href没有href，也补充href if xlink_href and not href: image.set('href', xlink_href) changed = True # 如果都没有，尝试找data-href等 if not href and not xlink_href: for key in image.attrib: if 'href' in key: val = image.attrib[key] image.set('{http://www.w3.org/1999/xlink}href', val) image.set('href', val) changed = True break if changed: tree.write(svg_path, encoding="utf-8", xml_declaration=True) except Exception as e: print(f"SVG命名空间检查/修正失败: {svg_path}，错误: {e}") # 如果SVG存在但PNG不存在，自动转换 if not os.path.exists(png_path): try: subprocess.run([ 'rsvg-convert', '-f', 'png', '-o', png_path, '--dpi-x', '300', '--dpi-y', '300', '--background-color', '#ffffff', svg_path ], check=True) print(f"已自动将SVG转为PNG: {svg_path} -> {png_path}") except Exception as e: print(f"SVG转PNG失败: {svg_path}，错误: {e}") return { "original_path": folder_path, "reason": "svg_to_png_failed", "discarded": True } # 再检查其他必需文件 for file in ["data.json", "info.json"]: if not os.path.exists(os.path.join(folder_path, file)): print(f"跳过文件夹 {folder_path}，缺少文件 {file}") return None # 如果keep_all为True，直接复制并保留，无需任何判断 if keep_all: info_path = os.path.join(folder_path, "info.json") with open(info_path, 'r') as f: info_data = json.load(f) chart_variation = info_data.get("chart_variation", "") # 转换metadata new_info = { "chart_variation": info_data.get("chart_variation", ""), "image_mode": info_data.get("image_mode", ""), "background_color": info_data.get("background_color", ""), "chart_type": info_data.get("chart_type", "") } original_data_source = info_data.get("data_source") if isinstance(original_data_source, str) and original_data_source: new_info["data_source"] = original_data_source.split('/')[-1] else: new_info["data_source"] = "" title_to_chart = info_data.get("title_to_chart", "") image_to_chart = info_data.get("image_to_chart", "") if image_to_chart is None or image_to_chart == "none" or str(image_to_chart).lower() == "none": image_to_chart = "none" new_info["layout_template"] = f"{title_to_chart}_{image_to_chart}" with lock: folder_id = counter.value counter.value += 1 new_folder_name = f"{folder_id:08d}" new_folder_path = os.path.join(output_dir, new_folder_name) os.makedirs(new_folder_path, exist_ok=True) for file in ["chart.svg", "chart.png", "data.json", "info.json"]: src_file = os.path.join(folder_path, file) dst_file = os.path.join(new_folder_path, file) shutil.copy2(src_file, dst_file) with open(os.path.join(new_folder_path, "info.json"), 'w') as f: json.dump(new_info, f, indent=2) try: chart_variation_to_log = new_info.get("chart_variation", "") layout_template_to_log = new_info.get("layout_template", "") chart_variation_log_path = os.path.join(output_dir, "chart_variations.txt") layout_template_log_path = os.path.join(output_dir, "layout_templates.txt") with open(chart_variation_log_path, 'a', encoding='utf-8') as cv_file: cv_file.write(f"{chart_variation_to_log}\n") with open(layout_template_log_path, 'a', encoding='utf-8') as lt_file: lt_file.write(f"{layout_template_to_log}\n") except Exception as e: print(f"警告：无法写入日志文件到 {output_dir}。chart_variation: {chart_variation_to_log}, layout_template: {layout_template_to_log}。错误: {e}") return { "original_path": folder_path, "new_path": new_folder_path, "overlap_rate": 0, "discarded": False } # 新增：检查SVG文件大小 svg_path = os.path.join(folder_path, "chart.svg") try: svg_file_size = os.path.getsize(svg_path) if svg_file_size > 15 * 1024 * 1024: # 10MB print(f"跳过文件夹 {folder_path}，SVG文件大小 ({svg_file_size / (1024*1024):.2f}MB) 超过15MB") return { "original_path": folder_path, "reason": "svg_too_large", "discarded": True } except OSError as e: print(f"无法获取SVG文件大小 {svg_path}：{e}，跳过此文件夹检查") return { "original_path": folder_path, "reason": "svg_size_check_error", "discarded": True } # 先检查chart_name是否在mapping中 info_path = os.path.join(folder_path, "info.json") try: with open(info_path, 'r') as f: info_data = json.load(f) except json.JSONDecodeError as e: print(f"跳过文件夹 {folder_path}，解析 info.json 失败：{e}") return { "original_path": folder_path, "reason": "info_json_decode_error", "discarded": True } # 检查data.json中的数据数量 data_path = os.path.join(folder_path, "data.json") try: with open(data_path, 'r') as f: data = json.load(f) if isinstance(data["data"], list) and len(data) <= 2: print(f"跳过文件夹 {folder_path}，data.json中的数据数量不超过2") return { "original_path": folder_path, "reason": "insufficient_data", "discarded": True } except json.JSONDecodeError as e: print(f"跳过文件夹 {folder_path}，解析 data.json 失败：{e}") return { "original_path": folder_path, "reason": "data_json_decode_error", "discarded": True } svg_path = os.path.join(folder_path, "chart.svg") # 检查SVG文件是否包含image标签 with open(svg_path, 'r', encoding='utf-8') as f: svg_content = f.read() # 如果SVG不包含image标签，归类为丢弃 if not has_image_tag(svg_content): print(f"跳过文件夹 {folder_path}，SVG不包含image标签") return { "original_path": folder_path, "reason": "no_image_tag", "discarded": True } try: # 先修复可能的重复命名空间声明 fix_duplicate_xmlns(svg_path) tree = etree.parse(svg_path) except etree.XMLSyntaxError as e: print(f"跳过文件夹 {folder_path}，解析 chart.svg 失败：{e}") return { "original_path": folder_path, "reason": "svg_parse_error", "discarded": True } # 新增：检查最小字体大小 min_font_size_violation = False for text_elem in tree.xpath("//text | //*[local-name()='text']"): font_size_attr = text_elem.get('font-size', '16') # 默认16px font_size_str = str(font_size_attr) # Ensure it's a string for processing try: # Common case: '12px' or '12' if 'px' in font_size_str: current_font_size = float(font_size_str.replace('px', '')) else: current_font_size = float(font_size_str) if current_font_size <= 9: min_font_size_violation = True break except ValueError: # Handle cases where font-size might be non-numeric e.g. "inherit", "medium" print(f"警告：文件夹 {folder_path} 中的文本元素存在无法解析的font-size值 '{font_size_attr}'。此文本元素的字体大小检查将被跳过。") pass # Or, depending on policy: min_font_size_violation = True if min_font_size_violation: print(f"跳过文件夹 {folder_path}，存在字体小于7px的文本。") return { "original_path": folder_path, "reason": "min_font_size_violation", "discarded": True } overlapping_pairs = check_text_overlap(tree) # 打印重叠文本对信息 if overlapping_pairs and False: print(f"\n在文件夹 {folder_path} 中发现重叠文本：") for i, pair in enumerate(overlapping_pairs, 1): print(f"重叠对 {i}:") print(f" 文本1: '{pair['text1']}'") print(f" 边界框1: x1={pair['box1']['x1']:.1f}, y1={pair['box1']['y1']:.1f}, x2={pair['box1']['x2']:.1f}, y2={pair['box1']['y2']:.1f}, w={pair['box1']['width']:.1f}, h={pair['box1']['height']:.1f}") # 检查并打印rotate信息 elem1 = pair['elem1'] has_rotate1 = has_rotate_transform(elem1, tree.getroot()) print(f" 是否有rotate变换1: {has_rotate1}") # 打印原始transform信息 transform1 = elem1.get("transform", "无transform") print(f" 原始transform1: {transform1}") # 打印transform层级链 print(f" transform链1:") current = elem1 level = 1 root = tree.getroot() while current is not None and current != root.getparent(): transform = current.get("transform") if transform: print(f" 层级{level}: {transform}") level += 1 current = current.getparent() print(f" 文本2: '{pair['text2']}'") print(f" 边界框2: x1={pair['box2']['x1']:.1f}, y1={pair['box2']['y1']:.1f}, x2={pair['box2']['x2']:.1f}, y2={pair['box2']['y2']:.1f}, w={pair['box2']['width']:.1f}, h={pair['box2']['height']:.1f}") # 检查并打印rotate信息 elem2 = pair['elem2'] has_rotate2 = has_rotate_transform(elem2, tree.getroot()) print(f" 是否有rotate变换2: {has_rotate2}") # 打印原始transform信息 transform2 = elem2.get("transform", "无transform") print(f" 原始transform2: {transform2}") # 打印transform层级链 print(f" transform链2:") current = elem2 level = 1 while current is not None and current != root.getparent(): transform = current.get("transform") if transform: print(f" 层级{level}: {transform}") level += 1 current = current.getparent() print(f" 重叠比例: {pair['overlap_ratio']:.2f}") # 计算重叠率 # 首先获取所有文本元素数量 root = tree.getroot() total_texts = len(root.xpath("//text | //*[local-name()='text']")) # 计算有多少不同的文本元素存在重叠 overlapping_texts = set() for pair in overlapping_pairs: overlapping_texts.add(pair['text1']) overlapping_texts.add(pair['text2']) overlap_rate = len(overlapping_texts) * 100 / total_texts if total_texts > 0 else 0 # 判断是否丢弃 if info_data.get("chart_variation", "") not in wanted_variations: is_discarded = (overlap_rate > 5 or len(overlapping_pairs) > 3) elif info_data.get("chart_variation", "") in skip_variations: is_discarded = True else: is_discarded = (overlap_rate > 15 or len(overlapping_pairs) > 5) # 如果需要丢弃，直接返回不复制文件 if is_discarded: return { "original_path": folder_path, "reason": "text_overlap", "discarded": True } # 只处理需要保留的文件 # 处理info.json - 因为已经在前面读取过，这里可以直接使用info_data # 转换metadata new_info = { "chart_variation": info_data.get("chart_variation", ""), "image_mode": info_data.get("image_mode", ""), "background_color": info_data.get("background_color", "") } # 新增：保留并处理data_source original_data_source = info_data.get("data_source") # Get value, could be None if isinstance(original_data_source, str) and original_data_source: # Check if it's a non-empty string new_info["data_source"] = original_data_source.split('/')[-1] else: new_info["data_source"] = "" # Default to empty string if not found, None, or empty # 处理layout_template title_to_chart = info_data.get("title_to_chart", "") image_to_chart = info_data.get("image_to_chart", "") # 如果image_to_chart是none，改成"none"字符串 if image_to_chart is None or image_to_chart == "none" or image_to_chart.lower() == "none": image_to_chart = "none" new_info["layout_template"] = f"{title_to_chart}_{image_to_chart}" # 获取并增加计数器 with lock: folder_id = counter.value counter.value += 1 new_folder_name = f"{folder_id:08d}" # 创建目标文件夹 new_folder_path = os.path.join(output_dir, new_folder_name) os.makedirs(new_folder_path, exist_ok=True) # 复制文件到新文件夹 for file in ["chart.svg", "chart.png", "data.json", "info.json"]: src_file = os.path.join(folder_path, file) dst_file = os.path.join(new_folder_path, file) shutil.copy2(src_file, dst_file) # 保存新的info.json with open(os.path.join(new_folder_path, "info.json"), 'w') as f: json.dump(new_info, f, indent=2) # 新增：记录 chart_variation 和 layout_template 到txt文件 try: chart_variation_to_log = new_info.get("chart_variation", "") layout_template_to_log = new_info.get("layout_template", "") chart_variation_log_path = os.path.join(output_dir, "chart_variations.txt") layout_template_log_path = os.path.join(output_dir, "layout_templates.txt") with open(chart_variation_log_path, 'a', encoding='utf-8') as cv_file: cv_file.write(f"{chart_variation_to_log}\n") with open(layout_template_log_path, 'a', encoding='utf-8') as lt_file: lt_file.write(f"{layout_template_to_log}\n") except Exception as e: print(f"警告：无法写入日志文件到 {output_dir}。chart_variation: {chart_variation_to_log}, layout_template: {layout_template_to_log}。错误: {e}") return { "original_path": folder_path, "new_path": new_folder_path, "overlap_rate": overlap_rate, "discarded": False } def get_max_folder_number(directory): """获取目录中最大的数字文件夹编号""" if not os.path.exists(directory): return -1 max_num = -1 for folder_name in os.listdir(directory): folder_path = os.path.join(directory, folder_name) if os.path.isdir(folder_path): try: num = int(folder_name) max_num = max(max_num, num) except ValueError: # 如果文件夹名不是纯数字，则忽略 pass return max_num def process_folders(): """处理文件夹并检测SVG文本重叠""" parser = argparse.ArgumentParser(description="处理SVG文件并检测文本重叠") parser.add_argument("--input", "-i", nargs='+', required=True, help="输入文件夹路径，可以指定多个") parser.add_argument("--output", "-o", default="./converted2", help="输出文件夹路径") parser.add_argument("--processes", "-p", type=int, default=1, help="进程数") parser.add_argument("--keep-all", action="store_true", help="无条件保留所有文件夹，不做任何丢弃判断") args = parser.parse_args() # 创建输出目录 os.makedirs(args.output, exist_ok=True) # 收集所有输入文件夹中的子文件夹 subfolders = [] for input_dir in args.input: if not os.path.exists(input_dir): print(f"警告：输入文件夹 {input_dir} 不存在，已跳过") continue for item in os.listdir(input_dir): item_path = os.path.join(input_dir, item) if os.path.isdir(item_path): subfolders.append(item_path) if not subfolders: print("错误：未找到任何有效的输入文件夹") return print(f"找到 {len(subfolders)} 个待处理文件夹") # 扫描输出目录获取最大编号 max_number = get_max_folder_number(args.output) start_number = max_number + 1 if max_number >= 0 else 0 print(f"将从编号 {start_number} 开始") # 初始化计数器和锁 manager = Manager() counter = manager.Value('i', start_number) lock = manager.Lock() # 准备多进程参数 process_args = [(folder, args.output, counter, lock, args.keep_all) for folder in subfolders] # 使用多进程处理 with Pool(processes=args.processes) as pool: results = list(tqdm(pool.imap(process_single_folder, process_args), total=len(subfolders))) # 统计处理结果 processed = [r for r in results if r is not None] discarded = [r for r in processed if r["discarded"]] no_image = [r for r in discarded if r.get("reason") == "no_image_tag"] text_overlap = [r for r in discarded if r.get("reason") == "text_overlap"] not_in_mapping = [r for r in discarded if r.get("reason") == "chart_not_in_mapping"] chart_excluded = [r for r in discarded if r.get("reason") == "chart_type_excluded"] min_font_size_violations = [r for r in discarded if r.get("reason") == "min_font_size_violation"] svg_too_large_skips = [r for r in discarded if r.get("reason") == "svg_too_large"] svg_size_check_error_skips = [r for r in discarded if r.get("reason") == "svg_size_check_error"] info_json_decode_errors = [r for r in discarded if r.get("reason") == "info_json_decode_error"] svg_parse_errors = [r for r in discarded if r.get("reason") == "svg_parse_error"] insufficient_data = [r for r in discarded if r.get("reason") == "insufficient_data"] data_json_decode_errors = [r for r in discarded if r.get("reason") == "data_json_decode_error"] print(f"处理完成! 共处理 {len(processed)} 个文件夹，其中保留 {len(processed) - len(discarded)} 个，丢弃 {len(discarded)} 个。") print(f"丢弃原因: 没有图像标签 {len(no_image)} 个，文本重叠 {len(text_overlap)} 个，" f"chart不在mapping中 {len(not_in_mapping)} 个，chart类型被排除 {len(chart_excluded)} 个，" f"最小字体违规 {len(min_font_size_violations)} 个，SVG文件过大 {len(svg_too_large_skips)} 个，SVG大小检查错误 {len(svg_size_check_error_skips)} 个，" f"info.json解析错误 {len(info_json_decode_errors)} 个，SVG解析错误 {len(svg_parse_errors)} 个，" f"数据量不足 {len(insufficient_data)} 个，data.json解析错误 {len(data_json_decode_errors)} 个。") def has_rotate_transform(elem, root): """检查元素或其父元素是否存在rotate变换""" current = elem while current is not None: transform = current.get("transform", "") if transform and "rotate" in transform: return True if current == root: break current = current.getparent() return False if __name__ == "__main__": process_folders()