yumoqing/pipeline-opportunity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mining): 挖掘批次/类别落项目归属(用户确认商机产线分项目)——opp_mining_batches/opp_clusters加project_id(空=平台级挖掘),可见性对齐opp_reports口径(挂项目仅本项目会话可见/平台级本机构可见/无项目只见平台级),_check_batch_visible单一规则源status/list_clusters/cluster_detail复用;补列脚本information_schema守卫幂等

Browse Source
This commit is contained in:
yumoqing 2026-09-12 09:02:04 +08:00
parent 058c5f5235
commit 8b7626dc83
5 changed files with 109 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
models/opp_clusters.json
View File

@ -31,6 +31,12 @@
"length": 32,
"nullable": "no"
},
{
"name": "project_id",
"title": "项目ID(空=平台级挖掘)",
"type": "str",
"length": 32
},
{
"name": "name",
"title": "类别名称(LLM命名)",

6
models/opp_mining_batches.json
View File

@ -24,6 +24,12 @@
"length": 32,
"nullable": "no"
},
{
"name": "project_id",
"title": "项目ID(空=平台级挖掘)",
"type": "str",
"length": 32
},
{
"name": "scope",
"title": "挖掘范围(top=TopX全量/targeted=指定类型)",

2
mysql.ddl.sql
View File

@ -85,6 +85,7 @@ CREATE TABLE opp_clusters
`id` VARCHAR(32) NOT NULL comment '主键ID',
`batch_id` VARCHAR(32) NOT NULL comment '批次ID',
`org_id` VARCHAR(32) NOT NULL comment '机构ID',
`project_id` VARCHAR(32) comment '项目ID(空=平台级挖掘)',
`name` VARCHAR(128) comment '类别名称(LLM命名)',
`doc_count` int NOT NULL DEFAULT '0' comment '类内需求数',
`share` double(6,4) DEFAULT '0' comment '占比',
@ -119,6 +120,7 @@ CREATE TABLE opp_mining_batches
`id` VARCHAR(32) NOT NULL comment '主键ID',
`org_id` VARCHAR(32) NOT NULL comment '机构ID',
`project_id` VARCHAR(32) comment '项目ID(空=平台级挖掘)',
`scope` VARCHAR(16) NOT NULL DEFAULT 'top' comment '挖掘范围(top=TopX全量/targeted=指定类型)',
`params_json` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci comment '批次参数(days/sources/record_type/keyword/category)',
`status` VARCHAR(24) NOT NULL DEFAULT 'pulling' comment '批次状态(pulling/embedding/clustering/naming/done/failed)',

62
pipeline_opportunity/opp_mining_flow.py
View File

@ -49,7 +49,9 @@ async def start_mining(sor, ctx, scope="top", days=365, sources="", keyword="",
params_json = json.dumps({"scope": scope, "days": days, "sources": sources,
"keyword": keyword, "category": category}, ensure_ascii=False)
await sor.C("opp_mining_batches", {
"id": bid, "org_id": org_id, "scope": scope, "params_json": params_json,
"id": bid, "org_id": org_id,
"project_id": ctx.get("project_id") or "", # 项目归属（空=平台级挖掘，对齐报告口径）
"scope": scope, "params_json": params_json,
"status": "pulling", "vdb_col": "", "stats_json": "", "error_msg": "",
"created_by": created_by or ctx.get("user_id") or "",
})
@ -147,13 +149,14 @@ async def _run_mining_inner(sor, batch_id, ctx):
# ── 4. naming（LLM utility + 规则兜底）──
named = await name_clusters(sor, batch, clusters, vecs, text_map, cfg, ctx)
org_id = ctx.get("org_id") or ""
proj_id = str(batch.get("project_id") or "") # 簇归属跟随批次（写入时已落库）
# vid → snap_id 反查
vid2snap = {v: s for s, v in embed_res["id_map"].items()}
for cl in named:
cid = new_id()
cent_snap = vid2snap.get(cl["centroid_vid"], "")
await sor.C("opp_clusters", {
"id": cid, "batch_id": batch_id, "org_id": org_id,
"id": cid, "batch_id": batch_id, "org_id": org_id, "project_id": proj_id,
"name": cl["name"][:128], "doc_count": cl["size"], "share": cl["share"],
"heat_rank": cl["rank"], "naming_evidence": json.dumps(
{"samples": cl["samples"], "by": cl["evidence"]}, ensure_ascii=False),
@ -198,34 +201,62 @@ async def _cleanup_old_collections(sor, base, org_id, cfg):
async def mining_status(sor, ctx, batch_id=""):
"""查批次状态（org 隔离）。batch_id 空 = 本机构最近批次列表。"""
"""查批次状态（隔离：本机构 + 本项目/平台级）。batch_id 空 = 可见批次列表。
可见性口径与 opp_reports 对齐（2026-09-12 用户确认挖掘也分项目）：
- 挂项目的批次：仅当前会话项目内可见（project_id 相等）
- 平台级批次（project_id 空）：本机构登录可见
- 会话无当前项目时：只见平台级批次
"""
org_id = ctx.get("org_id") or ""
proj_id = ctx.get("project_id") or ""
if batch_id:
recs = await sor.sqlExe(
"SELECT id, scope, status, stats_json, error_msg, vdb_col, created_at "
"SELECT id, scope, status, project_id, stats_json, error_msg, vdb_col, created_at "
"FROM opp_mining_batches WHERE id=${b}$ AND org_id=${o}$",
{"b": batch_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return False, "批次不存在或不属于当前机构"
return True, rows_to_dicts(recs, limit=1)[0]
row = rows_to_dicts(recs, limit=1)[0]
bproj = str(row.get("project_id") or "")
if bproj and bproj != proj_id:
return False, "批次属于其他项目，当前会话不可见"
return True, row
# 列表：本项目批次 + 平台级批次
recs = await sor.sqlExe(
"SELECT id, scope, status, stats_json, error_msg, created_at "
"FROM opp_mining_batches WHERE org_id=${o}$ ORDER BY created_at DESC LIMIT 10",
{"o": org_id})
"SELECT id, scope, status, project_id, stats_json, error_msg, created_at "
"FROM opp_mining_batches WHERE org_id=${o}$ "
"AND (project_id='' OR project_id=${p}$) "
"ORDER BY created_at DESC LIMIT 10",
{"o": org_id, "p": proj_id})
await sor.sqlExe("COMMIT", {})
return True, rows_to_dicts(recs, limit=10)
async def list_clusters(sor, ctx, batch_id):
"""类别排名表（org 隔离：批次必须属于本机构）。"""
async def _check_batch_visible(sor, ctx, batch_id):
"""批次可见性单一规则源（status/list_clusters 复用，语义不漂移）。
返回 (ok, err)。"""
org_id = ctx.get("org_id") or ""
proj_id = ctx.get("project_id") or ""
recs = await sor.sqlExe(
"SELECT id FROM opp_mining_batches WHERE id=${b}$ AND org_id=${o}$",
"SELECT id, project_id FROM opp_mining_batches WHERE id=${b}$ AND org_id=${o}$",
{"b": batch_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return False, "批次不存在或不属于当前机构"
bproj = str(getattr(recs[0], "project_id", "") or "")
if bproj and bproj != proj_id:
return False, "批次属于其他项目，当前会话不可见"
return True, ""
async def list_clusters(sor, ctx, batch_id):
"""类别排名表（隔离：批次必须本机构+本项目/平台级可见）。"""
ok, err = await _check_batch_visible(sor, ctx, batch_id)
if not ok:
return False, err
org_id = ctx.get("org_id") or ""
recs = await sor.sqlExe(
"SELECT id, name, doc_count, share, heat_rank, naming_evidence, centroid_snap_id "
"FROM opp_clusters WHERE batch_id=${b}$ AND org_id=${o}$ ORDER BY heat_rank ASC",
@ -235,16 +266,19 @@ async def list_clusters(sor, ctx, batch_id):
async def cluster_detail(sor, ctx, cluster_id, limit=20):
"""类内需求明细（样例，带来源 URL 可查证）。"""
"""类内需求明细（样例，带来源 URL 可查证；经批次可见性校验）。"""
org_id = ctx.get("org_id") or ""
recs = await sor.sqlExe(
"SELECT id, name, doc_count, share, heat_rank, naming_evidence "
"FROM opp_clusters WHERE id=${c}$ AND org_id=${o}$",
"SELECT c.id, c.name, c.doc_count, c.share, c.heat_rank, c.naming_evidence, c.batch_id "
"FROM opp_clusters c WHERE c.id=${c}$ AND c.org_id=${o}$",
{"c": cluster_id, "o": org_id})
await sor.sqlExe("COMMIT", {})
if not recs:
return False, "类别不存在或不属于当前机构"
cl = rows_to_dicts(recs, limit=1)[0]
ok, err = await _check_batch_visible(sor, ctx, cl.get("batch_id") or "")
if not ok:
return False, err
items = rows_to_dicts(await sor.sqlExe(
"SELECT id, title, source, budget_wan, url, publish_time FROM opp_demand_snap "
"WHERE cluster_id=${c}$ ORDER BY budget_wan DESC LIMIT ${n}$",

47
scripts/p1_add_project_id.py Normal file
View File

@ -0,0 +1,47 @@
# -*- coding:utf-8 -*-
"""幂等补列：opp_mining_batches / opp_clusters 加 project_id（information_schema 守卫，绝不 DROP）。"""
import asyncio, os, sys
W = "/d/pipeline/pipeline-app"; os.chdir(W); sys.path.insert(0, W)
from appPublic.folderUtils import ProgramPath
from appPublic.jsonConfig import getConfig
from appPublic.event_dispatcher import EventDispatcher
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
p = ProgramPath(); c = getConfig(W, {"workdir": W, "ProgramPath": p}); DBPools(c.databases)
se = ServerEnv(); se.event_dispatcher = EventDispatcher(); se.get_module_dbname = lambda m: "pipeline"
TARGETS = [("opp_mining_batches", "project_id"), ("opp_clusters", "project_id")]
async def main():
async with DBPools().sqlorContext("pipeline") as sor:
for table, col in TARGETS:
chk = await sor.sqlExe(
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
"WHERE table_schema=DATABASE() AND table_name=${t}$ AND column_name=${c}$",
{"t": table, "c": col})
await sor.sqlExe("COMMIT", {})
if chk:
print("SKIP(已存在): %s.%s" % (table, col))
continue
await sor.sqlExe(
"ALTER TABLE %s ADD COLUMN `%s` VARCHAR(32) DEFAULT '' "
"COMMENT '项目ID(空=平台级挖掘)'" % (table, col), {})
await sor.sqlExe("COMMIT", {})
print("ADDED: %s.%s" % (table, col))
# 索引：批次按 org+project 查询
try:
await sor.sqlExe("CREATE INDEX idx_opp_batches_orgproj ON opp_mining_batches (org_id, project_id)", {})
await sor.sqlExe("COMMIT", {})
print("index added")
except Exception as e:
print("index skip:", str(e)[:60])
# 验证
for table, col in TARGETS:
chk = await sor.sqlExe(
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
"WHERE table_schema=DATABASE() AND table_name=${t}$ AND column_name=${c}$",
{"t": table, "c": col})
await sor.sqlExe("COMMIT", {})
print("VERIFY %s.%s:" % (table, col), "OK" if chk else "MISSING")
asyncio.run(main())