feat: 商机产线能力包——招标数据(爬取平台内网HTTP)→热点推荐→研发报告→人工确认→研发审批

build.sh

@@ -0,0 +1,110 @@
+#!/bin/bash
+# pipeline-opportunity 一键部署脚本（在宿主应用根目录执行或从本目录定位宿主）。
+# 职责：建表（运行期禁止 schema 变更，建表只在部署期）+ 种子数据（appcodes + pipelines 产线记录）
+#      + wwwroot 软链 + pip install 本包。
+# 参考模式：pipeline-bidding/build.sh。
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+# 定位宿主应用根目录（含 wwwroot + py3）
+APP_ROOT=""
+for candidate in "$SCRIPT_DIR/../../pipeline-app" "$SCRIPT_DIR/../.." "$HOME/work/pipeline/pipeline-app"; do
+    if [ -d "$candidate/wwwroot" ] && [ -d "$candidate/py3" ]; then
+        APP_ROOT="$(cd "$candidate" && pwd)"
+        break
+    fi
+done
+if [ -z "$APP_ROOT" ]; then
+    echo "ERROR: host app (pipeline-app) root not found"
+    exit 1
+fi
+PYTHON="$APP_ROOT/py3/bin/python"
+echo "Host app root: $APP_ROOT"
+
+# ── 1. 建表（幂等：CREATE TABLE IF NOT EXISTS）──
+cd "$SCRIPT_DIR"
+"$PYTHON" - "$APP_ROOT" <<'PYEOF'
+import re
+ddl = open('mysql.ddl.sql', encoding='utf-8').read()
+ddl = re.sub(r'drop table if exists (\w+);', '', ddl)
+ddl = re.sub(r'CREATE TABLE (\w+)\s*\(', r'CREATE TABLE IF NOT EXISTS \1 (', ddl)
+open('/tmp/pipeline_opportunity_ddl.sql', 'w', encoding='utf-8').write(ddl)
+print('DDL prepared:', ddl.count('CREATE TABLE IF NOT EXISTS'), 'tables')
+PYEOF
+
+"$PYTHON" - "$APP_ROOT" <<'PYEOF'
+import sys, os, subprocess
+sys.path.insert(0, os.getcwd())
+from appPublic.jsonConfig import getConfig
+from appPublic.aes import aes_decode_b64
+cfg = getConfig(sys.argv[1], {'workdir': sys.argv[1]})
+kw = cfg.databases['pipeline'].kwargs
+pwd = aes_decode_b64(cfg.password_key, kw.password)
+with open('/tmp/pipeline_opportunity_ddl.sql', 'rb') as f:
+    r = subprocess.run(['mysql', '-h', str(kw.host), '-P', str(kw.port),
+                        '-u', str(kw.user), '-p%s' % pwd, str(kw.db)],
+                       stdin=f, capture_output=True)
+if r.returncode == 0:
+    print('tables applied')
+else:
+    print('ERROR:', r.stderr.decode('utf-8', 'replace').strip()[:300])
+    sys.exit(1)
+PYEOF
+
+# ── 2. 种子数据：appcodes 字典 + pipelines 产线记录（数据源 init/data.json，幂等）──
+cd "$APP_ROOT"
+"$PYTHON" - "$SCRIPT_DIR" "$APP_ROOT" <<'PYEOF'
+import sys, os, json, asyncio
+CDIR = sys.argv[1]
+sys.path.insert(0, os.getcwd())
+from appPublic.jsonConfig import getConfig
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+
+data = json.load(open(os.path.join(CDIR, 'init', 'data.json')))
+
+async def seed():
+    config = getConfig(sys.argv[2], {'workdir': sys.argv[2]})
+    db = DBPools(config.databases)
+    async with db.sqlorContext('pipeline') as sor:
+        for pl in data.get('pipelines', []):
+            recs = await sor.sqlExe("SELECT id FROM pipelines WHERE id=${i}$", {"i": pl['id']})
+            if not recs:
+                await sor.C('pipelines', {
+                    'id': pl['id'], 'name': pl['name'],
+                    'description': pl.get('description', ''),
+                    'pipeline_type': pl.get('pipeline_type', 'opportunity'),
+                    'version': '1.0.0', 'status': pl.get('status', 'published'),
+                    'org_id': pl.get('org_id', '0'), 'created_by': 'system',
+                })
+                print('pipeline registered:', pl['id'])
+        for ac in data.get('appcodes', []):
+            pid = ac['parentid']
+            rows = await sor.R('appcodes', {'id': pid})
+            if not rows:
+                await sor.C('appcodes', {'id': pid, 'name': ac.get('parentname', pid),
+                                         'hierarchy_flg': '0'})
+            for item in ac.get('items', []):
+                chk = await sor.sqlExe(
+                    "SELECT id FROM appcodes_kv WHERE parentid=${p}$ AND k=${k}$",
+                    {'p': pid, 'k': item['k']})
+                if not chk:
+                    await sor.C('appcodes_kv', {'id': getID(), 'parentid': pid,
+                                                'k': item['k'], 'v': item['v']})
+        await sor.sqlExe("COMMIT", {})
+        print('seeded %d code groups' % len(data.get('appcodes', [])))
+
+asyncio.run(seed())
+PYEOF
+
+# ── 3. wwwroot 软链 + pip install ──
+rm -f "$APP_ROOT/wwwroot/pipeline-opportunity"
+ln -sf "$SCRIPT_DIR/wwwroot" "$APP_ROOT/wwwroot/pipeline-opportunity"
+echo "wwwroot linked: pipeline-opportunity"
+
+"$APP_ROOT/py3/bin/pip" install "$SCRIPT_DIR/" 2>&1 | tail -1
+
+echo "pipeline-opportunity module deploy complete."
+echo "后续（宿主负责）：app 入口 import load_pipeline_opportunity；RBAC 用 scripts/load_path.py。"
+echo "爬虫平台接入配置：appbase params 表设 tender_api_base / tender_api_token（默认 http://192.168.16.2:9085）。"

init/data.json

@@ -0,0 +1,34 @@
+{
+    "pipelines": [
+        {
+            "id": "opportunity_general",
+            "name": "商机产线",
+            "pipeline_type": "opportunity",
+            "description": "招标市场情报(数据爬取平台)→热点软件推荐→每日AI/Agent招标→研发报告编写→人工确认→研发审批流程",
+            "status": "published",
+            "org_id": "0"
+        }
+    ],
+    "appcodes": [
+        {
+            "parentid": "opp_report_status",
+            "parentname": "研发报告状态",
+            "items": [
+                {"k": "draft", "v": "草稿"},
+                {"k": "confirmed", "v": "人工确认通过"},
+                {"k": "approval_initiated", "v": "已发起研发审批"},
+                {"k": "approved", "v": "审批通过"},
+                {"k": "rejected", "v": "审批驳回"}
+            ]
+        },
+        {
+            "parentid": "opp_approval_status",
+            "parentname": "研发审批状态",
+            "items": [
+                {"k": "initiated", "v": "已发起"},
+                {"k": "approved", "v": "审批通过"},
+                {"k": "rejected", "v": "审批驳回"}
+            ]
+        }
+    ]
+}

json/build.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+# 由 models/*.json + json/*.json 生成 wwwroot 下 CRUD 页面
+# 注意: py3/bin/xls2ui 的 shebang 可能是旧路径，直接用 python 调
+cd "$(dirname "$0")"
+PY=../../pipeline-app/py3/bin/python
+$PY ../../pipeline-app/py3/bin/xls2ui -m ../models -o ../wwwroot pipeline_opportunity opp_reports.json opp_approvals.json

json/opp_approvals.json

@@ -0,0 +1,36 @@
+{
+  "tblname": "opp_approvals",
+  "title": "研发审批",
+  "params": {
+    "sortby": [
+      "created_at desc"
+    ],
+    "confidential_fields": [],
+    "browserfields": {
+      "exclouded": [
+        "note"
+      ]
+    },
+    "editexclouded": [
+      "id",
+      "created_at",
+      "resolved_at",
+      "resolved_by",
+      "report_id"
+    ],
+    "editable": {
+      "new_data_url": "{{entire_url('add_opp_approvals.dspy')}}",
+      "update_data_url": "{{entire_url('update_opp_approvals.dspy')}}",
+      "delete_data_url": "{{entire_url('delete_opp_approvals.dspy')}}"
+    },
+    "data_filter": {
+      "AND": [
+        {
+          "field": "status",
+          "op": "=",
+          "var": "status_input"
+        }
+      ]
+    }
+  }
+}

json/opp_reports.json

@@ -0,0 +1,47 @@
+{
+  "tblname": "opp_reports",
+  "title": "研发报告",
+  "params": {
+    "sortby": [
+      "created_at desc"
+    ],
+    "confidential_fields": [],
+    "browserfields": {
+      "exclouded": [
+        "content",
+        "confirm_task_id"
+      ]
+    },
+    "editexclouded": [
+      "id",
+      "created_at",
+      "updated_at",
+      "confirm_task_id",
+      "confirmed_by"
+    ],
+    "editable": {
+      "new_data_url": "{{entire_url('add_opp_reports.dspy')}}",
+      "update_data_url": "{{entire_url('update_opp_reports.dspy')}}",
+      "delete_data_url": "{{entire_url('delete_opp_reports.dspy')}}"
+    },
+    "data_filter": {
+      "AND": [
+        {
+          "field": "title",
+          "op": "LIKE",
+          "var": "title_input"
+        },
+        {
+          "field": "software",
+          "op": "LIKE",
+          "var": "software_input"
+        },
+        {
+          "field": "status",
+          "op": "=",
+          "var": "status_input"
+        }
+      ]
+    }
+  }
+}

models/opp_approvals.json

@@ -0,0 +1,70 @@
+{
+    "summary": [
+        {
+            "name": "opp_approvals",
+            "title": "研发审批",
+            "primary": [
+                "id"
+            ],
+            "catelog": "entity"
+        }
+    ],
+    "fields": [
+        {
+            "name": "id",
+            "title": "主键ID",
+            "type": "str",
+            "length": 32,
+            "nullable": "no"
+        },
+        {
+            "name": "report_id",
+            "title": "报告ID",
+            "type": "str",
+            "length": 32,
+            "nullable": "no"
+        },
+        {
+            "name": "project_id",
+            "title": "项目ID",
+            "type": "str",
+            "length": 32
+        },
+        {
+            "name": "status",
+            "title": "审批状态",
+            "type": "str",
+            "length": 16,
+            "nullable": "no",
+            "default": "initiated"
+        },
+        {
+            "name": "note",
+            "title": "审批说明",
+            "type": "text"
+        },
+        {
+            "name": "created_by",
+            "title": "发起人",
+            "type": "str",
+            "length": 64
+        },
+        {
+            "name": "resolved_by",
+            "title": "审批人",
+            "type": "str",
+            "length": 64
+        },
+        {
+            "name": "created_at",
+            "title": "创建时间",
+            "type": "timestamp",
+            "nullable": "no"
+        },
+        {
+            "name": "resolved_at",
+            "title": "结论时间",
+            "type": "timestamp"
+        }
+    ]
+}

models/opp_reports.json

@@ -0,0 +1,83 @@
+{
+    "summary": [
+        {
+            "name": "opp_reports",
+            "title": "研发报告",
+            "primary": [
+                "id"
+            ],
+            "catelog": "entity"
+        }
+    ],
+    "fields": [
+        {
+            "name": "id",
+            "title": "主键ID",
+            "type": "str",
+            "length": 32,
+            "nullable": "no"
+        },
+        {
+            "name": "project_id",
+            "title": "项目ID",
+            "type": "str",
+            "length": 32
+        },
+        {
+            "name": "software",
+            "title": "软件/主题",
+            "type": "str",
+            "length": 128,
+            "nullable": "no"
+        },
+        {
+            "name": "title",
+            "title": "报告标题",
+            "type": "str",
+            "length": 255,
+            "nullable": "no"
+        },
+        {
+            "name": "content",
+            "title": "报告正文",
+            "type": "text"
+        },
+        {
+            "name": "status",
+            "title": "报告状态",
+            "type": "str",
+            "length": 24,
+            "nullable": "no",
+            "default": "draft"
+        },
+        {
+            "name": "confirm_task_id",
+            "title": "确认任务ID",
+            "type": "str",
+            "length": 32
+        },
+        {
+            "name": "confirmed_by",
+            "title": "确认人",
+            "type": "str",
+            "length": 64
+        },
+        {
+            "name": "created_by",
+            "title": "创建人",
+            "type": "str",
+            "length": 64
+        },
+        {
+            "name": "created_at",
+            "title": "创建时间",
+            "type": "timestamp",
+            "nullable": "no"
+        },
+        {
+            "name": "updated_at",
+            "title": "更新时间",
+            "type": "timestamp"
+        }
+    ]
+}

mysql.ddl.sql

@@ -0,0 +1,70 @@
+
+-- /home/ymq/work/pipeline/pipeline-opportunity/models/opp_reports.json
+
+
+
+
+
+-- 建库时请用以下语句，支持emoji字符
+-- CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+drop table if exists opp_reports;
+CREATE TABLE opp_reports
+(
+
+  `id` VARCHAR(32) NOT NULL  comment '主键ID',
+  `project_id` VARCHAR(32)   comment '项目ID',
+  `software` VARCHAR(128) NOT NULL  comment '软件/主题',
+  `title` VARCHAR(255) NOT NULL  comment '报告标题',
+  `content` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci   comment '报告正文',
+  `status` VARCHAR(24) NOT NULL  DEFAULT 'draft' comment '报告状态',
+  `confirm_task_id` VARCHAR(32)   comment '确认任务ID',
+  `confirmed_by` VARCHAR(64)   comment '确认人',
+  `created_by` VARCHAR(64)   comment '创建人',
+  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL  comment '创建时间',
+  `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP   comment '更新时间'
+
+
+,primary key(id)
+
+
+)
+CHARACTER SET utf8mb4
+COLLATE utf8mb4_unicode_ci
+engine=innodb 
+comment '研发报告'
+;
+
+
+-- /home/ymq/work/pipeline/pipeline-opportunity/models/opp_approvals.json
+
+
+
+
+
+-- 建库时请用以下语句，支持emoji字符
+-- CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+drop table if exists opp_approvals;
+CREATE TABLE opp_approvals
+(
+
+  `id` VARCHAR(32) NOT NULL  comment '主键ID',
+  `report_id` VARCHAR(32) NOT NULL  comment '报告ID',
+  `project_id` VARCHAR(32)   comment '项目ID',
+  `status` VARCHAR(16) NOT NULL  DEFAULT 'initiated' comment '审批状态',
+  `note` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci   comment '审批说明',
+  `created_by` VARCHAR(64)   comment '发起人',
+  `resolved_by` VARCHAR(64)   comment '审批人',
+  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL  comment '创建时间',
+  `resolved_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP   comment '结论时间'
+
+
+,primary key(id)
+
+
+)
+CHARACTER SET utf8mb4
+COLLATE utf8mb4_unicode_ci
+engine=innodb 
+comment '研发审批'
+;
+

pipeline_opportunity/__init__.py

@@ -0,0 +1 @@
+from .init import load_pipeline_opportunity  # noqa: F401

pipeline_opportunity/init.py

@@ -0,0 +1,35 @@
+"""pipeline-opportunity 模块装载入口。
+
+宿主应用（pipeline-app）在 init() 中调用 load_pipeline_opportunity() 即挂载商机产线：
+能力包注册（opp_ability）。未装本模块的宿主完全零接触。
+
+平台边界：本模块属于产线平台，只通过内网 HTTP 消费数据爬取平台的只读数据接口；
+报告/确认/审批状态落产线自己的库（opp_* 表），绝不回写爬虫库。
+
+表结构不在此处建（铁律：运行期不做 schema 变更），建表走部署期 build.sh。
+"""
+
+import logging
+
+logger = logging.getLogger("pipeline.opportunity")
+
+_loaded = False
+
+
+def load_pipeline_opportunity():
+    """挂载商机产线（幂等）。"""
+    global _loaded
+    if _loaded:
+        return True
+
+    # 主 agent 能力包 + slash 命令（import 即注册，与投标产线同机制）
+    try:
+        from . import opp_ability  # noqa: F401
+        logger.info("[pipeline_opportunity] ability registered: opportunity_general")
+    except Exception as e:
+        logger.warning("[pipeline_opportunity] ability 注册失败: %s", str(e)[:200])
+        return False
+
+    _loaded = True
+    logger.info("[pipeline_opportunity] v0.1.0 loaded")
+    return True

pipeline_opportunity/opp_ability.py

