From 8b7626dc8365e489aa6dae2bef2d3bbe6a2b5119 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 12 Sep 2026 09:02:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(mining):=20=E6=8C=96=E6=8E=98=E6=89=B9?= =?UTF-8?q?=E6=AC=A1/=E7=B1=BB=E5=88=AB=E8=90=BD=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E5=BD=92=E5=B1=9E(=E7=94=A8=E6=88=B7=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?=E5=95=86=E6=9C=BA=E4=BA=A7=E7=BA=BF=E5=88=86=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?)=E2=80=94=E2=80=94opp=5Fmining=5Fbatches/opp=5Fclusters?= =?UTF-8?q?=E5=8A=A0project=5Fid(=E7=A9=BA=3D=E5=B9=B3=E5=8F=B0=E7=BA=A7?= =?UTF-8?q?=E6=8C=96=E6=8E=98),=E5=8F=AF=E8=A7=81=E6=80=A7=E5=AF=B9?= =?UTF-8?q?=E9=BD=90opp=5Freports=E5=8F=A3=E5=BE=84(=E6=8C=82=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E4=BB=85=E6=9C=AC=E9=A1=B9=E7=9B=AE=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E5=8F=AF=E8=A7=81/=E5=B9=B3=E5=8F=B0=E7=BA=A7=E6=9C=AC?= =?UTF-8?q?=E6=9C=BA=E6=9E=84=E5=8F=AF=E8=A7=81/=E6=97=A0=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E5=8F=AA=E8=A7=81=E5=B9=B3=E5=8F=B0=E7=BA=A7),=5Fchec?= =?UTF-8?q?k=5Fbatch=5Fvisible=E5=8D=95=E4=B8=80=E8=A7=84=E5=88=99?= =?UTF-8?q?=E6=BA=90status/list=5Fclusters/cluster=5Fdetail=E5=A4=8D?= =?UTF-8?q?=E7=94=A8;=E8=A1=A5=E5=88=97=E8=84=9A=E6=9C=ACinformation=5Fsch?= =?UTF-8?q?ema=E5=AE=88=E5=8D=AB=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- models/opp_clusters.json | 6 +++ models/opp_mining_batches.json | 6 +++ mysql.ddl.sql | 2 + pipeline_opportunity/opp_mining_flow.py | 62 +++++++++++++++++++------ scripts/p1_add_project_id.py | 47 +++++++++++++++++++ 5 files changed, 109 insertions(+), 14 deletions(-) create mode 100644 scripts/p1_add_project_id.py diff --git a/models/opp_clusters.json b/models/opp_clusters.json index c4b1251..deba5dc 100644 --- a/models/opp_clusters.json +++ b/models/opp_clusters.json @@ -31,6 +31,12 @@ "length": 32, "nullable": "no" }, + { + "name": "project_id", + "title": "项目ID(空=平台级挖掘)", + "type": "str", + "length": 32 + }, { "name": "name", "title": "类别名称(LLM命名)", diff --git a/models/opp_mining_batches.json b/models/opp_mining_batches.json index f69e061..5984db2 100644 --- a/models/opp_mining_batches.json +++ b/models/opp_mining_batches.json @@ -24,6 +24,12 @@ "length": 32, "nullable": "no" }, + { + "name": "project_id", + "title": "项目ID(空=平台级挖掘)", + "type": "str", + "length": 32 + }, { "name": "scope", "title": "挖掘范围(top=TopX全量/targeted=指定类型)", diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 49be3d3..8c12a72 100644 --- a/mysql.ddl.sql +++ b/mysql.ddl.sql @@ -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)', diff --git a/pipeline_opportunity/opp_mining_flow.py b/pipeline_opportunity/opp_mining_flow.py index 4d4f108..112791f 100644 --- a/pipeline_opportunity/opp_mining_flow.py +++ b/pipeline_opportunity/opp_mining_flow.py @@ -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}$", diff --git a/scripts/p1_add_project_id.py b/scripts/p1_add_project_id.py new file mode 100644 index 0000000..2702538 --- /dev/null +++ b/scripts/p1_add_project_id.py @@ -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())