commit b5cf9599f42ed80df5e07cd32d7149e1c6e5f1fc Author: yumoqing Date: Fri Aug 28 17:00:58 2026 +0800 feat: 商机产线能力包——招标数据(爬取平台内网HTTP)→热点推荐→研发报告→人工确认→研发审批 diff --git a/build.sh b/build.sh new file mode 100644 index 0000000..31bf63a --- /dev/null +++ b/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）。" diff --git a/init/data.json b/init/data.json new file mode 100644 index 0000000..1f61780 --- /dev/null +++ b/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": "审批驳回"} + ] + } + ] +} diff --git a/json/build.sh b/json/build.sh new file mode 100644 index 0000000..a1d95a9 --- /dev/null +++ b/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 diff --git a/json/opp_approvals.json b/json/opp_approvals.json new file mode 100644 index 0000000..a1ce9c6 --- /dev/null +++ b/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" + } + ] + } + } +} diff --git a/json/opp_reports.json b/json/opp_reports.json new file mode 100644 index 0000000..bb7f442 --- /dev/null +++ b/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" + } + ] + } + } +} diff --git a/models/opp_approvals.json b/models/opp_approvals.json new file mode 100644 index 0000000..4b9caeb --- /dev/null +++ b/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" + } + ] +} diff --git a/models/opp_reports.json b/models/opp_reports.json new file mode 100644 index 0000000..c9a77a3 --- /dev/null +++ b/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" + } + ] +} diff --git a/mysql.ddl.sql b/mysql.ddl.sql new file mode 100644 index 0000000..d92c1f6 --- /dev/null +++ b/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 '研发审批' +; + diff --git a/pipeline_opportunity/__init__.py b/pipeline_opportunity/__init__.py new file mode 100644 index 0000000..b93bf1e --- /dev/null +++ b/pipeline_opportunity/__init__.py @@ -0,0 +1 @@ +from .init import load_pipeline_opportunity # noqa: F401 diff --git a/pipeline_opportunity/__pycache__/__init__.cpython-310.pyc b/pipeline_opportunity/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..a116116 Binary files /dev/null and b/pipeline_opportunity/__pycache__/__init__.cpython-310.pyc differ diff --git a/pipeline_opportunity/__pycache__/init.cpython-310.pyc b/pipeline_opportunity/__pycache__/init.cpython-310.pyc new file mode 100644 index 0000000..60d2eb1 Binary files /dev/null and b/pipeline_opportunity/__pycache__/init.cpython-310.pyc differ diff --git a/pipeline_opportunity/__pycache__/opp_ability.cpython-310.pyc b/pipeline_opportunity/__pycache__/opp_ability.cpython-310.pyc new file mode 100644 index 0000000..4ef77fc Binary files /dev/null and b/pipeline_opportunity/__pycache__/opp_ability.cpython-310.pyc differ diff --git a/pipeline_opportunity/__pycache__/opp_common.cpython-310.pyc b/pipeline_opportunity/__pycache__/opp_common.cpython-310.pyc new file mode 100644 index 0000000..1679ebf Binary files /dev/null and b/pipeline_opportunity/__pycache__/opp_common.cpython-310.pyc differ diff --git a/pipeline_opportunity/__pycache__/opp_data_capability.cpython-310.pyc b/pipeline_opportunity/__pycache__/opp_data_capability.cpython-310.pyc new file mode 100644 index 0000000..786f667 Binary files /dev/null and b/pipeline_opportunity/__pycache__/opp_data_capability.cpython-310.pyc differ diff --git a/pipeline_opportunity/__pycache__/opp_report_capability.cpython-310.pyc b/pipeline_opportunity/__pycache__/opp_report_capability.cpython-310.pyc new file mode 100644 index 0000000..b050d1b Binary files /dev/null and b/pipeline_opportunity/__pycache__/opp_report_capability.cpython-310.pyc differ diff --git a/pipeline_opportunity/init.py b/pipeline_opportunity/init.py new file mode 100644 index 0000000..1b3d4b4 --- /dev/null +++ b/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 diff --git a/pipeline_opportunity/opp_ability.py b/pipeline_opportunity/opp_ability.py new file mode 100644 index 0000000..b806b80 --- /dev/null +++ b/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() diff --git a/pipeline_opportunity/opp_common.py b/pipeline_opportunity/opp_common.py new file mode 100644 index 0000000..d335a5c --- /dev/null +++ b/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, ()) diff --git a/pipeline_opportunity/opp_data_capability.py b/pipeline_opportunity/opp_data_capability.py new file mode 100644 index 0000000..c2b6786 --- /dev/null +++ b/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") diff --git a/pipeline_opportunity/opp_report_capability.py b/pipeline_opportunity/opp_report_capability.py new file mode 100644 index 0000000..948f175 --- /dev/null +++ b/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) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..94770bd --- /dev/null +++ b/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*"] diff --git a/scripts/__pycache__/load_path.cpython-310.pyc b/scripts/__pycache__/load_path.cpython-310.pyc new file mode 100644 index 0000000..3f6fb8e Binary files /dev/null and b/scripts/__pycache__/load_path.cpython-310.pyc differ diff --git a/scripts/gen_ddl.sh b/scripts/gen_ddl.sh new file mode 100644 index 0000000..408ffc7 --- /dev/null +++ b/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" diff --git a/scripts/load_path.py b/scripts/load_path.py new file mode 100644 index 0000000..5fede5b --- /dev/null +++ b/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 && py3/bin/python pkgs/pipeline-opportunity/scripts/load_path.py +""" +import os +import subprocess +import sys + +MOD = "pipeline-opportunity" +# set_role_perm.py 在宿主应用根目录（本脚本位于 /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() diff --git a/wwwroot/opp_approvals/add_opp_approvals.dspy b/wwwroot/opp_approvals/add_opp_approvals.dspy new file mode 100644 index 0000000..e384b12 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/wwwroot/opp_approvals/delete_opp_approvals.dspy b/wwwroot/opp_approvals/delete_opp_approvals.dspy new file mode 100644 index 0000000..4e451f8 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/wwwroot/opp_approvals/get_opp_approvals.dspy b/wwwroot/opp_approvals/get_opp_approvals.dspy new file mode 100644 index 0000000..22ad591 --- /dev/null +++ b/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":[] +} \ No newline at end of file diff --git a/wwwroot/opp_approvals/index.ui b/wwwroot/opp_approvals/index.ui new file mode 100644 index 0000000..580e33b --- /dev/null +++ b/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":[] + +}] +} \ No newline at end of file diff --git a/wwwroot/opp_approvals/update_opp_approvals.dspy b/wwwroot/opp_approvals/update_opp_approvals.dspy new file mode 100644 index 0000000..8270447 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/wwwroot/opp_reports/add_opp_reports.dspy b/wwwroot/opp_reports/add_opp_reports.dspy new file mode 100644 index 0000000..c797c3f --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/wwwroot/opp_reports/delete_opp_reports.dspy b/wwwroot/opp_reports/delete_opp_reports.dspy new file mode 100644 index 0000000..75eb0ee --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/wwwroot/opp_reports/get_opp_reports.dspy b/wwwroot/opp_reports/get_opp_reports.dspy new file mode 100644 index 0000000..79dbb8c --- /dev/null +++ b/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":[] +} \ No newline at end of file diff --git a/wwwroot/opp_reports/index.ui b/wwwroot/opp_reports/index.ui new file mode 100644 index 0000000..2678852 --- /dev/null +++ b/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":[] + +}] +} \ No newline at end of file diff --git a/wwwroot/opp_reports/update_opp_reports.dspy b/wwwroot/opp_reports/update_opp_reports.dspy new file mode 100644 index 0000000..ae87c86 --- /dev/null +++ b/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" + } +} \ No newline at end of file