cykj/shanxia2.0
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(crm): 客户与线索模块接入数据权限过滤

Browse Source 
列表按 owner_user_id/create_by 过滤，详情与 CUD 校验 scope；公海客户列表保持租户内全员可见。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-06-28 20:30:41 +08:00
1 parent f5ab023ab5
commit 3107d537f6
9 files changed
+483 -103
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/api/crm.py
+155 -30
View File
@@ -14,6 +14,11 @@ from app.core.database import get_db
from app.core.menu_guard import require_admin_or_super_admin
from app.core.response import error_response, success_response
from app.core.tenant_context import get_tenant_no
from app.crud.crm.data_scope_helpers import (
crm_user_info,
ensure_crm_create_by_row_in_scope,
ensure_crm_owner_row_in_scope,
)
from app.utils.crm_export_columns import (
CLUE_EXPORT_COLUMNS,
CONTACT_EXPORT_COLUMNS,
@@ -352,12 +357,14 @@ async def customer_list(
"""客户列表（支持多种视图模式）"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
rows, total = list_customers(
db,
tenant_no,
user_id,
user_info=user_info,
page=body.page,
page_size=body.pageSize,
**body.to_list_kwargs(),
@@ -389,6 +396,9 @@ async def customer_detail(
row = get_customer(db, customer_id)
if not row:
return error_response(404, "客户不存在")
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该客户")
detail = customer_row_to_dict(row)
enrich_customer_user_names(db, [detail])
return success_response(detail)
@@ -436,6 +446,7 @@ async def customer_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(exclude_unset=True)
@@ -444,7 +455,7 @@ async def customer_update(
if customer_name_exists(db, data["customer_name"], tenant_no, exclude_id=customer_id):
return error_response(400, "客户名称已存在")
obj = update_customer(db, customer_id, user_id=user_id, **data)
obj = update_customer(db, customer_id, user_id=user_id, tenant_no=tenant_no, user_info=user_info, **data)
return success_response(customer_row_to_dict(obj), message="更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -460,11 +471,13 @@ async def customer_delete(
):
"""删除客户（支持批量）"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
success_ids = []
fail_items = []
for cid in body.ids:
try:
soft_delete_customer(db, cid)
soft_delete_customer(db, cid, user_info=user_info, tenant_no=tenant_no)
success_ids.append(cid)
except ValueError as e:
fail_items.append({"id": cid, "reason": str(e)})
@@ -899,6 +912,7 @@ async def customer_transfer(
"""批量转移客户"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
ids = body.get("ids", [])
owner_user_id = body.get("ownerUserId", "")
@@ -906,7 +920,7 @@ async def customer_transfer(
if not ids or not owner_user_id:
return error_response(400, "缺少必要参数 ids 或 ownerUserId")
for cid in ids:
transfer_customer(db, cid, owner_user_id, owner_user_name, user_id=user_id)
transfer_customer(db, cid, owner_user_id, owner_user_name, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
return success_response(None, message="转移成功")
except ValueError as e:
return error_response(400, str(e))
@@ -922,13 +936,15 @@ async def customer_into_pool(
):
"""批量放入公海池"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
ids = body.get("ids", [])
reason = body.get("reason", "")
if not ids:
return error_response(400, "缺少客户ID列表")
for cid in ids:
into_pool(db, cid, reason=reason, user_id=user_id)
into_pool(db, cid, reason=reason, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
return success_response(None, message="已放入公海池")
except ValueError as e:
return error_response(400, str(e))
@@ -955,6 +971,8 @@ async def customer_deal(
deal_amount=deal_amount,
deal_time=deal_time,
user_id=user_id,
user_info=crm_user_info(_user),
tenant_no=get_tenant_no(db, _user),
)
return success_response(customer_row_to_dict(obj), message="成交操作成功")
except ValueError as e:
@@ -994,12 +1012,14 @@ async def customer_export(
"""导出客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
list_body = body or CustomerListBody()
rows, _ = list_customers(
db,
tenant_no,
user_id,
user_info=user_info,
page=1,
page_size=10000,
**list_body.to_list_kwargs(),
@@ -1067,12 +1087,14 @@ async def customer_mine(
"""我的客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
list_body = body or CustomerListBody()
rows, total = list_customers(
db,
tenant_no,
user_id,
user_info=user_info,
page=list_body.page,
page_size=list_body.pageSize,
**list_body.to_list_kwargs(view_mode_override="mine"),
@@ -1099,12 +1121,14 @@ async def customer_to_follow(
"""待跟进客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
list_body = body or CustomerListBody()
rows, total = list_customers(
db,
tenant_no,
user_id,
user_info=user_info,
page=list_body.page,
page_size=list_body.pageSize,
**list_body.to_list_kwargs(view_mode_override="toFollow"),
@@ -1131,12 +1155,14 @@ async def customer_key(
"""重点客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
list_body = body or CustomerListBody()
rows, total = list_customers(
db,
tenant_no,
user_id,
user_info=user_info,
page=list_body.page,
page_size=list_body.pageSize,
**list_body.to_list_kwargs(view_mode_override="key"),
@@ -1163,11 +1189,13 @@ async def customer_dealed(
"""已成交客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
page = body.page if body else 1
page_size = body.pageSize if body else 10
rows, total = list_customers(
db, tenant_no, user_id, page=page, page_size=page_size,
user_info=user_info,
view_mode="dealed",
keyword=body.keyword if body else None,
)
@@ -1193,11 +1221,13 @@ async def customer_week_new(
"""本周新增客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
page = body.page if body else 1
page_size = body.pageSize if body else 10
rows, total = list_customers(
db, tenant_no, user_id, page=page, page_size=page_size,
user_info=user_info,
view_mode="weekNew",
keyword=body.keyword if body else None,
)
@@ -1223,11 +1253,13 @@ async def customer_long_time_no_follow(
"""长期未跟进客户列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
page = body.page if body else 1
page_size = body.pageSize if body else 10
rows, total = list_customers(
db, tenant_no, user_id, page=page, page_size=page_size,
user_info=user_info,
view_mode="longTimeNoFollow",
keyword=body.keyword if body else None,
)
@@ -1303,8 +1335,9 @@ async def pool_customer_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", "")
assign_customer(db, body.ids, body.ownerUserId, body.ownerUserName, user_id=user_id)
assign_customer(db, body.ids, body.ownerUserId, body.ownerUserName, user_id=user_id, user_info=user_info, tenant_no=tenant_no)
return success_response(message="分配成功")
except ValueError as e:
return error_response(400, str(e))
@@ -1666,10 +1699,11 @@ async def follow_list(
"""跟进记录分页列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
page = body.page if body else 1
page_size = body.pageSize if body else 10
rows, total = list_follows(
db, tenant_no, page=page, page_size=page_size,
db, tenant_no, page=page, page_size=page_size, user_info=user_info,
keyword=body.keyword if body else None,
customer_id=body.customer_id if body else None,
follow_type=body.follow_type if body else None,
@@ -1696,12 +1730,17 @@ async def follow_detail(
):
"""跟进记录详情"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
follow_id = body.get("id")
if not follow_id:
return error_response(400, "缺少跟进记录ID")
row = get_follow(db, follow_id)
if not row:
return error_response(404, "跟进记录不存在")
ensure_crm_create_by_row_in_scope(
db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
)
return success_response(follow_row_to_dict(row))
except ValueError as e:
return error_response(400, str(e))
@@ -1718,10 +1757,11 @@ async def follow_by_customer(
"""按客户ID查询跟进记录"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
customer_id = body.get("customerId", "")
if not customer_id:
return error_response(400, "缺少客户ID")
rows = list_follows_by_customer(db, tenant_no, customer_id)
rows = list_follows_by_customer(db, tenant_no, customer_id, user_info=user_info)
return success_response(rows)
except ValueError as e:
return error_response(400, str(e))
@@ -1738,9 +1778,12 @@ async def 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", "")
user_name = _user.get("nick_name", "") or _user.get("name", "")
obj = create_follow(
db, tenant_no=tenant_no, user_id=user_id,
db, tenant_no=tenant_no, user_id=user_id, user_name=user_name,
user_info=user_info,
**body.model_dump(exclude_unset=True),
)
return success_response(follow_row_to_dict(obj), message="创建成功")
@@ -1761,7 +1804,7 @@ async def follow_update(
user_id = _user.get("user_id", "") or _user.get("id", "")
data = body.model_dump(exclude_unset=True)
follow_id = data.pop("id")
obj = update_follow(db, follow_id, user_id=user_id, **data)
obj = update_follow(db, follow_id, user_id=user_id, tenant_no=get_tenant_no(db, _user), user_info=crm_user_info(_user), **data)
return success_response(follow_row_to_dict(obj), message="更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -1778,7 +1821,7 @@ async def follow_delete(
"""删除跟进记录"""
try:
for fid in body.ids:
soft_delete_follow(db, fid)
soft_delete_follow(db, fid, user_info=crm_user_info(_user), tenant_no=get_tenant_no(db, _user))
return success_response(None, message="删除成功")
except ValueError as e:
return error_response(400, str(e))
@@ -1795,6 +1838,7 @@ async def follow_export(
"""导出跟进记录"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
rows, _ = list_follows(
db, tenant_no, page=1, page_size=10000,
keyword=body.keyword if body else None,
@@ -1826,10 +1870,11 @@ async def pool_rule_list(
):
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
page = body.page if body else 1
page_size = body.pageSize if body else 10
rows, total = list_pool_rules(
db, tenant_no,
db, tenant_no, user_info=user_info,
page=page, page_size=page_size,
pool_name=body.poolName if body else None,
rule_type=body.ruleType if body else None,
@@ -1856,10 +1901,12 @@ async def pool_rule_detail(
):
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
from app.crud.crm.pool_rule import _row_to_dict as pool_rule_row_to_dict
row = get_pool_rule(db, body.id, tenant_no)
if not row:
return error_response(404, "公海池规则不存在")
ensure_crm_create_by_row_in_scope(db, row, user_info, tenant_no, message="无权访问该公海规则")
return success_response(pool_rule_row_to_dict(row))
except ValueError as e:
return error_response(400, str(e))
@@ -1902,10 +1949,12 @@ async def pool_rule_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", "")
update_pool_rule(
db, body.id, tenant_no,
user_id=user_id,
user_info=user_info,
pool_name=body.poolName,
rule_type=body.ruleType,
timeout_days=body.timeoutDays,
@@ -1929,7 +1978,8 @@ async def pool_rule_delete(
):
try:
tenant_no = get_tenant_no(db, _user)
count = delete_pool_rules(db, body.ids, tenant_no)
user_info = crm_user_info(_user)
count = delete_pool_rules(db, body.ids, tenant_no, user_info=user_info)
return success_response({"count": count}, message="删除成功")
except ValueError as e:
return error_response(400, str(e))
@@ -1945,8 +1995,11 @@ async def pool_rule_update_status(
):
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
update_pool_rule_status(db, body.id, tenant_no, body.status, user_id=user_id)
update_pool_rule_status(
db, body.id, tenant_no, body.status, user_id=user_id, user_info=user_info
)
return success_response(message="状态更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -1982,7 +2035,10 @@ async def statistics_overview(
"""客户统计概览"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_overview_statistics(db, tenant_no, user_id=body.user_id if body else None)
user_info = crm_user_info(_user)
data = get_overview_statistics(
db, tenant_no, user_id=body.user_id if body else None, user_info=user_info
)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -1999,7 +2055,8 @@ async def statistics_source(
"""客户来源统计"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_source_statistics(db, tenant_no)
user_info = crm_user_info(_user)
data = get_source_statistics(db, tenant_no, user_info=user_info)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -2016,7 +2073,8 @@ async def statistics_level(
"""客户级别分布"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_level_statistics(db, tenant_no)
user_info = crm_user_info(_user)
data = get_level_statistics(db, tenant_no, user_info=user_info)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -2033,7 +2091,8 @@ async def statistics_industry(
"""客户行业分布"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_industry_statistics(db, tenant_no)
user_info = crm_user_info(_user)
data = get_industry_statistics(db, tenant_no, user_info=user_info)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -2050,10 +2109,12 @@ async def statistics_trend(
"""成交趋势"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
data = get_trend_statistics(
db, tenant_no,
start_date=body.start_date if body else None,
end_date=body.end_date if body else None,
user_info=user_info,
)
return success_response(data)
except ValueError as e:
@@ -2071,7 +2132,8 @@ async def statistics_owner(
"""负责人统计"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_owner_statistics(db, tenant_no)
user_info = crm_user_info(_user)
data = get_owner_statistics(db, tenant_no, user_info=user_info)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -2088,10 +2150,12 @@ async def statistics_follow(
"""跟进统计"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
data = get_follow_statistics(
db, tenant_no,
start_date=body.start_date if body else None,
end_date=body.end_date if body else None,
user_info=user_info,
)
return success_response(data)
except ValueError as e:
@@ -3248,12 +3312,14 @@ async def clue_list(
"""线索列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
rows, total = list_clues(
db,
tenant_no,
user_id,
user_info=user_info,
page=body.page,
page_size=body.pageSize,
keyword=body.keyword or body.search_keyword,
@@ -3293,12 +3359,15 @@ async def clue_detail(
"""线索详情"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
clue_id = body.get("id")
if not clue_id:
return error_response(400, "缺少线索ID")
user_info = crm_user_info(_user)
row = get_clue(db, clue_id)
if not row or row.tenant_no != tenant_no:
return error_response(404, "线索不存在")
ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该线索")
detail = clue_row_to_dict(row)
enrich_clue_user_names(db, [detail])
return success_response(detail)
@@ -3340,11 +3409,12 @@ async def clue_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", "")
clue_id = body.id
data = body.model_dump(by_alias=False, exclude_unset=True)
data.pop("id", None)
obj = update_clue(db, clue_id, data, user_id, tenant_no=tenant_no)
obj = update_clue(db, clue_id, data, user_id, tenant_no=tenant_no, user_info=user_info)
return success_response(clue_row_to_dict(obj), message="更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -3361,12 +3431,13 @@ async def clue_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", "")
success_ids = []
fail_items = []
for cid in body.ids:
try:
soft_delete_clue(db, cid, user_id=user_id, tenant_no=tenant_no)
soft_delete_clue(db, cid, user_id=user_id, tenant_no=tenant_no, user_info=user_info)
success_ids.append(cid)
except ValueError as e:
fail_items.append({"id": cid, "reason": str(e)})
@@ -3391,6 +3462,7 @@ async def clue_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_clues(
db,
@@ -3400,6 +3472,7 @@ async def clue_assign(
current_user_id=user_id,
remark=body.remark,
tenant_no=tenant_no,
user_info=user_info,
)
return success_response({"count": count}, message=f"分配成功{count}条")
except ValueError as e:
@@ -3417,6 +3490,7 @@ async def clue_convert(
"""线索转化"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
customer_data = body.model_dump(by_alias=False, exclude_unset=True)
customer_data.pop("id", None)
@@ -3426,6 +3500,7 @@ async def clue_convert(
customer_data,
current_user_id=user_id,
tenant_no=tenant_no,
user_info=user_info,
)
return success_response(result, message="转化成功")
except ValueError as e:
@@ -3463,8 +3538,11 @@ async def clue_update_status(
"""更新线索状态"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
obj = update_clue_status(db, body.id, body.status, user_id, tenant_no=tenant_no)
obj = update_clue_status(
db, body.id, body.status, user_id, tenant_no=tenant_no, user_info=user_info
)
return success_response(clue_row_to_dict(obj), message="状态更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -3481,6 +3559,7 @@ async def clue_invalid(
"""线索标记无效"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
id_list = list(body.ids or [])
if body.id:
@@ -3490,7 +3569,9 @@ async def clue_invalid(
return error_response(400, "缺少线索ID")
results = []
for cid in id_list:
obj = invalid_clue(db, cid, user_id, remark=body.remark, tenant_no=tenant_no)
obj = invalid_clue(
db, cid, user_id, remark=body.remark, tenant_no=tenant_no, user_info=user_info
)
results.append(clue_row_to_dict(obj))
data = results[0] if len(results) == 1 else {"list": results, "count": len(results)}
return success_response(data, message="标记无效成功")
@@ -3509,8 +3590,11 @@ async def clue_update_level(
"""批量更新线索级别"""
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 = update_clue_level(db, body.ids, body.level, user_id, tenant_no=tenant_no)
count = update_clue_level(
db, body.ids, body.level, user_id, tenant_no=tenant_no, user_info=user_info
)
return success_response({"count": count}, message=f"更新成功{count}条")
except ValueError as e:
return error_response(400, str(e))
@@ -3527,12 +3611,14 @@ async def clue_mine(
"""我的线索"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
rows, total = list_clues(
db,
tenant_no,
user_id,
user_info=user_info,
page=body.page,
page_size=body.pageSize,
keyword=body.keyword or body.search_keyword,
@@ -3571,12 +3657,14 @@ async def clue_to_follow(
"""待跟进线索"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
rows, total = list_clues(
db,
tenant_no,
user_id,
user_info=user_info,
page=body.page,
page_size=body.pageSize,
keyword=body.keyword or body.search_keyword,
@@ -3615,12 +3703,14 @@ async def clue_today_new(
"""今日新增线索"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
rows, total = list_clues(
db,
tenant_no,
user_id,
user_info=user_info,
page=body.page,
page_size=body.pageSize,
keyword=body.keyword or body.search_keyword,
@@ -3673,11 +3763,15 @@ async def clue_follow_list(
):
"""线索跟进记录列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
rows, total = list_clue_follows(
db,
body.clue_id,
page=body.page,
page_size=body.pageSize,
user_info=user_info,
tenant_no=tenant_no,
)
enrich_clue_user_names(db, rows)
return success_response({
@@ -3702,6 +3796,7 @@ async def clue_follow_create(
"""新增线索跟进记录"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
user_name = (
_user.get("real_name", "")
@@ -3715,6 +3810,7 @@ async def clue_follow_create(
body,
user_id,
user_name,
user_info=user_info,
)
return success_response(result, message="跟进记录创建成功")
except ValueError as e:
@@ -3731,11 +3827,13 @@ async def clue_follow_delete(
):
"""删除线索跟进记录（支持批量）"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
success_ids = []
fail_items = []
for fid in body.ids:
try:
ok = delete_clue_follow(db, fid)
ok = delete_clue_follow(db, fid, user_info=user_info, tenant_no=tenant_no)
if ok:
success_ids.append(fid)
else:
@@ -3760,10 +3858,14 @@ async def clue_follow_statistics(
):
"""线索跟进统计"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
clue_id = body.get("clueId")
if not clue_id:
return error_response(400, "缺少线索ID")
result = get_clue_follow_statistics(db, clue_id)
result = get_clue_follow_statistics(
db, clue_id, user_info=user_info, tenant_no=tenant_no
)
return success_response(result)
except ValueError as e:
return error_response(400, str(e))
@@ -3826,12 +3928,14 @@ async def clue_export(
"""导出线索列表"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
list_body = body or ClueListBody()
rows, _ = list_clues(
db,
tenant_no,
user_id,
user_info=user_info,
page=1,
page_size=10000,
keyword=list_body.keyword or list_body.search_keyword,
@@ -3991,10 +4095,12 @@ async def clue_follow_detail(
):
"""线索跟进记录详情"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
follow_id = body.get("id")
if not follow_id:
return error_response(400, "缺少跟进记录ID")
detail = get_clue_follow(db, follow_id)
detail = get_clue_follow(db, follow_id, user_info=user_info, tenant_no=tenant_no)
if not detail:
return error_response(404, "跟进记录不存在")
enrich_clue_user_names(db, [detail])
@@ -4013,8 +4119,12 @@ async def clue_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", "")
result = update_clue_follow(db, body.id, body, user_id)
result = update_clue_follow(
db, body.id, body, user_id, user_info=user_info, tenant_no=tenant_no
)
return success_response(result, message="更新成功")
except ValueError as e:
return error_response(400, str(e))
@@ -4031,11 +4141,13 @@ async def clue_statistics_overview(
"""线索统计概览"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
user_id = _user.get("user_id", "") or _user.get("id", "")
payload = body or ClueStatisticsBody()
data = get_clue_overview_statistics(
db,
tenant_no,
user_info=user_info,
user_id=user_id,
owner_user_id=payload.owner_user_id,
begin_time=payload.begin_time,
@@ -4057,9 +4169,14 @@ async def clue_statistics_source(
"""线索来源统计"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
payload = body or ClueStatisticsBody()
data = get_clue_source_statistics(
db, tenant_no, begin_time=payload.begin_time, end_time=payload.end_time
db,
tenant_no,
begin_time=payload.begin_time,
end_time=payload.end_time,
user_info=user_info,
)
return success_response(data)
except ValueError as e:
@@ -4076,7 +4193,8 @@ async def clue_statistics_level(
"""线索等级分布"""
try:
tenant_no = get_tenant_no(db, _user)
data = get_clue_level_statistics(db, tenant_no)
user_info = crm_user_info(_user)
data = get_clue_level_statistics(db, tenant_no, user_info=user_info)
return success_response(data)
except ValueError as e:
return error_response(400, str(e))
@@ -4093,11 +4211,13 @@ async def clue_statistics_trend(
"""线索转化趋势"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
payload = body or ClueStatisticsBody()
time_unit = payload.time_unit or payload.type or "day"
data = get_clue_trend_statistics(
db,
tenant_no,
user_info=user_info,
begin_time=payload.begin_time,
end_time=payload.end_time,
time_unit=time_unit,
@@ -4118,9 +4238,14 @@ async def clue_statistics_owner(
"""负责人线索统计"""
try:
tenant_no = get_tenant_no(db, _user)
user_info = crm_user_info(_user)
payload = body or ClueStatisticsBody()
data = get_clue_owner_statistics(
db, tenant_no, begin_time=payload.begin_time, end_time=payload.end_time
db,
tenant_no,
begin_time=payload.begin_time,
end_time=payload.end_time,
user_info=user_info,
)
return success_response(data)
except ValueError as e:
backend/app/crud/crm/clue.py
+42 -4
View File
@@ -11,6 +11,11 @@ from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.crm_models import CrmClue
from app.models.models import SysUser
from app.crud.crm.data_scope_helpers import apply_crm_owner_scope, ensure_crm_owner_row_in_scope
def _ensure_clue_in_scope(db: Session, clue: CrmClue, user_info, tenant_no: str) -> None:
ensure_crm_owner_row_in_scope(db, clue, user_info, tenant_no, message="无权访问该线索")
def total_pages(total: int, page_size: int) -> int:
@@ -160,6 +165,7 @@ def list_clues(
owner_user_id: str = None,
create_time_start: str = None,
create_time_end: str = None,
user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
q = db.query(CrmClue).filter(
CrmClue.tenant_no == tenant_no,
@@ -216,6 +222,9 @@ def list_clues(
if create_time_end:
q = q.filter(CrmClue.create_time <= create_time_end)
if user_info:
q = apply_crm_owner_scope(q, CrmClue, user_info, db, tenant_no)
total = q.order_by(None).with_entities(func.count(CrmClue.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
@@ -312,13 +321,19 @@ def create_clue(
def update_clue(
db: Session, clue_id: str, data: dict, user_id: str = None, tenant_no: str = None
db: Session,
clue_id: str,
data: dict,
user_id: str = None,
tenant_no: str = None,
user_info=None,
) -> CrmClue:
obj = get_clue(db, clue_id)
if not obj:
raise ValueError("线索不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, obj, user_info, obj.tenant_no)
if obj.is_converted == 1:
raise ValueError("已转化的线索不可修改")
@@ -370,13 +385,18 @@ def update_clue(
def soft_delete_clue(
db: Session, clue_id: str, user_id: str = None, tenant_no: str = None
db: Session,
clue_id: str,
user_id: str = None,
tenant_no: str = None,
user_info=None,
) -> bool:
obj = get_clue(db, clue_id)
if not obj:
raise ValueError("线索不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, obj, user_info, obj.tenant_no)
if obj.is_converted == 1:
raise ValueError("已转化的线索不可删除")
obj.del_flag = 1
@@ -394,6 +414,7 @@ def assign_clues(
current_user_id: str = None,
remark: str = None,
tenant_no: str = None,
user_info=None,
) -> int:
owner_name = (owner_user_name or "").strip()
if not owner_name:
@@ -413,6 +434,7 @@ def assign_clues(
now = datetime.now()
remark_text = (remark or "").strip()
for row in rows:
_ensure_clue_in_scope(db, row, user_info, row.tenant_no)
row.owner_user_id = owner_user_id
row.owner_user_name = owner_name
row.assign_time = now
@@ -434,6 +456,7 @@ def convert_clue(
customer_data: dict,
current_user_id: str,
tenant_no: str,
user_info=None,
) -> dict:
from app.crud.crm.contact import create_contact
from app.crud.crm.customer import check_duplicate_customer, generate_customer_no
@@ -444,6 +467,7 @@ def convert_clue(
raise ValueError("线索不存在")
if clue.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, clue, user_info, tenant_no)
_validate_convertible(clue)
if clue.mobile:
@@ -555,12 +579,14 @@ def invalid_clue(
user_id: str = None,
remark: str = None,
tenant_no: str = None,
user_info=None,
) -> CrmClue:
obj = get_clue(db, clue_id)
if not obj:
raise ValueError("线索不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, obj, user_info, obj.tenant_no)
if obj.is_converted == 1:
raise ValueError("已转化的线索不可标记无效")
obj.status = "invalid"
@@ -694,7 +720,12 @@ def import_clues(
def update_clue_level(
db: Session, ids: List[str], level: str, user_id: str = None, tenant_no: str = None
db: Session,
ids: List[str],
level: str,
user_id: str = None,
tenant_no: str = None,
user_info=None,
) -> int:
q = db.query(CrmClue).filter(CrmClue.id.in_(ids), CrmClue.del_flag == 0)
if tenant_no:
@@ -707,6 +738,7 @@ def update_clue_level(
raise ValueError("已转化的线索不可修改等级")
now = datetime.now()
for row in rows:
_ensure_clue_in_scope(db, row, user_info, row.tenant_no)
row.level = level
row.update_by = user_id
row.update_time = now
@@ -715,13 +747,19 @@ def update_clue_level(
def update_clue_status(
db: Session, clue_id: str, status: str, user_id: str = None, tenant_no: str = None
db: Session,
clue_id: str,
status: str,
user_id: str = None,
tenant_no: str = None,
user_info=None,
) -> CrmClue:
obj = get_clue(db, clue_id)
if not obj:
raise ValueError("线索不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, obj, user_info, obj.tenant_no)
if obj.is_converted == 1:
raise ValueError("已转化的线索不可修改状态")
obj.status = status
backend/app/crud/crm/clue_follow.py
+71 -9
View File
@@ -9,6 +9,8 @@ from sqlalchemy import func
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.crud.crm.clue import get_clue, _ensure_clue_in_scope
from app.crud.crm.data_scope_helpers import apply_crm_create_by_scope, ensure_crm_create_by_row_in_scope
from app.models.crm_models import CrmClueFollow
@@ -46,13 +48,15 @@ def create_clue_follow(
data: Any,
user_id: str,
user_name: str,
user_info=None,
) -> Dict[str, Any]:
"""创建线索跟进记录，并级联更新父线索的跟进相关字段"""
from app.models.crm_models import CrmClue
clue = db.query(CrmClue).filter(CrmClue.id == data.clue_id).first()
clue = get_clue(db, data.clue_id)
if not clue:
raise ValueError("线索不存在")
if clue.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, clue, user_info, tenant_no)
if clue.is_converted == 1:
raise ValueError("已转化的线索不允许添加跟进记录")
@@ -97,6 +101,8 @@ def update_clue_follow(
follow_id: str,
data: Any,
user_id: str,
user_info=None,
tenant_no: str = None,
) -> Dict[str, Any]:
"""更新线索跟进记录，必要时同步线索 next_time"""
from app.models.crm_models import CrmClue
@@ -107,6 +113,11 @@ def update_clue_follow(
).first()
if not row:
raise ValueError("跟进记录不存在")
if tenant_no and row.tenant_no != tenant_no:
raise ValueError("跟进记录不存在")
ensure_crm_create_by_row_in_scope(
db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
)
if data.follow_type is not None:
row.follow_type = data.follow_type
@@ -136,12 +147,24 @@ def list_clue_follows(
clue_id: str,
page: int = 1,
page_size: int = 10,
user_info=None,
tenant_no: str = None,
) -> Tuple[List[Dict[str, Any]], int]:
"""分页查询线索跟进记录"""
clue = get_clue(db, clue_id)
if not clue:
raise ValueError("线索不存在")
if tenant_no and clue.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, clue, user_info, clue.tenant_no)
q = db.query(CrmClueFollow).filter(
CrmClueFollow.clue_id == clue_id,
CrmClueFollow.del_flag == 0,
)
if user_info:
q = apply_crm_create_by_scope(q, CrmClueFollow, user_info, db, clue.tenant_no)
total = q.count()
rows = (
q.order_by(CrmClueFollow.create_time.desc())
@@ -152,16 +175,33 @@ def list_clue_follows(
return [_row_to_dict(r) for r in rows], total
def get_clue_follow(db: Session, follow_id: str) -> Optional[Dict[str, Any]]:
def get_clue_follow(
db: Session,
follow_id: str,
user_info=None,
tenant_no: str = None,
) -> Optional[Dict[str, Any]]:
"""获取单条跟进记录"""
row = db.query(CrmClueFollow).filter(
CrmClueFollow.id == follow_id,
CrmClueFollow.del_flag == 0,
).first()
return _row_to_dict(row) if row else None
if not row:
return None
if tenant_no and row.tenant_no != tenant_no:
return None
ensure_crm_create_by_row_in_scope(
db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
)
return _row_to_dict(row)
def delete_clue_follow(db: Session, follow_id: str) -> bool:
def delete_clue_follow(
db: Session,
follow_id: str,
user_info=None,
tenant_no: str = None,
) -> bool:
"""软删除跟进记录，并回写线索 follow_count / last_time"""
from app.models.crm_models import CrmClue
@@ -171,6 +211,12 @@ def delete_clue_follow(db: Session, follow_id: str) -> bool:
).first()
if not row:
return False
if tenant_no and row.tenant_no != tenant_no:
return False
ensure_crm_create_by_row_in_scope(
db, row, user_info, row.tenant_no, message="无权访问该跟进记录"
)
clue_id = row.clue_id
row.del_flag = 1
row.update_time = datetime.now()
@@ -196,15 +242,31 @@ def delete_clue_follow(db: Session, follow_id: str) -> bool:
return True
def get_follow_statistics(db: Session, clue_id: str) -> Dict[str, Any]:
def get_follow_statistics(
db: Session,
clue_id: str,
user_info=None,
tenant_no: str = None,
) -> Dict[str, Any]:
"""统计线索跟进次数和最后跟进时间"""
result = db.query(
clue = get_clue(db, clue_id)
if not clue:
raise ValueError("线索不存在")
if tenant_no and clue.tenant_no != tenant_no:
raise ValueError("线索不存在")
_ensure_clue_in_scope(db, clue, user_info, clue.tenant_no)
q = db.query(
func.count(CrmClueFollow.id).label("total"),
func.max(CrmClueFollow.create_time).label("lastFollowTime"),
).filter(
CrmClueFollow.clue_id == clue_id,
CrmClueFollow.del_flag == 0,
).first()
)
if user_info:
q = apply_crm_create_by_scope(q, CrmClueFollow, user_info, db, clue.tenant_no)
result = q.first()
return {
"total": result.total or 0,
backend/app/crud/crm/clue_statistics.py
+30 -21
View File
@@ -8,15 +8,18 @@ from sqlalchemy import func
from sqlalchemy.orm import Session
from app.models.crm_models import CrmClue
from app.crud.crm.data_scope_helpers import apply_crm_owner_scope
def _base_query(db: Session, tenant_no: str, owner_user_id: str = None):
def _base_query(db: Session, tenant_no: str, owner_user_id: str = None, user_info=None):
q = db.query(CrmClue).filter(
CrmClue.tenant_no == tenant_no,
CrmClue.del_flag == 0,
)
if owner_user_id:
q = q.filter(CrmClue.owner_user_id == owner_user_id)
if user_info:
q = apply_crm_owner_scope(q, CrmClue, user_info, db, tenant_no)
return q
@@ -35,16 +38,17 @@ def get_clue_overview_statistics(
owner_user_id: str = None,
begin_time: str = None,
end_time: str = None,
user_info=None,
) -> Dict[str, Any]:
"""线索统计概览（默认当前用户负责范围）"""
effective_owner = owner_user_id or user_id
base = _base_query(db, tenant_no, effective_owner)
base = _base_query(db, tenant_no, effective_owner, user_info=user_info)
ranged = _apply_create_range(base, begin_time, end_time)
total_count = ranged.count()
to_follow_count = (
_base_query(db, tenant_no, effective_owner)
_base_query(db, tenant_no, effective_owner, user_info=user_info)
.filter(
CrmClue.is_converted == 0,
CrmClue.next_time.isnot(None),
@@ -53,7 +57,7 @@ def get_clue_overview_statistics(
.count()
)
converted_q = _base_query(db, tenant_no, effective_owner).filter(CrmClue.is_converted == 1)
converted_q = _base_query(db, tenant_no, effective_owner, user_info=user_info).filter(CrmClue.is_converted == 1)
if begin_time:
converted_q = converted_q.filter(CrmClue.converted_time >= begin_time)
if end_time:
@@ -64,7 +68,7 @@ def get_clue_overview_statistics(
invalid_count = (
_apply_create_range(
_base_query(db, tenant_no, effective_owner).filter(CrmClue.status == "invalid"),
_base_query(db, tenant_no, effective_owner, user_info=user_info).filter(CrmClue.status == "invalid"),
begin_time,
end_time,
).count()
@@ -72,7 +76,7 @@ def get_clue_overview_statistics(
a_level_count = (
_apply_create_range(
_base_query(db, tenant_no, effective_owner).filter(CrmClue.level == "A"),
_base_query(db, tenant_no, effective_owner, user_info=user_info).filter(CrmClue.level == "A"),
begin_time,
end_time,
).count()
@@ -80,13 +84,13 @@ def get_clue_overview_statistics(
week_start = datetime.now() - timedelta(days=7)
week_new_count = (
_base_query(db, tenant_no, effective_owner)
_base_query(db, tenant_no, effective_owner, user_info=user_info)
.filter(CrmClue.create_time >= week_start)
.count()
)
follow_status_count = (
_base_query(db, tenant_no, effective_owner)
_base_query(db, tenant_no, effective_owner, user_info=user_info)
.filter(CrmClue.status == "follow", CrmClue.is_converted == 0)
.count()
)
@@ -112,8 +116,9 @@ def get_clue_source_statistics(
tenant_no: str,
begin_time: str = None,
end_time: str = None,
user_info=None,
) -> List[Dict[str, Any]]:
q = _base_query(db, tenant_no)
q = _base_query(db, tenant_no, user_info=user_info)
q = _apply_create_range(q, begin_time, end_time)
rows = (
q.with_entities(CrmClue.source, func.count(CrmClue.id).label("count"))
@@ -125,7 +130,7 @@ def get_clue_source_statistics(
for source, count in rows:
if not source:
continue
converted_q = _base_query(db, tenant_no).filter(
converted_q = _base_query(db, tenant_no, user_info=user_info).filter(
CrmClue.source == source,
CrmClue.is_converted == 1,
)
@@ -148,8 +153,10 @@ def get_clue_source_statistics(
return result
def get_clue_level_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
total = _base_query(db, tenant_no).count()
def get_clue_level_statistics(
db: Session, tenant_no: str, user_info=None
) -> List[Dict[str, Any]]:
total = _base_query(db, tenant_no, user_info=user_info).count()
if total == 0:
return []
@@ -162,11 +169,11 @@ def get_clue_level_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any
]
result: List[Dict[str, Any]] = []
for level, level_name in levels:
count = _base_query(db, tenant_no).filter(CrmClue.level == level).count()
count = _base_query(db, tenant_no, user_info=user_info).filter(CrmClue.level == level).count()
if count == 0 and level == "unrated":
continue
converted_count = (
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(CrmClue.level == level, CrmClue.is_converted == 1)
.count()
)
@@ -191,6 +198,7 @@ def get_clue_trend_statistics(
begin_time: str = None,
end_time: str = None,
time_unit: str = "day",
user_info=None,
) -> List[Dict[str, Any]]:
end_dt = datetime.strptime(end_time, "%Y-%m-%d") if end_time else datetime.now()
start_dt = (
@@ -219,7 +227,7 @@ def get_clue_trend_statistics(
next_point = current + timedelta(days=1)
new_count = (
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(
CrmClue.create_time >= range_start,
CrmClue.create_time <= range_end,
@@ -227,7 +235,7 @@ def get_clue_trend_statistics(
.count()
)
converted_count = (
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(
CrmClue.is_converted == 1,
CrmClue.converted_time >= range_start,
@@ -256,9 +264,10 @@ def get_clue_owner_statistics(
tenant_no: str,
begin_time: str = None,
end_time: str = None,
user_info=None,
) -> List[Dict[str, Any]]:
owners = (
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(CrmClue.owner_user_id.isnot(None), CrmClue.owner_user_id != "")
.with_entities(CrmClue.owner_user_id, CrmClue.owner_user_name)
.group_by(CrmClue.owner_user_id, CrmClue.owner_user_name)
@@ -266,17 +275,17 @@ def get_clue_owner_statistics(
)
result: List[Dict[str, Any]] = []
for owner_id, owner_name in owners:
total_q = _base_query(db, tenant_no).filter(CrmClue.owner_user_id == owner_id)
total_q = _base_query(db, tenant_no, user_info=user_info).filter(CrmClue.owner_user_id == owner_id)
total_q = _apply_create_range(total_q, begin_time, end_time)
total_count = total_q.count()
a_level_count = (
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(CrmClue.owner_user_id == owner_id, CrmClue.level == "A")
.count()
)
converted_q = _base_query(db, tenant_no).filter(
converted_q = _base_query(db, tenant_no, user_info=user_info).filter(
CrmClue.owner_user_id == owner_id,
CrmClue.is_converted == 1,
)
@@ -290,7 +299,7 @@ def get_clue_owner_statistics(
follow_count = (
_apply_create_range(
_base_query(db, tenant_no)
_base_query(db, tenant_no, user_info=user_info)
.filter(
CrmClue.owner_user_id == owner_id,
CrmClue.status == "follow",
backend/app/crud/crm/customer.py
+49 -1
View File
@@ -11,6 +11,15 @@ from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.crm_models import CrmCustomer
from app.models.models import SysUser
from app.crud.crm.data_scope_helpers import (
apply_crm_create_by_scope,
apply_crm_owner_scope,
ensure_crm_owner_row_in_scope,
)
def _ensure_customer_in_scope(db: Session, customer: CrmCustomer, user_info, tenant_no: str) -> None:
ensure_crm_owner_row_in_scope(db, customer, user_info, tenant_no, message="无权访问该客户")
def total_pages(total: int, page_size: int) -> int:
@@ -169,6 +178,8 @@ def list_customers(
owner_user_id: str = None,
create_time_start: str = None,
create_time_end: str = None,
user_info=None,
skip_data_scope: bool = False,
) -> Tuple[List[Dict[str, Any]], int]:
q = db.query(CrmCustomer).filter(
CrmCustomer.tenant_no == tenant_no,
@@ -238,6 +249,10 @@ def list_customers(
if create_time_end:
q = q.filter(CrmCustomer.create_time <= create_time_end)
pool_scope_exempt = skip_data_scope or view_mode == "pool" or status == "pool"
if user_info and not pool_scope_exempt:
q = apply_crm_owner_scope(q, CrmCustomer, user_info, db, tenant_no)
total = q.order_by(None).with_entities(func.count(CrmCustomer.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
@@ -267,6 +282,7 @@ def list_pool_customers(
level: str = None,
industry: str = None,
park_area: str = None,
user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
return list_customers(
db,
@@ -278,6 +294,8 @@ def list_pool_customers(
level=level,
industry=industry,
park_area=park_area,
user_info=user_info,
skip_data_scope=True,
)
@@ -432,11 +450,16 @@ def update_customer(
customer_id: str,
*,
user_id: str = None,
user_info=None,
tenant_no: str = None,
**kwargs,
) -> CrmCustomer:
obj = get_customer(db, customer_id)
if not obj:
raise ValueError("客户不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
_updatable = (
"customer_name", "short_name", "customer_type", "mobile", "telephone",
@@ -469,10 +492,15 @@ def update_customer(
return obj
def soft_delete_customer(db: Session, customer_id: str, user_id: str = None) -> bool:
def soft_delete_customer(
db: Session, customer_id: str, user_id: str = None, user_info=None, tenant_no: str = None
) -> bool:
obj = get_customer(db, customer_id)
if not obj:
raise ValueError("客户不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
obj.del_flag = 1
obj.update_by = user_id
obj.update_time = datetime.now()
@@ -486,10 +514,15 @@ def transfer_customer(
new_owner_id: str,
new_owner_name: str,
user_id: str = None,
user_info=None,
tenant_no: str = None,
) -> CrmCustomer:
obj = get_customer(db, customer_id)
if not obj:
raise ValueError("客户不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
old_owner_id = obj.owner_user_id or ""
old_owner_name = obj.owner_user_name or ""
@@ -523,10 +556,15 @@ def deal_customer(
deal_amount: float = 0,
deal_time: str = None,
user_id: str = None,
user_info=None,
tenant_no: str = None,
) -> CrmCustomer:
obj = get_customer(db, customer_id)
if not obj:
raise ValueError("客户不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
obj.deal_status = "dealed"
if deal_amount:
@@ -545,10 +583,15 @@ def into_pool(
customer_id: str,
reason: str = None,
user_id: str = None,
user_info=None,
tenant_no: str = None,
) -> CrmCustomer:
obj = get_customer(db, customer_id)
if not obj:
raise ValueError("客户不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
old_owner_id = obj.owner_user_id or ""
old_owner_name = obj.owner_user_name or ""
@@ -635,12 +678,17 @@ def assign_customer(
owner_user_id: str,
owner_user_name: str,
user_id: str = None,
user_info=None,
tenant_no: str = None,
) -> List[CrmCustomer]:
results = []
for cid in ids:
obj = get_customer(db, cid)
if not obj:
continue
if tenant_no and obj.tenant_no != tenant_no:
continue
_ensure_customer_in_scope(db, obj, user_info, obj.tenant_no)
old_owner_id = obj.owner_user_id or ""
old_owner_name = obj.owner_user_name or ""
backend/app/crud/crm/customer_follow.py
+28 -5
View File
@@ -9,8 +9,9 @@ from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.crud.crm.customer import _resolve_user_display_name
from app.crud.crm.customer import _resolve_user_display_name, get_customer, _ensure_customer_in_scope
from app.models.crm_models import CrmCustomerFollow
from app.crud.crm.data_scope_helpers import apply_crm_create_by_scope, ensure_crm_create_by_row_in_scope
def total_pages(total: int, page_size: int) -> int:
@@ -67,6 +68,7 @@ def list_follows(
owner_user_id: str = None,
follow_time_start: str = None,
follow_time_end: str = None,
user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
q = db.query(CrmCustomerFollow).filter(
CrmCustomerFollow.tenant_no == tenant_no,
@@ -95,6 +97,9 @@ def list_follows(
if follow_time_end:
q = q.filter(CrmCustomerFollow.follow_time <= follow_time_end)
if user_info:
q = apply_crm_create_by_scope(q, CrmCustomerFollow, user_info, db, tenant_no)
total = q.order_by(None).with_entities(func.count(CrmCustomerFollow.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
@@ -112,7 +117,12 @@ def list_follows_by_customer(
db: Session,
tenant_no: str,
customer_id: str,
user_info=None,
) -> List[Dict[str, Any]]:
cust = get_customer(db, customer_id)
if not cust or cust.tenant_no != tenant_no:
return []
_ensure_customer_in_scope(db, cust, user_info, tenant_no)
rows = (
db.query(CrmCustomerFollow)
.filter(
@@ -144,14 +154,17 @@ def create_follow(
user_id: str = None,
user_name: str = None,
dept_id: str = "",
user_info=None,
**kwargs,
) -> CrmCustomerFollow:
customer_id = kwargs.get("customer_id")
customer_name = kwargs.get("customer_name")
if customer_id and not customer_name:
from app.crud.crm.customer import get_customer
if customer_id:
cust = get_customer(db, customer_id)
if cust:
if not cust or cust.tenant_no != tenant_no:
raise ValueError("客户不存在")
_ensure_customer_in_scope(db, cust, user_info, tenant_no)
if not customer_name:
customer_name = cust.customer_name
follow_time = kwargs.get("follow_time") or datetime.now()
@@ -192,11 +205,16 @@ def update_follow(
follow_id: str,
*,
user_id: str = None,
user_info=None,
tenant_no: str = None,
**kwargs,
) -> CrmCustomerFollow:
obj = get_follow(db, follow_id)
if not obj:
raise ValueError("跟进记录不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("跟进记录不存在")
ensure_crm_create_by_row_in_scope(db, obj, user_info, obj.tenant_no, message="无权访问该跟进记录")
_updatable = (
"contact_id", "contact_name", "follow_type", "follow_content",
@@ -213,10 +231,15 @@ def update_follow(
return obj
def soft_delete_follow(db: Session, follow_id: str, user_id: str = None) -> bool:
def soft_delete_follow(
db: Session, follow_id: str, user_id: str = None, user_info=None, tenant_no: str = None
) -> bool:
obj = get_follow(db, follow_id)
if not obj:
raise ValueError("跟进记录不存在")
if tenant_no and obj.tenant_no != tenant_no:
raise ValueError("跟进记录不存在")
ensure_crm_create_by_row_in_scope(db, obj, user_info, obj.tenant_no, message="无权访问该跟进记录")
obj.del_flag = 1
obj.update_by = user_id
obj.update_time = datetime.now()
backend/app/crud/crm/data_scope_helpers.py
+54
View File
@@ -0,0 +1,54 @@
"""CRM 数据权限辅助（负责人 owner_user_id + 标准 create_by/create_dept_id）。"""
from __future__ import annotations
from typing import Any, Dict, Optional
from sqlalchemy.orm import Session
from app.core.data_scope import apply_data_scope, row_in_data_scope
CRM_OWNER_FIELD = "owner_user_id"
def crm_user_info(user: Optional[dict]) -> Optional[Dict[str, Any]]:
if not user:
return None
uid = (user.get("user_id") or user.get("id") or "").strip()
return {
"user_id": uid or None,
"role": user.get("role"),
"dept_id": user.get("dept_id") or user.get("region_id"),
"region_id": user.get("region_id") or user.get("dept_id"),
}
def apply_crm_owner_scope(query, model, user_info, db: Session, tenant_no: str):
if not user_info:
return query
return apply_data_scope(
query, model, user_info, db, tenant_no, owner_field=CRM_OWNER_FIELD
)
def apply_crm_create_by_scope(query, model, user_info, db: Session, tenant_no: str):
if not user_info:
return query
return apply_data_scope(query, model, user_info, db, tenant_no)
def ensure_crm_owner_row_in_scope(
db: Session, row, user_info, tenant_no: str, *, message: str = "无权访问该数据"
) -> None:
if not user_info or row is None:
return
if not row_in_data_scope(db, row, user_info, tenant_no, owner_field=CRM_OWNER_FIELD):
raise ValueError(message)
def ensure_crm_create_by_row_in_scope(
db: Session, row, user_info, tenant_no: str, *, message: str = "无权访问该数据"
) -> None:
if not user_info or row is None:
return
if not row_in_data_scope(db, row, user_info, tenant_no):
raise ValueError(message)
backend/app/crud/crm/pool_rule.py
+16 -1
View File
@@ -11,6 +11,10 @@ from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.crm_models import CrmCustomer, CrmCustomerPool, CrmCustomerPoolRule
from app.crud.crm.data_scope_helpers import (
apply_crm_create_by_scope,
ensure_crm_create_by_row_in_scope,
)
logger = logging.getLogger(__name__)
@@ -49,6 +53,7 @@ def list_pool_rules(
pool_name: Optional[str] = None,
rule_type: Optional[str] = None,
status: Optional[str] = None,
user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
q = db.query(CrmCustomerPoolRule).filter(
CrmCustomerPoolRule.tenant_no == tenant_no,
@@ -62,6 +67,9 @@ def list_pool_rules(
if status:
q = q.filter(CrmCustomerPoolRule.status == status)
if user_info:
q = apply_crm_create_by_scope(q, CrmCustomerPoolRule, user_info, db, tenant_no)
total = q.count()
q = q.order_by(CrmCustomerPoolRule.sort.asc(), CrmCustomerPoolRule.create_time.desc())
offset = (page - 1) * page_size
@@ -119,6 +127,7 @@ def update_pool_rule(
tenant_no: str,
*,
user_id: Optional[str] = None,
user_info=None,
pool_name: Optional[str] = None,
rule_type: Optional[str] = None,
timeout_days: Optional[int] = None,
@@ -130,6 +139,7 @@ def update_pool_rule(
row = get_pool_rule(db, rule_id, tenant_no)
if not row:
raise ValueError("公海池规则不存在")
ensure_crm_create_by_row_in_scope(db, row, user_info, tenant_no, message="无权访问该公海规则")
if pool_name is not None:
row.pool_name = pool_name
@@ -151,12 +161,15 @@ def update_pool_rule(
return row
def delete_pool_rules(db: Session, rule_ids: List[str], tenant_no: str) -> int:
def delete_pool_rules(
db: Session, rule_ids: List[str], tenant_no: str, user_info=None
) -> int:
count = 0
for rid in rule_ids:
row = get_pool_rule(db, rid, tenant_no)
if not row:
continue
ensure_crm_create_by_row_in_scope(db, row, user_info, tenant_no, message="无权访问该公海规则")
if row.status == "enable":
raise ValueError(f"公海池规则【{row.pool_name}】正在启用中，请先禁用后再删除")
row.del_flag = 1
@@ -173,10 +186,12 @@ def update_pool_rule_status(
status: str,
*,
user_id: Optional[str] = None,
user_info=None,
) -> CrmCustomerPoolRule:
row = get_pool_rule(db, rule_id, tenant_no)
if not row:
raise ValueError("公海池规则不存在")
ensure_crm_create_by_row_in_scope(db, row, user_info, tenant_no, message="无权访问该公海规则")
row.status = status
row.update_by = user_id
row.update_time = datetime.now()
backend/app/crud/crm/statistics.py
+38 -32
View File
@@ -8,16 +8,24 @@ from sqlalchemy import and_, func
from sqlalchemy.orm import Session
from app.models.crm_models import CrmCustomer, CrmCustomerFollow
from app.crud.crm.data_scope_helpers import apply_crm_create_by_scope, apply_crm_owner_scope
def get_overview_statistics(
db: Session, tenant_no: str, user_id: str = None
) -> Dict[str, Any]:
"""客户统计概览：客户总数、公海池数、本月新增、成交数"""
base = db.query(CrmCustomer).filter(
def _customer_base_query(db: Session, tenant_no: str, user_info=None):
q = db.query(CrmCustomer).filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
)
if user_info:
q = apply_crm_owner_scope(q, CrmCustomer, user_info, db, tenant_no)
return q
def get_overview_statistics(
db: Session, tenant_no: str, user_id: str = None, user_info=None
) -> Dict[str, Any]:
"""客户统计概览：客户总数、公海池数、本月新增、成交数"""
base = _customer_base_query(db, tenant_no, user_info=user_info)
total = base.count()
pool_count = base.filter(CrmCustomer.status == "pool").count()
@@ -35,42 +43,39 @@ def get_overview_statistics(
}
def get_source_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
def get_source_statistics(
db: Session, tenant_no: str, user_info=None
) -> List[Dict[str, Any]]:
"""按客户来源统计"""
rows = (
db.query(CrmCustomer.source, func.count(CrmCustomer.id).label("count"))
.filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
)
_customer_base_query(db, tenant_no, user_info=user_info)
.with_entities(CrmCustomer.source, func.count(CrmCustomer.id).label("count"))
.group_by(CrmCustomer.source)
.all()
)
return [{"source": r.source or "未知", "count": r.count} for r in rows]
def get_level_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
def get_level_statistics(
db: Session, tenant_no: str, user_info=None
) -> List[Dict[str, Any]]:
"""按客户级别统计"""
rows = (
db.query(CrmCustomer.level, func.count(CrmCustomer.id).label("count"))
.filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
)
_customer_base_query(db, tenant_no, user_info=user_info)
.with_entities(CrmCustomer.level, func.count(CrmCustomer.id).label("count"))
.group_by(CrmCustomer.level)
.all()
)
return [{"level": r.level or "未知", "count": r.count} for r in rows]
def get_industry_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
def get_industry_statistics(
db: Session, tenant_no: str, user_info=None
) -> List[Dict[str, Any]]:
"""按行业统计"""
rows = (
db.query(CrmCustomer.industry, func.count(CrmCustomer.id).label("count"))
.filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
)
_customer_base_query(db, tenant_no, user_info=user_info)
.with_entities(CrmCustomer.industry, func.count(CrmCustomer.id).label("count"))
.group_by(CrmCustomer.industry)
.all()
)
@@ -82,14 +87,13 @@ def get_trend_statistics(
tenant_no: str,
start_date: str = None,
end_date: str = None,
user_info=None,
) -> List[Dict[str, Any]]:
"""成交趋势（按月统计成交客户数）"""
query = db.query(
query = _customer_base_query(db, tenant_no, user_info=user_info).with_entities(
func.date_format(CrmCustomer.deal_time, "%Y-%m").label("month"),
func.count(CrmCustomer.id).label("count"),
).filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
CrmCustomer.deal_status == "dealed",
CrmCustomer.deal_time.isnot(None),
)
@@ -101,17 +105,16 @@ def get_trend_statistics(
return [{"month": r.month, "count": r.count} for r in rows]
def get_owner_statistics(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
def get_owner_statistics(
db: Session, tenant_no: str, user_info=None
) -> List[Dict[str, Any]]:
"""按负责人统计客户数"""
rows = (
db.query(
_customer_base_query(db, tenant_no, user_info=user_info)
.with_entities(
CrmCustomer.owner_user_name,
func.count(CrmCustomer.id).label("count"),
)
.filter(
CrmCustomer.tenant_no == tenant_no,
CrmCustomer.del_flag == 0,
)
.group_by(CrmCustomer.owner_user_id, CrmCustomer.owner_user_name)
.all()
)
@@ -123,12 +126,15 @@ def get_follow_statistics(
tenant_no: str,
start_date: str = None,
end_date: str = None,
user_info=None,
) -> Dict[str, Any]:
"""跟进统计"""
base = db.query(CrmCustomerFollow).filter(
CrmCustomerFollow.tenant_no == tenant_no,
CrmCustomerFollow.del_flag == 0,
)
if user_info:
base = apply_crm_create_by_scope(base, CrmCustomerFollow, user_info, db, tenant_no)
if start_date:
base = base.filter(CrmCustomerFollow.follow_time >= start_date)
if end_date: