"""标书合成能力（agent.bid_compositor）：把已通过的章节合成完整标书（md + docx）。 对应业务：最后合成标书。 产出落 bid_documents（version 递增）+ 工作空间文件，并登记为交付件。 docx 生成用 python-docx；不可用时降级为 md（诚实降级，不假装成功）。 """ import json import logging import os from .bid_common import ( get_db, new_id, rec_to_dict, rows_to_dicts, to_float, to_int, record, get_project, get_tender_by_project, resolve_project_id, CH_APPROVED, DOC_DRAFT, ) logger = logging.getLogger("pipeline.bidding.compose") async def _workspace_dir(sor, project_id): """项目工作目录：统一走引擎 get_project_dir_by_id（读库，创建时已写死）。 不走 get_workspace_dir（那个依赖会话当前项目），角色 agent 场景下按项目自身字段解析。 旧实现自行拼 {base}/{org}/{pipeline_id}/{项目名}（旧平铺口径）导致与项目目录错位， 2026-09-01 收敛到引擎统一函数根治。 铁律（输出文件必须在项目空间内）：解析失败绝不回退 /tmp—— 标书是交付件，写到项目外就违反"输出文件必须在项目空间内"。宁可合成失败。 """ from pipeline_service.workspace import get_project_dir_by_id pdir, _ = await get_project_dir_by_id(sor, project_id) if pdir: os.makedirs(pdir, exist_ok=True) return pdir return '' def _fetch_image_bytes(url, timeout=30): """下载配图（章节正文里的 markdown 图片 URL → 本地字节流）。 URL 是 invoke_model 产物经 downloadfile2url 落地的本地持久地址 （/idfile 静态路径，上游 24h 时效 URL 已在推理层落地，这里拿到的是 平台自身地址）。http(s) 下载；本地绝对路径直读。失败返回 None（诚实 降级：docx 里保留文字说明，不假装成功）。 """ try: u = (url or '').strip() if not u: return None if u.startswith('http://') or u.startswith('https://'): import urllib.request req = urllib.request.Request(u, headers={'User-Agent': 'pipeline-bidding'}) with urllib.request.urlopen(req, timeout=timeout) as resp: data = resp.read() return data if data else None if u.startswith('/') and os.path.isfile(u): with open(u, 'rb') as f: return f.read() except Exception as e: logger.warning("compose: 配图下载失败 url=%s: %s", (url or '')[:120], e) return None # markdown 图片行：![alt](url)（章节写作者按角色 prompt 规范嵌入 invoke_model 产物） _MD_IMG_RE = None def _parse_md_image(line): """解析一行 markdown 图片语法，返回 (alt, url) 或 None。""" global _MD_IMG_RE if _MD_IMG_RE is None: import re _MD_IMG_RE = re.compile(r'^\s*!\[([^\]]*)\]\(([^)\s]+)\)\s*$') m = _MD_IMG_RE.match(line or '') if not m: return None return m.group(1), m.group(2) def _write_docx(path, title, subtitle, chapters): """用 python-docx 写标书。返回 (ok, page_hint_or_err)。 章节正文支持 markdown 图片行 ![alt](url)（写作者经 invoke_model 生成的 真实配图）：下载后 add_picture 嵌入；下载失败保留文字占位（诚实降级）。 """ try: from docx import Document from docx.shared import Pt, Inches from docx.enum.text import WD_ALIGN_PARAGRAPH except ImportError: return False, "python-docx 未安装" try: doc = Document() # 中文字体（python-docx 对 CJK 需显式设置 eastasia） style = doc.styles["Normal"] style.font.name = "宋体" style.font.size = Pt(12) try: from docx.oxml.ns import qn style.element.rPr.rFonts.set(qn("w:eastAsia"), "宋体") except Exception: pass p = doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.CENTER r = p.add_run(title) r.bold = True r.font.size = Pt(26) if subtitle: p2 = doc.add_paragraph() p2.alignment = WD_ALIGN_PARAGRAPH.CENTER r2 = p2.add_run(subtitle) r2.font.size = Pt(14) doc.add_page_break() doc.add_heading("目　录", level=1) for c in chapters: doc.add_paragraph("%s　%s" % (c.get("chapter_no", ""), c.get("title", ""))) doc.add_page_break() img_ok = img_fail = 0 for c in chapters: doc.add_heading("%s　%s" % (c.get("chapter_no", ""), c.get("title", "")), level=1) for line in (c.get("content") or "").split("\n"): t = line.rstrip() if not t: continue img = _parse_md_image(t) if img: alt, url = img data = _fetch_image_bytes(url) if data: import io pic_p = doc.add_paragraph() pic_p.alignment = WD_ALIGN_PARAGRAPH.CENTER try: pic_p.add_run().add_picture(io.BytesIO(data), width=Inches(5.5)) if alt: cap = doc.add_paragraph(alt) cap.alignment = WD_ALIGN_PARAGRAPH.CENTER img_ok += 1 except Exception as e: doc.add_paragraph("［配图 %s 嵌入失败：%s］" % (alt or url[:60], str(e)[:80])) img_fail += 1 else: doc.add_paragraph("［配图缺失（下载失败）：%s %s］" % (alt or "", url[:80])) img_fail += 1 continue if t.startswith("### "): doc.add_heading(t[4:], level=3) elif t.startswith("## "): doc.add_heading(t[3:], level=2) elif t.startswith("# "): doc.add_heading(t[2:], level=2) elif t.lstrip().startswith(("- ", "* ")): doc.add_paragraph(t.lstrip()[2:], style="List Bullet") else: doc.add_paragraph(t) doc.add_page_break() doc.save(path) if img_ok or img_fail: logger.info("compose: docx 配图嵌入 成功%d 失败%d", img_ok, img_fail) return True, "" except Exception as e: return False, "%s: %s" % (type(e).__name__, str(e)[:200]) async def compose_bid(project_id, doc_name="", who=None, agent_id=None, task_id=""): """合成标书：要求全部章节 approved（否则拒绝，避免拿半成品合成）。""" try: project_id = await resolve_project_id(project_id) except ValueError as e: return False, str(e) db, dbname = get_db() async with db.sqlorContext(dbname) as sor: proj = await get_project(sor, project_id) if not proj: return False, "项目不存在: %s" % project_id recs = await sor.sqlExe( "SELECT id, chapter_no, title, status, content, max_score, review_score " "FROM bid_chapters WHERE project_id=${pid}$ ORDER BY order_no, chapter_no", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) chapters = rows_to_dicts(recs, limit=500) if not chapters: return False, "项目下无章节，无法合成（请先生成章节骨架并编写）" bad = [c for c in chapters if c.get("status") != CH_APPROVED] if bad: names = ", ".join("%s %s(%s)" % (c.get("chapter_no"), c.get("title"), c.get("status")) for c in bad[:8]) return False, ("仍有 %d 个章节未评审通过，不可合成：%s%s" % (len(bad), names, " ..." if len(bad) > 8 else "")) tender = await get_tender_by_project(sor, project_id) vrec = await sor.sqlExe( "SELECT COALESCE(MAX(version),0) AS v FROM bid_documents WHERE project_id=${pid}$", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) ver = to_int(getattr(vrec[0], "v", 0) if vrec else 0) + 1 ws = await _workspace_dir(sor, project_id) if not ws: return False, "无法解析项目工作目录（输出文件必须落项目空间内，已中止合成）" title = doc_name or ("%s 投标文件" % (tender.get("title") or proj.get("name") or "项目")) subtitle = "投标人：（本单位）　　招标编号：%s" % (tender.get("tender_no") or "-") md_lines = ["# " + title, "", subtitle, "", "## 目录", ""] for c in chapters: md_lines.append("- %s %s" % (c.get("chapter_no"), c.get("title"))) md_lines.append("") for c in chapters: md_lines.append("# %s %s" % (c.get("chapter_no"), c.get("title"))) md_lines.append("") md_lines.append(c.get("content") or "") md_lines.append("") md = "\n".join(md_lines) md_path = os.path.join(ws, "bid_v%d.md" % ver) with open(md_path, "w", encoding="utf-8") as f: f.write(md) docx_path = os.path.join(ws, "bid_v%d.docx" % ver) ok_docx, err = _write_docx(docx_path, title, subtitle, chapters) file_path = docx_path if ok_docx else md_path total_chars = sum(len(c.get("content") or "") for c in chapters) page_est = max(1, total_chars // 900) # 约 900 字/页（中文小四单倍行距） max_sum = sum(to_float(c.get("max_score"), 0.0) for c in chapters) did = new_id() await sor.C("bid_documents", { "id": did, "project_id": project_id, "version": ver, "doc_name": title[:250], "file_path": file_path[:480], "page_count": page_est, "chapter_count": len(chapters), "max_score": max_sum or None, "status": DOC_DRAFT, "task_id": task_id or "", "created_by": who or "agent.bid_compositor", }) await sor.sqlExe("COMMIT", {}) await record(sor, project_id, "bid_documents", did, "compose", to_state=DOC_DRAFT, who=who, agent_id=agent_id, detail="v%d %d章 %d字" % (ver, len(chapters), total_chars)) note = "" if ok_docx else ("（docx 生成失败，已产出 markdown：%s）" % err) return True, ("标书已合成 v%d：%s\n章节 %d 个，约 %d 字，估算 %d 页，对应满分 %.1f%s\n" "系统将自动派发整书评分任务。" % (ver, file_path, len(chapters), total_chars, page_est, max_sum, note)) async def list_bid_documents(project_id, limit=20): try: project_id = await resolve_project_id(project_id) except ValueError as e: return [] db, dbname = get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT id, version, doc_name, file_path, page_count, chapter_count, " "total_score, max_score, status, created_at FROM bid_documents " "WHERE project_id=${pid}$ ORDER BY version DESC LIMIT " + str(to_int(limit, 20)), {"pid": project_id}) await sor.sqlExe("COMMIT", {}) return rows_to_dicts(recs) async def read_bid_document(document_id, offset=0, length=15000): """读合成标书正文（分段），供整书评审用。""" db, dbname = get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT id, project_id, version, file_path FROM bid_documents WHERE id=${did}$", {"did": document_id}) await sor.sqlExe("COMMIT", {}) if not recs: return "标书不存在: %s" % document_id d = rec_to_dict(recs[0]) path = d.get("file_path") or "" md = os.path.splitext(path)[0] + ".md" src = md if os.path.isfile(md) else path if not src or not os.path.isfile(src): return "标书文件不存在于磁盘: %s" % path if src.endswith(".docx"): from .bid_analysis_capability import _extract_file_text txt = _extract_file_text(src) else: with open(src, "r", encoding="utf-8", errors="ignore") as f: txt = f.read() o = to_int(offset, 0) n = to_int(length, 15000) return ("【标书 v%s】字符 %d-%d / 共 %d\n%s" % (d.get("version"), o, o + n, len(txt), txt[o:o + n]))