import hashlib import re from collections import Counter from typing import Any, Dict, Iterable, List STOPWORDS = { "的", "了", "和", "与", "及", "或", "在", "中", "对", "为", "是", "有", "进行", "使用", "相关", "根据", "如下", "可以", "需要", "应当", "以及", } def enhance_slices( slices: Iterable[Dict[str, Any]], file_info: Dict[str, Any], max_slices: int, ) -> List[Dict[str, Any]]: refs: List[Dict[str, Any]] = [] filename = str(file_info.get("filename") or file_info.get("name") or "") kb_id = str(file_info.get("kb_id") or "") file_id = str(file_info.get("file_id") or file_info.get("id") or "") for idx, raw in enumerate(list(slices)[:max_slices]): text = _slice_text(raw) if not text.strip(): continue slice_id = str(raw.get("id") or raw.get("slice_id") or raw.get("uuid") or f"{file_id}:{idx}") title_path = _infer_title_path(text, filename) keywords = extract_keywords(text, limit=12) refs.append({ "kb_id": str(raw.get("kb_id") or kb_id), "file_id": str(raw.get("file_id") or file_id), "slice_id": slice_id, "ordinal": int(raw.get("ordinal") or raw.get("index") or idx), "filename": str(raw.get("filename") or filename), "title_path": title_path, "keywords": keywords, "brief": _brief(text), "chunk_type": _infer_chunk_type(text), "content_hash": hashlib.sha1(text.encode("utf-8", errors="ignore")).hexdigest(), "_text": text, }) return refs def extract_keywords(text: str, limit: int = 10) -> List[str]: text = normalize_text(text) candidates = re.findall(r"[\u4e00-\u9fff]{2,12}|[A-Za-z0-9][A-Za-z0-9_\-]{1,30}", text) normalized: List[str] = [] for token in candidates: token = token.strip("::,,。.;;、()()[]【】") if len(token) < 2 or token in STOPWORDS: continue if re.fullmatch(r"\d+", token): continue normalized.append(token) counts = Counter(normalized) return [term for term, _ in counts.most_common(limit)] def normalize_text(text: str) -> str: return re.sub(r"\s+", " ", text or "").strip() def _slice_text(raw: Dict[str, Any]) -> str: for key in ("content", "text", "page_content", "slice_content", "markdown"): value = raw.get(key) if isinstance(value, str) and value.strip(): return value return "" def _brief(text: str, limit: int = 360) -> str: return normalize_text(text)[:limit] def _infer_title_path(text: str, filename: str) -> List[str]: path = [] if filename: path.append(filename) lines = [line.strip() for line in (text or "").splitlines() if line.strip()] for line in lines[:6]: line = re.sub(r"^#+\s*", "", line).strip() if 2 <= len(line) <= 50 and not line.endswith(("。", ".", ";", ";")): path.append(line) break return path[:4] if path else [] def _infer_chunk_type(text: str) -> str: compact = normalize_text(text) if "|" in text and "---" in text: return "table" if re.search(r"(步骤|流程|操作|检查|确认|打开|关闭|启动|停止)", compact): return "procedure" if re.search(r"(故障|异常|报警|失效|原因|处理|排查)", compact): return "fault" if re.search(r"!\[.*?\]\(.*?\)", text): return "image" return "text"