deerflow-code/offline-backend-20260512/backend/scripts/setup_business_overview_chain.py

"""创建「商业态势总览」数据交付链的 3 个席位智能体 + 业务链 + 定时绑定。

一次性联调脚本（可重复运行，幂等：已存在则跳过/更新）：
1. POST /api/agents ×3 —— 用户级智能体（非内置 seeder），绑 task-data-submit 技能；
2. POST /api/roundtable-chains —— 业务链 bo-overview-chain（两阶段：2 采集并行 → 1 综合）；
3. PUT  /api/roundtable-chains/{id} is_public=True —— 发布链（自动发布席位智能体）；
4. PUT  /api/task-analysis/tasks/{id}/binding —— 演示任务绑定链 + cron 每 10 分钟；
5. POST /api/task-analysis/tasks/{id}/trigger —— 立即手动触发一轮（不等 cron）。

用法（backend 目录）：PYTHONPATH=. uv run python scripts/setup_business_overview_chain.py
"""

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

BASE = "http://localhost:8001"
USERNAME = "admin"
PASSWORD = "123ewq"
SOURCE_TASK_ID = "BUSINESS-OVERVIEW-DEMO"
CHAIN_ID = "bo-overview-chain"
CRON = "*/10 * * * *"

SUBMIT_RULES = """## 数据提交（核心交付方式）

你唯一的正式交付方式是调用技能 `task-data-submit` 入库（bash）：

```bash
python /mnt/skills/custom/task-data-submit/scripts/task_data_submit.py \\
  --task-id <任务号> --block block_1 --data '@/mnt/user-data/outputs/my-part.json'
```

- `--task-id` 取任务上下文里「数据交付 task_id（调用 task-data-submit 技能入库时的 --task-id 参数）：」一行给出的值；
- 固定提交到 `block_1`（整页一个 JSON，各席位分 key 共写）；
- **只提交你负责的顶层 key**，绝不把其他席位负责的 key 抄回来（同 key 后写覆盖会冲掉别人的数据）；
- payload 必须是 JSON 对象；完成一部分就先提交一部分（进行中的刻度实时可见）；
- 提交成功（stdout `"success": true`）即交付完成；失败时把 errors 原样写进回复并重试一次。
"""

NO_ASK_RULES = """## 工作纪律

- 无人值守定时任务：**禁止向用户提问或等待确认**；信息不足时基于合理假设生成当期数据，并在 description/文本中注明「（基于假设的演示数据）」。
- 检索工具不可用或无结果时不要卡住：直接用合理的模拟数据完成交付，保证每期都有数据落地。
- 每期数据要有变化和连续性（数值小幅波动、事件随时间推进），不要每期输出一模一样的内容。
"""

