chengma
b9d84e71ec
实现 Excel 旧标题与旧封面路径回写原文件，按 source_file_abs/source_sheet/source_row 定位行，支持输出列自动追加、xlsm 保留 VBA、文件占用中文错误提示与 export_copy 另存副本。 Tab① 增加回写旧数据按钮与 WriteBackWorker，后台调用 excel.write_back，锁文件时提示关闭后重试；补充 Excel 与 GUI 单测覆盖回写、锁文件、另存副本和 worker 调用。 同步 harness 文档：T-204 标记完成，新增 T-204b 记录采集完成自动回写缺口，扩展 T-205 覆盖 Chrome 未启动/未登录引导保护，并更新 current-state、routes、api、architecture、requirements 与 progress。
498 lines
15 KiB
Python
498 lines
15 KiB
Python
"""Excel import and write-back helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from typing import Iterable, Optional
|
|
|
|
from . import db
|
|
|
|
|
|
try:
|
|
from openpyxl import load_workbook
|
|
|
|
OPENPYXL_IMPORT_ERROR = None
|
|
except ModuleNotFoundError as exc:
|
|
load_workbook = None
|
|
OPENPYXL_IMPORT_ERROR = exc
|
|
|
|
|
|
REQUIRED_FIELDS = ("alias", "item_id")
|
|
HEADER_ALIASES = {
|
|
"account_name": {"账号名", "账号名称", "店铺", "店铺名", "shop", "accountname"},
|
|
"alias": {"别名", "账号别名", "alias"},
|
|
"item_id": {"商品id", "商品编号", "商品id号", "itemid", "item"},
|
|
}
|
|
OUTPUT_HEADERS = {
|
|
"old_title": "旧标题",
|
|
"old_cover_path": "旧封面图片路径",
|
|
"new_title": "新标题",
|
|
"new_cover_path": "新封面图片路径",
|
|
"update_status": "更新状态",
|
|
}
|
|
OLD_WRITE_BACK_FIELDS = ("old_title", "old_cover_path")
|
|
WRITEABLE_STAGES = {"collected", "generated", "applied"}
|
|
LOCK_WINERRORS = {5, 32, 33}
|
|
LOCK_ERRNOS = {13, 16}
|
|
|
|
|
|
class ExcelError(RuntimeError):
|
|
"""Raised when Excel import or export cannot complete."""
|
|
|
|
|
|
def _require_openpyxl() -> None:
|
|
if OPENPYXL_IMPORT_ERROR is not None:
|
|
raise ExcelError("openpyxl 未安装，无法读取 Excel 文件")
|
|
|
|
|
|
def _text(value) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, float) and value.is_integer():
|
|
return str(int(value))
|
|
return str(value).strip()
|
|
|
|
|
|
def _normalize_header(value) -> str:
|
|
return re.sub(r"[\s_]+", "", _text(value).lower())
|
|
|
|
|
|
def _field_for_header(value) -> Optional[str]:
|
|
normalized = _normalize_header(value)
|
|
for field, names in HEADER_ALIASES.items():
|
|
if normalized in {_normalize_header(name) for name in names}:
|
|
return field
|
|
return None
|
|
|
|
|
|
def _normalize_item_id(value) -> str:
|
|
text = _text(value)
|
|
if re.fullmatch(r"\d+\.0", text):
|
|
text = text[:-2]
|
|
return text
|
|
|
|
|
|
def _valid_item_id(value) -> bool:
|
|
return bool(re.fullmatch(r"\d+", value or ""))
|
|
|
|
|
|
def _row_key(batch_id, source_file_abs, source_sheet, source_row) -> str:
|
|
return f"{batch_id}:{source_file_abs}:{source_sheet}:{source_row}"
|
|
|
|
|
|
def _empty_row(values) -> bool:
|
|
return all(_text(value) == "" for value in values)
|
|
|
|
|
|
def _headers_for_sheet(sheet):
|
|
first_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), ())
|
|
if _empty_row(first_row):
|
|
return None
|
|
headers = {}
|
|
for index, value in enumerate(first_row):
|
|
field = _field_for_header(value)
|
|
if field and field not in headers:
|
|
headers[field] = index
|
|
return headers
|
|
|
|
|
|
def _stats():
|
|
return {
|
|
"files": 0,
|
|
"total": 0,
|
|
"valid": 0,
|
|
"invalid": 0,
|
|
"inserted": 0,
|
|
"file_errors": [],
|
|
"row_errors": [],
|
|
}
|
|
|
|
|
|
def _file_error(file_path, message, sheet=None, missing_columns=None):
|
|
error = {
|
|
"file": file_path,
|
|
"error": message,
|
|
}
|
|
if sheet is not None:
|
|
error["sheet"] = sheet
|
|
if missing_columns:
|
|
error["missing_columns"] = list(missing_columns)
|
|
return error
|
|
|
|
|
|
def _row_error(file_path, sheet, row_number, message):
|
|
return {
|
|
"file": file_path,
|
|
"sheet": sheet,
|
|
"row": row_number,
|
|
"error": message,
|
|
}
|
|
|
|
|
|
def _read_file(file_path, stats):
|
|
source_file = str(file_path)
|
|
source_file_abs = os.path.abspath(source_file)
|
|
try:
|
|
workbook = load_workbook(source_file_abs, read_only=True, data_only=True)
|
|
except Exception as exc:
|
|
stats["file_errors"].append(_file_error(source_file, f"读取失败: {exc}"))
|
|
return []
|
|
|
|
try:
|
|
usable_sheets = []
|
|
sheet_errors = []
|
|
for sheet in workbook.worksheets:
|
|
headers = _headers_for_sheet(sheet)
|
|
if headers is None:
|
|
continue
|
|
missing = [
|
|
"别名" if field == "alias" else "商品id"
|
|
for field in REQUIRED_FIELDS
|
|
if field not in headers
|
|
]
|
|
if missing:
|
|
sheet_errors.append(
|
|
_file_error(
|
|
source_file,
|
|
"缺少必需列",
|
|
sheet=sheet.title,
|
|
missing_columns=missing,
|
|
)
|
|
)
|
|
else:
|
|
usable_sheets.append((sheet, headers))
|
|
|
|
if sheet_errors:
|
|
stats["file_errors"].extend(sheet_errors)
|
|
return []
|
|
if not usable_sheets:
|
|
stats["file_errors"].append(_file_error(source_file, "没有可解析工作表"))
|
|
return []
|
|
|
|
rows = []
|
|
for sheet, headers in usable_sheets:
|
|
max_column = max(headers.values()) + 1
|
|
for row_number, values in enumerate(
|
|
sheet.iter_rows(
|
|
min_row=2,
|
|
max_col=max_column,
|
|
values_only=True,
|
|
),
|
|
start=2,
|
|
):
|
|
if _empty_row(values):
|
|
continue
|
|
stats["total"] += 1
|
|
account_name = _text(values[headers["account_name"]]) if "account_name" in headers else ""
|
|
alias = _text(values[headers["alias"]])
|
|
item_id = _normalize_item_id(values[headers["item_id"]])
|
|
if not alias:
|
|
stats["invalid"] += 1
|
|
stats["row_errors"].append(
|
|
_row_error(source_file, sheet.title, row_number, "别名不能为空")
|
|
)
|
|
continue
|
|
if not item_id:
|
|
stats["invalid"] += 1
|
|
stats["row_errors"].append(
|
|
_row_error(source_file, sheet.title, row_number, "商品id不能为空")
|
|
)
|
|
continue
|
|
if not _valid_item_id(item_id):
|
|
stats["invalid"] += 1
|
|
stats["row_errors"].append(
|
|
_row_error(source_file, sheet.title, row_number, "商品id必须是数字")
|
|
)
|
|
continue
|
|
|
|
rows.append(
|
|
{
|
|
"source_file": source_file,
|
|
"source_file_abs": source_file_abs,
|
|
"source_sheet": sheet.title,
|
|
"source_row": row_number,
|
|
"account_name": account_name or None,
|
|
"alias": alias,
|
|
"item_id": item_id,
|
|
}
|
|
)
|
|
stats["valid"] += 1
|
|
return rows
|
|
finally:
|
|
workbook.close()
|
|
|
|
|
|
def import_tasks(file_paths: Iterable[str], path=None, note=None, write_db=True) -> dict:
|
|
"""Parse input Excel files and optionally insert rows into SQLite."""
|
|
|
|
_require_openpyxl()
|
|
if isinstance(file_paths, (str, os.PathLike)):
|
|
paths = [str(file_paths)]
|
|
else:
|
|
paths = [str(file_path) for file_path in file_paths]
|
|
stats = _stats()
|
|
stats["files"] = len(paths)
|
|
rows = []
|
|
for file_path in paths:
|
|
rows.extend(_read_file(file_path, stats))
|
|
|
|
batch_id = None
|
|
if write_db and rows:
|
|
db.init_db(path)
|
|
batch_id = db.create_batch(paths, note=note, path=path)
|
|
for row in rows:
|
|
row["row_key"] = _row_key(
|
|
batch_id,
|
|
row["source_file_abs"],
|
|
row["source_sheet"],
|
|
row["source_row"],
|
|
)
|
|
stats["inserted"] = db.insert_tasks(batch_id, rows, path=path)
|
|
return {
|
|
"batch_id": batch_id,
|
|
"rows": rows,
|
|
"stats": stats,
|
|
}
|
|
|
|
|
|
def _account_alias(account) -> str:
|
|
if isinstance(account, dict):
|
|
return str(account.get("alias") or "").strip()
|
|
return str(getattr(account, "alias", "") or "").strip()
|
|
|
|
|
|
def match_summary(rows: list[dict], accounts: list) -> dict:
|
|
"""Summarize whether imported rows match configured account aliases."""
|
|
|
|
known_aliases = {_account_alias(account) for account in accounts if _account_alias(account)}
|
|
by_account = {}
|
|
unmatched_aliases = []
|
|
matched = 0
|
|
unmatched = 0
|
|
for row in rows:
|
|
alias = str(row.get("alias") or "").strip()
|
|
if alias in known_aliases:
|
|
matched += 1
|
|
by_account[alias] = by_account.get(alias, 0) + 1
|
|
else:
|
|
unmatched += 1
|
|
if alias and alias not in unmatched_aliases:
|
|
unmatched_aliases.append(alias)
|
|
return {
|
|
"matched": matched,
|
|
"unmatched": unmatched,
|
|
"by_account": by_account,
|
|
"unmatched_aliases": unmatched_aliases,
|
|
}
|
|
|
|
|
|
def _batch_tasks(batch_id, path=None):
|
|
if not batch_id:
|
|
raise ExcelError("缺少批次 ID，无法回写 Excel")
|
|
db.init_db(path)
|
|
tasks = db.list_tasks(batch_id=batch_id, path=path)
|
|
if not tasks:
|
|
raise ExcelError(f"批次没有可回写任务: {batch_id}")
|
|
return tasks
|
|
|
|
|
|
def _filter_tasks_by_excel_path(tasks, excel_path):
|
|
if excel_path is None:
|
|
return list(tasks)
|
|
target = os.path.abspath(str(excel_path))
|
|
filtered = [
|
|
task for task in tasks
|
|
if os.path.abspath(task.source_file_abs) == target
|
|
]
|
|
if not filtered:
|
|
raise ExcelError(f"批次中没有来自该 Excel 的任务: {target}")
|
|
return filtered
|
|
|
|
|
|
def _has_write_back_data(task, fields) -> bool:
|
|
if getattr(task, "stage", None) in WRITEABLE_STAGES:
|
|
return True
|
|
return any(getattr(task, field, None) not in (None, "") for field in fields)
|
|
|
|
|
|
def _group_by_source_file(tasks, fields):
|
|
groups = {}
|
|
for task in tasks:
|
|
if not _has_write_back_data(task, fields):
|
|
continue
|
|
source_file_abs = os.path.abspath(task.source_file_abs)
|
|
groups.setdefault(source_file_abs, []).append(task)
|
|
return groups
|
|
|
|
|
|
def _is_locked_error(exc) -> bool:
|
|
return (
|
|
isinstance(exc, PermissionError)
|
|
or getattr(exc, "winerror", None) in LOCK_WINERRORS
|
|
or getattr(exc, "errno", None) in LOCK_ERRNOS
|
|
)
|
|
|
|
|
|
def _excel_suffix(path_value) -> str:
|
|
return os.path.splitext(str(path_value))[1].lower()
|
|
|
|
|
|
def _load_write_workbook(source_path):
|
|
kwargs = {}
|
|
if _excel_suffix(source_path) == ".xlsm":
|
|
kwargs["keep_vba"] = True
|
|
return load_workbook(source_path, **kwargs)
|
|
|
|
|
|
def _ensure_output_columns(sheet, fields):
|
|
header_by_normalized = {}
|
|
for column in range(1, sheet.max_column + 1):
|
|
header = _text(sheet.cell(row=1, column=column).value)
|
|
if header:
|
|
header_by_normalized.setdefault(_normalize_header(header), column)
|
|
|
|
columns = {}
|
|
next_column = sheet.max_column + 1
|
|
for field in fields:
|
|
header = OUTPUT_HEADERS[field]
|
|
normalized = _normalize_header(header)
|
|
column = header_by_normalized.get(normalized)
|
|
if column is None:
|
|
column = next_column
|
|
sheet.cell(row=1, column=column).value = header
|
|
header_by_normalized[normalized] = column
|
|
next_column += 1
|
|
columns[field] = column
|
|
return columns
|
|
|
|
|
|
def _write_tasks_to_workbook(workbook, tasks, fields) -> int:
|
|
rows = 0
|
|
missing_sheets = []
|
|
by_sheet = {}
|
|
for task in tasks:
|
|
by_sheet.setdefault(task.source_sheet, []).append(task)
|
|
|
|
for sheet_name, sheet_tasks in by_sheet.items():
|
|
if sheet_name not in workbook.sheetnames:
|
|
missing_sheets.append(sheet_name)
|
|
continue
|
|
sheet = workbook[sheet_name]
|
|
columns = _ensure_output_columns(sheet, fields)
|
|
for task in sheet_tasks:
|
|
if not _has_write_back_data(task, fields):
|
|
continue
|
|
for field in fields:
|
|
sheet.cell(
|
|
row=int(task.source_row),
|
|
column=columns[field],
|
|
).value = getattr(task, field, None) or ""
|
|
rows += 1
|
|
|
|
if missing_sheets:
|
|
names = ", ".join(sorted(set(missing_sheets)))
|
|
raise ExcelError(f"原 Excel 缺少导入时的工作表: {names}")
|
|
return rows
|
|
|
|
|
|
def _write_source_to_path(source_path, tasks, target_path, fields):
|
|
workbook = None
|
|
try:
|
|
workbook = _load_write_workbook(source_path)
|
|
except Exception as exc:
|
|
if _is_locked_error(exc):
|
|
raise ExcelError(f"Excel 文件被占用，请关闭后重试: {source_path}") from exc
|
|
raise ExcelError(f"读取 Excel 失败: {source_path}: {exc}") from exc
|
|
|
|
try:
|
|
rows = _write_tasks_to_workbook(workbook, tasks, fields)
|
|
try:
|
|
workbook.save(target_path)
|
|
except Exception as exc:
|
|
if _is_locked_error(exc):
|
|
raise ExcelError(f"Excel 文件被占用，请关闭后重试: {target_path}") from exc
|
|
raise ExcelError(f"保存 Excel 失败: {target_path}: {exc}") from exc
|
|
return rows
|
|
finally:
|
|
if workbook is not None:
|
|
workbook.close()
|
|
|
|
|
|
def _write_result(batch_id, groups, target_for_source, fields):
|
|
rows = 0
|
|
written_files = []
|
|
for source_path, tasks in groups.items():
|
|
target_path = target_for_source(source_path)
|
|
rows += _write_source_to_path(source_path, tasks, target_path, fields)
|
|
written_files.append(os.path.abspath(target_path))
|
|
return {
|
|
"ok": True,
|
|
"batch_id": batch_id,
|
|
"files": len(written_files),
|
|
"rows": rows,
|
|
"written_files": written_files,
|
|
}
|
|
|
|
|
|
def write_back(batch_id, excel_path=None, path=None) -> dict:
|
|
"""Write collected old-title/old-cover fields back to the original Excel file."""
|
|
|
|
_require_openpyxl()
|
|
fields = OLD_WRITE_BACK_FIELDS
|
|
tasks = _filter_tasks_by_excel_path(_batch_tasks(batch_id, path=path), excel_path)
|
|
groups = _group_by_source_file(tasks, fields)
|
|
return _write_result(
|
|
batch_id,
|
|
groups,
|
|
lambda source_path: source_path,
|
|
fields,
|
|
)
|
|
|
|
|
|
def _is_excel_output_path(path_value) -> bool:
|
|
return _excel_suffix(path_value) in {".xlsx", ".xlsm"}
|
|
|
|
|
|
def _unique_copy_path(out_dir, source_path):
|
|
base_name = os.path.basename(source_path)
|
|
stem, suffix = os.path.splitext(base_name)
|
|
candidate = os.path.join(out_dir, f"{stem}_cmshopee回写{suffix}")
|
|
index = 1
|
|
while os.path.exists(candidate):
|
|
candidate = os.path.join(out_dir, f"{stem}_cmshopee回写_{index}{suffix}")
|
|
index += 1
|
|
return candidate
|
|
|
|
|
|
def export_copy(batch_id, out_dir_or_path, path=None) -> dict:
|
|
"""Export a write-back copy without touching the original Excel file."""
|
|
|
|
_require_openpyxl()
|
|
if not out_dir_or_path:
|
|
raise ExcelError("缺少另存路径，无法导出 Excel 副本")
|
|
|
|
fields = OLD_WRITE_BACK_FIELDS
|
|
tasks = _batch_tasks(batch_id, path=path)
|
|
groups = _group_by_source_file(tasks, fields)
|
|
output_path = os.path.abspath(str(out_dir_or_path))
|
|
output_is_file = _is_excel_output_path(output_path)
|
|
if output_is_file and len(groups) > 1:
|
|
raise ExcelError("多个源 Excel 另存副本时，目标必须是目录")
|
|
if output_is_file:
|
|
output_dir = os.path.dirname(output_path)
|
|
if output_dir:
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
def target_for_source(source_path):
|
|
if os.path.abspath(source_path) == output_path:
|
|
raise ExcelError("另存副本路径不能与原 Excel 相同")
|
|
return output_path
|
|
else:
|
|
os.makedirs(output_path, exist_ok=True)
|
|
|
|
def target_for_source(source_path):
|
|
return _unique_copy_path(output_path, source_path)
|
|
|
|
return _write_result(batch_id, groups, target_for_source, fields)
|