@@ -0,0 +1,338 @@
+"""商机产线能力包（opportunity_general）：主 agent（驾驶舱）工具 + 角色定义 + slash 命令。
+
+与开发/投标产线同构，通过 pipeline-core 的 PipelineAbility 注册表挂载，
+零侵入 core / AgentExecutor：项目 sd_projects.pipeline_id='opportunity_general' 时自动生效。
+
+数据边界：招标数据只从数据爬取平台（内网 HTTP）读取，本产线不爬取、不存招标原始数据；
+本产线自己的状态（研发报告/审批）落产线库的 opp_reports / opp_approvals 表。
+
+商机闭环：热点软件推荐 → 选定方向 → 编写研发报告 → 人工确认（门禁）→ 发起研发审批 → 审批结论。
+"""
+
+import json
+import logging
+
+from pipeline_core import (
+    ToolDefinition, PipelineAbility, RoleSpec, register_ability,
+    SlashCommand, register_slash_command,
+)
+
+logger = logging.getLogger("pipeline.opportunity.ability")
+
+PIPELINE_ID = "opportunity_general"
+
+# ══════════════════ 工具定义（主 agent / 驾驶舱）══════════════════
+
+OPP_TOOLS = [
+    # ── 数据（来自数据爬取平台，只读）──
+    ToolDefinition(name="opp_hot_software", description="热点软件推荐：近N天招标市场按软件主题聚合排名（招标量/预算/热点地区/样例项目），用于判断哪些软件方向最值得做",
+                   parameters={"days": "统计窗口天数(默认30)", "top": "返回前几名(默认10)"}, category="data"),
+    ToolDefinition(name="opp_daily_ai_tenders", description="每日AI/Agent软件类产品或项目招标信息展示：指定日期的AI相关招标清单+预算统计",
+                   parameters={"date": "日期YYYY-MM-DD(默认今天)"}, category="data"),
+    ToolDefinition(name="opp_search_tenders", description="按关键词搜索招标信息（IT项目）",
+                   parameters={"keyword": "标题关键词", "days": "回溯天数(默认7)",
+                               "source": "来源(可选 ccgp/ggzy)", "limit": "条数(默认50)"}, category="data"),
+    ToolDefinition(name="opp_tender_detail", description="查看单条招标信息完整详情（含公告原文、预算、联系人、来源链接）",
+                   parameters={"tender_id": "招标信息ID"}, category="data"),
+    ToolDefinition(name="opp_crawler_stats", description="查看数据爬取平台采集健康度（总量/24h新增/最近运行）",
+                   parameters={}, category="data"),
+
+    # ── 报告与审批（商机闭环）──
+    ToolDefinition(name="opp_create_report", description="创建软件研发报告草稿。从热门软件中选定的方向写入报告，数据基础章节自动从爬取平台拉取",
+                   parameters={"software": "软件/主题名", "title": "报告标题(可选)",
+                               "analysis": "市场分析正文(可选，可先建骨架后补写)"}, category="report"),
+    ToolDefinition(name="opp_update_report", description="更新草稿报告内容（仅草稿状态可改）",
+                   parameters={"report_id": "报告ID", "content": "新的报告全文"}, category="report"),
+    ToolDefinition(name="opp_list_reports", description="列出研发报告及状态（draft/confirmed/approval_initiated/approved/rejected）",
+                   parameters={"status": "状态过滤(可选)"}, category="report"),
+    ToolDefinition(name="opp_get_report", description="查看报告全文",
+                   parameters={"report_id": "报告ID"}, category="report"),
+    ToolDefinition(name="opp_submit_report", description="报告写完后提交人工确认（创建确认任务，等人工决策）",
+                   parameters={"report_id": "报告ID", "note": "给确认人的说明(可选)"}, category="report"),
+    ToolDefinition(name="opp_list_approvals", description="列出研发审批单及状态",
+                   parameters={"status": "状态过滤(可选)"}, category="report"),
+    ToolDefinition(name="opp_diagnose", description="诊断商机产线：报告/审批分布、待办人工任务、爬取平台连通性",
+                   parameters={}, category="agent"),
+]
+
+# ══════════════════ 产线 prompt 片段 ══════════════════
+
+OPP_PROMPT = """你是「商机产线」的驾驶舱 agent，负责把招标市场情报转化为软件研发商机：
+热点软件推荐 → 选定方向 → 编写软件研发报告 → 人工确认 → 发起研发审批 → 跟踪审批结论。
+
+## 数据来源（重要边界）
+招标数据全部来自「数据爬取平台」（内网 HTTP，已采集入库、已去重、每条带来源链接）。
+你只消费数据（opp_hot_software / opp_daily_ai_tenders / opp_search_tenders / opp_tender_detail），
+**不要**自己去爬网站，也不要怀疑数据新鲜度——先看 opp_crawler_stats 的采集时间。
+
+## 商机闭环与门禁（不要绕过）
+1. 用 opp_hot_software 看哪个软件方向招标多、预算大，给用户推荐。
+2. 用户选定方向后：opp_create_report 建草稿 → 你基于真实招标数据补写分析
+   （市场空间/代表项目/竞争格局/研发建议）→ opp_update_report 更新全文。
+3. 写完用 opp_submit_report 提交**人工确认**（门禁）。确认前不许发起审批。
+4. 人工确认通过后，你才用（或由人工任务驱动）发起研发审批；审批结论由人工回流，
+   **禁止**自己批准自己的审批。
+
+## 硬规则
+- 报告里的招标数量、预算、项目案例必须来自工具返回的真实数据，**禁止编造**；
+  查不到就如实写"数据未覆盖"。
+- 每条引用的招标信息带来源链接（爬虫平台已提供 url / source_url）。
+- 报告没被人工确认，不许当作可发起审批；审批没过，不许当作可立项。
+
+## 典型场景
+- 用户问「最近什么软件最火/最值得做」→ opp_hot_software(days=30) 后给结论。
+- 用户问「今天有什么AI招标」→ opp_daily_ai_tenders()。
+- 用户说「给XX写个研发报告」→ opp_create_report → 拉数据补写 → opp_submit_report。
+- 用户问进展 → opp_diagnose（报告/审批分布 + 待办人工任务）。"""
+
+# ══════════════════ 角色集（商机产线轻量：分析+撰写两角色）══════════════════
+
+OPP_ROLES = [
+    RoleSpec(
+        name="agent.opp_analyst",
+        description="商机分析师",
+        aliases=["opp_analyst", "analyst"],
+        system_prompt="你是商机分析师。先 load_skill 加载 role 技能，按其中的职责与规范执行任务。",
+        next_role="",
+        task_title="商机分析",
+    ),
+    RoleSpec(
+        name="agent.opp_writer",
+        description="研发报告撰写工程师",
+        aliases=["opp_writer", "writer"],
+        system_prompt="你是研发报告撰写工程师。先 load_skill 加载 role 技能，按其中的职责与规范执行任务。",
+        next_role="",
+        task_title="研发报告编写",
+    ),
+]
+
+# ══════════════════ handler（签名 async def handler(sor, params, ctx) -> str）══════════════════
+
+
+def _fmt(ok, msg):
+    return ("OK: " if ok else "ERROR: ") + str(msg)
+
+
+def _fmt_rows(rows, empty="（空）"):
+    if not rows:
+        return empty
+    if isinstance(rows, str):
+        return rows
+    return "\n".join(json.dumps(r, ensure_ascii=False, default=str) for r in rows[:80])
+
+
+async def _h_hot_software(sor, p, ctx):
+    from .opp_data_capability import hot_software
+    ok, res = await hot_software(sor, days=int(p.get("days") or 30),
+                                 top=int(p.get("top") or 10))
+    if not ok:
+        return _fmt(False, res)
+    out = {"窗口": "%d天" % res.get("window_days"),
+           "窗口内招标总数": res.get("tenders_in_window"),
+           "排名": res.get("ranking")}
+    return json.dumps(out, ensure_ascii=False, default=str)
+
+
+async def _h_daily_ai(sor, p, ctx):
+    from .opp_data_capability import daily_ai_tenders
+    ok, res = await daily_ai_tenders(sor, target_date=p.get("date") or "")
+    if not ok:
+        return _fmt(False, res)
+    return json.dumps({
+        "date": res.get("date"), "count": res.get("count"),
+        "有预算条数": res.get("with_budget"),
+        "预算合计(万元)": res.get("total_budget_wan"),
+        "items": res.get("items"),
+    }, ensure_ascii=False, default=str)
+
+
+async def _h_search(sor, p, ctx):
+    from .opp_data_capability import search_tenders
+    ok, res = await search_tenders(sor, keyword=p.get("keyword") or "",
+                                   days=int(p.get("days") or 7),
+                                   source=p.get("source") or "",
+                                   limit=int(p.get("limit") or 50))
+    if not ok:
+        return _fmt(False, res)
+    return json.dumps({"count": res.get("count"), "items": res.get("items")},
+                      ensure_ascii=False, default=str)
+
+
+async def _h_detail(sor, p, ctx):
+    from .opp_data_capability import tender_detail
+    ok, res = await tender_detail(sor, p.get("tender_id") or "")
+    if not ok:
+        return _fmt(False, res)
+    return json.dumps(res, ensure_ascii=False, default=str)
+
+
+async def _h_crawler_stats(sor, p, ctx):
+    from .opp_data_capability import crawler_stats
+    ok, res = await crawler_stats(sor)
+    if not ok:
+        return _fmt(False, res)
+    return json.dumps(res, ensure_ascii=False, default=str)
+
+
+async def _h_create_report(sor, p, ctx):
+    from .opp_report_capability import create_report
+    ok, rid = await create_report(
+        ctx.get("project_id") or "", p.get("software") or "",
+        title=p.get("title") or "", analysis=p.get("analysis") or "",
+        created_by="agent.main_agent")
+    return _fmt(ok, ("report_id=%s（草稿，用 opp_update_report 补写，完成后 opp_submit_report）" % rid) if ok else rid)
+
+
+async def _h_update_report(sor, p, ctx):
+    from .opp_report_capability import update_report
+    ok, msg = await update_report(p.get("report_id") or "", p.get("content") or "")
+    return _fmt(ok, msg)
+
+
+async def _h_list_reports(sor, p, ctx):
+    from .opp_report_capability import list_reports
+    rows = await list_reports(project_id=ctx.get("project_id") or "",
+                              status=p.get("status") or "")
+    return _fmt_rows(rows, "尚无研发报告")
+
+
+async def _h_get_report(sor, p, ctx):
+    from .opp_report_capability import get_report_full
+    rep = await get_report_full(p.get("report_id") or "")
+    if not rep:
+        return _fmt(False, "报告不存在")
+    return json.dumps(rep, ensure_ascii=False, default=str)
+
+
+async def _h_submit_report(sor, p, ctx):
+    from .opp_report_capability import submit_for_confirmation
+    ok, msg = await submit_for_confirmation(p.get("report_id") or "",
+                                            note=p.get("note") or "")
+    return _fmt(ok, msg)
+
+
+async def _h_list_approvals(sor, p, ctx):
+    from .opp_report_capability import list_approvals
+    rows = await list_approvals(project_id=ctx.get("project_id") or "",
+                                status=p.get("status") or "")
+    return _fmt_rows(rows, "尚无审批单")
+
+
+async def _h_diagnose(sor, p, ctx):
+    from .opp_report_capability import list_reports, list_approvals
+    from .opp_data_capability import crawler_stats
+    from .opp_common import find_human_task, HT_REPORT_CONFIRM, HT_DEV_APPROVAL
+    reps = await list_reports()
+    apps = await list_approvals()
+    ok, cs = await crawler_stats(sor)
+    dist = {}
+    for r in reps:
+        dist[r.get("status")] = dist.get(r.get("status"), 0) + 1
+    ht_conf = await find_human_task(sor, ctx.get("project_id") or "",
+                                    HT_REPORT_CONFIRM, status="pending")
+    ht_appr = await find_human_task(sor, ctx.get("project_id") or "",
+                                    HT_DEV_APPROVAL, status="pending")
+    return json.dumps({
+        "报告状态分布": dist,
+        "审批单": len(apps),
+        "待确认报告任务": ht_conf.get("title") if ht_conf else None,
+        "待审批任务": ht_appr.get("title") if ht_appr else None,
+        "爬取平台": ("连通，共 %s 条，24h 新增 %s" % (
+            cs.get("total"), cs.get("collected_24h"))) if ok else "不可达: %s" % cs,
+    }, ensure_ascii=False, default=str)
+
+
+OPP_HANDLERS = {
+    "opp_hot_software": _h_hot_software,
+    "opp_daily_ai_tenders": _h_daily_ai,
+    "opp_search_tenders": _h_search,
+    "opp_tender_detail": _h_detail,
+    "opp_crawler_stats": _h_crawler_stats,
+    "opp_create_report": _h_create_report,
+    "opp_update_report": _h_update_report,
+    "opp_list_reports": _h_list_reports,
+    "opp_get_report": _h_get_report,
+    "opp_submit_report": _h_submit_report,
+    "opp_list_approvals": _h_list_approvals,
+    "opp_diagnose": _h_diagnose,
+}
+
+
+def _merge_shared():
+    """并入引擎级通用任务/交付件/问答/项目工具（不重复实现，见 pipeline_service.shared_ability）。"""
+    try:
+        from pipeline_service.shared_ability import SHARED_TOOLS, SHARED_HANDLERS
+    except Exception as e:      # pipeline_service 未安装/版本旧 → 只用商机专属工具，诚实降级
+        logger.warning("shared_ability 不可用，商机产线仅挂载专属工具: %s", str(e)[:120])
+        return list(OPP_TOOLS), dict(OPP_HANDLERS)
+    names = {t.name for t in OPP_TOOLS}
+    tools = list(OPP_TOOLS) + [t for t in SHARED_TOOLS if t.name not in names]
+    handlers = dict(SHARED_HANDLERS)
+    handlers.update(OPP_HANDLERS)      # 同名以商机产线实现为准
+    return tools, handlers
+
+
+def register_opp_ability():
+    """注册商机产线能力包（幂等）。"""
+    tools, handlers = _merge_shared()
+    ability = PipelineAbility(
+        pipeline_id=PIPELINE_ID,
+        name="商机产线",
+        tools=tools,
+        system_prompt=OPP_PROMPT,
+        handlers=handlers,
+        roles=OPP_ROLES,
+        menus=[
+            {"label": "📊 研发报告", "icon": "", "url": "/pipeline-opportunity/opp_reports/index.ui",
+             "type": "popup", "width": "88%", "height": "82%"},
+            {"label": "✅ 研发审批", "icon": "", "url": "/pipeline-opportunity/opp_approvals/index.ui",
+             "type": "popup", "width": "88%", "height": "82%"},
+        ],
+    )
+    register_ability(ability)
+    return ability
+
+
+# ══════════════════ slash 命令（产线专属）══════════════════
+
+async def _slash_hot(args, ctx):
+    ex = ctx.get("executor")
+    if not ex:
+        return "无执行器上下文"
+    return await ex._execute_ability_tool("opp_hot_software", {"days": args or 30})
+
+
+async def _slash_ai(args, ctx):
+    ex = ctx.get("executor")
+    if not ex:
+        return "无执行器上下文"
+    return await ex._execute_ability_tool("opp_daily_ai_tenders", {"date": args or ""})
+
+
+async def _slash_reports(args, ctx):
+    ex = ctx.get("executor")
+    if not ex:
+        return "无执行器上下文"
+    return await ex._execute_ability_tool("opp_list_reports", {})
+
+
+async def _slash_oppdiag(args, ctx):
+    ex = ctx.get("executor")
+    if not ex:
+        return "无执行器上下文"
+    return await ex._execute_ability_tool("opp_diagnose", {})
+
+
+def register_opp_slash_commands():
+    for cmd in [
+        SlashCommand("hot", "热点软件推荐（近30天招标排名）", _slash_hot, "pipeline", PIPELINE_ID),
+        SlashCommand("ai", "今日AI/Agent招标清单", _slash_ai, "pipeline", PIPELINE_ID),
+        SlashCommand("reports", "列出研发报告", _slash_reports, "pipeline", PIPELINE_ID),
+        SlashCommand("oppdiag", "商机产线诊断", _slash_oppdiag, "pipeline", PIPELINE_ID),
+    ]:
+        register_slash_command(cmd)
+
+
+# import 即注册（与投标产线一致：宿主 import pipeline_opportunity 即生效）
+register_opp_ability()
+register_opp_slash_commands()

