77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
# ============================================================
|
||
# 通用 Excel 现场导入器（moduleId: importers-excel, 可重生 ✅）
|
||
# 列名映射/工厂命名/工时推断等客户差异 → profiles/*.json；
|
||
# 解析逻辑复用 domain-kangni-intake（首个 profile 实现），
|
||
# 导入结果同时落 world dict 与 SQLite（项目隔离）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
PROFILES_DIR = Path(__file__).parent / "profiles"
|
||
|
||
|
||
def list_profiles() -> list[dict[str, Any]]:
|
||
"""列出可用导入 profile（id/名称/描述）。"""
|
||
out = []
|
||
for f in sorted(PROFILES_DIR.glob("*.json")):
|
||
try:
|
||
p = json.loads(f.read_text(encoding="utf-8"))
|
||
out.append({"profileId": p.get("profileId") or f.stem,
|
||
"projectName": p.get("projectName") or f.stem,
|
||
"description": p.get("description") or ""})
|
||
except (json.JSONDecodeError, OSError):
|
||
continue
|
||
return out
|
||
|
||
|
||
def load_profile(profile_id: str = "kangni") -> dict[str, Any]:
|
||
"""加载导入 profile；不存在时返回空 dict（调用方走内置默认）。"""
|
||
path = PROFILES_DIR / f"{profile_id}.json"
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
return {}
|
||
|
||
|
||
def import_site_excel(
|
||
world: dict[str, Any],
|
||
*,
|
||
profile_id: str = "kangni",
|
||
route_path: str | Path | None = None,
|
||
data_dir: str | Path | None = None,
|
||
include_sibling_orders: bool = False,
|
||
station_count: int | None = None,
|
||
) -> dict[str, Any]:
|
||
"""按 profile 导入现场 Excel → world + SQLite（激活对应项目）。"""
|
||
from server.aps_domain.kangni_intake import load_site_into_world
|
||
|
||
profile = load_profile(profile_id)
|
||
meta = load_site_into_world(
|
||
world,
|
||
route_path=route_path,
|
||
data_dir=data_dir,
|
||
include_sibling_orders=include_sibling_orders,
|
||
station_count=station_count or int(profile.get("stationCount") or 4),
|
||
profile=profile,
|
||
)
|
||
# 导入结果入库：激活该客户项目并替换其主数据
|
||
if not os.environ.get("APS_DB_DISABLED"):
|
||
try:
|
||
from server.db.sync import set_active_project, world_to_db
|
||
set_active_project(
|
||
profile.get("projectCode") or profile_id,
|
||
profile.get("projectName") or profile_id,
|
||
profile.get("industry") or "machining",
|
||
)
|
||
world_to_db(world)
|
||
meta["dbSynced"] = True
|
||
except Exception:
|
||
meta["dbSynced"] = False
|
||
return meta
|