585 lines
25 KiB
Python
585 lines
25 KiB
Python
"""
|
||
订单服务模块
|
||
---------
|
||
提供订单处理服务，协调Excel处理和订单合并流程。
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
import shutil
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple, Union, Any, Callable
|
||
|
||
from app.config.settings import ConfigManager
|
||
from app.core.utils.log_utils import get_logger
|
||
from app.core.excel.processor import ExcelProcessor
|
||
from app.core.excel.merger import PurchaseOrderMerger
|
||
from app.core.db.product_db import ProductDatabase
|
||
from app.core.db.order_metadata_db import OrderMetadataDB
|
||
from app.core.ocr.metadata_extractor import OrderMetadataExtractor, sanitize_for_filename
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
class OrderService:
|
||
"""
|
||
订单服务：协调Excel处理和订单合并流程
|
||
"""
|
||
|
||
def __init__(self, config: Optional[ConfigManager] = None, missing_barcodes_cb: Optional[Callable[[List[str]], None]] = None):
|
||
"""
|
||
初始化订单服务
|
||
|
||
Args:
|
||
config: 配置管理器，如果为None则创建新的
|
||
missing_barcodes_cb: 缺失条码的回调函数
|
||
"""
|
||
logger.info("初始化OrderService")
|
||
self.config = config or ConfigManager()
|
||
self.missing_barcodes_cb = missing_barcodes_cb
|
||
|
||
# 创建共享的商品数据库实例
|
||
db_path = self.config.get_path('Paths', 'product_db', fallback='data/product_cache.db') if hasattr(self.config, 'get_path') else 'data/product_cache.db'
|
||
tpl_folder = self.config.get('Paths', 'template_folder', fallback='templates')
|
||
item_data = self.config.get('Templates', 'item_data', fallback='商品资料.xlsx')
|
||
tpl_path = os.path.join(tpl_folder, item_data)
|
||
self.product_db = ProductDatabase(db_path, tpl_path)
|
||
|
||
# 创建Excel处理器和采购单合并器
|
||
self.excel_processor = ExcelProcessor(
|
||
self.config,
|
||
product_db=self.product_db,
|
||
missing_barcodes_cb=self.missing_barcodes_cb
|
||
)
|
||
self.order_merger = PurchaseOrderMerger(self.config)
|
||
|
||
# 元信息识别器 + 单据元信息库
|
||
self.extractor = OrderMetadataExtractor()
|
||
self.metadata_db = OrderMetadataDB(db_path)
|
||
|
||
logger.info("OrderService初始化完成")
|
||
|
||
def get_latest_excel(self) -> Optional[str]:
|
||
"""
|
||
获取最新的Excel文件
|
||
|
||
Returns:
|
||
最新Excel文件路径，如果未找到则返回None
|
||
"""
|
||
return self.excel_processor.get_latest_excel()
|
||
|
||
def process_excel(self, file_path: Optional[str] = None, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]:
|
||
"""
|
||
处理Excel订单文件，生成标准采购单
|
||
|
||
Args:
|
||
file_path: Excel文件路径，如果为None则处理最新的文件
|
||
|
||
Returns:
|
||
输出采购单文件路径，如果处理失败则返回None
|
||
"""
|
||
if not file_path:
|
||
file_path = self.excel_processor.get_latest_excel()
|
||
if not file_path:
|
||
logger.warning("未找到可处理的Excel文件")
|
||
return None
|
||
logger.info("OrderService开始处理最新Excel文件")
|
||
else:
|
||
logger.info(f"OrderService开始处理指定Excel文件: {file_path}")
|
||
|
||
# 检查是否需要特殊的供应商预处理（如杨碧月）
|
||
try:
|
||
from app.services.special_suppliers_service import SpecialSuppliersService
|
||
special_service = SpecialSuppliersService(self.config)
|
||
|
||
# 尝试识别并预处理（注意：这里不再传入 progress_cb 避免无限递归或重复进度条，
|
||
# 或者我们在 special_service 内部逻辑中处理完后直接返回结果）
|
||
# 为了避免循环调用，我们在 SpecialSuppliersService 内部不再调用 process_excel，
|
||
# 而是让 process_excel 识别后自己决定是否处理预处理后的文件。
|
||
|
||
# 我们新增一个 check_and_preprocess 方法
|
||
preprocessed_path = self._check_special_preprocess(file_path)
|
||
if preprocessed_path:
|
||
logger.info(f"检测到特殊供应商，已生成预处理文件: {preprocessed_path}")
|
||
file_path = preprocessed_path
|
||
except Exception as e:
|
||
logger.error(f"检查特殊预处理时出错: {e}")
|
||
|
||
result_path = self.excel_processor.process_specific_file(file_path, progress_cb=progress_cb)
|
||
if not result_path:
|
||
return None
|
||
|
||
# 应用单据元信息识别 + 重命名 result 与原图
|
||
try:
|
||
meta = self._extract_and_save_metadata(file_path)
|
||
if meta:
|
||
result_path = self._apply_metadata_to_filenames(
|
||
result_path, file_path, meta
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"应用单据元信息失败（不影响 result 文件）: {e}")
|
||
|
||
return result_path
|
||
|
||
def _check_special_preprocess(self, file_path: str) -> Optional[str]:
|
||
"""检查并执行特殊的预处理（支持杨碧月、烟草公司、蓉城易购）"""
|
||
try:
|
||
from app.core.utils.file_utils import smart_read_excel
|
||
import pandas as pd
|
||
import re
|
||
|
||
# 仅读取前 50 行进行智能识别 (header=None 确保能读到第一行内容)
|
||
df_head = smart_read_excel(file_path, nrows=50, header=None)
|
||
df_str = df_head.astype(str)
|
||
|
||
# 1. 识别：烟草公司 (Tobacco)
|
||
# 特征：内容中包含“专卖证号”或特定证号“510109104938”
|
||
is_tobacco = df_str.apply(lambda x: x.str.contains('专卖证号|510109104938')).any().any()
|
||
if is_tobacco:
|
||
logger.info("识别到烟草公司订单，执行专用预处理...")
|
||
from app.services.tobacco_service import TobaccoService
|
||
tobacco_svc = TobaccoService(self.config)
|
||
return tobacco_svc.preprocess_tobacco_order(file_path)
|
||
|
||
# 2. 识别：蓉城易购 (Rongcheng Yigou)
|
||
# 特征：内容中包含单号标识“RCDH”
|
||
is_rongcheng = df_str.apply(lambda x: x.str.contains('RCDH')).any().any()
|
||
if is_rongcheng:
|
||
logger.info("识别到蓉城易购订单，执行专用预处理...")
|
||
from app.services.special_suppliers_service import SpecialSuppliersService
|
||
special_svc = SpecialSuppliersService(self.config)
|
||
return special_svc.preprocess_rongcheng_yigou(file_path)
|
||
|
||
# 3. 识别：杨碧月 (Yang Biyue)
|
||
# 特征：经手人列包含“杨碧月”
|
||
handler_col = None
|
||
for col in df_head.columns:
|
||
# 在前50行中搜索“经手人”关键字
|
||
if df_head[col].astype(str).str.contains('经手人').any():
|
||
handler_col = col
|
||
break
|
||
|
||
if handler_col is not None:
|
||
# 检查该列是否有“杨碧月”
|
||
if df_head[handler_col].astype(str).str.contains('杨碧月').any():
|
||
logger.info("识别到杨碧月订单，执行专用预处理...")
|
||
from app.services.special_suppliers_service import SpecialSuppliersService
|
||
special_svc = SpecialSuppliersService(self.config)
|
||
return special_svc.process_yang_biyue_only(file_path)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"智能预处理识别失败: {e}")
|
||
return None
|
||
|
||
def get_purchase_orders(self) -> List[str]:
|
||
"""
|
||
获取采购单文件列表
|
||
|
||
Returns:
|
||
采购单文件路径列表
|
||
"""
|
||
return self.order_merger.get_purchase_orders()
|
||
|
||
def merge_purchase_orders(self, file_paths: List[str], progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]:
|
||
"""
|
||
合并指定的采购单文件
|
||
|
||
Args:
|
||
file_paths: 采购单文件路径列表
|
||
|
||
Returns:
|
||
合并后的采购单文件路径，如果合并失败则返回None
|
||
"""
|
||
logger.info(f"OrderService开始合并指定采购单: {file_paths}")
|
||
return self.merge_orders(file_paths, progress_cb)
|
||
|
||
def merge_all_purchase_orders(self, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]:
|
||
"""
|
||
合并所有可用的采购单文件
|
||
|
||
Returns:
|
||
合并后的采购单文件路径，如果合并失败则返回None
|
||
"""
|
||
logger.info("OrderService开始合并所有采购单")
|
||
return self.merge_orders(None, progress_cb)
|
||
|
||
def merge_orders(self, file_paths: Optional[List[str]] = None, progress_cb: Optional[Callable[[int], None]] = None) -> Optional[str]:
|
||
"""
|
||
合并采购单
|
||
|
||
Args:
|
||
file_paths: 采购单文件路径列表，如果为None则处理所有采购单
|
||
|
||
Returns:
|
||
合并后的采购单文件路径，如果合并失败则返回None
|
||
"""
|
||
if file_paths:
|
||
logger.info(f"OrderService开始合并指定采购单: {file_paths}")
|
||
else:
|
||
logger.info("OrderService开始合并所有采购单")
|
||
|
||
return self.order_merger.process(file_paths, progress_cb)
|
||
|
||
def validate_unit_price(self, result_path: str) -> List[str]:
|
||
"""
|
||
校验采购单单价与商品资料进货价的差异
|
||
|
||
Args:
|
||
result_path: 待校验的采购单路径
|
||
|
||
Returns:
|
||
差异信息列表，无差异返回空列表
|
||
"""
|
||
try:
|
||
import pandas as pd
|
||
from app.core.utils.file_utils import smart_read_excel
|
||
from app.core.handlers.column_mapper import ColumnMapper as CM
|
||
|
||
# 使用共享的商品数据库实例
|
||
product_db = self.product_db
|
||
|
||
# 读取待校验的采购单
|
||
df_res = smart_read_excel(result_path)
|
||
|
||
res_barcode_col = CM.find_column(list(df_res.columns), 'barcode')
|
||
res_price_col = CM.find_column(list(df_res.columns), 'unit_price')
|
||
|
||
if not res_barcode_col or not res_price_col:
|
||
logger.warning("未能在采购单中找到条码或单价列")
|
||
return []
|
||
|
||
# 批量查询进货价
|
||
barcodes = df_res[res_barcode_col].astype(str).str.strip().tolist()
|
||
item_prices = product_db.get_prices(barcodes)
|
||
|
||
results = []
|
||
for _, row in df_res.iterrows():
|
||
bc = str(row[res_barcode_col]).strip()
|
||
if bc not in item_prices:
|
||
continue
|
||
|
||
try:
|
||
res_price = float(row[res_price_col])
|
||
except (ValueError, TypeError):
|
||
continue
|
||
|
||
item_price = item_prices[bc]
|
||
diff = abs(res_price - item_price)
|
||
if diff > 1.0:
|
||
results.append(f"条码 {bc}: 采购单价={res_price} vs 进货价={item_price} 差异={diff:.2f}")
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
logger.error(f"单价校验过程中发生错误: {e}")
|
||
return []
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# 单据元信息识别 + 文件重命名
|
||
# ══════════════════════════════════════════════════════════════
|
||
|
||
def _extract_and_save_metadata(self, ocr_excel_path: str) -> Optional[Any]:
|
||
"""从 OCR 输出的 xlsx 同目录的 .meta.json 提取元信息，写入 SQLite。
|
||
|
||
Returns:
|
||
OrderMetadata 或 None（失败时）
|
||
"""
|
||
try:
|
||
base = Path(ocr_excel_path)
|
||
file_hash = base.stem
|
||
meta_path = base.with_suffix('.meta.json')
|
||
|
||
ocr_text = ''
|
||
ocr_rows: List[List[str]] = []
|
||
source_image = ''
|
||
|
||
if meta_path.exists():
|
||
try:
|
||
payload = json.loads(meta_path.read_text(encoding='utf-8'))
|
||
# 区分表格 OCR 文本和通用识别文本
|
||
ocr_text = payload.get('ocr_text', '') or ''
|
||
general_text = payload.get('general_text', '') or ''
|
||
ocr_rows = payload.get('ocr_rows', []) or []
|
||
source_image = payload.get('image_path', '') or ''
|
||
except Exception as e:
|
||
logger.warning(f"读 meta.json 失败: {e}")
|
||
|
||
# 兜底：从 xlsx 拼文本（与 OCRService._write_meta_json 的兜底一致）
|
||
if not ocr_text and not general_text:
|
||
try:
|
||
import xlrd
|
||
rb = xlrd.open_workbook(str(ocr_excel_path))
|
||
ws = rb.sheet_by_index(0)
|
||
lines = []
|
||
for r in range(ws.nrows):
|
||
row_vals = [str(ws.cell_value(r, c)) for c in range(ws.ncols)]
|
||
ocr_rows.append(row_vals)
|
||
lines.append(' '.join(row_vals))
|
||
ocr_text = '\n'.join(lines)
|
||
except Exception as e:
|
||
logger.debug(f"从 xlsx 拼 OCR 文本失败: {e}")
|
||
|
||
meta = self.extractor.extract(ocr_text, ocr_rows, general_text=general_text)
|
||
self.metadata_db.save(
|
||
file_hash=file_hash,
|
||
supplier=meta.supplier,
|
||
bill_date=meta.bill_date,
|
||
total_amount=meta.total_amount,
|
||
raw_supplier_text=meta.raw_supplier_text,
|
||
source_image=source_image,
|
||
)
|
||
logger.info(
|
||
f"元信息识别: hash={file_hash} supplier={meta.supplier!r} "
|
||
f"bill_date={meta.bill_date!r} total_amount={meta.total_amount:.2f}"
|
||
)
|
||
return meta
|
||
except Exception as e:
|
||
logger.error(f"_extract_and_save_metadata 失败: {e}")
|
||
return None
|
||
|
||
def _apply_metadata_to_filenames(self, result_path: str,
|
||
ocr_excel_path: str,
|
||
meta) -> str:
|
||
"""应用新文件名规则（按供应商+日期）：
|
||
- result (xls)：{供应商}_{日期}.xls
|
||
- output xlsx： {供应商}_{日期}.xlsx
|
||
- 原图： {供应商}_{日期}.{ext}
|
||
|
||
冲突时加 _2 / _3 ...；任一步骤失败不影响整体。
|
||
|
||
Returns:
|
||
新 result 路径（无论重命名是否成功都返回；失败时返回原路径）
|
||
"""
|
||
try:
|
||
# 1. 提取并清理元数据
|
||
supplier_clean = sanitize_for_filename(meta.supplier) or '未知供应商'
|
||
date_part = meta.bill_date or '未知日期'
|
||
|
||
# 2. 构造新的基础文件名：[供应商]_[日期]
|
||
# 按照用户要求：按照供应商名称加日期进行修改，且不含“采购单”等冗余字眼
|
||
base_name = f"{supplier_clean}_{date_part}"
|
||
|
||
file_hash = Path(ocr_excel_path).stem # 仅用作 SQLite PK，不入文件名
|
||
logger.info(f"生成标准化文件名: {base_name}")
|
||
|
||
# ── 1. result 重命名 ──
|
||
new_result_path = self._safe_rename(
|
||
result_path, f"{base_name}.xls"
|
||
)
|
||
|
||
# ── 2. output xlsx 重命名（保证 output/ 与 result/ 文件名一致）──
|
||
new_output_path = self._safe_rename(
|
||
ocr_excel_path, f"{base_name}.xlsx"
|
||
)
|
||
# 同步 meta.json
|
||
old_meta = Path(ocr_excel_path).with_suffix('.meta.json')
|
||
if old_meta.exists():
|
||
new_meta = Path(new_output_path).with_suffix('.meta.json')
|
||
if str(new_meta) != str(old_meta):
|
||
try:
|
||
if new_meta.exists(): os.remove(new_meta)
|
||
os.rename(old_meta, new_meta)
|
||
except Exception as e:
|
||
logger.warning(f"meta.json 重命名失败: {e}")
|
||
|
||
# ── 2.5 同步 processed_files.json（避免下次 OCR 重新识别）──
|
||
if str(new_output_path) != str(ocr_excel_path):
|
||
self._sync_processed_record(Path(ocr_excel_path), Path(new_output_path))
|
||
# 也要同步 ExcelProcessor 的记录
|
||
self._update_excel_process_record(Path(ocr_excel_path), Path(new_output_path))
|
||
|
||
# ── 3. 原图重命名 ──
|
||
try:
|
||
# 尝试从 processed_files.json 反查原图路径
|
||
src_image = self._find_input_image_for_output(str(ocr_excel_path))
|
||
|
||
# 如果没找到，再从 DB 拿
|
||
if not src_image:
|
||
meta_row = self.metadata_db.get(file_hash)
|
||
src_image = (meta_row or {}).get('source_image', '')
|
||
|
||
if src_image and os.path.exists(src_image):
|
||
src_p = Path(src_image)
|
||
ext = src_p.suffix
|
||
# 严禁包含原文件名，统一格式: 采购单_YYYYMMDD_供应商.ext
|
||
new_image_name = f"{base_name}{ext}"
|
||
new_image_path = self._safe_rename(src_image, new_image_name)
|
||
|
||
# ── 3.5 同步 processed_files.json 的 Key (原图路径) ──
|
||
if str(new_image_path) != str(src_p):
|
||
self._sync_processed_key(src_p, Path(new_image_path))
|
||
|
||
# 更新 source_image 路径到 DB
|
||
if str(new_image_path) != str(src_p):
|
||
self.metadata_db.save(
|
||
file_hash=file_hash,
|
||
supplier=meta.supplier,
|
||
bill_date=meta.bill_date,
|
||
total_amount=meta.total_amount,
|
||
raw_supplier_text=meta.raw_supplier_text,
|
||
source_image=str(new_image_path),
|
||
)
|
||
else:
|
||
logger.warning(f"找不到原图，跳过重命名: {src_image}")
|
||
except Exception as e:
|
||
logger.warning(f"原图重命名失败: {e}")
|
||
|
||
return str(new_result_path)
|
||
except Exception as e:
|
||
logger.error(f"_apply_metadata_to_filenames 失败: {e}")
|
||
return result_path
|
||
|
||
@staticmethod
|
||
def _safe_rename(src: str, new_basename: str) -> str:
|
||
"""把 src 重命名为 src.parent / new_basename。
|
||
|
||
- 目标已存在时加 _2 / _3 ... 后缀避免覆盖
|
||
- 失败返回原路径，不抛异常
|
||
"""
|
||
try:
|
||
src_p = Path(src)
|
||
if not src_p.exists():
|
||
return src
|
||
new_p = src_p.parent / new_basename
|
||
if str(new_p) == str(src_p):
|
||
return src
|
||
if new_p.exists():
|
||
# 冲突去重
|
||
stem, suffix = new_p.stem, new_p.suffix
|
||
n = 2
|
||
while True:
|
||
cand = src_p.parent / f"{stem}_{n}{suffix}"
|
||
if not cand.exists():
|
||
new_p = cand
|
||
break
|
||
n += 1
|
||
os.rename(src_p, new_p)
|
||
logger.info(f"重命名: {src_p.name} -> {new_p.name}")
|
||
return str(new_p)
|
||
except Exception as e:
|
||
logger.warning(f"重命名失败 {src} -> {new_basename}: {e}")
|
||
return src
|
||
|
||
@staticmethod
|
||
def _dedup_path(p: Path) -> Path:
|
||
"""路径冲突时加 _N 后缀。"""
|
||
stem, suffix = p.stem, p.suffix
|
||
parent = p.parent
|
||
n = 1
|
||
while True:
|
||
cand = parent / f"{stem}_{n}{suffix}"
|
||
if not cand.exists():
|
||
return cand
|
||
n += 1
|
||
|
||
def _find_input_image_for_output(self, output_xlsx: str) -> Optional[str]:
|
||
"""从 processed_files.json 反查生成该 output_xlsx 的原图路径。"""
|
||
try:
|
||
output_dir = self.config.get_path('Paths', 'output_folder', fallback='data/output')
|
||
record_file = os.path.join(output_dir, 'processed_files.json')
|
||
|
||
if not os.path.exists(record_file):
|
||
logger.debug(f"记录文件不存在: {record_file}")
|
||
return None
|
||
|
||
with open(record_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
target_output = os.path.normpath(output_xlsx)
|
||
found_img = None
|
||
|
||
# 1. 尝试直接匹配和规范化路径匹配
|
||
for img_path, out_path in data.items():
|
||
if os.path.normpath(out_path) == target_output:
|
||
found_img = img_path
|
||
break
|
||
|
||
if found_img:
|
||
if os.path.exists(found_img):
|
||
return found_img
|
||
|
||
# 2. 模糊匹配：如果记录中的路径不存在，尝试在同目录下找 stem 匹配的文件
|
||
# (解决因之前重命名导致的路径不一致问题)
|
||
img_p = Path(found_img)
|
||
parent = img_p.parent
|
||
if parent.exists():
|
||
stem = img_p.stem
|
||
logger.debug(f"尝试模糊匹配原图: stem={stem[:20]}... in {parent}")
|
||
# 尝试寻找以原 stem 开头的文件
|
||
for cand in parent.iterdir():
|
||
if cand.is_file() and cand.stem.startswith(stem[:20]):
|
||
logger.info(f"模糊匹配成功: {cand.name}")
|
||
return str(cand)
|
||
logger.debug("模糊匹配失败")
|
||
else:
|
||
logger.debug(f"在 processed_files.json 中未找到输出文件 {output_xlsx} 对应的原图记录")
|
||
except Exception as e:
|
||
logger.debug(f"反查原图路径失败: {e}")
|
||
return None
|
||
|
||
def _sync_processed_record(self, old_output: Path, new_output: Path) -> None:
|
||
"""重命名 output xlsx 后，把 processed_files.json 里所有 value 为旧路径的项改成新路径。"""
|
||
self._update_processed_json(old_path=old_output, new_path=new_output, is_key=False)
|
||
|
||
def _sync_processed_key(self, old_input: Path, new_input: Path) -> None:
|
||
"""重命名原图后，把 processed_files.json 里的 key 从旧路径改为新路径。"""
|
||
self._update_processed_json(old_path=old_input, new_path=new_input, is_key=True)
|
||
|
||
def _update_excel_process_record(self, old_xlsx: Path, new_xlsx: Path) -> None:
|
||
"""重命名 xlsx 后，同步 excel_process_records.json (ExcelProcessor 使用的记录)。"""
|
||
self._update_processed_json(
|
||
old_path=old_xlsx,
|
||
new_path=new_xlsx,
|
||
is_key=True,
|
||
filename='excel_process_records.json'
|
||
)
|
||
|
||
def _update_processed_json(self, old_path: Path, new_path: Path, is_key: bool = False, filename: str = 'processed_files.json') -> None:
|
||
"""更新处理记录 JSON。
|
||
|
||
Args:
|
||
old_path: 旧路径
|
||
new_path: 新路径
|
||
is_key: True 更新 key, False 更新 value
|
||
filename: JSON 文件名
|
||
"""
|
||
try:
|
||
# 确定文件路径
|
||
output_dir = self.config.get_path('Paths', 'output_folder', fallback='data/output')
|
||
record_file = os.path.join(output_dir, filename)
|
||
|
||
if not os.path.exists(record_file):
|
||
return
|
||
|
||
with open(record_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
old_str = str(old_path)
|
||
new_str = str(new_path)
|
||
changed = False
|
||
|
||
# 规范化路径以便匹配
|
||
old_norm = os.path.normpath(old_str)
|
||
|
||
if is_key:
|
||
# 更新 Key
|
||
for k in list(data.keys()):
|
||
if k == old_str or os.path.normpath(k) == old_norm:
|
||
data[new_str] = data.pop(k)
|
||
changed = True
|
||
break
|
||
else:
|
||
# 更新 Value
|
||
for k, v in list(data.items()):
|
||
if v == old_str or os.path.normpath(v) == old_norm:
|
||
data[k] = new_str
|
||
changed = True
|
||
|
||
if changed:
|
||
with open(record_file, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
logger.info(f"{filename} 已同步 ({'Key' if is_key else 'Value'}): {old_str} -> {new_str}")
|
||
except Exception as e:
|
||
logger.warning(f"更新 {filename} 失败: {e}")
|