pipeline_opportunity/opp_common.py

@@ -0,0 +1,189 @@
+"""商机产线公共层：DB 上下文、爬虫平台 HTTP 客户端、报告/审批状态、人工任务。
+
+所有 opp_* capability 模块共用本模块，禁止各自重复实现。
+
+爬虫平台接入：内网 HTTP（http_api.py），配置优先级
+  1. appbase params 表: tender_api_base / tender_api_token（系统级配置禁硬编码）
+  2. 兜底默认: http://192.168.16.2:9085（内网地址，仅默认值）
+"""
+
+import json
+import logging
+import urllib.parse
+
+from sqlor.dbpools import DBPools
+from appPublic.uniqueID import getID
+
+DBNAME = "pipeline"
+PIPELINE_ID = "opportunity_general"
+
+logger = logging.getLogger("pipeline.opportunity")
+
+# 爬虫平台默认接入点（内网网卡地址，公网不可达）；可被 params 表覆盖
+DEFAULT_API_BASE = "http://192.168.16.2:9085"
+
+# ── 报告状态 ──
+RP_DRAFT = "draft"                    # 草稿（agent 编写中/待审阅）
+RP_CONFIRMED = "confirmed"            # 人工确认通过（门禁）
+RP_APPROVAL_INITIATED = "approval_initiated"  # 已发起研发审批
+RP_APPROVED = "approved"              # 审批通过
+RP_REJECTED = "rejected"              # 审批驳回
+
+# ── 审批状态 ──
+AP_INITIATED = "initiated"
+AP_APPROVED = "approved"
+AP_REJECTED = "rejected"
+
+# ── 人工任务类型 ──
+HT_REPORT_CONFIRM = "opp_report_confirm"    # 报告人工确认门禁
+HT_DEV_APPROVAL = "opp_dev_approval"        # 研发审批人工决策
+
+# 报告状态流转（合法迁移）
+REPORT_TRANSITIONS = {
+    RP_DRAFT: (RP_CONFIRMED,),
+    RP_CONFIRMED: (RP_APPROVAL_INITIATED,),
+    RP_APPROVAL_INITIATED: (RP_APPROVED, RP_REJECTED),
+    RP_REJECTED: (RP_DRAFT,),       # 驳回可改回草稿重来
+}
+
+
+def get_db():
+    """取 DBPools（databases 为空时从 config 兜底注入，与引擎其他模块一致）。"""
+    db = DBPools()
+    if not db.databases:
+        from appPublic.jsonConfig import getConfig
+        config = getConfig()
+        if config and config.databases:
+            db.databases = config.databases
+    return db, DBNAME
+
+
+def new_id():
+    return getID()
+
+
+def rec_to_dict(rec):
+    """sqlor 行对象 → dict（过滤方法名）。"""
+    if rec is None:
+        return {}
+    if isinstance(rec, dict):
+        return {k: v for k, v in rec.items() if not callable(v)}
+    d = {}
+    for k, v in vars(rec).items():
+        if k.startswith("_") or callable(v):
+            continue
+        d[k] = v
+    return d
+
+
+def rows_to_dicts(recs, limit=200):
+    return [rec_to_dict(r) for r in (recs or [])[:limit]]
+
+
+def json_loads(s, default=None):
+    if not s:
+        return default if default is not None else []
+    if isinstance(s, (list, dict)):
+        return s
+    try:
+        return json.loads(s)
+    except (json.JSONDecodeError, TypeError, ValueError):
+        return default if default is not None else []
+
+
+async def get_param(sor, key, default=""):
+    """读 appbase params 表配置 + 默认兜底（系统级配置禁硬编码）。"""
+    try:
+        from pipeline_service.workspace import get_param as _gp
+        v = await _gp(sor, key, "")
+        if v not in (None, ""):
+            return str(v)
+    except Exception:
+        pass
+    return default
+
+
+async def get_crawler_config(sor):
+    """爬虫平台接入配置 (base, token)。params 表优先，默认内网地址兜底。"""
+    base = await get_param(sor, "tender_api_base", DEFAULT_API_BASE)
+    token = await get_param(sor, "tender_api_token", "")
+    return (base or DEFAULT_API_BASE).rstrip("/"), token
+
+
+async def crawler_get(sor, path, params=None, timeout=20):
+    """调爬虫平台只读接口。返回 (ok, payload|错误信息)。"""
+    base, token = await get_crawler_config(sor)
+    qs = ""
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            qs = "?" + urllib.parse.urlencode(clean)
+    url = base + path + qs
+    headers = {"X-API-Token": token} if token else {}
+    try:
+        import httpx
+    except ImportError:
+        httpx = None
+    try:
+        if httpx is not None:
+            async with httpx.AsyncClient(timeout=timeout) as cli:
+                r = await cli.get(url, headers=headers)
+                if r.status_code != 200:
+                    return False, "爬虫平台返回 %d: %s" % (r.status_code, r.text[:200])
+                return True, r.json()
+        # 无 httpx 时退回 urllib（同步阻塞，仅在缺依赖时用）
+        import asyncio
+        import urllib.request
+
+        def _blocking():
+            req = urllib.request.Request(url, headers=headers)
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return json.loads(resp.read().decode("utf-8"))
+
+        return True, await asyncio.get_event_loop().run_in_executor(None, _blocking)
+    except Exception as e:
+        logger.warning("crawler_get %s failed: %s", path, str(e)[:200])
+        return False, "爬虫平台调用失败: %s" % str(e)[:200]
+
+
+async def find_human_task(sor, project_id, task_type, status=None):
+    """查项目下指定类型的人工任务（供门禁判断与幂等创建）。"""
+    sql = ("SELECT id, title, status, task_type FROM pipeline_human_tasks "
+           "WHERE project_id=${pid}$ AND task_type=${tt}$")
+    args = {"pid": project_id, "tt": task_type}
+    if status:
+        sql += " AND status=${st}$"
+        args["st"] = status
+    sql += " ORDER BY created_at DESC LIMIT 1"
+    recs = await sor.sqlExe(sql, args)
+    await sor.sqlExe("COMMIT", {})
+    return rec_to_dict(recs[0]) if recs else None
+
+
+async def create_human_task(sor, project_id, task_type, title, description="",
+                            assignee_role="owner"):
+    """创建项目级人工任务（确认/审批门禁）。幂等由调用方先 find 保证。"""
+    hid = new_id()
+    await sor.C("pipeline_human_tasks", {
+        "id": hid,
+        "project_id": project_id,
+        "task_type": task_type,
+        "title": title,
+        "description": description,
+        "status": "pending",
+        "assignee_role": assignee_role,
+        "created_by": "agent.opportunity",
+    })
+    await sor.sqlExe("COMMIT", {})
+    return hid
+
+
+async def get_report(sor, report_id):
+    recs = await sor.sqlExe(
+        "SELECT * FROM opp_reports WHERE id=${i}$", {"i": report_id})
+    await sor.sqlExe("COMMIT", {})
+    return rec_to_dict(recs[0]) if recs else None
+
+
+async def can_transition(cur_status, new_status):
+    return new_status in REPORT_TRANSITIONS.get(cur_status, ())

