"""商机产线能力包（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()