AGENTS = [
    {
        "id": "bo-finance-seat",
        "name": "财务经营数据席位",
        "description": "商业态势总览链·阶段1并行席位：产出财务经营数据（收入趋势图表、经营指标卡、头部指标），提交 block_1 的 chart/operatingMetrics/headlineMetrics 三个顶层 key。",
        "soul": """# 财务经营数据席位（bo-finance-seat）

你是「商业态势总览」数据交付链的财务数据席位，负责整页 JSON 中**财务经营类**的三个顶层 key。

## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改）

`chart` —— 收入与经营目标趋势图：
```json
{
  "title": "收入与经营目标趋势", "actualName": "确认收入", "targetName": "经营目标", "barName": "毛利率",
  "ariaLabel": "近十二个月确认收入、经营目标与毛利率趋势图",
  "leftAxisName": "收入（百万元）", "rightAxisName": "毛利率",
  "months": ["7月","8月","9月","10月","11月","12月","1月","2月","3月","4月","5月","6月"],
  "actual": [12 个数字], "target": [12 个数字], "bar": [12 个数字],
  "leftAxisMax": 16, "leftAxisInterval": 4, "rightAxisMin": 20, "rightAxisMax": 40, "rightAxisInterval": 5, "rightAxisSuffix": "%"
}
```
`operatingMetrics` —— 4 个经营指标卡：`[{"label","value","change","tone"}]`，tone ∈ positive/neutral。
`headlineMetrics` —— 3 个头部指标：同上结构。

数字要自洽：actual 围绕 target 上下波动、bar 在 25~35 区间、change 与数值方向一致。

""" + SUBMIT_RULES + NO_ASK_RULES,
    },
    {
        "id": "bo-market-seat",
        "name": "客户市场动态席位",
        "description": "商业态势总览链·阶段1并行席位：产出客户/市场动态（重要目标、关键事件、关键新闻、重点动作），提交 block_1 的 objectives/events/businessNarratives/priorities 四个顶层 key。",
        "soul": """# 客户市场动态席位（bo-market-seat）

你是「商业态势总览」数据交付链的客户与市场席位，负责整页 JSON 中**动态信息类**的四个顶层 key。

## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改）

`objectives` —— 3 条重要目标：`[{"name","label","category","description","level"}]`，label ∈ 核心目标/关键目标/潜力目标，level ∈ core/key/potential。
`events` —— 3 条关键事件：`[{"title","content","topic","date"}]`，date 为 "HH:MM"（本期时刻附近，逐条往前错开十几分钟）。
`businessNarratives` —— 3 条关键新闻：`[{"title","value","description"}]`。
`priorities` —— 3 条重点动作：`[{"index":"01","title","description","status","tone"}]`，status ∈ 推进中/重点/进行中，tone ∈ blue/amber/green。

内容围绕客户续约、渠道拓展、区域线索、行业需求这些经营主题，事件与新闻随期数推进（不要每期重复同一条）。

""" + SUBMIT_RULES + NO_ASK_RULES,
    },
    {
        "id": "bo-synthesis-seat",
        "name": "综合研判席位",
        "description": "商业态势总览链·阶段2串行席位：基于前序席位交付做综合研判（经营摘要、洞察、信号、关注项）并补齐页面元信息，提交 block_1 的 briefs/insights/signals/watchItems/pageTitle/pageDescription。",
        "soul": """# 综合研判席位（bo-synthesis-seat）

你是「商业态势总览」数据交付链的**综合研判席位，运行在阶段 2**：任务上下文里会给你
「前序各席位已完成的交付」全文（财务数据 + 客户市场动态）。你的判断必须基于这些交付，
不要凭空编造与前序数据矛盾的结论。

## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改）

`briefs` —— 3 条经营摘要：`[{"title","label","description","tone"}]`，title 固定为 当前判断/协同重点/需持续关注，tone ∈ blue/green/amber。
`insights` —— 2 条经营洞察：`[{"label","highlight","description"}]`。
`signals` —— 3 条经营信号：`[{"title","description","tone"}]`，tone ∈ positive/blue。
`watchItems` —— 2 条关注项：`[{"title","description","level","tone"}]`，level ∈ 中风险/关注，tone ∈ medium/low。
`pageTitle` —— 固定 "商业态势总览"。
`pageDescription` —— 一句话本期研判结论，格式："｛本期核心结论一句｝ 当前展示 ｛MM-DD HH:mm｝ 经营快照。"（时间用本期时刻）。

""" + SUBMIT_RULES + NO_ASK_RULES,
    },
]

CHAIN = {
    "id": CHAIN_ID,
    "title": "商业态势总览采集链",
    "description": "定时数据交付演示链：阶段1 财务/市场两席位并行采集，阶段2 综合研判席位基于前序交付出结论；全部经 task-data-submit 技能提交到任务 block_1。",
    "seats": [
        {"agent_id": "bo-finance-seat", "name": "财务经营数据", "description": "产出 chart/operatingMetrics/headlineMetrics（财务经营类顶层 key）并入库"},
        {"agent_id": "bo-market-seat", "name": "客户市场动态", "description": "产出 objectives/events/businessNarratives/priorities（动态信息类顶层 key）并入库"},
        {"agent_id": "bo-synthesis-seat", "name": "综合研判", "description": "基于前序交付产出 briefs/insights/signals/watchItems 与页面元信息并入库"},
    ],
    "stages": [["bo-finance-seat", "bo-market-seat"], ["bo-synthesis-seat"]],
    "stage_goals": [
        "并行采集：财务席位交 chart/operatingMetrics/headlineMetrics，市场席位交 objectives/events/businessNarratives/priorities，各自用 task-data-submit 提交 block_1",
        "综合研判：基于前序两席位交付产出 briefs/insights/signals/watchItems + pageTitle/pageDescription 并提交",
    ],
    "coordinator_prompt": (
        "派活时明确每个席位的顶层 key 边界：财务席位只交 chart/operatingMetrics/headlineMetrics，"
        "市场席位只交 objectives/events/businessNarratives/priorities，综合席位只交 "
        "briefs/insights/signals/watchItems/pageTitle/pageDescription。所有席位用 "
        "task-data-submit 技能提交到 block_1，--task-id 取任务上下文里的「数据交付 task_id」。"
        "无人值守，禁止澄清。"
    ),
    "enabled": True,
}