pipeline_opportunity/opp_data_capability.py

@@ -0,0 +1,42 @@
+"""商机产线数据能力：从数据爬取平台拉取招标数据（只读）。
+
+所有函数签名统一 (sor, **params)，返回 (ok, payload|错误信息)。
+薄封装：查询条件透传给爬虫平台，本层不做二次加工（热点排序由爬虫平台算）。
+"""
+
+from .opp_common import crawler_get
+
+
+async def search_tenders(sor, keyword="", days=7, source="", only_it=True, limit=50):
+    """关键词检索招标信息。"""
+    params = {"days": days, "limit": limit}
+    if keyword:
+        params["keyword"] = keyword
+    if source:
+        params["source"] = source
+    params["only_it"] = "1" if only_it else "0"
+    return await crawler_get(sor, "/api/tenders", params)
+
+
+async def tender_detail(sor, tender_id):
+    """单条招标详情（含公告全文、来源 url）。"""
+    return await crawler_get(sor, "/api/tenders/%s" % tender_id)
+
+
+async def daily_ai_tenders(sor, target_date=""):
+    """每日 AI/Agent 软件类招标展示数据。"""
+    params = {}
+    if target_date:
+        params["date"] = target_date
+    return await crawler_get(sor, "/api/ai_tenders", params)
+
+
+async def hot_software(sor, days=30, top=10):
+    """热点软件主题排名（数量+预算+热点地区+样例）。"""
+    return await crawler_get(sor, "/api/hot_software",
+                             {"days": days, "top": top})
+
+
+async def crawler_stats(sor):
+    """爬虫平台采集健康度。"""
+    return await crawler_get(sor, "/api/stats")

pipeline_opportunity/opp_report_capability.py

