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()