wecom-fullstack
467472311d
task dama-entity-isolation-sweep-20260923 (sections 2-2 4/5, merged by projectBot 2026-09-23 18:2x)
4) operation_logs write side: every audit write carries the account (no DDL, no column added)
- new app/utils/oplog.py::log_op() single wrapper: sets detail.entity_id; scope=global for
by-design non-entity writes (platform dict / role-level); scope=batch + OPLOG_BATCH_NO_ENTITY
sentinel for cross-account batches; no context -> OPLOG_NO_ENTITY sentinel (never guess)
- 16 remaining construction points converted (data/alert_rules/bi_reports/kpi_causality/
bot_bridge x2/budget_generate/driver_budget/kpis _log x3/templates _log x2/erp_sync);
pre-4 state had 20 sites without entity_id, now 0 outside the wrapper
- read side untouched (knowledge_service still user_id fallback); history not backfilled
5) docs/entity-dimension-data-dictionary.md: 8-table three-state registry
(kpi_templates/driver_factor_templates/scenario_suggestions/bot_bridge_config = non-entity;
kpi_subject_map = inherited via kpi_id; analysis_results/subjects = account dimension,
legacy rows still NULL -> other task; operation_logs = non-entity column + detail.entity_id)
Evidence: RED 2 failed/7 passed (reverted call sites) -> GREEN full 1428 passed; 9 new tests
including two read-back rows; app md5 restore verified identical; no cross-account writes
LOCAL commit only - push is gated (ahead contains the P1 entity-dim batch commit fb4609b)
108 lines
4.9 KiB
Python
108 lines
4.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""操作审计写入统一入口（2026-09-23，`dama-entity-isolation-sweep-20260923` §二之二④）。
|
||
|
||
**约定（只动写侧；不加列、不改读侧口径、不回填历史）**：
|
||
|
||
1. 每条 `operation_logs` **尽量带 `detail.entity_id`** —— 语义是「**本动作所属账套**」：
|
||
会话账套（`Depends(get_entity_id)`）/ Bot 入参账套 / 被处理对象自身的账套。
|
||
2. 确实没有账套上下文的写入口 → `entity_id=None`，并打一条 `OPLOG_NO_ENTITY` 警告：
|
||
把覆盖缺口留成**可 grep 的哨兵**（下一轮审计收口有据可查），
|
||
**绝不猜账套** —— 猜错比留空更有害（错账套会让 `knowledge_service` 归集时又串一次）。
|
||
3. 新的审计写入口一律走 `log_op()`，不要新写 `OperationLog(...)` 构造（第 28 处必漏）。
|
||
|
||
**读侧（本轮不动）**：`app/services/knowledge_service.py::extract_events` 仍按
|
||
`user_id → user_entities` 兜底过滤 —— `user_id IS NULL` 的 Bot/脚本日志因此进不了记忆抽取
|
||
（**覆盖缺口，非泄漏**）。`detail.entity_id` 到位后下一轮才切读侧。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
|
||
from app.models import OperationLog
|
||
|
||
logger = logging.getLogger("cma.oplog")
|
||
|
||
#: 哨兵前缀（供 grep / 告警链订阅）：无账套上下文的写入口
|
||
NO_ENTITY_TAG = "OPLOG_NO_ENTITY"
|
||
#: 哨兵前缀：跨账套批量写（一次动作覆盖多账套，无法归单一账套）——覆盖缺口，待下一轮收口
|
||
BATCH_NO_ENTITY_TAG = "OPLOG_BATCH_NO_ENTITY"
|
||
|
||
#: `scope` 取值：显式声明「该写入口不是账套维度」（平台字典表 / 角色级配置），不再报缺口
|
||
SCOPE_GLOBAL = "global"
|
||
#: `scope` 取值：一次动作跨多账套批量（无法归单账套）——照实报警，不猜账套
|
||
SCOPE_BATCH = "batch"
|
||
|
||
|
||
def _as_detail_dict(detail) -> dict:
|
||
"""把三种历史 detail 写法归一到 dict（**不改变已有存储形态语义**）。
|
||
|
||
- ``dict`` → 直接用（复制一份，不就地改调用方的对象）
|
||
- JSON **对象**字符串（历史写法 ``json.dumps({...})``）→ 解析为 dict，与 entity_id 合并进同一对象
|
||
- 其他字符串 → ``{"note": <原文>}``（原先以标量字符串存的行，升级为对象并附账套）
|
||
"""
|
||
if detail is None:
|
||
return {}
|
||
if isinstance(detail, dict):
|
||
return dict(detail)
|
||
if isinstance(detail, str):
|
||
s = detail.strip()
|
||
if s.startswith("{"):
|
||
try:
|
||
v = json.loads(s)
|
||
except (ValueError, TypeError):
|
||
v = None
|
||
if isinstance(v, dict):
|
||
return v
|
||
return {"note": detail}
|
||
return {"note": str(detail)}
|
||
|
||
|
||
def log_op(db, action, *, target_type=None, target_id=None, detail=None,
|
||
user_id=None, entity_id=None, scope=None, commit=False):
|
||
"""写一条 `operation_logs`，统一带账套（`detail.entity_id`）。
|
||
|
||
:param db: SQLAlchemy Session
|
||
:param action: 动作名（DB 非空列）
|
||
:param target_type: 目标类型（DB 非空列）
|
||
:param target_id: 目标 id（可空）
|
||
:param detail: dict（推荐）/ JSON 对象字符串 / 普通 str（存为 ``{"note": <原文>}``）/ None
|
||
:param user_id: 操作人；Bot/脚本无用户时 None（读侧本轮仍按 user_id 兜底）
|
||
:param entity_id: **本动作所属账套**；无上下文时 None（触发 OPLOG_NO_ENTITY 哨兵）
|
||
:param scope: ``SCOPE_GLOBAL`` = 该写入口按设计非账套维度（写 ``detail.entity_scope``，不报警）；
|
||
``SCOPE_BATCH`` = 跨账套批量（写 ``entity_scope=batch`` **并**报警，不猜账套）；
|
||
None = 未知（报警，留可 grep 的覆盖缺口）
|
||
:param commit: 是否立即提交（默认 False，交由调用方事务边界）
|
||
"""
|
||
d = _as_detail_dict(detail)
|
||
|
||
# 调用方若已把账套写在 detail 里（历史写法），沿用之，避免同一个键两套来源
|
||
if entity_id is None and d.get("entity_id") is not None:
|
||
entity_id = d["entity_id"]
|
||
|
||
if entity_id is not None:
|
||
d["entity_id"] = int(entity_id)
|
||
elif scope == SCOPE_GLOBAL:
|
||
d["entity_scope"] = SCOPE_GLOBAL
|
||
elif scope == SCOPE_BATCH:
|
||
d["entity_scope"] = SCOPE_BATCH
|
||
logger.warning(
|
||
"%s action=%s target_type=%s target_id=%s — 跨账套批量写，detail.entity_id 留空"
|
||
"（禁止猜账套）", BATCH_NO_ENTITY_TAG, action, target_type, target_id)
|
||
else:
|
||
logger.warning(
|
||
"%s action=%s target_type=%s target_id=%s — 该写入口无账套上下文，"
|
||
"detail.entity_id 留空（禁止猜账套）", NO_ENTITY_TAG, action, target_type, target_id)
|
||
|
||
row = OperationLog(
|
||
user_id=user_id,
|
||
action=action,
|
||
target_type=target_type,
|
||
target_id=target_id,
|
||
detail=d or None,
|
||
)
|
||
db.add(row)
|
||
if commit:
|
||
db.commit()
|
||
return row
|