from __future__ import annotations import re from typing import Any _HEADING = re.compile(r"(?m)^\s*#{1,6}\s+(.+?)\s*$") _TARGET_SENTENCE = re.compile(r"(?im)^\s*[-*]?\s*目标语句\s*[::]\s*(.+?)\s*$") _FIELD_LINE = re.compile(r"(?im)^\s*[-*]?\s*([^::\n]{1,24})\s*[::]\s*(.+?)\s*$") _BULLET = re.compile(r"^\s*(?:[-*+]\s+|[①②③④⑤⑥⑦⑧⑨⑩]\s*|\d{1,2}[.、)]\s*)") def parse_objective_text(value: Any) -> dict[str, Any]: """Extract business sections from the persisted direction without inventing text.""" text = _text(value) if not text: return { "headline": "创作方向未记录", "targets": [], "topicValue": None, "accountValue": None, "domainDimensions": [], "constraints": [], "constraintItems": [], "constraintGroups": [], "targetCount": 0, "domainDimensionCount": 0, "structured": False, "raw": "", } sections = _markdown_sections(text) targets = [_clean(match.group(1)) for match in _TARGET_SENTENCE.finditer(text)] if not targets: target_section = _first_section(sections, "创作总目标", "创作目标", "总目标") targets = _meaningful_lines(target_section)[:3] constraint_groups = _objective_constraint_groups(text) domain_dimensions = [ str(group.get("dimensionName")) for group in constraint_groups if group.get("kind") == "domain" and group.get("dimensionName") ] if not domain_dimensions: domain_section = _first_section(sections, "领域评估维度", "领域维度") domain_dimensions = [ line for line in _meaningful_lines(domain_section) if not re.match( r"^(?:评估内容|通过标准|强制性|前置条件)\s*[::]", line ) ] constraint_items = [ {"group": group["title"], **item} for group in constraint_groups for item in group.get("items") or [] if item.get("label") in {"通过标准", "强制性", "前置条件", "约束条件"} ] constraints = [item["value"] for item in constraint_items] topic_value = _section_preview(_first_section(sections, "选题价值主张")) account_value = _section_preview(_first_section(sections, "账号价值主张")) headline = targets[0] if targets else _first_business_sentence(text) return { "headline": headline or _clean(text), "targets": _unique(targets), "topicValue": topic_value, "accountValue": account_value, "domainDimensions": _unique(domain_dimensions), "constraints": _unique(constraints), "constraintItems": _unique_records(constraint_items), "constraintGroups": constraint_groups, "targetCount": len(_unique(targets)), "domainDimensionCount": len(_unique(domain_dimensions)), "structured": bool(sections or targets), "raw": text, } def parse_goal_text(value: Any) -> dict[str, Any]: """Return the persisted round goal as one authoritative business statement.""" text = _text(value) if not text: return { "headline": "本轮目标未记录", "focusItems": [], "raw": "", } return { "headline": text, "focusItems": [], "raw": text, } def plan_path_summary(path: Any, index: int) -> str: if not isinstance(path, dict): return _clean(path) or f"方案 {index}" path_index = path.get("path_index") or index target = _clean(path.get("target")) action = _business_plan_term(path.get("action")) method = _business_plan_term(path.get("method")) detail = plan_path_detail(path) return f"方案 {path_index}:{detail}" if detail else f"方案 {path_index}" def plan_path_detail(path: Any) -> str: if not isinstance(path, dict): return _clean(path) target = _clean(path.get("target")) action = _business_plan_term(path.get("action")) method = _business_plan_term(path.get("method")) return " · ".join(value for value in (action, target, method) if value) def plan_mode_label(mode: Any, path_count: int) -> str: if path_count == 1: return "单路" value = _clean(mode) aliases = {"race": "竞争", "赛马": "竞争", "divide": "分工"} return aliases.get(value.lower(), value or "方式未记录") def preview(value: Any, limit: int = 140) -> str | None: text = _clean(value) if not text: return None if len(text) <= limit: return text cut = text[: limit + 1] punctuation = max(cut.rfind(token) for token in "。!?;") if punctuation >= max(24, int(limit * 0.55)): return cut[: punctuation + 1] return text[: limit - 1].rstrip() + "…" def _markdown_sections(text: str) -> dict[str, str]: matches = list(_HEADING.finditer(text)) sections: dict[str, str] = {} for index, match in enumerate(matches): title = _clean(match.group(1)).replace("**", "") end = matches[index + 1].start() if index + 1 < len(matches) else len(text) body = text[match.end() : end].strip() if title and body: sections[title] = body return sections def _first_section(sections: dict[str, str], *labels: str) -> str | None: for label in labels: for title, body in sections.items(): if label in title: return body return None def _objective_constraint_groups(text: str) -> list[dict[str, Any]]: headings = list(_HEADING.finditer(text)) groups: list[dict[str, Any]] = [] for index, heading in enumerate(headings): title = _clean(heading.group(1)).replace("**", "") if not re.match(r"^(?:目标\s*\d+|领域维度\s*\d+)", title): continue end = headings[index + 1].start() if index + 1 < len(headings) else len(text) body = text[heading.end() : end] fields = [ {"label": label.strip(), "value": _clean(value)} for label, value in _FIELD_LINE.findall(body) if label.strip() in { "name", "维度名", "名称", "评估内容", "通过标准", "强制性", "前置条件", "约束条件", } and _clean(value) ] dimension_name = next( ( item["value"] for item in fields if item["label"].lower() in {"name", "维度名", "名称"} ), None, ) business_items = [ item for item in fields if item["label"].lower() not in {"name", "维度名", "名称"} ] if not business_items and not dimension_name: continue groups.append( { "title": title, "kind": "domain" if title.startswith("领域维度") else "target", "dimensionName": dimension_name, "items": business_items, } ) return groups def _section_preview(value: str | None) -> str | None: lines = _meaningful_lines(value) return lines[0] if lines else None def _meaningful_lines(value: Any) -> list[str]: result: list[str] = [] for raw in _text(value).splitlines(): line = _clean(_BULLET.sub("", raw)) if not line or line.startswith("#"): continue if re.match(r"^(?:评估维度|评估内容|通过标准|强制性|前置条件|name)\s*[::]", line, re.I): continue result.append(line) return _unique(result) def _first_business_sentence(text: str) -> str: without_headings = _HEADING.sub("", text) lines = _meaningful_lines(without_headings) return lines[0] if lines else _clean(text) def _clean(value: Any) -> str: text = str(value or "").replace("\\n", "\n") text = re.sub(r"[`*_]", "", text) text = re.sub(r"\s+", " ", text).strip(" #::") return text def _text(value: Any) -> str: return str(value or "").replace("\r\n", "\n").replace("\\n", "\n").strip() def _unique(values: list[str]) -> list[str]: result: list[str] = [] for value in values: if value and value not in result: result.append(value) return result def _unique_records(values: list[dict[str, str]]) -> list[dict[str, str]]: result: list[dict[str, str]] = [] seen: set[tuple[str, str, str]] = set() for value in values: key = ( value.get("group") or "", value.get("label") or "", value.get("value") or "", ) if key not in seen: seen.add(key) result.append(value) return result def _business_plan_term(value: Any) -> str: text = _clean(value) return { "增": "新增", "改": "修改", "删": "删除", "组": "组合", "产生脉络": "形成段落结构", "产生元素": "补充脚本元素", }.get(text, text)