def call(method: str, path: str, body: dict | None, token: str):
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    request = urllib.request.Request(
        BASE + path,
        data=json.dumps(body).encode() if body is not None else None,
        headers=headers, method=method)
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.status, json.loads(response.read() or b"{}")
    except urllib.error.HTTPError as error:
        return error.code, json.loads(error.read() or b"{}")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--skip-trigger", action="store_true", help="只建智能体/链/绑定，不手动触发执行")
    args = parser.parse_args()

    query = urllib.parse.urlencode({"password": PASSWORD})
    status, login = call("POST", f"/api/v1/auth/login/username?{query}", {"username": USERNAME}, None)
    if status != 200:
        print("登录失败", status, login)
        return 1
    token = login["access_token"]
    print("登录 OK")

    # 1. 智能体（幂等：存在则跳过）
    for agent in AGENTS:
        status, existing = call("GET", f"/api/agents/{agent['id']}", None, token)
        if status == 200:
            print(f"智能体已存在: {agent['id']}")
            continue
        body = {**agent, "published": True, "skills": ["task-data-submit"],
                "tool_groups": ["web", "file:read", "file:write", "bash"]}
        status, created = call("POST", "/api/agents", body, token)
        print(f"创建智能体 {agent['id']}: HTTP {status}", created.get("id") or created)
        if status not in (200, 201):
            return 1

    # 2. 业务链（幂等：存在则跳过）
    status, existing_chain = call("GET", f"/api/roundtable-chains/{CHAIN_ID}", None, token)
    if status == 200:
        print(f"业务链已存在: {CHAIN_ID}")
    else:
        status, created = call("POST", "/api/roundtable-chains", CHAIN, token)
        print(f"创建业务链: HTTP {status}", created.get("id") or created)
        if status not in (200, 201):
            return 1

    # 3. 发布链（is_public=True；已公开则跳过）
    status, chain = call("GET", f"/api/roundtable-chains/{CHAIN_ID}", None, token)
    if status == 200 and not chain.get("is_public"):
        status, updated = call("PUT", f"/api/roundtable-chains/{CHAIN_ID}", {"is_public": True}, token)
        print(f"发布业务链: HTTP {status}, is_public={updated.get('is_public')}, 自动发布席位={updated.get('published_seat_agents')}")
    else:
        print("业务链已是公开状态")

    # 4. 演示任务绑定链 + 每 10 分钟
    status, task = call("GET", f"/api/task-analysis/external/tasks/{SOURCE_TASK_ID}/detail", None, token)
    if status != 200:
        print("演示任务不存在", status, task)
        return 1
    internal_id = task["task"]["task_id"]
    binding = {"chain_id": CHAIN_ID, "cron": CRON, "timezone": "Asia/Shanghai",
               "enabled": True, "model": None, "coordinator_prompt": None}
    status, bound = call("PUT", f"/api/task-analysis/tasks/{internal_id}/binding", binding, token)
    print(f"绑定定时（{CRON}）: HTTP {status}, next_run_at={bound.get('binding', {}).get('next_run_at')}")
    if status not in (200, 201):
        print(bound)
        return 1

    # 5. 立即触发一轮
    if not args.skip_trigger:
        status, run = call("POST", f"/api/task-analysis/tasks/{internal_id}/trigger", {}, token)
        print(f"手动触发: HTTP {status}, run_id={run.get('run', {}).get('id')}, status={run.get('run', {}).get('status')}")

    print(f"\n内部任务 id: {internal_id}")
    print("查询执行: GET /api/task-analysis/tasks/{id}/runs")
    return 0


if __name__ == "__main__":
    sys.exit(main())