#!/usr/bin/env python3 # -*- coding: utf-8 -*- """am 密集段落转换器 —— 把生产报告里的长散文/密集清单转成 am 原生组件 2026-10-19 用户要求：所有 HTML 一律走 am；复盘里"太密、没序列、分不清主次"的段落 要用 am 组件（callout / limits / kv）重排，而不是手写 CSS。 本模块被 am_bridge.py 调用，处理三类模式： 1. `### 行业分布` 里的 "A14只(股1/股2)、B6只(...)" → limits 横条 2. `核心矛盾: xxx vs yyy ... 决定变量 = zzz` → callout + kv 3. `双轨口径: A轨=... B轨=... 共振点=... 分歧点=...` → 三个 callout 4. 归因判断 3 层 bullet → 三个 callout """ import re # ── 通用工具 ────────────────────────────────────────── def _split_by_sep(txt, seps=r"[;；]"): parts = re.split(r"(?<=%s)\s*" % seps, txt) return [p.strip() for p in parts if p and p.strip()] def _fmt(txt): """去掉 md 强调符号, am 组件内用纯文本 + 可选粗体。""" return re.sub(r"\*\*(.+?)\*\*", r"\1", txt).strip() # ── 1. 行业分布 → limits ────────────────────────────── def industry_to_limits(txt): """'涨停行业分布(...): 汽车零部件14只(股1/股2...)、通用设备6只(...), 其余16个行业各1只 — 结构特征=...' → ```limits 块 + 结构特征 callout """ txt = _fmt(txt) head, rest = "", txt m = re.match(r"^([^：:]{4,80})[:：]\s*(.*)$", txt, re.S) if m: head, rest = m.group(1), m.group(2) tail = "" if "—" in rest: rest, tail = rest.split("—", 1) rows, rest_row = [], None for seg in re.split(r"[、,，]\s*(?=[^\s/、,，()（）]{2,10}\d+只)|(?=其余\d+个行业)", rest): seg = seg.strip() if not seg: continue mr = re.match(r"^其余\s*(\d+)\s*个行业\s*各\s*(\d+)\s*只$", seg) if mr: rest_row = (int(mr.group(1)), int(mr.group(2))) continue mm = re.match(r"^([^\d(（]{2,12})\s*(\d+)\s*只\s*(?:[（(](.*)[）)])?$", seg, re.S) if mm: rows.append([mm.group(1).strip(), int(mm.group(2)), (mm.group(3) or "").strip()]) elif rows: rows[-1][2] = (rows[-1][2] + "、" + seg).strip("、") if not rows: return None mx = max(r[1] for r in rows) out = [] if head: out.append(f"**{head}**\n") out.append("```limits") for ind, cnt, names in rows: note = "" if names: parts = [x for x in re.split(r"[/、]", names) if x.strip()] note = "、".join(parts[:4]) if len(parts) > 4: note += f" 等 {len(parts)} 只" out.append(f"{ind} | {cnt} / {mx} | 只 | {note}") if rest_row: out.append(f"其余 {rest_row[0]} 个行业 | {rest_row[1]} / {mx} | 只 | —") out.append("```") if tail.strip(): out.append("") out.append(f"```callout info 结构特征\n{tail.strip()}\n```") return "\n".join(out) # ── 2. 核心矛盾 → callout + kv ──────────────────────── def clash_to_callout(txt): """'核心矛盾: A vs B ... 决定变量 = X; 最早验证信号 = Y' → callout(warn) + kv""" txt = _fmt(txt) txt = re.sub(r"^核心矛盾\s*[:：]\s*", "", txt) txt = re.sub(r"^资金核心矛盾\s*[:：]\s*", "", txt) # 拆出 决定变量/验证信号 kv_rows = [] for pat, label in ((r"决定变量\s*=\s*", "决定变量"), (r"最早验证信号\s*=\s*", "最早验证信号"), (r"证伪条件\s*=\s*", "证伪条件")): m = re.search(pat + r"(.+?)(?:;|；|。|$)", txt) if m: kv_rows.append(f"{label}: {m.group(1).strip().rstrip(';；。')}") txt = txt.replace(m.group(0), "").strip() # 拆 vs 两侧 mv = re.search(r"(.+?)\s+vs\s+(.+?)(?:——|--|。|$)", txt, re.S) body = "" if mv: left = mv.group(1).strip(" 「」\"'") right = mv.group(2).strip(" 「」\"'") body = (f"**多方 / 现状**：{left}\n\n**空方 / 反证**：{right}\n\n" f"{txt[mv.end():].strip().lstrip('——').strip()}") else: body = txt out = ["```callout warn 核心矛盾", body.strip(), "```"] if kv_rows: out.append("") out.append("```kv") out.extend(kv_rows) out.append("```") return "\n".join(out) # ── 3. 双轨口径 → 三个 callout ──────────────────────── def dual_to_callouts(txt): """'双轨口径: A轨=... B轨=... 共振点=... 分歧点=... 判定为...' → callout×3""" txt = _fmt(txt) txt = re.sub(r"^双轨口径\s*[:：]\s*", "", txt) segs = {} # A轨/B轨 主体 # 2026-10-19 修正: 原文形如 "A轨=大盘与资金(量化),B轨=情绪与梯队。今日... A轨=七指数...; # B轨=涨停63只..." —— 前半是"口径定义"(括号说明), 后半才是真正的轨内容。 # 旧正则把定义部分当成主体 → B轨只剩"情绪与梯队"。 # 策略: 取最后一次出现的 A轨=/B轨= (真正的判断内容), 若只有一次则用它。 _a = list(re.finditer(r"A轨\s*=\s*", txt)) _b = list(re.finditer(r"B轨\s*=\s*", txt)) if _a: _m = _a[-1] _end = _b[-1].start() if _b and _b[-1].start() > _m.end() else len(txt) _seg = txt[_m.end():_end] # 若这段太短(<15字)说明命中的是定义, 回退用第一次 if len(_seg.strip()) < 15 and len(_a) > 1: _m = _a[0] _end = _b[-1].start() if _b and _b[-1].start() > _m.end() else len(txt) _seg = txt[_m.end():_end] segs["A"] = _seg.strip().strip(";；。,") if _b: _m = _b[-1] _endm = re.search(r"。\s*共振点|;\s*共振点|。\s*判定", txt[_m.end():]) _end = _m.end() + _endm.start() if _endm else len(txt) _seg = txt[_m.end():_end] if len(_seg.strip()) < 15 and len(_b) > 1: _m = _b[0] _endm = re.search(r"。\s*共振点|;\s*共振点", txt[_m.end():]) _end = _m.end() + _endm.start() if _endm else len(txt) _seg = txt[_m.end():_end] segs["B"] = _seg.strip().strip(";；。,") mc = re.search(r"共振点\s*=\s*(.+?)(?=;\s*分歧点|；\s*分歧点|。|$)", txt, re.S) md = re.search(r"分歧点\s*=\s*(.+?)(?=。|$)", txt, re.S) mj = re.search(r"判定为(.+?)(?:。|$)", txt, re.S) out = [] if segs.get("A"): out += ["```callout info A 轨 · 大盘与资金", segs["A"], "```", ""] if segs.get("B"): out += ["```callout info B 轨 · 情绪与梯队", segs["B"], "```", ""] if mc or md: body = [] if mc: body.append(f"**共振点**：{mc.group(1).strip()}") if md: body.append(f"**分歧点**：{md.group(1).strip()}") out += ["```callout warn 共振与分歧", "\n\n".join(body), "```", ""] if mj: out += ["```callout ok 判定", f"判定为{mj.group(1).strip()}", "```"] return "\n".join(out) if out else None # ── 4. 归因判断 3 层 → 三个 callout ─────────────────── def attribution_to_callouts(bullets): """[('今日定性', 'xxx'), ('主线状态', 'yyy'), ('明日验证', 'zzz')] → callout×3""" types = {"今日定性": "info", "主线状态": "warn", "明日验证": "ok"} out = [] for lab, body in bullets: t = types.get(lab.strip(), "info") body = _fmt(body) # 去掉重复标签 body = re.sub(r"^" + re.escape(lab) + r"\s*[:：]\s*", "", body) # 明日验证的 ①②③ 转列表 if "验证" in lab: items = re.split(r"(?=[①②③④⑤])", body) items = [x.strip() for x in items if x.strip()] if len(items) > 1: body = "\n".join("- " + re.sub(r"^[①②③④⑤]\s*", "", x) for x in items) out += [f"```callout {t} {lab}", body, "```", ""] return "\n".join(out).rstrip() # ── 主转换：按小节标题分派 ───────────────────────────── # ── 降密：长文本打包断行（只插空白/
，不改内容）2026-10-19 ────────── def _split_units(text, hard=200): """先按句末标点切；仍超长（>hard）的片段再按逗号切。""" units = [u for u in re.split(r"(?<=[。！？；;])", text) if u] res = [] for u in units: if len(u) <= hard: res.append(u) continue buf = "" for x in [p for p in re.split(r"(?<=[，,])", u) if p]: if buf and len(buf) + len(x) > hard: res.append(buf) buf = x else: buf += x if buf: res.append(buf) return res def _pack(text, target=140, hard=200): """把长文本贪心打包成多段（每段 ~target 字，上限 ~hard）。""" units = _split_units(text, hard) if len(units) <= 1: return [text] paras, cur = [], "" for u in units: if cur and len(cur) + len(u) > target: paras.append(cur.strip()) cur = u else: cur += u if cur.strip(): paras.append(cur.strip()) return paras def density_pass(md_text, limit=180, target=130): """长段降密：表格单元/列表项用

断行；正文与 callout 正文拆多段。 只插入断行标记（空白 /
），文本内容逐字不变。callout/limits/kv 等 围栏内的语法行不动，只处理 callout 的正文行。 """ lines = md_text.split("\n") out, i = [], 0 while i < len(lines): ln = lines[i] s = ln.strip() # callout 围栏：特殊处理正文行 if re.match(r"^\s*```callout\b", ln): out.append(ln) i += 1 while i < len(lines) and not lines[i].strip().startswith("```"): body = lines[i] bs = body.strip() if bs and len(bs) > limit: packed = _pack(bs, target) if len(packed) > 1: out.append("\n\n".join(packed)) i += 1 continue out.append(body) i += 1 if i < len(lines): out.append(lines[i]) # 收尾 ``` i += 1 continue # 其它围栏（代码/limits/kv/flow…）原样跳过 if s.startswith("```"): out.append(ln) i += 1 while i < len(lines) and not lines[i].strip().startswith("```"): out.append(lines[i]) i += 1 if i < len(lines): out.append(lines[i]) i += 1 continue # 表格行：长单元插

