139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
from __future__ import annotations
|
||
|
||
from .shared import *
|
||
|
||
from .signal_filters import _signal_title_text
|
||
|
||
def _safe_threshold(value: Any, default: float) -> float:
|
||
try:
|
||
number = float(value)
|
||
except (TypeError, ValueError):
|
||
number = float(default)
|
||
return min(1.0, max(0.0, number))
|
||
|
||
def _parse_json_object_text(raw: str) -> dict[str, Any]:
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
return {}
|
||
if text.startswith("```"):
|
||
text = text.strip("`")
|
||
if text.lower().startswith("json"):
|
||
text = text[4:].strip()
|
||
try:
|
||
parsed = json.loads(text)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
def _seed_text_snapshot(seed: dict[str, Any]) -> str:
|
||
pieces: list[str] = []
|
||
for value in seed.values():
|
||
if isinstance(value, str):
|
||
pieces.append(value)
|
||
elif isinstance(value, (list, tuple)):
|
||
pieces.extend([str(item) for item in value if item is not None])
|
||
return "\n".join([part for part in pieces if str(part).strip()]).strip()
|
||
|
||
def _needs_seed_localize_zh(seed: dict[str, Any]) -> bool:
|
||
text = _seed_text_snapshot(seed)
|
||
if not text:
|
||
return False
|
||
zh_chars = len(re.findall(r"[\u4e00-\u9fff]", text))
|
||
latin_chars = len(re.findall(r"[A-Za-z]", text))
|
||
return zh_chars < 20 and latin_chars > 40
|
||
|
||
async def _localize_seed_to_zh(
|
||
*,
|
||
db: AsyncSession,
|
||
tenant_id: int,
|
||
project_id: int,
|
||
seed: dict[str, Any],
|
||
) -> dict[str, Any]:
|
||
if not seed:
|
||
return seed
|
||
prompt = (
|
||
"将下列 JSON 草稿的自然语言内容翻译为简体中文。"
|
||
"要求：仅返回 JSON；键名与结构保持不变；模型名/专有技术名词可保留英文。\n\n"
|
||
f"{json.dumps(seed, ensure_ascii=False)}"
|
||
)
|
||
try:
|
||
raw = await llm_service.generate(
|
||
prompt,
|
||
function_name="sstc_charter_localize_zh",
|
||
db=db,
|
||
tenant_id=tenant_id,
|
||
project_id=project_id,
|
||
)
|
||
localized = _parse_json_object_text(raw)
|
||
return localized if localized else seed
|
||
except Exception:
|
||
return seed
|
||
|
||
def _to_string_list(value: Any) -> list[str]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, str):
|
||
lines = [line.strip(" -\t") for line in value.replace("\r", "").split("\n")]
|
||
return [line for line in lines if line]
|
||
if isinstance(value, (list, tuple)):
|
||
output: list[str] = []
|
||
for item in value:
|
||
text = str(item or "").strip(" -\t")
|
||
if text:
|
||
output.append(text)
|
||
return output
|
||
text = str(value).strip()
|
||
return [text] if text else []
|
||
|
||
def _to_bullet_text(items: list[str]) -> str:
|
||
cleaned = [str(item or "").strip() for item in items if str(item or "").strip()]
|
||
if not cleaned:
|
||
return ""
|
||
return "\n".join([f"- {item}" for item in cleaned])
|
||
|
||
def _extract_charter_signal_ids(content: dict[str, Any], charter_signal_id: Optional[int] = None) -> list[int]:
|
||
signal_ids: list[int] = []
|
||
raw_ids = content.get("signal_ids")
|
||
if isinstance(raw_ids, list):
|
||
for sid in raw_ids:
|
||
try:
|
||
signal_ids.append(int(sid))
|
||
except Exception:
|
||
continue
|
||
if charter_signal_id:
|
||
try:
|
||
cid = int(charter_signal_id)
|
||
if cid not in signal_ids:
|
||
signal_ids.append(cid)
|
||
except Exception:
|
||
pass
|
||
return sorted(set(signal_ids))
|
||
|
||
def _build_signal_context_text(
|
||
signal_rows: list[Signal],
|
||
extra_text: str = "",
|
||
raw_text: str = "",
|
||
) -> str:
|
||
parts: list[str] = []
|
||
for idx, signal in enumerate(signal_rows, start=1):
|
||
title = _signal_title_text(signal) or f"信号{idx}"
|
||
parts.append(
|
||
"\n".join(
|
||
[
|
||
f"[信号{idx}] 标题：{title}",
|
||
f"来源：{signal.source or '-'}",
|
||
f"优先级：{signal.priority or '-'}",
|
||
f"内容：{str(signal.content or '').strip()}",
|
||
]
|
||
).strip()
|
||
)
|
||
extra_blocks: list[str] = []
|
||
for block in (extra_text, raw_text):
|
||
text = str(block or "").strip()
|
||
if text and text not in extra_blocks:
|
||
extra_blocks.append(text)
|
||
for text in extra_blocks:
|
||
parts.append(f"[补充上下文]\n{text}")
|
||
return "\n\n".join([part for part in parts if part]).strip()
|
||
|