@@ -0,0 +1,212 @@
+"""商机产线报告/审批能力：研发报告编写 → 人工确认 → 研发审批流程。
+
+状态机（REPORT_TRANSITIONS，opp_common）：
+  draft → confirmed → approval_initiated → approved / rejected
+                                                    └ rejected → draft（可重来）
+
+门禁纪律：
+- confirm 只能 draft → confirmed，且是人工决策（由人工任务回流驱动）
+- initiate_approval 只能 confirmed → approval_initiated（未经确认禁止发起审批）
+- 审批结论由人工决策回流（approve/reject），agent 不得自批自审
+"""
+
+import logging
+
+from .opp_common import (
+    get_db, new_id, rec_to_dict, rows_to_dicts, can_transition,
+    RP_DRAFT, RP_CONFIRMED, RP_APPROVAL_INITIATED, RP_APPROVED, RP_REJECTED,
+    AP_INITIATED, AP_APPROVED, AP_REJECTED,
+    HT_REPORT_CONFIRM, HT_DEV_APPROVAL,
+    find_human_task, create_human_task, get_report,
+)
+
+logger = logging.getLogger("pipeline.opportunity.report")
+
+
+async def create_report(project_id, software, title="", analysis="",
+                        created_by="agent.opportunity"):
+    """创建研发报告草稿。返回 (True, report_id) 或 (False, 错误)。"""
+    if not software:
+        return False, "缺少 software（软件/主题名）"
+    if not title:
+        title = "%s 研发机会分析报告" % software
+    db, dbname = get_db()
+    rid = new_id()
+    async with db.sqlorContext(dbname) as sor:
+        await sor.C("opp_reports", {
+            "id": rid,
+            "project_id": project_id or "",
+            "software": software,
+            "title": title,
+            "content": analysis or "",
+            "status": RP_DRAFT,
+            "created_by": created_by,
+        })
+        await sor.sqlExe("COMMIT", {})
+    return True, rid
+
+
+async def update_report(report_id, content):
+    """更新草稿报告内容（仅 draft 可改）。"""
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        rep = await get_report(sor, report_id)
+        if not rep:
+            return False, "报告不存在: %s" % report_id
+        if rep.get("status") != RP_DRAFT:
+            return False, "状态 %s 不可修改（仅草稿可编辑）" % rep.get("status")
+        await sor.U("opp_reports", {"id": report_id, "content": content})
+        await sor.sqlExe("COMMIT", {})
+    return True, "已更新"
+
+
+async def list_reports(project_id="", status=""):
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ("SELECT id, project_id, software, title, status, created_by, "
+               "created_at, updated_at FROM opp_reports WHERE 1=1")
+        args = {}
+        if project_id:
+            sql += " AND project_id=${pid}$"
+            args["pid"] = project_id
+        if status:
+            sql += " AND status=${st}$"
+            args["st"] = status
+        sql += " ORDER BY created_at DESC LIMIT 50"
+        recs = await sor.sqlExe(sql, args)
+        await sor.sqlExe("COMMIT", {})
+        return rows_to_dicts(recs)
+
+
+async def get_report_full(report_id):
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        rep = await get_report(sor, report_id)
+        return rep
+
+
+async def submit_for_confirmation(report_id, note=""):
+    """agent 完成报告后提交待人工确认：发人工确认任务（门禁）。
+
+    报告保持 draft，创建/复用 pending 人工任务。人工确认后才进 confirmed。
+    """
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        rep = await get_report(sor, report_id)
+        if not rep:
+            return False, "报告不存在: %s" % report_id
+        if rep.get("status") != RP_DRAFT:
+            return False, "状态 %s 不可提交确认" % rep.get("status")
+        exist = await find_human_task(sor, rep.get("project_id", ""),
+                                      HT_REPORT_CONFIRM, status="pending")
+        if exist:
+            return True, "已有待确认任务: %s" % exist.get("id")
+        hid = await create_human_task(
+            sor, rep.get("project_id", ""), HT_REPORT_CONFIRM,
+            "确认研发报告：%s" % rep.get("title", ""),
+            note or "请审阅研发报告《%s》，确认后发起研发审批。" % rep.get("title", ""))
+        await sor.U("opp_reports", {
+            "id": report_id, "confirm_task_id": hid})
+        await sor.sqlExe("COMMIT", {})
+        return True, "已提交人工确认任务: %s" % hid
+
+
+async def confirm_report(report_id, ok, operator=""):
+    """人工确认回流（门禁）。ok=True → confirmed；ok=False → 退回草稿待改。"""
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        rep = await get_report(sor, report_id)
+        if not rep:
+            return False, "报告不存在: %s" % report_id
+        if rep.get("status") != RP_DRAFT:
+            return False, "状态 %s 无法确认（仅草稿可确认）" % rep.get("status")
+        new_status = RP_CONFIRMED if ok else RP_DRAFT
+        await sor.U("opp_reports", {
+            "id": report_id, "status": new_status,
+            "confirmed_by": operator or "human"})
+        # 完结对应的人工任务
+        hid = rep.get("confirm_task_id")
+        if hid:
+            await sor.U("pipeline_human_tasks", {
+                "id": hid, "status": "done" if ok else "rejected"})
+        await sor.sqlExe("COMMIT", {})
+    return True, "已确认" if ok else "已退回草稿"
+
+
+async def initiate_approval(report_id, note="", created_by="agent.opportunity"):
+    """发起研发审批：要求报告已 confirmed（人工确认门禁）。"""
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        rep = await get_report(sor, report_id)
+        if not rep:
+            return False, "报告不存在: %s" % report_id
+        if not await can_transition(rep.get("status"), RP_APPROVAL_INITIATED):
+            return False, "状态 %s 不可发起审批（须先人工确认）" % rep.get("status")
+        aid = new_id()
+        await sor.C("opp_approvals", {
+            "id": aid,
+            "report_id": report_id,
+            "project_id": rep.get("project_id", ""),
+            "status": AP_INITIATED,
+            "note": note,
+            "created_by": created_by,
+        })
+        await sor.U("opp_reports", {
+            "id": report_id, "status": RP_APPROVAL_INITIATED})
+        # 发人工审批任务
+        await create_human_task(
+            sor, rep.get("project_id", ""), HT_DEV_APPROVAL,
+            "研发审批：%s" % rep.get("title", ""),
+            note or "请审批《%s》的研发立项。" % rep.get("title", ""))
+        await sor.sqlExe("COMMIT", {})
+    return True, aid
+
+
+async def resolve_approval(approval_id, result, operator=""):
+    """审批结论回流（人工决策）。result=approved/rejected。"""
+    if result not in (AP_APPROVED, AP_REJECTED):
+        return False, "result 必须是 approved 或 rejected"
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "SELECT id, report_id, status FROM opp_approvals WHERE id=${i}$",
+            {"i": approval_id})
+        await sor.sqlExe("COMMIT", {})
+        if not recs:
+            return False, "审批单不存在: %s" % approval_id
+        rec = rec_to_dict(recs[0])
+        if rec.get("status") != AP_INITIATED:
+            return False, "审批单状态 %s 无法结论" % rec.get("status")
+        report_status = RP_APPROVED if result == AP_APPROVED else RP_REJECTED
+        await sor.U("opp_approvals", {
+            "id": approval_id, "status": result,
+            "resolved_by": operator or "human"})
+        await sor.U("opp_reports", {
+            "id": rec.get("report_id"), "status": report_status})
+        # 完结人工审批任务
+        ht = await find_human_task(sor, rec.get("project_id", ""),
+                                   HT_DEV_APPROVAL, status="pending")
+        if ht:
+            await sor.U("pipeline_human_tasks", {
+                "id": ht.get("id"),
+                "status": "done" if result == AP_APPROVED else "rejected"})
+        await sor.sqlExe("COMMIT", {})
+    return True, result
+
+
+async def list_approvals(project_id="", status=""):
+    db, dbname = get_db()
+    async with db.sqlorContext(dbname) as sor:
+        sql = ("SELECT id, report_id, project_id, status, note, created_by, "
+               "created_at, resolved_at FROM opp_approvals WHERE 1=1")
+        args = {}
+        if project_id:
+            sql += " AND project_id=${pid}$"
+            args["pid"] = project_id
+        if status:
+            sql += " AND status=${st}$"
+            args["st"] = status
+        sql += " ORDER BY created_at DESC LIMIT 50"
+        recs = await sor.sqlExe(sql, args)
+        await sor.sqlExe("COMMIT", {})
+        return rows_to_dicts(recs)

pyproject.toml

@@ -0,0 +1,9 @@
+[project]
+name = "pipeline_opportunity"
+version = "0.1.0"
+description = "商机产线：热点软件推荐→每日AI/Agent招标→研发报告→人工确认→研发审批。数据来自数据爬取平台(内网HTTP)。"
+dependencies = []
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["pipeline_opportunity*"]

scripts/gen_ddl.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+# 生成 pipeline-opportunity 的 MySQL DDL（json2ddl 需要 sqlor + appPublic 在 PYTHONPATH）
+set -e
+R=/home/ymq/work/repos
+export PYTHONPATH="$R/xls2ddl:$R/sqlor:$R/apppublic"
+python3 "$R/xls2ddl/gen_opportunity_ddl.py"

