zhanghongyu_0626
793c0307f8
- Updated `RiskListAccess` and `ThresholdWriteAccess` to enforce access control in the risk repository and threshold repository, ensuring only authorized roles can perform sensitive operations. - Introduced new methods in `RiskRepository` for counting pending alerts and listing alerts with access checks, improving data security and compliance. - Enhanced the `chat.py` and `deps.py` files to integrate compliance roles into the risk management matrix, allowing for more granular access control. - Updated documentation to reflect the new testing baseline of 825 passed tests, indicating improved stability and functionality across the application. This update significantly strengthens the risk management capabilities, ensuring robust access control and compliance with organizational policies.
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""客户亏损阈值配置与提醒留痕（jinrong_agent.customer_threshold_config / customer_notify_log）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from app.repository.repo_access import ThresholdWriteAccess
|
|
from sqlalchemy import text
|
|
|
|
from app.config.database import get_agent_engine
|
|
|
|
|
|
class ThresholdRepository:
|
|
def __init__(self, engine=None) -> None:
|
|
self._engine = engine or get_agent_engine()
|
|
|
|
def list_enabled(self, customer_id: str) -> list[dict[str, Any]]:
|
|
sql = text(
|
|
"""
|
|
SELECT id, customer_id, scope_type, scope_ref, loss_threshold_pct,
|
|
notify_channel, is_enabled
|
|
FROM customer_threshold_config
|
|
WHERE customer_id = :cid AND is_enabled = 1
|
|
"""
|
|
)
|
|
with self._engine.connect() as conn:
|
|
return [dict(r) for r in conn.execute(sql, {"cid": customer_id}).mappings()]
|
|
|
|
def list_customer_ids_with_threshold(self) -> list[str]:
|
|
sql = text(
|
|
"""
|
|
SELECT DISTINCT customer_id
|
|
FROM customer_threshold_config
|
|
WHERE is_enabled = 1
|
|
"""
|
|
)
|
|
with self._engine.connect() as conn:
|
|
return [str(r[0]) for r in conn.execute(sql).fetchall()]
|
|
|
|
def upsert_portfolio(
|
|
self,
|
|
customer_id: str,
|
|
loss_threshold_pct: Decimal,
|
|
*,
|
|
access: ThresholdWriteAccess,
|
|
) -> int:
|
|
"""组合级阈值：同一客户仅保留一条 portfolio 配置（更新或插入）。"""
|
|
sel = text(
|
|
"""
|
|
SELECT id FROM customer_threshold_config
|
|
WHERE customer_id = :cid AND scope_type = 'portfolio' AND scope_ref IS NULL
|
|
LIMIT 1
|
|
"""
|
|
)
|
|
with self._engine.begin() as conn:
|
|
row = conn.execute(sel, {"cid": customer_id}).mappings().first()
|
|
if row:
|
|
upd = text(
|
|
"""
|
|
UPDATE customer_threshold_config
|
|
SET loss_threshold_pct = :pct, is_enabled = 1
|
|
WHERE id = :id
|
|
"""
|
|
)
|
|
conn.execute(upd, {"pct": loss_threshold_pct, "id": row["id"]})
|
|
return int(row["id"])
|
|
ins = text(
|
|
"""
|
|
INSERT INTO customer_threshold_config
|
|
(customer_id, scope_type, scope_ref, loss_threshold_pct, notify_channel, is_enabled)
|
|
VALUES (:cid, 'portfolio', NULL, :pct, 'app', 1)
|
|
"""
|
|
)
|
|
result = conn.execute(ins, {"cid": customer_id, "pct": loss_threshold_pct})
|
|
return int(result.lastrowid)
|
|
|
|
def insert_notify_log(
|
|
self,
|
|
*,
|
|
customer_id: str,
|
|
trace_id: str,
|
|
threshold_config_id: int | None,
|
|
payload: dict,
|
|
channel: str = "app",
|
|
send_status: str = "sent",
|
|
) -> None:
|
|
sql = text(
|
|
"""
|
|
INSERT INTO customer_notify_log
|
|
(customer_id, trace_id, notify_type, threshold_config_id, payload, channel, send_status)
|
|
VALUES (:cid, :tid, 'loss_threshold', :cfg_id, :payload, :channel, :status)
|
|
"""
|
|
)
|
|
with self._engine.begin() as conn:
|
|
conn.execute(
|
|
sql,
|
|
{
|
|
"cid": customer_id,
|
|
"tid": trace_id or "threshold-check",
|
|
"cfg_id": threshold_config_id,
|
|
"payload": json.dumps(payload, ensure_ascii=False),
|
|
"channel": channel,
|
|
"status": send_status,
|
|
},
|
|
)
|