"""预存「商业态势总览」演示数据（task-data 数据交付 · 第一块对接）。 对齐前端页面（code-1 仓库）``#/situation-overview/action/business-overview?positionId=planning`` 的 mock 逻辑（``features/business-overview/mock/businessTimelineSnapshots.js``）： - 默认任务 ``BUSINESS-OVERVIEW-DEMO``（对方系统任务号，前端对接用它）； - 13 期时间轴刻度 = 页面小时刻度：2026-06-17 22:00 → 2026-06-18 10:00（每小时一期， Beijing；最后一期为「当前期」）； - 每期 ``block_1`` payload = 页面 ``createBusinessTimelineSnapshot`` 同构 JSON （整页 dashboard 对象，前端拿到即可直接渲染），按同一 slot 公式选相位； - ``block_1`` 标题置为「商业态势总览」，其余块保持默认。 幂等：重复运行会先级联删除演示任务再重建（不影响其它任务）。 用法（在 backend 目录） ====================== :: PYTHONPATH=. uv run python scripts/seed_business_overview_demo.py # 预存 PYTHONPATH=. uv run python scripts/seed_business_overview_demo.py --dry-run # 只打印不写库 """ from __future__ import annotations import argparse import asyncio import logging from datetime import datetime, timedelta, timezone logging.basicConfig(level=logging.INFO, format="%(message)s") logger = logging.getLogger("seed_business_overview_demo") SOURCE_TASK_ID = "BUSINESS-OVERVIEW-DEMO" TZ_BEIJING = timezone(timedelta(hours=8)) # 前端 BUSINESS_TIMELINE_ANCHOR = new Date(2026, 5, 18, 10, 30)（本地时区，页面按北京时区使用）。 ANCHOR = datetime(2026, 6, 18, 10, 30, tzinfo=TZ_BEIJING) # 页面小时刻度：13 个刻度、终点 = 锚点整点，逐小时回退（与截图范围一致）。 TICK_COUNT = 13 def timeline_ticks() -> list[datetime]: end = ANCHOR.replace(minute=0, second=0, microsecond=0) return [end - timedelta(hours=TICK_COUNT - 1 - index) for index in range(TICK_COUNT)] # ── 前端 dashboardViews.overview（BusinessOverviewDashboard.vue，逐字对齐） ──── BASE_DASHBOARD: dict = { "leftTitle": "经营概览", "signalTitle": "经营信号", "actionTitle": "本月重点动作", "watchTitle": "经营关注", "pageTitle": "商业态势总览", "pageDescription": "围绕收入增长、回款效率与商机质量，实时掌握经营节奏。", "briefs": [ {"title": "当前判断", "label": "经营主线", "description": "收入增长保持稳定，当前工作重点从获取增量转向提升重点项目的签约确定性。", "tone": "blue"}, {"title": "协同重点", "label": "本周推进", "description": "销售、交付与客户成功团队围绕到期客户建立共跟机制，确保信息与行动同步。", "tone": "green"}, {"title": "需持续关注", "label": "节奏风险", "description": "部分项目的审批周期仍有不确定性，应提前准备替代推进路径。", "tone": "amber"}, ], "objectives": [ {"name": "年度收入目标", "label": "核心目标", "category": "财务增长", "description": "保持收入目标达成率在 82% 以上，确保二季度末进入集中签约窗口。", "level": "core"}, {"name": "重点客户续约", "label": "关键目标", "category": "客户经营", "description": "提前锁定 12 个到期项目的续约与增购机会，提升存量收入质量。", "level": "key"}, {"name": "回款效率提升", "label": "潜力目标", "category": "现金流", "description": "分层推进逾期账款治理，保障季度现金流目标稳定达成。", "level": "potential"}, ], "events": [], "operatingMetrics": [ {"label": "年度确认收入", "value": "¥128.6M", "change": "+18.4%", "tone": "positive"}, {"label": "回款达成率", "value": "94.7%", "change": "+3.1pct", "tone": "positive"}, {"label": "活跃客户数", "value": "286", "change": "+22", "tone": "positive"}, {"label": "重点商机金额", "value": "¥348.2M", "change": "环比 +9.6%", "tone": "neutral"}, ], "headlineMetrics": [ {"label": "年度收入目标", "value": "¥156.0M", "change": "达成 82.4%", "tone": "positive"}, {"label": "综合毛利率", "value": "31.8%", "change": "同比 +1.7pct", "tone": "positive"}, {"label": "销售费用率", "value": "12.6%", "change": "预算内", "tone": "neutral"}, ], "businessNarratives": [], "insights": [], "priorities": [ {"index": "01", "title": "华东渠道拓展", "description": "完成 3 家核心伙伴的联合方案签约，扩大区域覆盖。", "status": "推进中", "tone": "blue"}, {"index": "02", "title": "大客户续约提升", "description": "聚焦 12 个到期项目，提前锁定续约与增购机会。", "status": "重点", "tone": "amber"}, {"index": "03", "title": "回款专项治理", "description": "分层跟进逾期账款，确保季度现金流目标达成。", "status": "进行中", "tone": "blue"}, ], "signals": [ {"title": "商机储备充足", "description": "中高概率商机覆盖未来两个季度收入目标的 2.2 倍。", "tone": "positive"}, {"title": "产品组合持续优化", "description": "高毛利订阅业务占比提升至 46%，盈利结构更加稳健。", "tone": "positive"}, {"title": "客户活跃度提升", "description": "核心客户月活与增购意向同步上行，续费基础良好。", "tone": "blue"}, ], "watchItems": [ {"title": "两项重点商机审批周期延长", "description": "预计影响 ¥16.8M 的当季确认节奏，已进入管理层协同跟进。", "level": "中风险", "tone": "medium"}, {"title": "区域回款进度存在差异", "description": "西南区域 30 天以上应收占比高于整体均值 4.2 个百分点。", "level": "关注", "tone": "low"}, ], "visualization": {"type": "trend", "title": "收入与经营目标趋势"}, "chart": { "title": "收入与经营目标趋势", "actualName": "确认收入", "targetName": "经营目标", "barName": "毛利率", "ariaLabel": "近十二个月确认收入、经营目标与毛利率趋势图", "leftAxisName": "收入（百万元）", "rightAxisName": "毛利率", "months": ["7月", "8月", "9月", "10月", "11月", "12月", "1月", "2月", "3月", "4月", "5月", "6月"], "actual": [7.2, 8.1, 8.8, 9.6, 10.4, 11.8, 8.7, 9.3, 10.5, 11.2, 12.1, 13.4], "target": [7.5, 8.3, 9.1, 9.8, 10.7, 12.2, 9.0, 9.6, 10.9, 11.5, 12.6, 13.0], "bar": [28.1, 28.7, 29.2, 30.1, 29.8, 30.6, 30.2, 30.9, 31.3, 31.1, 31.6, 31.8], "leftAxisMax": 16, "leftAxisInterval": 4, "rightAxisMin": 20, "rightAxisMax": 40, "rightAxisInterval": 5, "rightAxisSuffix": "%", }, } # ── 前端 timelinePhases（businessTimelineSnapshots.js，逐字对齐） ───────────── TIMELINE_PHASES: list[dict] = [ { "name": "实时校准", "summary": "重点项目签约口径完成最新校准，经营节奏保持在目标区间。", "chartDelta": 0, "marginDelta": 0, "briefs": [ {"label": "实时判断", "description": "重点项目签约口径完成最新校准，确认收入与经营目标之间的差距继续收窄。"}, {"label": "联合跟进", "description": "销售、交付与客户成功团队已同步最新客户反馈，关键行动进入同频推进。"}, {"label": "审批观察", "description": "两项重点商机仍等待客户内部审批，需要保留替代推进路径与资源窗口。"}, ], "objectives": [ {"description": "根据最新签约反馈，将年度收入目标达成率稳定在 82% 以上，并锁定季度末确认窗口。"}, {"description": "完成 12 个重点续约项目的责任人与下一步动作校准，优先保障高价值客户。"}, {"description": "按账龄与区域更新回款清单，保障本月现金流目标可追踪、可纠偏。"}, ], "events": [ {"title": "重点商机口径完成实时校准", "content": "华东 3 个渠道项目更新商务状态，预计新增确认收入 ¥8.6M。", "topic": "签约推进"}, {"title": "续约客户反馈同步至共跟清单", "content": "客户成功团队补齐关键使用场景，销售侧已更新下一次沟通动作。", "topic": "客户经营"}, {"title": "区域回款风险完成重新分层", "content": "西南区域两笔应收进入专项协同，责任人与预计到账节点已明确。", "topic": "现金流"}, ], "narratives": [ {"title": "收入节奏", "value": "健康", "description": "最新确认口径显示，重点项目仍按季度目标窗口稳定推进。"}, {"title": "客户结构", "value": "优化", "description": "订阅与增购机会占比继续提升，收入质量保持改善趋势。"}, {"title": "资金回笼", "value": "可控", "description": "高风险应收已进入专项协同，当前未形成新增重大敞口。"}, ], "insights": [ {"label": "增长动能", "highlight": "华东渠道与重点客户续约", "description": "仍是当前最确定的增量来源。"}, {"label": "即时建议", "highlight": "维持重点项目逐单校准，", "description": "避免审批变化影响季度确认节奏。"}, ], }, { "name": "签约提速", "summary": "渠道项目释放积极信号，当前窗口应优先推动商务条款确认。", "chartDelta": 0.24, "marginDelta": 0.18, "briefs": [ {"label": "机会增强", "description": "两项高概率商机完成决策人确认，预计签约节奏较上一时点提前。"}, {"label": "商务协同", "description": "法务与交付资源已提前介入，联合处理付款条件和实施边界。"}, {"label": "资源提醒", "description": "售前资源短时集中，需要优先保障本周具备成交条件的重点项目。"}, ], "objectives": [ {"description": "抓住高概率项目的商务窗口，优先完成收入确认条件与合同边界核验。"}, {"description": "围绕续约增购项目补齐决策链信息，推动 4 个项目进入报价确认。"}, {"description": "将回款条款前置到合同协商环节，降低后续账期拉长风险。"}, ], "events": [ {"title": "华东渠道项目进入商务确认", "content": "核心伙伴已接受联合报价框架，合同条款进入最终核验。", "topic": "渠道拓展"}, {"title": "战略客户增购范围得到确认", "content": "客户新增两个业务场景，预计提升本次续约合同价值。", "topic": "客户增购"}, {"title": "售前资源启动优先级调度", "content": "两名解决方案专家转入高概率项目，保障本周方案澄清效率。", "topic": "资源协同"}, ], "narratives": [ {"title": "签约窗口", "value": "提速", "description": "高概率项目的商务条件趋于清晰，适合集中推动确认。"}, {"title": "增购空间", "value": "扩大", "description": "客户新增使用场景带动服务包与订阅组合机会。"}, {"title": "资源负荷", "value": "偏高", "description": "售前能力短时集中，需控制低概率项目的并行投入。"}, ], "insights": [ {"label": "最佳动作", "highlight": "优先完成付款条件和交付边界确认，", "description": "可进一步缩短合同流转周期。"}, {"label": "转化预测", "highlight": "高概率项目若按计划签约，", "description": "本月确认收入有望上调约 ¥2.1M。"}, ], }, { "name": "续约推进", "summary": "重点客户续约反馈集中回流，客户经营成为当前主要推进线。", "chartDelta": 0.12, "marginDelta": 0.1, "briefs": [ {"label": "客户反馈", "description": "续约客户对服务稳定性评价积极，但希望进一步明确增购后的交付节奏。"}, {"label": "场景共创", "description": "客户成功团队正在补齐业务收益案例，为增购报价提供量化依据。"}, {"label": "决策风险", "description": "部分客户尚未确认最终预算归属，需要加强业务与采购双线沟通。"}, ], "objectives": [ {"description": "保持年度目标节奏，同时将存量客户贡献作为本周收入确定性的主要支撑。"}, {"description": "围绕 12 个重点客户逐一确认预算、决策人和续约范围，提升续约确定性。"}, {"description": "把续约付款节点纳入共跟清单，避免合同确认与实际到账节奏脱节。"}, ], "events": [ {"title": "重点续约客户完成价值复盘", "content": "客户确认现有服务达到预期，并提出新增数据分析场景。", "topic": "客户经营"}, {"title": "增购方案进入联合测算", "content": "销售与交付共同评估资源投入，准备输出分阶段报价方案。", "topic": "方案共创"}, {"title": "客户预算路径完成补充核验", "content": "4 个续约项目已明确预算归属，后续转入采购沟通。", "topic": "预算确认"}, ], "narratives": [ {"title": "客户意愿", "value": "积极", "description": "重点客户续约意愿稳定，增购讨论逐步进入量化阶段。"}, {"title": "价值证明", "value": "补强", "description": "业务收益案例正在完善，有利于缩短客户内部决策时间。"}, {"title": "预算路径", "value": "待确认", "description": "少量项目仍需同步业务负责人和采购侧预算口径。"}, ], "insights": [ {"label": "客户策略", "highlight": "用业务收益案例支撑增购报价，", "description": "比单纯价格优惠更有利于稳定续约价值。"}, {"label": "协同建议", "highlight": "销售与客户成功共同参加下一轮沟通，", "description": "同步解决价值与交付问题。"}, ], }, { "name": "回款预警", "summary": "区域到账节奏出现短时偏差，当前应优先压实高账龄应收责任。", "chartDelta": -0.32, "marginDelta": -0.26, "briefs": [ {"label": "现金流提醒", "description": "西南区域两笔重点应收预计到账时间后移，月度回款预测需要下修。"}, {"label": "专项处置", "description": "财务与客户负责人已建立逐笔跟进机制，优先处理高账龄款项。"}, {"label": "目标影响", "description": "若本周未形成明确付款承诺，可能影响月末现金流安全边际。"}, ], "objectives": [ {"description": "收入目标保持不变，但新增合同需同步核验付款条件和客户信用情况。"}, {"description": "续约项目继续推进，并将历史回款表现纳入商务条款决策。"}, {"description": "集中处置 30 天以上应收，明确管理层协同事项与最晚到账节点。"}, ], "events": [ {"title": "区域到账预测出现偏差", "content": "两笔应收未按原计划进入付款流程，财务已发起风险提示。", "topic": "回款预警"}, {"title": "高账龄款项启动管理层协同", "content": "客户负责人补充付款障碍与决策链信息，形成逐笔处置方案。", "topic": "专项治理"}, {"title": "新增合同付款条件复核", "content": "商务团队统一检查首付款比例与账期，防止新增现金流风险。", "topic": "合同风控"}, ], "narratives": [ {"title": "到账节奏", "value": "承压", "description": "区域回款偏差扩大，需要以逐笔承诺替代笼统预测。"}, {"title": "风险敞口", "value": "中等", "description": "当前问题集中在少数高价值款项，仍具备专项化解条件。"}, {"title": "处置进度", "value": "已启动", "description": "财务、销售与管理层已进入联合跟进状态。"}, ], "insights": [ {"label": "风险动作", "highlight": "优先取得客户书面付款承诺，", "description": "并同步准备分期或节点化替代方案。"}, {"label": "经营影响", "highlight": "收入确认暂不受影响，", "description": "但现金流预测需保留约 ¥3.4M 风险缓冲。"}, ], }, { "name": "资源协同", "summary": "跨团队依赖集中显现，经营重点转向资源排期与关键节点保障。", "chartDelta": 0.08, "marginDelta": -0.08, "briefs": [ {"label": "执行判断", "description": "商机质量保持稳定，交付与售前资源能否按期到位成为主要约束。"}, {"label": "排期协同", "description": "重点项目已按收入贡献与成交概率重新排序，资源将向前两级项目倾斜。"}, {"label": "交付提醒", "description": "两个项目存在并行上线需求，需要提前确认实施负责人和验收条件。"}, ], "objectives": [ {"description": "通过资源优先级调度保障高价值项目按期签约与确认，稳定年度目标节奏。"}, {"description": "对重点续约项目同步售前、交付和客户成功排期，减少内部等待时间。"}, {"description": "将资源投入与回款节点联动管理，优先保障付款条件清晰的项目。"}, ], "events": [ {"title": "重点项目资源优先级完成调整", "content": "售前与交付排期按项目价值重新排序，优先保障高概率机会。", "topic": "资源调度"}, {"title": "并行上线项目启动依赖核验", "content": "实施负责人、环境准备与验收标准已进入联合确认。", "topic": "交付协同"}, {"title": "跨部门问题转入日清机制", "content": "未闭环依赖将在每日经营例会上更新责任人与处理时限。", "topic": "机制优化"}, ], "narratives": [ {"title": "资源供给", "value": "趋紧", "description": "重点项目进入集中阶段，售前与交付资源需要动态调度。"}, {"title": "项目排序", "value": "已校准", "description": "当前优先级已综合收入贡献、成交概率、付款质量。"}, {"title": "协同效率", "value": "改善", "description": "跨部门问题进入日清机制，等待时间开始缩短。"}, ], "insights": [ {"label": "资源原则", "highlight": "优先保障高概率、高价值、付款清晰项目，", "description": "暂停低确定性定制投入。"}, {"label": "节奏建议", "highlight": "每日校准一次关键依赖，", "description": "确保资源变化及时反映到经营预测。"}, ], }, ] # ── 前端 createBusinessTimelineSnapshot 的 Python 复刻 ─────────────────────── def _positive_modulo(value: int, divisor: int) -> int: return value % divisor def _two_digits(value: int) -> str: return f"{value:02d}" def _format_snapshot_time(moment: datetime) -> str: return f"{_two_digits(moment.month)}-{_two_digits(moment.day)} {_two_digits(moment.hour)}:{_two_digits(moment.minute)}" def _merge_rows(source: list[dict], patches: list[dict]) -> list[dict]: return [{**item, **(patches[index] if index < len(patches) else {})} for index, item in enumerate(source)] def _adjust_series(values: list[float], delta: float) -> list[float]: length = max(len(values), 1) return [round(float(value) + delta * ((index + 1) / length), 1) for index, value in enumerate(values)] def build_snapshot(moment: datetime) -> dict: """复刻 ``createBusinessTimelineSnapshot(baseDashboard, timestamp)``。""" minute_slot = round((moment - ANCHOR).total_seconds() / 60) slot = minute_slot + minute_slot // 5 + minute_slot // 60 phase = TIMELINE_PHASES[_positive_modulo(slot, len(TIMELINE_PHASES))] time_label = _format_snapshot_time(moment) events = [ {**event, "date": (moment - timedelta(minutes=17 * index)).strftime("%H:%M")} for index, event in enumerate(phase["events"]) ] snapshot = dict(BASE_DASHBOARD) snapshot.update( { "pageDescription": f"{phase['summary']} 当前展示 {time_label} 经营快照。", "snapshotName": phase["name"], "snapshotTime": moment.isoformat(), "briefs": _merge_rows(BASE_DASHBOARD["briefs"], phase["briefs"]), "objectives": _merge_rows(BASE_DASHBOARD["objectives"], phase["objectives"]), "events": events, "businessNarratives": phase["narratives"], "insights": phase["insights"], "chart": { **BASE_DASHBOARD["chart"], "ariaLabel": f"{time_label} {BASE_DASHBOARD['chart']['ariaLabel']}", "actual": _adjust_series(BASE_DASHBOARD["chart"]["actual"], phase["chartDelta"]), "bar": _adjust_series(BASE_DASHBOARD["chart"]["bar"], phase["marginDelta"]), }, } ) return snapshot # ── 预存主流程 ─────────────────────────────────────────────────────────────── async def _run(dry_run: bool) -> int: from deerflow.config import get_app_config from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config from deerflow.persistence.task_analysis import make_task_analysis_store from deerflow.runtime.task_analysis.data_delivery import submit_block_data config = get_app_config() await init_engine_from_config(config.database) sf = get_session_factory() store = make_task_analysis_store(sf) ticks = timeline_ticks() snapshots = {tick: build_snapshot(tick) for tick in ticks} if dry_run: for tick in ticks: snap = snapshots[tick] logger.info("期 %s → %s · %s · %s", ticks.index(tick) + 1, tick.isoformat(), snap["snapshotName"], snap["pageDescription"][:40]) await close_engine() return 0 # 幂等：演示任务已存在则级联删除后重建。 existing = await store.get_task_by_source(SOURCE_TASK_ID) if existing is not None: logger.info("演示任务已存在（id=%s），删除后重建", existing["id"]) await store.delete_task_cascade(existing["id"]) task = await store.create_task( { "source_task_id": SOURCE_TASK_ID, "name": "商业态势总览（演示）", "description": "定时数据交付演示任务：13 期小时刻度（2026-06-17 22:00 → 2026-06-18 10:00），block_1 = 商业态势总览页面数据。", } ) await store.update_block_defs(task["id"], [{"block_key": "block_1", "title": "商业态势总览"}]) for index, tick in enumerate(ticks, start=1): period = await store.open_period(task["id"], source="cron", opened_at=tick) result = await submit_block_data(store, task_id=task["id"], block_key="block_1", data=snapshots[tick]) logger.info( "期 %s/%s opened_at=%s 相位=%s keys=%s", period["period_index"], TICK_COUNT, tick.isoformat(), snapshots[tick]["snapshotName"], result["keys"], ) logger.info("") logger.info("预存完成。前端对接：") logger.info(" 任务号 source_task_id = %s", SOURCE_TASK_ID) logger.info(" 块定义 GET /api/task-analysis/external/tasks/%s/blocks", SOURCE_TASK_ID) logger.info(" 时间轴 GET /api/task-analysis/external/tasks/%s/periods", SOURCE_TASK_ID) logger.info(" 单期数据 GET /api/task-analysis/external/tasks/%s/periods/{n|latest}", SOURCE_TASK_ID) await close_engine() return 0 def main() -> int: parser = argparse.ArgumentParser(description="预存商业态势总览演示数据（13 期小时刻度 + block_1 快照）") parser.add_argument("--dry-run", action="store_true", help="只打印将生成的期与相位，不写库") args = parser.parse_args() try: return asyncio.run(_run(args.dry_run)) except Exception: logger.exception("预存失败") return 1 if __name__ == "__main__": raise SystemExit(main())