"""预存「力量态势」演示任务详情（force-situation，新类型任务 002）。 一个任务一份自由 JSON 文档（force_situation_details 表，结构不定死字段）： - 时间轴（importantTimeNodes 重要时间节点）+ 当前研判 + 重要目标 + 洞察区块； - 任务衔接（taskConnections）：任务字段与任务列表页对齐 （taskName/taskDirection/taskLevel/taskStatus/taskContent/sourceType/createTime）， 另带 connectionType（进攻型/防御型/正面塑势）作为泳道分组。 用法（backend 目录）：PYTHONPATH=. uv run python scripts/seed_force_situation_demo.py [--dry-run] """ from __future__ import annotations import argparse import asyncio import json import sys from pathlib import Path TASK_ID = "002" _LOCATION_SCRIPTS = Path(__file__).resolve().parents[1] / "skills" / "custom" / "fs-location-info" / "scripts" sys.path.insert(0, str(_LOCATION_SCRIPTS)) from fs_location_info import attach_coordinates # noqa: E402 def _connection_rows() -> list[dict]: """任务衔接演示行（字段与任务列表页对齐）。""" rows = [ ("T01", "进攻型", "核心客户竞对反制续约", "W方向", 1, 20, "针对竞对报价渗透，输出两家战略客户反制口径并推动续签。", 0, "2026-09-03 20:08:35"), ("T02", "进攻型", "华东核心商圈抢点铺货", "W方向", 2, 10, "江苏、浙江核心商圈新增网点 40 个，抢在竞对促销前完成布点。", 0, "2026-09-05 10:12:00"), ("T03", "进攻型", "话题热度集中投放", "E方向", 2, 25, "账号编组与媒资就绪后 48 小时内完成话题集中投放。", 1, "2026-09-06 15:40:11"), ("T04", "进攻型", "重点群体定向触达", "E方向", 2, 20, "面向关键决策人群体输出定向内容并跟踪触达效果。", 1, "2026-09-08 09:20:45"), ("T05", "防御型", "竞对挖角渠道防护", "W方向", 1, 20, "梳理核心经销商合作风险清单，逐户明确稳控责任人与话术。", 0, "2026-09-04 11:30:00"), ("T06", "防御型", "低效网点流失止损", "W方向", 3, 10, "对 30 家低效网点分批整改，明确止损退出标准。", 0, "2026-09-09 14:05:22"), ("T07", "防御型", "负面舆情稳态控制", "N方向", 1, 25, "7×24 监测渠道与社群负面声量，触发阈值即启动应急预案。", 1, "2026-09-02 08:00:00"), ("T08", "正面塑势", "品牌口碑正向传播", "E方向", 2, 25, "围绕客户成功案例组织系列正面传播，扩大口碑基本盘。", 1, "2026-09-07 16:44:30"), ("T09", "正面塑势", "战略客户成功案例放大", "E方向", 2, 20, "提炼 3 个战略客户标杆案例，形成多语种传播素材。", 1, "2026-09-10 10:55:00"), ("T10", "正面塑势", "经销商信心公开塑造", "W方向", 3, 10, "年中大会前发布渠道政策预告，稳定经销商预期。", 0, "2026-09-11 09:15:40"), ] return [ { "id": row[0], "connectionType": row[1], "taskName": row[2], "taskDirection": row[3], "taskLevel": row[4], "taskStatus": row[5], "taskContent": row[6], "sourceType": row[7], "createTime": row[8], } for row in rows ] DETAIL: dict = { # 当前判断（研判卡） "currentJudgments": [ {"id": "J01", "content": "核心客户续约窗口正在收窄，竞对报价已进入两家战略客户决策层，若不在本周完成口径对齐，续约率可能跌破 90%。", "confidence": 86, "conclusion": "风险上升"}, {"id": "J02", "content": "华东铺货节奏总体符合计划，江苏、浙江核心商圈网点拓展仍可按既定节点推进，需盯紧安徽渠道落地质量。", "confidence": 78, "conclusion": "持续关注"}, {"id": "J03", "content": "经销商活跃度回升至近三个月高位，可借势优化低效网点并扩大覆盖，形成巩固渠道的窗口。", "confidence": 74, "conclusion": "存在战机"}, {"id": "J04", "content": "海外发布通道尚未就绪，重点信息触达存在 3–7 天滞后，可能拖慢对目标群体的及时影响。", "confidence": 81, "conclusion": "风险上升"}, {"id": "J05", "content": "账号编组与媒体生产能力已就绪，具备本周集中投放条件，可支撑话题热度与公开发布动作。", "confidence": 88, "conclusion": "存在战机"}, {"id": "J06", "content": "效果评估数据闭环仍不完整，重点群体沟通成效难以量化，暂不影响当前动作，但需持续补齐监测口径。", "confidence": 69, "conclusion": "持续关注"}, ], # 重要目标 "importantTargets": [ {"id": "G01", "name": "G01 核心客户续约目标", "description": "Q3 完成 15 家战略客户合同续签，续约率不低于 90%，锁定年度营收基本盘。"}, {"id": "G02", "name": "G02 华东区域铺货目标", "description": "新增终端网点 120 个，覆盖江苏、浙江、安徽核心商圈，Q3 末铺货完成率达到 70% 以上。"}, {"id": "G03", "name": "G03 经销商网络巩固", "description": "提升经销商活跃度至 75%，优化低效网点 30 家，防范竞对挖角导致核心渠道松动。"}, ], # 时间轴：重要时间节点 "importantTimeNodes": [ {"title": "G01 战略客户续约窗口开启", "content": "15 家战略客户进入合同窗口期，需在两周内完成商务条款对齐，锁定年度营收基本盘。", "topic": "核心续约", "date": "09-18"}, {"title": "华东铺货完成率中期考核", "content": "江苏、浙江核心商圈铺货进度复核，目标完成率 70%，安徽渠道落地质量同步盯紧。", "topic": "区域铺货", "date": "09-22"}, {"title": "全国经销商年中大会", "content": "集中通报渠道政策与返点节奏，稳定核心经销商合作意愿，防范竞对挖角。", "topic": "渠道巩固", "date": "09-28"}, {"title": "海外发布通道开通截止", "content": "重点信息触达通道需在此节点前就绪，避免 3–7 天滞后拖慢对目标群体的影响。", "topic": "发布通道", "date": "10-05"}, ], # 洞察区块 "insightTopics": [ {"title": "核心客户续约窗口收窄", "heat": 96}, {"title": "华东渠道竞对促销加码", "heat": 88}, {"title": "经销商利润承压讨论升温", "heat": 76}, {"title": "品牌口碑正面案例扩散", "heat": 64}, ], "insightRisks": [ {"description": "外部扩散风险", "severity": "高", "urgency": "高"}, {"description": "群体情绪升温", "severity": "高", "urgency": "中"}, {"description": "负面叙事形成", "severity": "中", "urgency": "高"}, {"description": "敌对势力借题发挥", "severity": "中", "urgency": "中"}, ], "insightOpportunities": [ {"description": "权威发声窗口", "value": "高", "timing": "紧迫"}, {"description": "议题澄清窗口", "value": "高", "timing": "中"}, {"description": "正面议题引导", "value": "中", "timing": "中"}, ], # 任务衔接：任务字段与任务列表页（/situation-overview/action/task）对齐； # connectionType 为衔接类型（进攻型/防御型/正面塑势），taskStatus 用任务列表状态码 # （10=未开始 20=进行中 25=已完成分析 30=已结束）。不带坐标。 "taskConnections": _connection_rows(), } def _stamp_locations(detail: dict) -> dict: """除任务衔接外，按 fs-location-info 补 latitude/longitude。""" for item in detail.get("currentJudgments") or []: attach_coordinates(item, query=item.get("content", ""), item_id=item.get("id", "")) for item in detail.get("importantTargets") or []: attach_coordinates(item, query=item.get("name", ""), item_id=item.get("id", "")) for index, item in enumerate(detail.get("importantTimeNodes") or [], start=1): attach_coordinates(item, query=item.get("title", ""), item_id=f"node-{index}-{item.get('title', '')}") for index, item in enumerate(detail.get("insightTopics") or [], start=1): attach_coordinates(item, query=item.get("title", ""), item_id=f"topic-{index}") for index, item in enumerate(detail.get("insightRisks") or [], start=1): attach_coordinates(item, query=item.get("description", ""), item_id=f"risk-{index}") for index, item in enumerate(detail.get("insightOpportunities") or [], start=1): attach_coordinates(item, query=item.get("description", ""), item_id=f"opp-{index}") return detail DETAIL = _stamp_locations(DETAIL) 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.force_situation import make_force_situation_store if dry_run: print(json.dumps(DETAIL, ensure_ascii=False, indent=2)[:2000]) print("…（dry-run，未写库）") return 0 config = get_app_config() await init_engine_from_config(config.database) sf = get_session_factory() store = make_force_situation_store(sf) try: record = await store.upsert_detail(TASK_ID, DETAIL, updated_by="seed") print(f"已预存任务 {TASK_ID} 详情：keys={sorted(record['detail'].keys())}") print(f"updated_at={record.get('updated_at')}") finally: await close_engine() return 0 if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() sys.exit(asyncio.run(_run(args.dry_run)))