if s.startswith("|") and s.count("|") >= 3: cells = ln.split("|") new_cells = [] for c in cells: body = c.strip() if len(body) > limit and not re.match(r"^[\s:|-]+$", body): packed = _pack(body, target) if len(packed) > 1: new_cells.append(" " + "

".join(packed) + " ") continue new_cells.append(c) out.append("|".join(new_cells)) i += 1 continue # 列表项：长项插

（保留 - 前缀） m = re.match(r"^(\s*[-*+]\s+)(.*)$", ln) if m and len(m.group(2)) > limit: packed = _pack(m.group(2), target) if len(packed) > 1: out.append(m.group(1) + "

".join(packed)) i += 1 continue # 普通长段落：拆多段 if len(ln) > limit and not s.startswith(("#", ">", "|")) and not re.match(r"^\d+[.)]\s", s): packed = _pack(s, target) if len(packed) > 1: out.append("\n\n".join(packed)) i += 1 continue out.append(ln) i += 1 return "\n".join(out) def transform_sections(md_text): """按 ##/### 标题智能转换密集段落。返回新 md。 2026-10-19: 原只处理 ###/####, 漏掉模块级(## 下直接跟的长段, 如"双轨口径: ..."), 现按 ## 也切块(标题本身保留, 只转换其下内容)。 """ lines = md_text.split("\n") out = [] i = 0 cur_title = "" while i < len(lines): ln = lines[i] m = re.match(r"^(#{2,4})\s+(.*)$", ln) if m: cur_title = m.group(2).strip() out.append(ln) i += 1 # 收集本节内容直到下个标题(同级或更高级) block = [] while i < len(lines) and not re.match(r"^#{2,4}\s+", lines[i]): block.append(lines[i]) i += 1 out.extend(_convert_block(cur_title, block)) continue out.append(ln) i += 1 return density_pass("\n".join(out)) def _convert_dual_table(title, block): """方向A/B 的双轨表格: | 维度 | 判断 |, 其中 A轨/B轨/双轨结论 行是超长文本 → 抽成长文本 callout, 其余短行保留成表格。2026-10-19 """ # 找出表格行 rows = [l for l in block if l.strip().startswith("|")] if len(rows) < 3: return None # 必须含 A轨/B轨 或 双轨结论 长行 long_rows = [r for r in rows if len(r) > 250 and ("A轨" in r or "B轨" in r or "双轨结论" in r)] if not long_rows: return None # 解析表格: 单元格 parsed = [] for r in rows: cells = [c.strip() for c in r.strip().strip("|").split("|")] parsed.append(cells) if len(parsed) < 2: return None out = [] short_rows = [parsed[0], parsed[1]] # 表头 + 分隔 callout_blocks = [] for cells in parsed[2:]: if len(cells) < 2: continue k, v = cells[0], cells[1] if len(v) > 250 and ("A轨" in k or "B轨" in k or "双轨结论" in k): v = re.sub(r"^(A轨|B轨|双轨结论)\s*[（(][^）)]*[）)]\s*[:：]\s*", "", v) v = re.sub(r"^(A轨|B轨|双轨结论)\s*[:：]\s*", "", v) t = "warn" if "双轨结论" in k else "info" # 长句拆分: 分号/句号断句 segs = [s.strip() for s in re.split(r"(?<=[;；])", _fmt(v)) if s.strip()] body = "\n\n".join(segs) if len(segs) > 1 else _fmt(v) callout_blocks.append(f"```callout {t} {k}\n{body}\n```") else: short_rows.append(cells) # 短表格(如果有内容) if len(short_rows) > 2: tbl = [] for i, cells in enumerate(short_rows): tbl.append("| " + " | ".join(cells) + " |") out.append("\n".join(tbl)) out.append("") out.extend(callout_blocks) return out def _convert_block(title, block): """对单个小节的内容做转换。""" text = "\n".join(block) body_lines = [l for l in block if l.strip()] # 双轨方向表 (A轨/B轨/双轨结论 长单元格) conv_dual = _convert_dual_table(title, block) if conv_dual: return [""] + conv_dual + [""] # 行业分布 if "行业分布" in title: dense = next((l for l in body_lines if len(l) > 200 and "只" in l), None) if dense: conv = industry_to_limits(dense) if conv: others = [l for l in block if l.strip() and l is not dense] return ([""] + others + ["", conv, ""] if others else ["", conv, ""]) # 资金矛盾 / 核心矛盾 段落 for l in body_lines: s = l.strip() if s.startswith(("- ", "* ")) and ("核心矛盾" in s) and len(s) > 150: conv = clash_to_callout(s[2:]) if conv: head = [x for x in body_lines if x is not l] return ["", "\n".join(head), "", conv, ""] if head else ["", conv, ""] # 双轨口径 (可能在块内任意位置) for l in body_lines: if l.strip().startswith("双轨口径") and len(l) > 150: conv = dual_to_callouts(l.strip()) if conv: others = [x for x in block if x.strip() and x is not l] return ([""] + others + ["", conv, ""]) if others else ["", conv, ""] # 归因判断 3 层 if "归因" in title: bullets = [] for l in body_lines: mm = re.match(r"^[-*]\s*\*\*(.+?)\*\*\s*[:：]?\s*(.*)$", l.strip()) if mm: bullets.append((mm.group(1), mm.group(2))) if len(bullets) >= 2: conv = attribution_to_callouts(bullets) if conv: return ["", conv, ""] return block