feat(crm): 商机域接入数据权限过滤（PR3 Task 2.3）

主表列表/统计/看板/漏斗按 owner_user_id 过滤，CUD 与跟进/产品子表校验父商机 scope。

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/crm.py

@@ -2204,12 +2204,14 @@ async def opportunity_list(
     """商机分页列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         list_body = body or OpportunityListBody()
         user_id = _user.get("user_id", "") or _user.get("id", "")
         rows, total = list_opportunities(
             db,
             tenant_no,
             user_id,
+            user_info=user_info,
             **_opp_list_kwargs(list_body, user_id),
         )
         return success_response({
@@ -2235,12 +2237,14 @@ async def opportunity_mine_list(
     """我的商机列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         list_body = body or OpportunityListBody()
         user_id = _user.get("user_id", "") or _user.get("id", "")
         rows, total = list_opportunities(
             db,
             tenant_no,
             user_id,
+            user_info=user_info,
             **_opp_list_kwargs(list_body, user_id, view_type="mine"),
         )
         return success_response({
@@ -2266,6 +2270,17 @@ async def opportunity_detail(
     """商机详情"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
+        from app.models.crm_models import CrmOpportunity
+
+        row = db.query(CrmOpportunity).filter(
+            CrmOpportunity.id == body.id,
+            CrmOpportunity.del_flag == 0,
+            CrmOpportunity.tenant_no == tenant_no,
+        ).first()
+        if not row:
+            return error_response(404, "商机不存在")
+        ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该商机")
         detail = get_opportunity(db, body.id, tenant_no=tenant_no)
         if not detail:
             return error_response(404, "商机不存在")
@@ -2285,6 +2300,7 @@ async def opportunity_add(
     """新增商机"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = create_opportunity(
             db,
@@ -2308,10 +2324,11 @@ async def opportunity_update(
     """编辑商机"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         data = body.model_dump(by_alias=False, exclude_unset=True)
         oid = data.pop("id")
-        row = update_opportunity(db, oid, data, user_id=user_id, tenant_no=tenant_no)
+        row = update_opportunity(db, oid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
         return success_response(row, message="更新成功")
     except ValueError as e:
         return error_response(400, str(e))
@@ -2328,9 +2345,10 @@ async def opportunity_delete(
     """删除商机（软删）"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         count = soft_delete_opportunities(
-            db, body.ids, user_id=user_id, tenant_no=tenant_no
+            db, body.ids, user_id=user_id, tenant_no=tenant_no, user_info=user_info
         )
         return success_response({"count": count}, message="删除成功")
     except ValueError as e:
@@ -2348,6 +2366,7 @@ async def opportunity_assign(
     """分配商机"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         count = assign_opportunities(
             db,
@@ -2356,6 +2375,7 @@ async def opportunity_assign(
             user_id=user_id,
             remark=body.remark,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response({"count": count}, message="分配成功")
     except ValueError as e:
@@ -2373,6 +2393,7 @@ async def opportunity_change_stage(
     """变更商机阶段"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = change_stage(
             db,
@@ -2384,6 +2405,7 @@ async def opportunity_change_stage(
             remark=body.remark,
             user_id=user_id,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response(row, message="阶段已更新")
     except ValueError as e:
@@ -2401,6 +2423,7 @@ async def opportunity_mark_won(
     """标记赢单"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = mark_won(
             db,
@@ -2410,6 +2433,7 @@ async def opportunity_mark_won(
             remark=body.remark,
             user_id=user_id,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response(row, message="已标记赢单")
     except ValueError as e:
@@ -2427,6 +2451,7 @@ async def opportunity_mark_lost(
     """标记输单"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = mark_lost(
             db,
@@ -2437,6 +2462,7 @@ async def opportunity_mark_lost(
             improvement=body.improvement,
             user_id=user_id,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response(row, message="已标记输单")
     except ValueError as e:
@@ -2454,6 +2480,7 @@ async def opportunity_batch_update_stage(
     """批量变更阶段"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         result = batch_update_stage(
             db,
@@ -2464,6 +2491,7 @@ async def opportunity_batch_update_stage(
             body.change_reason,
             tenant_no=tenant_no,
             user_id=user_id,
+            user_info=user_info,
         )
         return success_response(result)
     except ValueError as e:
@@ -2481,6 +2509,7 @@ async def opportunity_batch_update_expect_date(
     """批量修改预计成交日期"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         result = batch_update_expect_date(
             db,
@@ -2490,6 +2519,7 @@ async def opportunity_batch_update_expect_date(
             days=body.days,
             tenant_no=tenant_no,
             user_id=user_id,
+            user_info=user_info,
         )
         return success_response(result)
     except ValueError as e:
@@ -2507,7 +2537,8 @@ async def opportunity_stage_log_list(
     """商机阶段变更日志"""
     try:
         tenant_no = get_tenant_no(db, _user)
-        logs = get_stage_log_list(db, body.opportunity_id, tenant_no=tenant_no)
+        user_info = crm_user_info(_user)
+        logs = get_stage_log_list(db, body.opportunity_id, tenant_no=tenant_no, user_info=user_info)
         return success_response(logs)
     except ValueError as e:
         return error_response(400, str(e))
@@ -2524,6 +2555,7 @@ async def opportunity_kanban(
     """看板数据"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         list_body = body or OpportunityKanbanBody()
         list_kwargs = _opp_list_kwargs(list_body, "")
         data = get_kanban_data(
@@ -2536,6 +2568,7 @@ async def opportunity_kanban(
                 "create_time_start": list_kwargs["create_time_start"],
                 "create_time_end": list_kwargs["create_time_end"],
             },
+            user_info=user_info,
         )
         return success_response(data)
     except ValueError as e:
@@ -2553,10 +2586,12 @@ async def opportunity_statistics_overview(
     """商机统计概览"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         stat_body = body or OpportunityStatisticsBody()
         data = get_statistics_overview(
             db,
             tenant_no,
+            user_info=user_info,
             owner_user_id=stat_body.owner_user_id,
             begin_time=stat_body.begin_time,
             end_time=stat_body.end_time,
@@ -2576,7 +2611,8 @@ async def opportunity_statistics_stage(
     """按阶段统计"""
     try:
         tenant_no = get_tenant_no(db, _user)
-        return success_response(get_statistics_by_stage(db, tenant_no))
+        user_info = crm_user_info(_user)
+        return success_response(get_statistics_by_stage(db, tenant_no, user_info=user_info))
     except ValueError as e:
         return error_response(400, str(e))
     except Exception as e:
@@ -2592,6 +2628,7 @@ async def opportunity_statistics_win_rate(
     """赢单率趋势"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         stat_body = body or OpportunityStatisticsBody()
         data = get_win_rate_statistics(
             db,
@@ -2599,6 +2636,7 @@ async def opportunity_statistics_win_rate(
             stat_type=stat_body.type or "month",
             begin_time=stat_body.begin_time,
             end_time=stat_body.end_time,
+            user_info=user_info,
         )
         return success_response(data)
     except ValueError as e:
@@ -2616,12 +2654,14 @@ async def opportunity_statistics_owner(
     """按负责人统计"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         stat_body = body or OpportunityStatisticsBody()
         data = get_statistics_by_owner(
             db,
             tenant_no,
             begin_time=stat_body.begin_time,
             end_time=stat_body.end_time,
+            user_info=user_info,
         )
         return success_response(data)
     except ValueError as e:
@@ -2639,6 +2679,7 @@ async def opportunity_statistics_funnel(
     """销售漏斗"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         stat_body = body or OpportunityStatisticsBody()
         data = get_sales_funnel(
             db,
@@ -2646,6 +2687,7 @@ async def opportunity_statistics_funnel(
             owner_user_id=stat_body.owner_user_id,
             begin_time=stat_body.begin_time,
             end_time=stat_body.end_time,
+            user_info=user_info,
         )
         return success_response(data)
     except ValueError as e:
@@ -2663,12 +2705,14 @@ async def opportunity_follow_list(
     """商机跟进列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         rows, total = list_opportunity_follows(
             db,
             opportunity_id=body.opportunity_id,
             page=body.page,
             page_size=body.pageSize,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response({
             "list": rows,
@@ -2693,12 +2737,14 @@ async def opportunity_follow_add(
     """新增商机跟进"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = create_opportunity_follow(
             db,
             tenant_no,
             body.model_dump(by_alias=False, exclude_unset=True),
             user_id=user_id,
+            user_info=user_info,
         )
         return success_response(row, message="创建成功")
     except ValueError as e:
@@ -2716,11 +2762,12 @@ async def opportunity_follow_update(
     """编辑商机跟进"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         data = body.model_dump(by_alias=False, exclude_unset=True)
         fid = data.pop("id")
         row = update_opportunity_follow(
-            db, fid, data, user_id=user_id, tenant_no=tenant_no
+            db, fid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info
         )
         return success_response(row, message="更新成功")
     except ValueError as e:
@@ -2738,11 +2785,12 @@ async def opportunity_follow_delete(
     """删除商机跟进"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         count = 0
         for fid in body.ids:
             if soft_delete_opportunity_follow(
-                db, fid, user_id=user_id, tenant_no=tenant_no
+                db, fid, user_id=user_id, tenant_no=tenant_no, user_info=user_info
             ):
                 count += 1
         return success_response({"count": count}, message="删除成功")
@@ -2761,7 +2809,8 @@ async def opportunity_follow_detail(
     """商机跟进详情"""
     try:
         tenant_no = get_tenant_no(db, _user)
-        row = get_opportunity_follow(db, body.id, tenant_no=tenant_no)
+        user_info = crm_user_info(_user)
+        row = get_opportunity_follow(db, body.id, tenant_no=tenant_no, user_info=user_info)
         if not row:
             return error_response(404, "跟进记录不存在")
         return success_response(row)
@@ -2780,6 +2829,7 @@ async def opportunity_product_list(
     """商机产品列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         rows, total = list_opportunity_products(
             db,
             opportunity_id=body.opportunity_id,
@@ -2787,6 +2837,7 @@ async def opportunity_product_list(
             page=body.page,
             page_size=body.pageSize,
             tenant_no=tenant_no,
+            user_info=user_info,
         )
         return success_response({
             "list": rows,
@@ -2811,10 +2862,11 @@ async def opportunity_product_list_by_opportunity(
     """按商机查询产品列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         opp_id = (body.opportunity_id or body.id) if body else None
         if not opp_id:
             return error_response(400, "缺少商机ID")
-        rows = list_products_by_opportunity(db, opp_id, tenant_no=tenant_no)
+        rows = list_products_by_opportunity(db, opp_id, tenant_no=tenant_no, user_info=user_info)
         return success_response(rows)
     except ValueError as e:
         return error_response(400, str(e))
@@ -2831,12 +2883,14 @@ async def opportunity_product_add(
     """新增商机产品"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         row = create_opportunity_product(
             db,
             tenant_no,
             body.model_dump(by_alias=False, exclude_unset=True),
             user_id=user_id,
+            user_info=user_info,
         )
         return success_response(row, message="创建成功")
     except ValueError as e:
@@ -2854,6 +2908,7 @@ async def opportunity_product_batch_add(
     """批量添加商机产品"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         product_list = [
             p.model_dump(by_alias=False, exclude_unset=True)
@@ -2865,6 +2920,7 @@ async def opportunity_product_batch_add(
             body.opportunity_id,
             product_list,
             user_id=user_id,
+            user_info=user_info,
         )
         return success_response({"count": count}, message="批量添加成功")
     except ValueError as e:
@@ -2882,11 +2938,12 @@ async def opportunity_product_update(
     """编辑商机产品"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         data = body.model_dump(by_alias=False, exclude_unset=True)
         pid = data.pop("id")
         row = update_opportunity_product(
-            db, pid, data, user_id=user_id, tenant_no=tenant_no
+            db, pid, data, user_id=user_id, tenant_no=tenant_no, user_info=user_info
         )
         return success_response(row, message="更新成功")
     except ValueError as e:
@@ -2904,9 +2961,10 @@ async def opportunity_product_delete(
     """删除商机产品"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         user_id = _user.get("user_id", "") or _user.get("id", "")
         count = soft_delete_opportunity_products(
-            db, body.ids, user_id=user_id, tenant_no=tenant_no
+            db, body.ids, user_id=user_id, tenant_no=tenant_no, user_info=user_info
         )
         return success_response({"count": count}, message="删除成功")
     except ValueError as e:
@@ -2924,7 +2982,8 @@ async def opportunity_product_detail(
     """商机产品详情"""
     try:
         tenant_no = get_tenant_no(db, _user)
-        row = get_opportunity_product(db, body.id, tenant_no=tenant_no)
+        user_info = crm_user_info(_user)
+        row = get_opportunity_product(db, body.id, tenant_no=tenant_no, user_info=user_info)
         if not row:
             return error_response(404, "商机产品不存在")
         return success_response(row)
@@ -2943,12 +3002,14 @@ async def opportunity_export(
     """导出商机列表"""
     try:
         tenant_no = get_tenant_no(db, _user)
+        user_info = crm_user_info(_user)
         list_body = body or OpportunityListBody()
         user_id = _user.get("user_id", "") or _user.get("id", "")
         rows, _ = list_opportunities(
             db,
             tenant_no,
             user_id,
+            user_info=user_info,
             **_opp_list_kwargs(list_body, user_id),
             page=1,
             page_size=10000,

backend/app/crud/crm/opportunity.py

@@ -17,6 +17,7 @@ from app.models.crm_models import (
     CrmOpportunityStageLog,
 )
 from app.models.models import SysUser
+from app.crud.crm.data_scope_helpers import apply_crm_owner_scope, ensure_crm_owner_row_in_scope
 
 _FMT = "%Y-%m-%d %H:%M:%S"
 _DATE_FMT = "%Y-%m-%d"
@@ -54,6 +55,27 @@ _FUNNEL_STAGE_NAMES = (
 )
 
 
+def _ensure_opportunity_in_scope(db: Session, row: CrmOpportunity, user_info, tenant_no: str) -> None:
+    ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该商机")
+
+
+def _scoped_opportunity_query(
+    db: Session,
+    tenant_no: str,
+    user_info=None,
+    owner_user_id: str = None,
+):
+    q = db.query(CrmOpportunity).filter(
+        CrmOpportunity.tenant_no == tenant_no,
+        CrmOpportunity.del_flag == 0,
+    )
+    if owner_user_id:
+        q = q.filter(CrmOpportunity.owner_user_id == owner_user_id)
+    if user_info:
+        q = apply_crm_owner_scope(q, CrmOpportunity, user_info, db, tenant_no)
+    return q
+
+
 def total_pages(total: int, page_size: int) -> int:
     if page_size <= 0:
         return 0
@@ -300,6 +322,7 @@ def list_opportunities(
     create_time_end: str = None,
     amount_min: Any = None,
     amount_max: Any = None,
+    user_info=None,
 ) -> Tuple[List[Dict[str, Any]], int]:
     q = db.query(CrmOpportunity)
     effective_owner = owner_user_id
@@ -324,6 +347,9 @@ def list_opportunities(
         amount_max=amount_max,
     )
 
+    if user_info:
+        q = apply_crm_owner_scope(q, CrmOpportunity, user_info, db, tenant_no)
+
     total = q.order_by(None).with_entities(func.count(CrmOpportunity.id)).scalar() or 0
     page = max(1, page)
     page_size = min(max(1, page_size), 100)
@@ -554,6 +580,7 @@ def update_opportunity(
     data: dict,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunity).filter(
         CrmOpportunity.id == opportunity_id,
@@ -564,6 +591,7 @@ def update_opportunity(
     obj = q.first()
     if not obj:
         raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, obj, user_info, obj.tenant_no)
     _guard_terminal_status(obj, "修改")
 
     _updatable = (
@@ -620,6 +648,7 @@ def soft_delete_opportunities(
     ids: List[str],
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> int:
     q = db.query(CrmOpportunity).filter(CrmOpportunity.id.in_(ids), CrmOpportunity.del_flag == 0)
     if tenant_no:
@@ -632,6 +661,7 @@ def soft_delete_opportunities(
 
     now = datetime.now()
     for row in rows:
+        _ensure_opportunity_in_scope(db, row, user_info, row.tenant_no)
         _guard_terminal_status(row, "删除")
         row.del_flag = 1
         row.update_by = user_id
@@ -647,6 +677,7 @@ def assign_opportunities(
     user_id: str = None,
     remark: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> int:
     owner_name = _resolve_user_display_name(db, owner_user_id)
     q = db.query(CrmOpportunity).filter(CrmOpportunity.id.in_(ids), CrmOpportunity.del_flag == 0)
@@ -656,6 +687,7 @@ def assign_opportunities(
     now = datetime.now()
     count = 0
     for row in rows:
+        _ensure_opportunity_in_scope(db, row, user_info, row.tenant_no)
         if row.status in ("won", "lost"):
             continue
         row.owner_user_id = owner_user_id
@@ -680,6 +712,7 @@ def change_stage(
     remark: str = None,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunity).filter(
         CrmOpportunity.id == opportunity_id,
@@ -690,6 +723,7 @@ def change_stage(
     obj = q.first()
     if not obj:
         raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, obj, user_info, obj.tenant_no)
     _guard_terminal_status(obj, "变更阶段")
 
     old_stage = obj.stage
@@ -749,6 +783,7 @@ def mark_won(
     remark: str = None,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunity).filter(
         CrmOpportunity.id == opportunity_id,
@@ -759,6 +794,7 @@ def mark_won(
     obj = q.first()
     if not obj:
         raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, obj, user_info, obj.tenant_no)
     if obj.status == "won":
         raise ValueError("商机已是赢单状态")
 
@@ -802,6 +838,7 @@ def mark_lost(
     improvement: str = None,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunity).filter(
         CrmOpportunity.id == opportunity_id,
@@ -812,6 +849,7 @@ def mark_lost(
     obj = q.first()
     if not obj:
         raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, obj, user_info, obj.tenant_no)
     if obj.status == "lost":
         raise ValueError("商机已是输单状态")
 
@@ -854,6 +892,7 @@ def batch_update_stage(
     change_reason: str,
     tenant_no: str = None,
     user_id: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     success_count = 0
     fail_count = 0
@@ -869,6 +908,7 @@ def batch_update_stage(
                 change_reason=change_reason,
                 user_id=user_id,
                 tenant_no=tenant_no,
+                user_info=user_info,
             )
             success_count += 1
         except Exception as e:
@@ -889,6 +929,7 @@ def batch_update_expect_date(
     days: int = None,
     tenant_no: str = None,
     user_id: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     success_count = 0
     fail_count = 0
@@ -905,6 +946,7 @@ def batch_update_expect_date(
             obj = q.first()
             if not obj:
                 raise ValueError("商机不存在")
+            _ensure_opportunity_in_scope(db, obj, user_info, obj.tenant_no)
             _guard_terminal_status(obj, "修改预计成交日期")
 
             new_date: Optional[date] = None
@@ -949,8 +991,22 @@ def batch_update_expect_date(
 
 
 def get_stage_log_list(
-    db: Session, opportunity_id: str, tenant_no: str = None
+    db: Session,
+    opportunity_id: str,
+    tenant_no: str = None,
+    user_info=None,
 ) -> List[Dict[str, Any]]:
+    opp_q = db.query(CrmOpportunity).filter(
+        CrmOpportunity.id == opportunity_id,
+        CrmOpportunity.del_flag == 0,
+    )
+    if tenant_no:
+        opp_q = opp_q.filter(CrmOpportunity.tenant_no == tenant_no)
+    opp = opp_q.first()
+    if not opp:
+        raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, opp, user_info, opp.tenant_no)
+
     q = db.query(CrmOpportunityStageLog).filter(
         CrmOpportunityStageLog.opportunity_id == opportunity_id,
         CrmOpportunityStageLog.del_flag == 0,
@@ -972,34 +1028,18 @@ def get_statistics_overview(
     owner_user_id: str = None,
     begin_time: str = None,
     end_time: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
-    base = [CrmOpportunity.tenant_no == tenant_no, CrmOpportunity.del_flag == 0]
-    if owner_user_id:
-        base.append(CrmOpportunity.owner_user_id == owner_user_id)
+    q = _scoped_opportunity_query(db, tenant_no, user_info, owner_user_id)
     if begin_time:
-        base.append(CrmOpportunity.create_time >= begin_time)
+        q = q.filter(CrmOpportunity.create_time >= begin_time)
     if end_time:
-        base.append(CrmOpportunity.create_time <= end_time)
+        q = q.filter(CrmOpportunity.create_time <= end_time)
 
-    total_count = db.query(func.count(CrmOpportunity.id)).filter(*base).scalar() or 0
-    following_count = (
-        db.query(func.count(CrmOpportunity.id))
-        .filter(*base, CrmOpportunity.status == "following")
-        .scalar()
-        or 0
-    )
-    won_count = (
-        db.query(func.count(CrmOpportunity.id))
-        .filter(*base, CrmOpportunity.status == "won")
-        .scalar()
-        or 0
-    )
-    lost_count = (
-        db.query(func.count(CrmOpportunity.id))
-        .filter(*base, CrmOpportunity.status == "lost")
-        .scalar()
-        or 0
-    )
+    total_count = q.count()
+    following_count = q.filter(CrmOpportunity.status == "following").count()
+    won_count = q.filter(CrmOpportunity.status == "won").count()
+    lost_count = q.filter(CrmOpportunity.status == "lost").count()
     win_rate = Decimal("0")
     if won_count + lost_count > 0:
         win_rate = (
@@ -1009,11 +1049,8 @@ def get_statistics_overview(
         ).quantize(Decimal("0.01"))
 
     total_amount = (
-        db.query(func.coalesce(func.sum(CrmOpportunity.amount), 0))
-        .filter(
-            *base,
-            CrmOpportunity.status.in_(("following", "won")),
-        )
+        q.with_entities(func.coalesce(func.sum(CrmOpportunity.amount), 0))
+        .filter(CrmOpportunity.status.in_(("following", "won")))
         .scalar()
     )
 
@@ -1028,19 +1065,18 @@ def get_statistics_overview(
     }
 
 
-def get_statistics_by_stage(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
+def get_statistics_by_stage(
+    db: Session, tenant_no: str, user_info=None
+) -> List[Dict[str, Any]]:
     rows = (
-        db.query(
+        _scoped_opportunity_query(db, tenant_no, user_info)
+        .filter(CrmOpportunity.status == "following")
+        .with_entities(
             CrmOpportunity.stage,
             func.count(CrmOpportunity.id).label("count"),
             func.coalesce(func.sum(CrmOpportunity.amount), 0).label("totalAmount"),
             func.coalesce(func.avg(CrmOpportunity.win_rate), 0).label("avgWinRate"),
         )
-        .filter(
-            CrmOpportunity.tenant_no == tenant_no,
-            CrmOpportunity.del_flag == 0,
-            CrmOpportunity.status == "following",
-        )
         .group_by(CrmOpportunity.stage)
         .all()
     )
@@ -1064,6 +1100,7 @@ def get_win_rate_statistics(
     stat_type: str = "month",
     begin_time: str = None,
     end_time: str = None,
+    user_info=None,
 ) -> List[Dict[str, Any]]:
     if stat_type == "day":
         fmt = "%Y-%m-%d"
@@ -1073,24 +1110,21 @@ def get_win_rate_statistics(
         fmt = "%Y-%m"
 
     date_expr = func.date_format(CrmOpportunity.create_time, fmt).label("date")
-    filters = [
-        CrmOpportunity.tenant_no == tenant_no,
-        CrmOpportunity.del_flag == 0,
-        CrmOpportunity.status.in_(("won", "lost")),
-    ]
+    q = _scoped_opportunity_query(db, tenant_no, user_info).filter(
+        CrmOpportunity.status.in_(("won", "lost"))
+    )
     if begin_time:
-        filters.append(CrmOpportunity.create_time >= begin_time)
+        q = q.filter(CrmOpportunity.create_time >= begin_time)
     if end_time:
-        filters.append(CrmOpportunity.create_time <= end_time)
+        q = q.filter(CrmOpportunity.create_time <= end_time)
 
     rows = (
-        db.query(
+        q.with_entities(
             date_expr,
             func.count(CrmOpportunity.id).label("totalCount"),
             func.sum(case((CrmOpportunity.status == "won", 1), else_=0)).label("wonCount"),
             func.sum(case((CrmOpportunity.status == "lost", 1), else_=0)).label("lostCount"),
         )
-        .filter(*filters)
         .group_by(date_expr)
         .order_by(date_expr)
         .all()
@@ -1117,15 +1151,16 @@ def get_statistics_by_owner(
     tenant_no: str,
     begin_time: str = None,
     end_time: str = None,
+    user_info=None,
 ) -> List[Dict[str, Any]]:
-    filters = [CrmOpportunity.tenant_no == tenant_no, CrmOpportunity.del_flag == 0]
+    q = _scoped_opportunity_query(db, tenant_no, user_info)
     if begin_time:
-        filters.append(CrmOpportunity.create_time >= begin_time)
+        q = q.filter(CrmOpportunity.create_time >= begin_time)
     if end_time:
-        filters.append(CrmOpportunity.create_time <= end_time)
+        q = q.filter(CrmOpportunity.create_time <= end_time)
 
     rows = (
-        db.query(
+        q.with_entities(
             CrmOpportunity.owner_user_id.label("ownerUserId"),
             CrmOpportunity.owner_user_name.label("ownerUserName"),
             func.count(CrmOpportunity.id).label("totalCount"),
@@ -1145,7 +1180,6 @@ def get_statistics_by_owner(
                 0,
             ).label("followingAmount"),
         )
-        .filter(*filters)
         .group_by(CrmOpportunity.owner_user_id, CrmOpportunity.owner_user_name)
         .order_by(
             func.sum(case((CrmOpportunity.status == "won", 1), else_=0)).desc(),
@@ -1182,13 +1216,12 @@ def get_kanban_data(
     db: Session,
     tenant_no: str,
     filters: dict = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     filters = filters or {}
     result: Dict[str, Any] = {}
     for stage in _KANBAN_STAGES:
-        q = db.query(CrmOpportunity).filter(
-            CrmOpportunity.tenant_no == tenant_no,
-            CrmOpportunity.del_flag == 0,
+        q = _scoped_opportunity_query(db, tenant_no, user_info).filter(
             CrmOpportunity.stage == stage,
             CrmOpportunity.status == "following",
         )
@@ -1228,28 +1261,22 @@ def get_sales_funnel(
     owner_user_id: str = None,
     begin_time: str = None,
     end_time: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     funnel_data: List[Dict[str, Any]] = []
     for i, stage in enumerate(_FUNNEL_STAGES):
-        filters = [
-            CrmOpportunity.tenant_no == tenant_no,
-            CrmOpportunity.del_flag == 0,
-        ]
-        if owner_user_id:
-            filters.append(CrmOpportunity.owner_user_id == owner_user_id)
+        q = _scoped_opportunity_query(db, tenant_no, user_info, owner_user_id)
         if begin_time:
-            filters.append(CrmOpportunity.create_time >= begin_time)
+            q = q.filter(CrmOpportunity.create_time >= begin_time)
         if end_time:
-            filters.append(CrmOpportunity.create_time <= end_time)
+            q = q.filter(CrmOpportunity.create_time <= end_time)
 
         if stage == "won":
-            filters.append(CrmOpportunity.status == "won")
+            q = q.filter(CrmOpportunity.status == "won")
         else:
-            filters.extend(
-                [CrmOpportunity.stage == stage, CrmOpportunity.status == "following"]
-            )
+            q = q.filter(CrmOpportunity.stage == stage, CrmOpportunity.status == "following")
 
-        rows = db.query(CrmOpportunity).filter(*filters).all()
+        rows = q.all()
         count = len(rows)
         total_amount = sum(float(r.amount or 0) for r in rows)
         avg_win_rate = Decimal("0")

backend/app/crud/crm/opportunity_follow.py

@@ -8,6 +8,8 @@ from typing import Any, Dict, List, Optional, Tuple
 from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
+from app.crud.crm.data_scope_helpers import apply_crm_create_by_scope, ensure_crm_create_by_row_in_scope
+from app.crud.crm.opportunity import _ensure_opportunity_in_scope
 from app.models.crm_models import CrmOpportunity, CrmOpportunityFollow
 
 
@@ -103,6 +105,24 @@ def _recalc_opportunity_follow_stats(db: Session, opportunity_id: str) -> None:
     opp.update_time = datetime.now()
 
 
+def _ensure_parent_opportunity(
+    db: Session, opportunity_id: str, tenant_no: str, user_info
+) -> CrmOpportunity:
+    opp = (
+        db.query(CrmOpportunity)
+        .filter(
+            CrmOpportunity.id == opportunity_id,
+            CrmOpportunity.tenant_no == tenant_no,
+            CrmOpportunity.del_flag == 0,
+        )
+        .first()
+    )
+    if not opp:
+        raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, opp, user_info, tenant_no)
+    return opp
+
+
 def list_opportunity_follows(
     db: Session,
     *,
@@ -110,12 +130,19 @@ def list_opportunity_follows(
     page: int = 1,
     page_size: int = 10,
     tenant_no: str = None,
+    user_info=None,
 ) -> Tuple[List[Dict[str, Any]], int]:
+    if opportunity_id and tenant_no:
+        _ensure_parent_opportunity(db, opportunity_id, tenant_no, user_info)
+
     q = db.query(CrmOpportunityFollow).filter(CrmOpportunityFollow.del_flag == 0)
     if tenant_no:
         q = q.filter(CrmOpportunityFollow.tenant_no == tenant_no)
     if opportunity_id:
         q = q.filter(CrmOpportunityFollow.opportunity_id == opportunity_id)
+    if user_info and tenant_no:
+        q = apply_crm_create_by_scope(q, CrmOpportunityFollow, user_info, db, tenant_no)
+
     total = q.count()
     rows = (
         q.order_by(CrmOpportunityFollow.create_time.desc())
@@ -127,7 +154,10 @@ def list_opportunity_follows(
 
 
 def list_follows_by_opportunity(
-    db: Session, opportunity_id: str, tenant_no: str = None
+    db: Session,
+    opportunity_id: str,
+    tenant_no: str = None,
+    user_info=None,
 ) -> List[Dict[str, Any]]:
     rows, _ = list_opportunity_follows(
         db,
@@ -135,12 +165,16 @@ def list_follows_by_opportunity(
         page=1,
         page_size=10000,
         tenant_no=tenant_no,
+        user_info=user_info,
     )
     return rows
 
 
 def get_opportunity_follow(
-    db: Session, follow_id: str, tenant_no: str = None
+    db: Session,
+    follow_id: str,
+    tenant_no: str = None,
+    user_info=None,
 ) -> Optional[Dict[str, Any]]:
     q = db.query(CrmOpportunityFollow).filter(
         CrmOpportunityFollow.id == follow_id,
@@ -149,7 +183,12 @@ def get_opportunity_follow(
     if tenant_no:
         q = q.filter(CrmOpportunityFollow.tenant_no == tenant_no)
     row = q.first()
-    return _row_to_dict(row) if row else None
+    if not row:
+        return None
+    ensure_crm_create_by_row_in_scope(
+        db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
+    )
+    return _row_to_dict(row)
 
 
 def create_opportunity_follow(
@@ -157,18 +196,9 @@ def create_opportunity_follow(
     tenant_no: str,
     data: dict,
     user_id: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
-    opp = (
-        db.query(CrmOpportunity)
-        .filter(
-            CrmOpportunity.id == data.get("opportunity_id"),
-            CrmOpportunity.tenant_no == tenant_no,
-            CrmOpportunity.del_flag == 0,
-        )
-        .first()
-    )
-    if not opp:
-        raise ValueError("商机不存在")
+    opp = _ensure_parent_opportunity(db, data.get("opportunity_id"), tenant_no, user_info)
     if opp.status in ("won", "lost"):
         raise ValueError("已赢单或已输单的商机不允许添加跟进")
 
@@ -213,6 +243,7 @@ def update_opportunity_follow(
     data: dict,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunityFollow).filter(
         CrmOpportunityFollow.id == follow_id,
@@ -223,6 +254,9 @@ def update_opportunity_follow(
     row = q.first()
     if not row:
         raise ValueError("跟进记录不存在")
+    ensure_crm_create_by_row_in_scope(
+        db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
+    )
 
     if data.get("follow_type") is not None:
         row.follow_type = data["follow_type"]
@@ -252,6 +286,7 @@ def soft_delete_opportunity_follow(
     follow_id: str,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> bool:
     q = db.query(CrmOpportunityFollow).filter(
         CrmOpportunityFollow.id == follow_id,
@@ -262,6 +297,9 @@ def soft_delete_opportunity_follow(
     row = q.first()
     if not row:
         return False
+    ensure_crm_create_by_row_in_scope(
+        db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
+    )
     row.del_flag = 1
     row.update_by = user_id
     row.update_time = datetime.now()

backend/app/crud/crm/opportunity_product.py

@@ -9,6 +9,7 @@ from typing import Any, Dict, List, Optional, Tuple
 from sqlalchemy.orm import Session
 
 from app.core.utils import generate_uuid
+from app.crud.crm.opportunity import _ensure_opportunity_in_scope
 from app.models.crm_models import CrmOpportunity, CrmOpportunityProduct, CrmProduct
 
 
@@ -96,6 +97,24 @@ def _fill_product_redundant(db: Session, entity: CrmOpportunityProduct, product_
         entity.unit = entity.unit or product.unit
 
 
+def _get_opportunity_for_product(
+    db: Session, opportunity_id: str, tenant_no: str, user_info=None
+) -> CrmOpportunity:
+    opp = (
+        db.query(CrmOpportunity)
+        .filter(
+            CrmOpportunity.id == opportunity_id,
+            CrmOpportunity.tenant_no == tenant_no,
+            CrmOpportunity.del_flag == 0,
+        )
+        .first()
+    )
+    if not opp:
+        raise ValueError("商机不存在")
+    _ensure_opportunity_in_scope(db, opp, user_info, tenant_no)
+    return opp
+
+
 def list_opportunity_products(
     db: Session,
     *,
@@ -104,7 +123,10 @@ def list_opportunity_products(
     page: int = 1,
     page_size: int = 10,
     tenant_no: str = None,
+    user_info=None,
 ) -> Tuple[List[Dict[str, Any]], int]:
+    if opportunity_id and tenant_no:
+        _get_opportunity_for_product(db, opportunity_id, tenant_no, user_info)
     q = db.query(CrmOpportunityProduct).filter(CrmOpportunityProduct.del_flag == 0)
     if tenant_no:
         q = q.filter(CrmOpportunityProduct.tenant_no == tenant_no)
@@ -123,7 +145,10 @@ def list_opportunity_products(
 
 
 def list_products_by_opportunity(
-    db: Session, opportunity_id: str, tenant_no: str = None
+    db: Session,
+    opportunity_id: str,
+    tenant_no: str = None,
+    user_info=None,
 ) -> List[Dict[str, Any]]:
     rows, _ = list_opportunity_products(
         db,
@@ -131,12 +156,16 @@ def list_products_by_opportunity(
         page=1,
         page_size=10000,
         tenant_no=tenant_no,
+        user_info=user_info,
     )
     return rows
 
 
 def get_opportunity_product(
-    db: Session, product_row_id: str, tenant_no: str = None
+    db: Session,
+    product_row_id: str,
+    tenant_no: str = None,
+    user_info=None,
 ) -> Optional[Dict[str, Any]]:
     q = db.query(CrmOpportunityProduct).filter(
         CrmOpportunityProduct.id == product_row_id,
@@ -145,7 +174,11 @@ def get_opportunity_product(
     if tenant_no:
         q = q.filter(CrmOpportunityProduct.tenant_no == tenant_no)
     row = q.first()
-    return _row_to_dict(row) if row else None
+    if not row:
+        return None
+    if tenant_no:
+        _get_opportunity_for_product(db, row.opportunity_id, tenant_no, user_info)
+    return _row_to_dict(row)
 
 
 def create_opportunity_product(
@@ -153,19 +186,10 @@ def create_opportunity_product(
     tenant_no: str,
     data: dict,
     user_id: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     opp_id = data.get("opportunity_id")
-    opp = (
-        db.query(CrmOpportunity)
-        .filter(
-            CrmOpportunity.id == opp_id,
-            CrmOpportunity.tenant_no == tenant_no,
-            CrmOpportunity.del_flag == 0,
-        )
-        .first()
-    )
-    if not opp:
-        raise ValueError("商机不存在")
+    _get_opportunity_for_product(db, opp_id, tenant_no, user_info)
 
     qty = _decimal_value(data.get("quantity"))
     price = _decimal_value(data.get("price"))
@@ -216,14 +240,16 @@ def batch_add_opportunity_products(
     opportunity_id: str,
     product_list: List[dict],
     user_id: str = None,
+    user_info=None,
 ) -> int:
     if not product_list:
         raise ValueError("产品列表不能为空")
+    _get_opportunity_for_product(db, opportunity_id, tenant_no, user_info)
     count = 0
     for item in product_list:
         item = dict(item)
         item["opportunity_id"] = opportunity_id
-        create_opportunity_product(db, tenant_no, item, user_id)
+        create_opportunity_product(db, tenant_no, item, user_id, user_info=user_info)
         count += 1
     return count
 
@@ -234,6 +260,7 @@ def update_opportunity_product(
     data: dict,
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> Dict[str, Any]:
     q = db.query(CrmOpportunityProduct).filter(
         CrmOpportunityProduct.id == product_row_id,
@@ -244,6 +271,8 @@ def update_opportunity_product(
     row = q.first()
     if not row:
         raise ValueError("商机产品不存在")
+    if tenant_no:
+        _get_opportunity_for_product(db, row.opportunity_id, tenant_no, user_info)
 
     if data.get("quantity") is not None:
         row.quantity = _decimal_value(data["quantity"])
@@ -271,6 +300,7 @@ def soft_delete_opportunity_products(
     ids: List[str],
     user_id: str = None,
     tenant_no: str = None,
+    user_info=None,
 ) -> int:
     q = db.query(CrmOpportunityProduct).filter(
         CrmOpportunityProduct.id.in_(ids),
@@ -282,6 +312,8 @@ def soft_delete_opportunity_products(
     if not rows:
         return 0
     opp_id = rows[0].opportunity_id
+    if tenant_no:
+        _get_opportunity_for_product(db, opp_id, tenant_no, user_info)
     now = datetime.now()
     for row in rows:
         row.del_flag = 1