scripts/load_path.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""RBAC path registration for pipeline-opportunity module.
+
+与平台惯例一致：页面/CRUD/API → logined。
+在宿主应用根目录执行（set_role_perm.py 位于宿主根）：
+  cd <APP_ROOT> && py3/bin/python pkgs/pipeline-opportunity/scripts/load_path.py
+"""
+import os
+import subprocess
+import sys
+
+MOD = "pipeline-opportunity"
+# set_role_perm.py 在宿主应用根目录（本脚本位于 <APP_ROOT>/pkgs/pipeline-opportunity/scripts/）
+APP_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+
+TABLES = [
+    "opp_reports", "opp_approvals",
+]
+
+PATHS_ANY = []
+
+PATHS_LOGINED = [
+    "/%s" % MOD,
+    "/%s/index.ui" % MOD,
+    "/%s/agent" % MOD,
+    "/%s/agent/index.ui" % MOD,
+]
+for t in TABLES:
+    PATHS_LOGINED += [
+        "/%s/%s/index.ui" % (MOD, t),
+        "/%s/%s/get_%s.dspy" % (MOD, t, t),
+        "/%s/%s/add_%s.dspy" % (MOD, t, t),
+        "/%s/%s/update_%s.dspy" % (MOD, t, t),
+        "/%s/%s/delete_%s.dspy" % (MOD, t, t),
+    ]
+
+
+def _run(role, path):
+    r = subprocess.run([sys.executable, os.path.join(APP_ROOT, "set_role_perm.py"), role, path],
+                       capture_output=True, text=True, cwd=APP_ROOT)
+    if r.returncode != 0:
+        print("  FAIL [%s] %s: %s" % (role, path, (r.stderr or "").strip()[:120]))
+    return r.returncode == 0
+
+
+def main():
+    print("=== %s RBAC registration ===" % MOD)
+    n = 0
+    for p in PATHS_ANY:
+        n += _run("any", p)
+        print("  any: %s" % p)
+    for p in PATHS_LOGINED:
+        n += _run("logined", p)
+        print("  logined: %s" % p)
+    print("Done. %d/%d paths registered" % (n, len(PATHS_ANY) + len(PATHS_LOGINED)))
+
+
+if __name__ == "__main__":
+    main()

wwwroot/opp_approvals/add_opp_approvals.dspy

@@ -0,0 +1,37 @@
+
+ns = params_kw.copy()
+for k,v in ns.items():
+	if v == 'NaN' or v == 'null':
+		ns[k] = None
+id = params_kw.id
+if not id or len(id) > 32:
+	id = uuid()
+ns['id'] = id
+
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.C('opp_approvals', ns.copy())
+	return {
+		"widgettype":"Message",
+		"options":{
+			"cwidth":16,
+			"cheight":9,
+			"title":"Add Success",
+			"timeout":3,
+			"message":"ok"
+		}
+	}
+
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Add Error",
+		"cwidth":16,
+		"cheight":9,
+		"timeout":3,
+		"message":"failed"
+	}
+}

wwwroot/opp_approvals/delete_opp_approvals.dspy

@@ -0,0 +1,33 @@
+
+ns = {
+	'id':params_kw['id'],
+}
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.D('opp_approvals', ns)
+	debug('delete success');
+	return {
+		"widgettype":"Message",
+		"options":{
+			"title":"Delete Success",
+			"timeout":3,
+			"cwidth":16,
+			"cheight":9,
+			"message":"ok"
+		}
+	}
+
+debug('Delete failed');
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Delete Error",
+		"timeout":3,
+		"cwidth":16,
+		"cheight":9,
+		"message":"failed"
+	}
+}

wwwroot/opp_approvals/get_opp_approvals.dspy

@@ -0,0 +1,112 @@
+
+ns = params_kw.copy()
+
+
+debug(f'get_opp_approvals.dspy:{ns=}')
+if not ns.get('page'):
+	ns['page'] = 1 
+if not ns.get('sort'):
+
+
+	ns['sort'] = ["created_at desc"]
+
+
+
+sql = '''select * from opp_approvals where 1=1  [[filterstr]]'''
+
+filterjson = params_kw.get('data_filter')
+if filterjson and isinstance(filterjson, str):
+	try:
+		filterjson = json.loads(filterjson)
+	except (json.JSONDecodeError, TypeError):
+		filterjson = None
+fields_str=r'''[
+    {
+        "name": "id",
+        "title": "主键ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no"
+    },
+    {
+        "name": "report_id",
+        "title": "报告ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no"
+    },
+    {
+        "name": "project_id",
+        "title": "项目ID",
+        "type": "str",
+        "length": 32
+    },
+    {
+        "name": "status",
+        "title": "审批状态",
+        "type": "str",
+        "length": 16,
+        "nullable": "no",
+        "default": "initiated"
+    },
+    {
+        "name": "note",
+        "title": "审批说明",
+        "type": "text"
+    },
+    {
+        "name": "created_by",
+        "title": "发起人",
+        "type": "str",
+        "length": 64
+    },
+    {
+        "name": "resolved_by",
+        "title": "审批人",
+        "type": "str",
+        "length": 64
+    },
+    {
+        "name": "created_at",
+        "title": "创建时间",
+        "type": "timestamp",
+        "nullable": "no"
+    },
+    {
+        "name": "resolved_at",
+        "title": "结论时间",
+        "type": "timestamp"
+    }
+]'''
+ori_fields = json.loads(fields_str)
+if not filterjson:
+	fields = [ f['name'] for f in ori_fields ]
+	filterjson = default_filterjson(fields, ns)
+
+filterdic = ns.copy()
+filterdic['filterstr'] = ''
+filterdic['userorgid'] = '${userorgid}$'
+filterdic['userid'] = '${userid}$'
+if filterjson:
+	dbf = DBFilter(filterjson)
+	conds = dbf.gen(ns)
+	if conds:
+		ns.update(dbf.consts)
+		conds = f' and {conds}'
+		filterdic['filterstr'] = conds
+ac = ArgsConvert('[[', ']]')
+vars = ac.findAllVariables(sql)
+NameSpace = {v:'${' + v + '}$' for v in vars if v != 'filterstr' }
+filterdic.update(NameSpace)
+sql = ac.convert(sql, filterdic)
+
+debug(f'{sql=}')
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.sqlPaging(sql, ns) 
+	return r
+return {
+	"total":0,
+	"rows":[]
+}

wwwroot/opp_approvals/index.ui

@@ -0,0 +1,176 @@
+
+{
+	"widgettype":"VBox",
+	"options":{"cheight":40,"width":"100%"},
+	"subwidgets":[{
+	"id":"opp_approvals_tbl",
+	"widgettype":"Tabular",
+	"options":{
+		"width":"100%",
+		"height":"100%",
+
+
+		"title":"研发审批",
+
+
+
+
+		"css":"card",
+
+
+		"editable":{
+
+			"new_data_url":"{{entire_url('add_opp_approvals.dspy')}}",
+
+
+			"delete_data_url":"{{entire_url('delete_opp_approvals.dspy')}}",
+
+
+			"update_data_url":"{{entire_url('update_opp_approvals.dspy')}}"
+
+		},
+
+
+		"data_url":"{{entire_url('./get_opp_approvals.dspy')}}",
+
+		"data_method":"GET",
+		"data_params":{{json.dumps(params_kw, indent=4, ensure_ascii=False)}},
+		"row_options":{
+
+
+
+			"browserfields": {
+    "exclouded": [
+        "note"
+    ]
+},
+
+
+			"editexclouded":[
+    "id",
+    "created_at",
+    "resolved_at",
+    "resolved_by",
+    "report_id"
+],
+
+			"fields":[
+    {
+        "name": "id",
+        "title": "主键ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "主键ID"
+    },
+    {
+        "name": "report_id",
+        "title": "报告ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "报告ID"
+    },
+    {
+        "name": "project_id",
+        "title": "项目ID",
+        "type": "str",
+        "length": 32,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "项目ID"
+    },
+    {
+        "name": "status",
+        "title": "审批状态",
+        "type": "str",
+        "length": 16,
+        "nullable": "no",
+        "default": "initiated",
+        "cwidth": 16,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "审批状态"
+    },
+    {
+        "name": "note",
+        "title": "审批说明",
+        "type": "text",
+        "length": 0,
+        "uitype": "text",
+        "datatype": "text",
+        "label": "审批说明"
+    },
+    {
+        "name": "created_by",
+        "title": "发起人",
+        "type": "str",
+        "length": 64,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "发起人"
+    },
+    {
+        "name": "resolved_by",
+        "title": "审批人",
+        "type": "str",
+        "length": 64,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "审批人"
+    },
+    {
+        "name": "created_at",
+        "title": "创建时间",
+        "type": "timestamp",
+        "nullable": "no",
+        "length": 0,
+        "uitype": "str",
+        "datatype": "timestamp",
+        "label": "创建时间"
+    },
+    {
+        "name": "resolved_at",
+        "title": "结论时间",
+        "type": "timestamp",
+        "length": 0,
+        "uitype": "str",
+        "datatype": "timestamp",
+        "label": "结论时间"
+    }
+]
+		},  
+
+
+
+		"data_filter":{
+    "AND": [
+        {
+            "field": "status",
+            "op": "=",
+            "var": "status_input"
+        }
+    ]
+},
+
+
+
+
+
+		"page_rows":160,
+		"cache_limit":5
+	}
+
+	,"binds":[]
+
+}]
+}

wwwroot/opp_approvals/update_opp_approvals.dspy

@@ -0,0 +1,36 @@
+
+ns = params_kw.copy()
+for k,v in ns.items():
+	if v == 'NaN' or v == 'null':
+		ns[k] = None
+
+
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+
+	r = await sor.U('opp_approvals', ns)
+	debug('update success');
+	return {
+		"widgettype":"Message",
+		"options":{
+			"title":"Update Success",
+			"cwidth":16,
+			"cheight":9,
+			"timeout":3,
+			"message":"ok"
+		} 
+	}
+		
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Update Error",
+		"cwidth":16,
+		"cheight":9,
+		"timeout":3,
+		"message":"failed"
+	}
+}

wwwroot/opp_reports/add_opp_reports.dspy

@@ -0,0 +1,37 @@
+
+ns = params_kw.copy()
+for k,v in ns.items():
+	if v == 'NaN' or v == 'null':
+		ns[k] = None
+id = params_kw.id
+if not id or len(id) > 32:
+	id = uuid()
+ns['id'] = id
+
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.C('opp_reports', ns.copy())
+	return {
+		"widgettype":"Message",
+		"options":{
+			"cwidth":16,
+			"cheight":9,
+			"title":"Add Success",
+			"timeout":3,
+			"message":"ok"
+		}
+	}
+
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Add Error",
+		"cwidth":16,
+		"cheight":9,
+		"timeout":3,
+		"message":"failed"
+	}
+}

wwwroot/opp_reports/delete_opp_reports.dspy

@@ -0,0 +1,33 @@
+
+ns = {
+	'id':params_kw['id'],
+}
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.D('opp_reports', ns)
+	debug('delete success');
+	return {
+		"widgettype":"Message",
+		"options":{
+			"title":"Delete Success",
+			"timeout":3,
+			"cwidth":16,
+			"cheight":9,
+			"message":"ok"
+		}
+	}
+
+debug('Delete failed');
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Delete Error",
+		"timeout":3,
+		"cwidth":16,
+		"cheight":9,
+		"message":"failed"
+	}
+}

wwwroot/opp_reports/get_opp_reports.dspy

@@ -0,0 +1,125 @@
+
+ns = params_kw.copy()
+
+
+debug(f'get_opp_reports.dspy:{ns=}')
+if not ns.get('page'):
+	ns['page'] = 1 
+if not ns.get('sort'):
+
+
+	ns['sort'] = ["created_at desc"]
+
+
+
+sql = '''select * from opp_reports where 1=1  [[filterstr]]'''
+
+filterjson = params_kw.get('data_filter')
+if filterjson and isinstance(filterjson, str):
+	try:
+		filterjson = json.loads(filterjson)
+	except (json.JSONDecodeError, TypeError):
+		filterjson = None
+fields_str=r'''[
+    {
+        "name": "id",
+        "title": "主键ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no"
+    },
+    {
+        "name": "project_id",
+        "title": "项目ID",
+        "type": "str",
+        "length": 32
+    },
+    {
+        "name": "software",
+        "title": "软件/主题",
+        "type": "str",
+        "length": 128,
+        "nullable": "no"
+    },
+    {
+        "name": "title",
+        "title": "报告标题",
+        "type": "str",
+        "length": 255,
+        "nullable": "no"
+    },
+    {
+        "name": "content",
+        "title": "报告正文",
+        "type": "text"
+    },
+    {
+        "name": "status",
+        "title": "报告状态",
+        "type": "str",
+        "length": 24,
+        "nullable": "no",
+        "default": "draft"
+    },
+    {
+        "name": "confirm_task_id",
+        "title": "确认任务ID",
+        "type": "str",
+        "length": 32
+    },
+    {
+        "name": "confirmed_by",
+        "title": "确认人",
+        "type": "str",
+        "length": 64
+    },
+    {
+        "name": "created_by",
+        "title": "创建人",
+        "type": "str",
+        "length": 64
+    },
+    {
+        "name": "created_at",
+        "title": "创建时间",
+        "type": "timestamp",
+        "nullable": "no"
+    },
+    {
+        "name": "updated_at",
+        "title": "更新时间",
+        "type": "timestamp"
+    }
+]'''
+ori_fields = json.loads(fields_str)
+if not filterjson:
+	fields = [ f['name'] for f in ori_fields ]
+	filterjson = default_filterjson(fields, ns)
+
+filterdic = ns.copy()
+filterdic['filterstr'] = ''
+filterdic['userorgid'] = '${userorgid}$'
+filterdic['userid'] = '${userid}$'
+if filterjson:
+	dbf = DBFilter(filterjson)
+	conds = dbf.gen(ns)
+	if conds:
+		ns.update(dbf.consts)
+		conds = f' and {conds}'
+		filterdic['filterstr'] = conds
+ac = ArgsConvert('[[', ']]')
+vars = ac.findAllVariables(sql)
+NameSpace = {v:'${' + v + '}$' for v in vars if v != 'filterstr' }
+filterdic.update(NameSpace)
+sql = ac.convert(sql, filterdic)
+
+debug(f'{sql=}')
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+	r = await sor.sqlPaging(sql, ns) 
+	return r
+return {
+	"total":0,
+	"rows":[]
+}

wwwroot/opp_reports/index.ui

@@ -0,0 +1,208 @@
+
+{
+	"widgettype":"VBox",
+	"options":{"cheight":40,"width":"100%"},
+	"subwidgets":[{
+	"id":"opp_reports_tbl",
+	"widgettype":"Tabular",
+	"options":{
+		"width":"100%",
+		"height":"100%",
+
+
+		"title":"研发报告",
+
+
+
+
+		"css":"card",
+
+
+		"editable":{
+
+			"new_data_url":"{{entire_url('add_opp_reports.dspy')}}",
+
+
+			"delete_data_url":"{{entire_url('delete_opp_reports.dspy')}}",
+
+
+			"update_data_url":"{{entire_url('update_opp_reports.dspy')}}"
+
+		},
+
+
+		"data_url":"{{entire_url('./get_opp_reports.dspy')}}",
+
+		"data_method":"GET",
+		"data_params":{{json.dumps(params_kw, indent=4, ensure_ascii=False)}},
+		"row_options":{
+
+
+
+			"browserfields": {
+    "exclouded": [
+        "content",
+        "confirm_task_id"
+    ]
+},
+
+
+			"editexclouded":[
+    "id",
+    "created_at",
+    "updated_at",
+    "confirm_task_id",
+    "confirmed_by"
+],
+
+			"fields":[
+    {
+        "name": "id",
+        "title": "主键ID",
+        "type": "str",
+        "length": 32,
+        "nullable": "no",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "主键ID"
+    },
+    {
+        "name": "project_id",
+        "title": "项目ID",
+        "type": "str",
+        "length": 32,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "项目ID"
+    },
+    {
+        "name": "software",
+        "title": "软件/主题",
+        "type": "str",
+        "length": 128,
+        "nullable": "no",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "软件/主题"
+    },
+    {
+        "name": "title",
+        "title": "报告标题",
+        "type": "str",
+        "length": 255,
+        "nullable": "no",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "报告标题"
+    },
+    {
+        "name": "content",
+        "title": "报告正文",
+        "type": "text",
+        "length": 0,
+        "uitype": "text",
+        "datatype": "text",
+        "label": "报告正文"
+    },
+    {
+        "name": "status",
+        "title": "报告状态",
+        "type": "str",
+        "length": 24,
+        "nullable": "no",
+        "default": "draft",
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "报告状态"
+    },
+    {
+        "name": "confirm_task_id",
+        "title": "确认任务ID",
+        "type": "str",
+        "length": 32,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "确认任务ID"
+    },
+    {
+        "name": "confirmed_by",
+        "title": "确认人",
+        "type": "str",
+        "length": 64,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "确认人"
+    },
+    {
+        "name": "created_by",
+        "title": "创建人",
+        "type": "str",
+        "length": 64,
+        "cwidth": 18,
+        "uitype": "str",
+        "datatype": "str",
+        "label": "创建人"
+    },
+    {
+        "name": "created_at",
+        "title": "创建时间",
+        "type": "timestamp",
+        "nullable": "no",
+        "length": 0,
+        "uitype": "str",
+        "datatype": "timestamp",
+        "label": "创建时间"
+    },
+    {
+        "name": "updated_at",
+        "title": "更新时间",
+        "type": "timestamp",
+        "length": 0,
+        "uitype": "str",
+        "datatype": "timestamp",
+        "label": "更新时间"
+    }
+]
+		},  
+
+
+
+		"data_filter":{
+    "AND": [
+        {
+            "field": "title",
+            "op": "LIKE",
+            "var": "title_input"
+        },
+        {
+            "field": "software",
+            "op": "LIKE",
+            "var": "software_input"
+        },
+        {
+            "field": "status",
+            "op": "=",
+            "var": "status_input"
+        }
+    ]
+},
+
+
+
+
+
+		"page_rows":160,
+		"cache_limit":5
+	}
+
+	,"binds":[]
+
+}]
+}

wwwroot/opp_reports/update_opp_reports.dspy

@@ -0,0 +1,36 @@
+
+ns = params_kw.copy()
+for k,v in ns.items():
+	if v == 'NaN' or v == 'null':
+		ns[k] = None
+
+
+
+
+db = DBPools()
+dbname = get_module_dbname('pipeline_opportunity')
+async with db.sqlorContext(dbname) as sor:
+
+	r = await sor.U('opp_reports', ns)
+	debug('update success');
+	return {
+		"widgettype":"Message",
+		"options":{
+			"title":"Update Success",
+			"cwidth":16,
+			"cheight":9,
+			"timeout":3,
+			"message":"ok"
+		} 
+	}
+		
+return {
+	"widgettype":"Error",
+	"options":{
+		"title":"Update Error",
+		"cwidth":16,
+		"cheight":9,
+		"timeout":3,
+		"message":"failed"
+	}
+}