# ============================================================ # 知识文档导入与切块(moduleId: knowledge-ingest, 可重生 ✅) # plan.md §8.2:PDF/docx/md/txt → chunk → 知识资产(版本化+审批) # ============================================================ from __future__ import annotations import os import re from typing import Any # 切块目标长度(字符)与重叠 _CHUNK_SIZE = 500 _CHUNK_OVERLAP = 50 def _ext(path: str) -> str: return os.path.splitext(path)[1].lower() def extract_text_from_bytes(filename: str, raw: bytes) -> list[dict[str, Any]]: """按扩展名解析文件为带页/段元数据的文本块列表(尚未切 chunk)。 Returns: [{text, page?, heading?}, ...] """ name = filename or "upload.txt" ext = _ext(name) if ext in (".md", ".txt", ".markdown"): text = raw.decode("utf-8", errors="replace") return _split_markdown_sections(text) if ext == ".pdf": return _extract_pdf(raw) if ext in (".docx",): return _extract_docx(raw) if ext in (".xlsx", ".xlsm"): return _extract_xlsx(raw) # 兜底当纯文本 try: return [{"text": raw.decode("utf-8"), "page": 1}] except UnicodeDecodeError as exc: raise ValueError(f"不支持的文件类型或编码:{name}") from exc def extract_text_from_path(path: str) -> list[dict[str, Any]]: """从本地路径读取并解析。""" if not os.path.isfile(path): raise FileNotFoundError(f"文件不存在:{path}") with open(path, "rb") as f: raw = f.read() return extract_text_from_bytes(os.path.basename(path), raw) def _split_markdown_sections(text: str) -> list[dict[str, Any]]: parts: list[dict[str, Any]] = [] current_heading = "" buf: list[str] = [] for line in text.splitlines(): if re.match(r"^#{1,3}\s+", line): if buf: parts.append({"text": "\n".join(buf).strip(), "heading": current_heading or None}) buf = [] current_heading = re.sub(r"^#{1,3}\s+", "", line).strip() buf.append(line) else: buf.append(line) if buf: parts.append({"text": "\n".join(buf).strip(), "heading": current_heading or None}) return [p for p in parts if p["text"]] def _extract_pdf(raw: bytes) -> list[dict[str, Any]]: try: import pdfplumber except ImportError as exc: raise ValueError("解析 PDF 需要安装 pdfplumber:pip install pdfplumber") from exc from io import BytesIO pages: list[dict[str, Any]] = [] with pdfplumber.open(BytesIO(raw)) as pdf: for i, page in enumerate(pdf.pages, start=1): t = (page.extract_text() or "").strip() if t: pages.append({"text": t, "page": i}) if not pages: raise ValueError("PDF 未提取到文本(可能是扫描件)") return pages def _extract_docx(raw: bytes) -> list[dict[str, Any]]: try: from docx import Document except ImportError as exc: raise ValueError("解析 docx 需要安装 python-docx:pip install python-docx") from exc from io import BytesIO doc = Document(BytesIO(raw)) paras = [p.text.strip() for p in doc.paragraphs if p.text and p.text.strip()] if not paras: raise ValueError("docx 未提取到文本") return [{"text": "\n".join(paras), "page": 1}] def _extract_xlsx(raw: bytes) -> list[dict[str, Any]]: from io import BytesIO from openpyxl import load_workbook wb = load_workbook(BytesIO(raw), data_only=True, read_only=True) parts: list[dict[str, Any]] = [] for sheet in wb.worksheets: rows = [] for row in sheet.iter_rows(values_only=True): cells = [str(c) for c in row if c is not None and str(c).strip()] if cells: rows.append(" | ".join(cells)) if rows: parts.append({"text": f"# {sheet.title}\n" + "\n".join(rows), "heading": sheet.title}) if not parts: raise ValueError("xlsx 无有效单元格") return parts def chunk_sections(sections: list[dict[str, Any]], *, size: int = _CHUNK_SIZE, overlap: int = _CHUNK_OVERLAP) -> list[dict[str, Any]]: """把段落/页切成固定长度 chunk(带重叠)。""" chunks: list[dict[str, Any]] = [] idx = 0 for sec in sections: text = (sec.get("text") or "").strip() if not text: continue heading = sec.get("heading") page = sec.get("page") if len(text) <= size: chunks.append({ "chunkId": f"c{idx:04d}", "text": text, "heading": heading, "page": page, "seq": idx, }) idx += 1 continue start = 0 while start < len(text): end = min(start + size, len(text)) piece = text[start:end].strip() if piece: chunks.append({ "chunkId": f"c{idx:04d}", "text": piece, "heading": heading, "page": page, "seq": idx, }) idx += 1 if end >= len(text): break start = max(end - overlap, start + 1) return chunks def preview_ingest(filename: str, raw: bytes | None = None, path: str | None = None, *, kind: str = "sop", title: str | None = None) -> dict[str, Any]: """预览导入:解析+切块,不写库。""" if path: sections = extract_text_from_path(path) fname = os.path.basename(path) elif raw is not None: sections = extract_text_from_bytes(filename, raw) fname = filename else: raise ValueError("需要提供 path 或 raw") chunks = chunk_sections(sections) title = title or os.path.splitext(fname)[0] return { "filename": fname, "title": title, "kind": kind, "sectionCount": len(sections), "chunkCount": len(chunks), "preview": [{"seq": c["seq"], "heading": c.get("heading"), "page": c.get("page"), "chars": len(c["text"]), "snippet": c["text"][:80]} for c in chunks[:8]], "chunks": chunks, } def confirmation_for_knowledge_import(preview: dict[str, Any]) -> tuple[str, list[str]]: """P2 确认卡文案。""" title = f"导入知识文档《{preview.get('title')}》" lines = [ f"文件:{preview.get('filename')}", f"类型:{preview.get('kind')}", f"切块:{preview.get('chunkCount')} 段(来自 {preview.get('sectionCount')} 节/页)", "入库后可检索;默认已审批(approved=true)。", ] return title, lines def apply_knowledge_import(preview: dict[str, Any], *, tags: list[str] | None = None, store: Any = None) -> dict[str, Any]: """把预览结果写入知识库(带 chunks)。""" from server.knowledge.assets import get_knowledge kb = store or get_knowledge() chunks = preview.get("chunks") or [] content = "\n\n".join( (f"【{c.get('heading') or '段落'}】\n" if c.get("heading") else "") + c["text"] for c in chunks ) if not content.strip(): raise ValueError("无正文可入库") asset = kb.add_with_chunks( kind=str(preview.get("kind") or "sop"), title=str(preview.get("title") or preview.get("filename") or "导入文档"), content=content[:8000], # 资产摘要正文(检索仍优先 chunk) chunks=chunks, tags=tags or ["导入", str(preview.get("filename") or "")], source=str(preview.get("filename") or ""), approved=True, ) return {"assetId": asset["assetId"], "title": asset["title"], "version": asset["version"], "chunkCount": len(asset.get("chunks") or [])} def bind_world_to_identity(world: dict[str, Any], *, sources: list[str] | None = None) -> dict[str, Any]: """把当前登录用户/租户写到世界元数据(分析/导入后绑定归属)。""" from datetime import datetime from server.auth.context import get_identity from server.timeutil import fmt_dt identity = get_identity() meta = dict(world.get("meta") or {}) meta.update({ "ownerUserId": identity.user_id, "ownerUsername": identity.username, "ownerFullname": identity.fullname, "tenantUuid": identity.tenant_uuid, "boundAt": fmt_dt(datetime.now()), "sources": list(sources or meta.get("sources") or []), }) world["meta"] = meta return meta def _knowledge_cell(value: Any) -> str: if value is None: return "" if isinstance(value, (list, tuple, set)): value = "、".join(str(item) for item in value if item not in (None, "")) return str(value).replace("\r", " ").replace("\n", " ").strip() def _append_rows_section( sections: list[dict[str, Any]], *, heading: str, rows: list[dict[str, Any]], columns: list[tuple[str, str]], ) -> None: if not rows: return lines = [f"# {heading}", f"记录数:{len(rows)}"] for row in rows: fields = [ f"{label}={_knowledge_cell(row.get(key))}" for key, label in columns if _knowledge_cell(row.get(key)) ] if fields: lines.append("- " + ";".join(fields)) sections.append({"heading": heading, "text": "\n".join(lines)}) def _project_source_sections(paths: list[str]) -> list[dict[str, Any]]: """保留源文件全部可读字段;标准化失败的列也能进入知识库检索。""" sections: list[dict[str, Any]] = [] seen: set[str] = set() for raw_path in paths[:32]: path = os.path.abspath(str(raw_path or "")) if not path or path in seen or not os.path.isfile(path): continue seen.add(path) if os.path.getsize(path) > 25 * 1024 * 1024: continue name = os.path.basename(path) try: extracted = extract_text_from_path(path) except Exception: if _ext(path) not in (".csv", ".txt"): continue try: with open(path, "rb") as source: raw = source.read() except OSError: continue text = "" for encoding in ("utf-8-sig", "gb18030"): try: text = raw.decode(encoding) break except UnicodeDecodeError: continue extracted = [{"text": text, "heading": name}] if text.strip() else [] for section in extracted: body = str(section.get("text") or "").strip() if not body: continue heading = str(section.get("heading") or "原始内容") sections.append({ "heading": f"源文件·{name}·{heading}", "page": section.get("page"), "text": f"# 源文件 {name}\n## {heading}\n{body}", }) return sections def ingest_project_analyze_report( deep: dict[str, Any], *, world: dict[str, Any] | None = None, ) -> dict[str, Any]: """项目分析结果 → 当前租户知识库(可检索)+ 绑定登录用户。 - 知识库按 tenant_uuid 隔离(get_knowledge) - 资产标签含用户/项目/来源,便于问答引用 """ from server.auth.context import get_identity from server.knowledge.embedding import index_units from server.knowledge.assets import get_knowledge identity = get_identity() if world is not None: bind_world_to_identity(world, sources=list(deep.get("sources") or [])) proj = str(deep.get("projectName") or "当前项目") summary = deep.get("summary") or {} sections: list[dict[str, Any]] = [] md = (deep.get("markdown") or "").strip() if md: sections.append({"text": md, "heading": f"项目分析·{proj}"}) # 原始表格作为数据知识保存,未知列不会因主数据映射失败而丢失。 sections.extend(_project_source_sections(list(deep.get("sourcePaths") or []))) # 卡片只展示抽样,知识库保存项目世界中的完整标准化数据集。 source = world or {} datasets: list[tuple[str, list[dict[str, Any]], list[tuple[str, str]]]] = [ (f"{proj} 物料主数据", list(source.get("flexMaterials") or deep.get("materials") or []), [ ("code", "物料编码"), ("name", "物料名称"), ("type", "类型"), ("unit", "单位"), ("stock", "库存"), ("sourcingType", "来源"), ]), (f"{proj} BOM", list(source.get("flexBom") or []), [ ("productCode", "父项"), ("materialCode", "子项"), ("quantity", "用量"), ("isKey", "关键件"), ]), (f"{proj} 工艺路线", list(source.get("flexRoutings") or deep.get("routings") or []), [ ("productCode", "产品"), ("seq", "顺序"), ("operationCode", "工序编码"), ("operationName", "工序名称"), ("stdTimePerUnit", "标准工时"), ("sourcingType", "来源"), ]), (f"{proj} 设备台账", list(source.get("flexEquipment") or deep.get("equipment") or []), [ ("code", "设备编码"), ("name", "设备名称"), ("status", "状态"), ("financeCode", "财务编号"), ("internalCode", "厂内编号"), ("assetCode", "固定资产编号"), ("factoryCode", "出厂编号"), ("capabilities", "能力"), ("zone", "区域"), ("availabilityRate", "可用率"), ]), (f"{proj} 项目订单", list(source.get("flexOrders") or deep.get("orders") or []), [ ("orderNo", "订单号"), ("productCode", "产品"), ("productName", "产品名称"), ("quantity", "数量"), ("dueDate", "交期"), ("status", "状态"), ("customerName", "客户"), ]), (f"{proj} 工序字典", list(source.get("flexOperations") or []), [ ("code", "工序编码"), ("name", "工序名称"), ("type", "类型"), ("sourcingType", "来源"), ("changeoverMin", "换型时间"), ]), (f"{proj} 生产区域", list(source.get("flexZones") or []), [ ("code", "区域编码"), ("name", "区域名称"), ]), ] for heading, rows, columns in datasets: _append_rows_section(sections, heading=heading, rows=rows, columns=columns) plan = deep.get("plan") or [] if plan: sections.append({ "heading": "补数计划", "text": f"# {proj} 补数 Plan\n" + "\n".join(f"- {p}" for p in plan[:20]), }) if not sections: return {"skipped": True, "reason": "无分析正文可入库"} chunks = chunk_sections(sections) title = f"项目分析·{proj}" tags = [ "项目分析", "自动入库", "完整数据集", f"user:{identity.username}", f"uid:{identity.user_id}", f"tenant:{identity.tenant_uuid}", proj, ] for src in (deep.get("sources") or [])[:6]: tags.append(str(src)[:48]) preview = { "filename": f"project-analyze-{identity.user_id}.md", "title": title, "kind": "case", "sectionCount": len(sections), "chunkCount": len(chunks), "chunks": chunks, } applied = apply_knowledge_import(preview, tags=tags) try: index_units(get_knowledge().iter_search_units()) applied["indexed"] = True except Exception: applied["indexed"] = False applied["ownerUserId"] = identity.user_id applied["tenantUuid"] = identity.tenant_uuid applied["summary"] = summary return applied