182 lines
5.3 KiB
Python
182 lines
5.3 KiB
Python
"""
|
|
报价模板访问。
|
|
|
|
真相在主账：后台上传/启用/停用 → 对象存储/本地上传路径 → 主账元数据 + mappingJson。
|
|
智能体生成前经 ledger 拉取已启用模板，再解析本地/下载字节后交 quote_fill。
|
|
|
|
禁止：把仓库 resources 当生产模板真相；禁止在本包发明价格或改 quoteVersion。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
from agent.ledger import LedgerClient, LedgerResponse
|
|
from agent.ledger.api_shell import LedgerApiShell
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TemplateEntry:
|
|
"""一条模板元数据（来自主账或联调缓存）。"""
|
|
|
|
code: str
|
|
filename: str = ""
|
|
object_key: str = ""
|
|
enabled: bool = False
|
|
source: str = "ledger"
|
|
template_id: str = ""
|
|
version_no: int = 0
|
|
version_id: str = ""
|
|
mapping_json: str = ""
|
|
biz_type: str = ""
|
|
category: str = ""
|
|
|
|
|
|
def local_cache_dir() -> Path:
|
|
"""
|
|
可选本机缓存目录（非生产真相）。
|
|
|
|
正式模板由后台上传；此目录仅开发落盘缓存，可空。
|
|
"""
|
|
here = Path(__file__).resolve()
|
|
return here.parents[2] / "resources" / "quote_templates"
|
|
|
|
|
|
def list_templates_from_ledger(
|
|
client: Optional[LedgerClient] = None,
|
|
*,
|
|
enabled_only: bool = True,
|
|
) -> LedgerResponse:
|
|
"""
|
|
向主账查询已启用报价模板列表。
|
|
|
|
正式路径：GET /inquiry/quoteTemplate/agent/listEnabled
|
|
"""
|
|
shell = LedgerApiShell(client=client)
|
|
return shell.list_quote_templates(enabled_only=enabled_only)
|
|
|
|
|
|
def entries_from_ledger_response(resp: LedgerResponse) -> list[TemplateEntry]:
|
|
"""把主账 listEnabled 结果转成 TemplateEntry。"""
|
|
if not resp.ok or resp.data is None:
|
|
return []
|
|
raw = resp.data
|
|
# Jeecg Result 可能包在 result / data
|
|
if isinstance(raw, dict):
|
|
items = raw.get("result") or raw.get("data") or raw.get("records") or []
|
|
if isinstance(raw.get("result"), list):
|
|
items = raw["result"]
|
|
elif isinstance(raw, list):
|
|
items = raw
|
|
else:
|
|
items = []
|
|
out: list[TemplateEntry] = []
|
|
for it in items:
|
|
if not isinstance(it, dict):
|
|
continue
|
|
tid = str(it.get("templateId") or it.get("id") or "")
|
|
out.append(
|
|
TemplateEntry(
|
|
code=tid or str(it.get("name") or ""),
|
|
filename=str(it.get("fileName") or ""),
|
|
object_key=str(it.get("objectKey") or ""),
|
|
enabled=True,
|
|
source="ledger",
|
|
template_id=tid,
|
|
version_no=int(it.get("versionNo") or 0),
|
|
version_id=str(it.get("versionId") or ""),
|
|
mapping_json=str(it.get("mappingJson") or ""),
|
|
biz_type=str(it.get("bizType") or ""),
|
|
category=str(it.get("category") or ""),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def list_local_cache_files() -> list[TemplateEntry]:
|
|
"""
|
|
列出本机缓存目录中的 xlsx（仅联调；可空）。
|
|
|
|
不等于后台已启用模板。
|
|
"""
|
|
root = local_cache_dir()
|
|
if not root.is_dir():
|
|
return []
|
|
out: list[TemplateEntry] = []
|
|
for path in sorted(root.iterdir()):
|
|
if not path.is_file():
|
|
continue
|
|
if path.suffix.lower() not in {".xlsx", ".xlsm", ".xltx"}:
|
|
continue
|
|
if path.name.startswith("."):
|
|
continue
|
|
out.append(
|
|
TemplateEntry(
|
|
code=path.stem,
|
|
filename=path.name,
|
|
object_key="",
|
|
enabled=False,
|
|
source="local_cache",
|
|
)
|
|
)
|
|
logger.info("quote_templates.local_cache count=%s", len(out))
|
|
return out
|
|
|
|
|
|
def list_template_registry() -> list[TemplateEntry]:
|
|
"""
|
|
优先主账；失败则回落本地缓存（可空）。
|
|
"""
|
|
resp = list_templates_from_ledger(enabled_only=True)
|
|
entries = entries_from_ledger_response(resp)
|
|
if entries:
|
|
return entries
|
|
return list_local_cache_files()
|
|
|
|
|
|
def registry_as_dicts() -> list[dict[str, Any]]:
|
|
"""便于 JSON/自检。"""
|
|
return [
|
|
{
|
|
"code": e.code,
|
|
"filename": e.filename,
|
|
"object_key": e.object_key,
|
|
"enabled": e.enabled,
|
|
"source": e.source,
|
|
"template_id": e.template_id,
|
|
"version_no": e.version_no,
|
|
"biz_type": e.biz_type,
|
|
"category": e.category,
|
|
}
|
|
for e in list_template_registry()
|
|
]
|
|
|
|
|
|
def pick_template_for_biz(
|
|
*,
|
|
biz_type: str,
|
|
keywords: Optional[list[str]] = None,
|
|
entries: Optional[list[TemplateEntry]] = None,
|
|
) -> Optional[TemplateEntry]:
|
|
"""
|
|
按业务线 + 关键词挑选模板（优先 is_default 由主账排序保证）。
|
|
|
|
关键词命中任一即优；否则同 biz 第一条。
|
|
"""
|
|
items = entries if entries is not None else list_template_registry()
|
|
same = [e for e in items if (e.biz_type == biz_type or not e.biz_type)]
|
|
if not same:
|
|
same = list(items)
|
|
if keywords:
|
|
kw_l = [k.lower() for k in keywords if k]
|
|
for e in same:
|
|
blob = f"{e.filename} {e.category} {e.code}".lower()
|
|
if any(k in blob for k in kw_l):
|
|
return e
|
|
return same[0] if same else None
|