chentianrui
376 lines
14 KiB
Python
376 lines
14 KiB
Python
"""
|
||
批量对比多个 calculation_results.json 与 project_data.json 中的费用
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from difflib import SequenceMatcher
|
||
|
||
|
||
def extract_guid_from_filename(filename):
|
||
"""从文件名中提取 GUID"""
|
||
pattern = r"([0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12})"
|
||
match = re.search(pattern, filename, re.IGNORECASE)
|
||
return match.group(1).upper() if match else None
|
||
|
||
|
||
def normalize_cost_name(name):
|
||
"""标准化费用名称，便于匹配"""
|
||
name = re.sub(r"_\w+$", "", name) # 去除后缀如 _GJJ
|
||
name = re.sub(r"[^\w]", "", name) # 去除符号
|
||
return name.lower()
|
||
|
||
|
||
def calculate_similarity(a, b):
|
||
"""计算两个字符串的相似度"""
|
||
return SequenceMatcher(None, a, b).ratio()
|
||
|
||
|
||
def load_calculation_results(json_file_path, project_type: str = "budget"):
|
||
"""读取计算结果 JSON，返回费用总和字典
|
||
|
||
- budget（预算工程）: 期望结构为 { node: { cost_name: number, ... }, ... }
|
||
逐项累加 cost_name -> number。
|
||
- inventory（清单工程）: 结构通常为 { category: { item: { metric_name: number, ... } } }
|
||
将最内层 metrics（如“合价/直接费/人工费/材料费/机械费/措施费/间接费/安全文明施工费”等）按 metric_name 汇总累计。
|
||
"""
|
||
with open(json_file_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
total_costs = {}
|
||
|
||
if project_type == "inventory":
|
||
# 三级结构：大类 -> 清单项 -> 指标dict
|
||
for _, items in (data or {}).items():
|
||
if isinstance(items, dict):
|
||
for _, metrics in items.items():
|
||
if isinstance(metrics, dict):
|
||
for metric_name, metric_val in metrics.items():
|
||
# 仅累计数值型
|
||
if isinstance(metric_val, (int, float)):
|
||
total_costs[metric_name] = total_costs.get(metric_name, 0) + float(metric_val)
|
||
else:
|
||
# 预算：两级结构：节点 -> 费用名: 数值
|
||
for _, cost_dict in (data or {}).items():
|
||
if isinstance(cost_dict, dict):
|
||
for cost_name, value in cost_dict.items():
|
||
if isinstance(value, (int, float)):
|
||
total_costs[cost_name] = total_costs.get(cost_name, 0) + float(value)
|
||
return total_costs
|
||
|
||
|
||
def find_node_by_guid(expense_preview, target_guid):
|
||
"""在 expensePreview 中递归查找 GUID 对应的节点，并返回其 sum 列表用于对比。
|
||
|
||
适配新结构：每个 GUID 节点包含三个列表：
|
||
- sum: 费用条目（含 id/cost）——用于对比
|
||
- children: 递归子节点列表（继续下钻）
|
||
- rcj: 其他数据（此处忽略）
|
||
"""
|
||
|
||
def search_recursive(node):
|
||
"""统一遍历任意层级结构（dict/list），匹配 GUID 并返回其 children 列表"""
|
||
if isinstance(node, dict):
|
||
# 当前节点如果带 GUID，先判断自身是否命中
|
||
current_guid = node.get("GUID", "")
|
||
if isinstance(current_guid, str) and current_guid:
|
||
current_guid = current_guid.strip("{}").upper()
|
||
if current_guid == target_guid:
|
||
# 命中后返回 sum 列表作为对比项来源
|
||
return node.get("sum", [])
|
||
|
||
# 若存在 children，优先深入 children（继续查找更深层 GUID）
|
||
if "children" in node and isinstance(node["children"], list):
|
||
result = search_recursive(node["children"])
|
||
if result is not None:
|
||
return result
|
||
|
||
# 继续遍历其它所有键的值，以覆盖类似 {"建筑工程": {"拆除": [ ... ]}} 的层级
|
||
for value in node.values():
|
||
result = search_recursive(value)
|
||
if result is not None:
|
||
return result
|
||
return None
|
||
|
||
if isinstance(node, list):
|
||
for item in node:
|
||
result = search_recursive(item)
|
||
if result is not None:
|
||
return result
|
||
return None
|
||
|
||
# 其他类型（如 str/number/None）无需处理
|
||
return None
|
||
|
||
return search_recursive(expense_preview)
|
||
|
||
|
||
def load_project_data_and_find_costs(project_json_path, target_guid):
|
||
"""读取 project_data.json 并查找对应 GUID 的费用列表
|
||
|
||
集成项目类型判断：
|
||
- 若判定为预算工程（budget）：沿用当前逻辑
|
||
- 若判定为清单工程（inventory）：暂时也沿用相同逻辑（如需差异处理可再扩展）
|
||
"""
|
||
with open(project_json_path, "r", encoding="utf-8") as f:
|
||
project_data = json.load(f)
|
||
|
||
# 判断项目类型
|
||
project_type = _determine_project_type(project_data)
|
||
if project_type == "budget":
|
||
print("项目类型：预算工程（按现有逻辑处理）")
|
||
else:
|
||
print("项目类型：清单工程（暂按预算同样逻辑处理，若需差异化请告知）")
|
||
|
||
expense_preview = project_data.get("projectData", {}).get("expensePreview", {})
|
||
children = find_node_by_guid(expense_preview, target_guid)
|
||
|
||
if not children:
|
||
print(f"未找到 GUID 为 {target_guid} 的节点")
|
||
return {}
|
||
|
||
external_costs = {}
|
||
for item in children:
|
||
item_id = item.get("id", "")
|
||
cost_str = item.get("cost", "0")
|
||
try:
|
||
cost_val = float(cost_str)
|
||
except ValueError:
|
||
cost_val = 0.0
|
||
|
||
if item_id:
|
||
clean_id = re.sub(r"_\w+$", "", item_id)
|
||
external_costs[clean_id] = cost_val
|
||
return external_costs
|
||
|
||
|
||
def match_and_compare_costs(calc_costs, ext_costs, similarity_threshold=0.6):
|
||
"""匹配两个费用列表并对比"""
|
||
comparison = []
|
||
matched_ext = set()
|
||
|
||
for calc_name, calc_value in calc_costs.items():
|
||
best_match = None
|
||
best_score = 0
|
||
|
||
for ext_name in ext_costs:
|
||
if ext_name in matched_ext:
|
||
continue
|
||
score = calculate_similarity(normalize_cost_name(calc_name), normalize_cost_name(ext_name))
|
||
if score > best_score:
|
||
best_score = score
|
||
best_match = ext_name
|
||
|
||
if best_match and best_score >= similarity_threshold:
|
||
ext_value = ext_costs[best_match]
|
||
difference = calc_value - ext_value
|
||
comparison.append(
|
||
{
|
||
"项目": calc_name,
|
||
"参考值": ext_value,
|
||
"计算值": calc_value,
|
||
"差异": difference,
|
||
"原数据项": best_match,
|
||
"相似度": best_score,
|
||
}
|
||
)
|
||
matched_ext.add(best_match)
|
||
else:
|
||
comparison.append(
|
||
{
|
||
"项目": calc_name,
|
||
"参考值": None,
|
||
"计算值": calc_value,
|
||
"差异": None,
|
||
"原数据项": None,
|
||
"相似度": best_score,
|
||
}
|
||
)
|
||
|
||
# 添加未匹配的参考项
|
||
for ext_name, ext_value in ext_costs.items():
|
||
if ext_name not in matched_ext:
|
||
comparison.append(
|
||
{
|
||
"项目": None,
|
||
"参考值": ext_value,
|
||
"计算值": None,
|
||
"差异": None,
|
||
"原数据项": ext_name,
|
||
"相似度": None,
|
||
}
|
||
)
|
||
|
||
return comparison
|
||
|
||
|
||
def save_comparison_to_txt(comparison, output_txt_path):
|
||
"""保存对比结果到 TXT 文件"""
|
||
with open(output_txt_path, "w", encoding="utf-8") as f:
|
||
f.write(f"{'项目':<20} {'参考值':<25} {'计算值':<25} {'差异':<25} {'原数据项':<30}\n")
|
||
f.write("-" * 120 + "\n")
|
||
|
||
for item in comparison:
|
||
project = (item["项目"] or "").ljust(20)[:20]
|
||
ref = f"{item['参考值']:.2f}" if item["参考值"] is not None else ""
|
||
ref = ref.ljust(25)[:25]
|
||
calc = f"{item['计算值']:.2f}" if item["计算值"] is not None else ""
|
||
calc = calc.ljust(25)[:25]
|
||
diff = f"{item['差异']:.2f}" if item["差异"] is not None else ""
|
||
diff = diff.ljust(25)[:25]
|
||
original = (item["原数据项"] or "").ljust(30)[:30]
|
||
f.write(f"{project}{ref}{calc}{diff}{original}\n")
|
||
print(f"✅ 对比结果已保存至: {output_txt_path}")
|
||
|
||
|
||
def _determine_project_type(data):
|
||
"""
|
||
根据basicData中的"项目类型"或"工程类型"判断工程类型
|
||
:param data: 项目数据
|
||
:return: 'inventory' 表示清单工程，'budget' 表示预算工程
|
||
"""
|
||
# 项目类型名称映射字典：将各种变体映射到标准类型（预算/清单）
|
||
PROJECT_TYPE_MAPPING = {
|
||
"概预算工程": "预算",
|
||
"初步设计概算": "预算",
|
||
"可行性研究投资估算": "预算",
|
||
"施工图预算": "预算",
|
||
"配网定额计价": "预算",
|
||
"招标控制价": "清单",
|
||
"投标报价": "清单",
|
||
"招投标工程": "清单",
|
||
"配网清单招投标计价": "清单",
|
||
}
|
||
|
||
# 获取 basicData
|
||
basic_data = data.get("basicData") or {}
|
||
|
||
# 尝试获取 "项目类型"，若不存在则尝试获取 "工程类型"
|
||
engineering_type = basic_data.get("项目类型") or basic_data.get("工程类型") or basic_data.get("工程类别")
|
||
|
||
if engineering_type:
|
||
# 去除前后空格
|
||
engineering_type = engineering_type.strip()
|
||
# 查找映射
|
||
mapped_type = PROJECT_TYPE_MAPPING.get(engineering_type)
|
||
if mapped_type == "预算":
|
||
print(f"根据项目类型 '{engineering_type}' 判断为预算工程")
|
||
return "budget"
|
||
elif mapped_type == "清单":
|
||
print(f"根据项目类型 '{engineering_type}' 判断为清单工程")
|
||
return "inventory"
|
||
else:
|
||
print(f"项目类型 '{engineering_type}' 未在映射中定义，跳过")
|
||
|
||
# 默认按预算工程处理，以保持当前对比逻辑不变
|
||
print("未能可靠判断项目类型，默认按预算工程处理")
|
||
return "budget"
|
||
|
||
|
||
def compare_costs_batch(calc_results_folder: str, project_data_json_path: str):
|
||
"""批量对比 calculation_results.json 与项目 JSON。
|
||
|
||
- 输出目录：在 calc_results_folder 下创建 comparison_results 保存结果。
|
||
- 根据 project_data_json_path 判定工程类型（预算/清单），以选择解析方式。
|
||
"""
|
||
# 输出对比结果的文件夹放在 calc_results_folder 内
|
||
output_folder = os.path.join(calc_results_folder, "comparison_results")
|
||
os.makedirs(output_folder, exist_ok=True)
|
||
|
||
# 支持的文件名关键词（可根据实际命名调整）
|
||
result_file_keyword = "_calculation_results.json"
|
||
|
||
# 预读取项目 JSON 并判定工程类型（供计算结果解析使用）
|
||
try:
|
||
with open(project_data_json_path, "r", encoding="utf-8") as f:
|
||
project_data_for_type = json.load(f)
|
||
project_type = _determine_project_type(project_data_for_type)
|
||
except Exception as e:
|
||
print(f"读取项目文件以判定类型失败，将默认按预算处理，错误: {e}")
|
||
project_type = "budget"
|
||
|
||
print(f"开始批量处理文件夹: {calc_results_folder}（项目类型: { '清单' if project_type=='inventory' else '预算' }）")
|
||
processed_count = 0
|
||
|
||
# 递归扫描 calc_results_folder 下所有子目录，寻找结果文件
|
||
matched_files = []
|
||
for root, _dirs, files in os.walk(calc_results_folder):
|
||
for filename in files:
|
||
if filename.endswith(".json") and result_file_keyword in filename:
|
||
matched_files.append(os.path.join(root, filename))
|
||
|
||
if not matched_files:
|
||
print("未在任何子目录中发现 '*_calculation_results.json' 文件，请确认 BCL 结果输出位置与命名。")
|
||
|
||
for calc_json_path in matched_files:
|
||
filename = os.path.basename(calc_json_path)
|
||
|
||
try:
|
||
# 提取 GUID
|
||
guid = extract_guid_from_filename(filename)
|
||
if not guid:
|
||
print(f"⚠️ 无法从文件名提取 GUID: {filename}")
|
||
continue
|
||
|
||
print(f"\n📄 处理文件: {filename}")
|
||
print(f" 提取 GUID: {guid}")
|
||
|
||
# 读取计算结果（按工程类型解析）
|
||
calc_costs = load_calculation_results(calc_json_path, project_type=project_type)
|
||
print(f" 加载 {len(calc_costs)} 个计算费用项")
|
||
|
||
# 从主 JSON 获取参考费用
|
||
ext_costs = load_project_data_and_find_costs(project_data_json_path, guid)
|
||
print(f" 找到 {len(ext_costs)} 个参考费用项")
|
||
|
||
# 对比
|
||
comparison = match_and_compare_costs(calc_costs, ext_costs, similarity_threshold=0.6)
|
||
|
||
# 生成输出文件名（与原 JSON 同名，但输出到指定文件夹）
|
||
base_name = os.path.splitext(filename)[0]
|
||
output_txt_path = os.path.join(output_folder, base_name + ".txt")
|
||
|
||
# 保存结果
|
||
save_comparison_to_txt(comparison, output_txt_path)
|
||
|
||
processed_count += 1
|
||
|
||
except Exception as e:
|
||
print(f"❌ 处理文件 {filename} 时出错: {e}")
|
||
|
||
print(f"\n✅ 批量处理完成！共处理 {processed_count} 个文件。")
|
||
print(f"📊 所有对比结果已保存至: {output_folder}")
|
||
|
||
|
||
# --------------------------
|
||
# 测试入口：直接运行本文件
|
||
# --------------------------
|
||
def _main():
|
||
"""直接运行费用对比，无需命令行或输入"""
|
||
|
||
# ✅ 在这里直接填写你要测试的路径（可自行修改）
|
||
calc_dir = r"data/input/bclresults/2022通用定额细分3.1.16"
|
||
proj_json = r"data/input/merged/2022通用定额细分3.1.16.json"
|
||
|
||
# 检查路径是否存在
|
||
if not os.path.exists(calc_dir):
|
||
print(f"❌ BCL 结果文件夹不存在: {calc_dir}")
|
||
return
|
||
|
||
if not os.path.exists(proj_json):
|
||
print(f"❌ 项目 JSON 文件不存在: {proj_json}")
|
||
return
|
||
|
||
print(f"✅ BCL 计算结果目录: {calc_dir}")
|
||
print(f"✅ 项目 JSON 文件: {proj_json}")
|
||
print("🚀 开始执行费用对比...")
|
||
|
||
# 调用你的主函数
|
||
compare_costs_batch(calc_dir, proj_json)
|
||
print("🎉 费用对比完成！")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_main()
|