Hermes CI Fix
8fa6c7f520
R2 rule-gate auto-promotion + R3 exception interception in the same collect transaction; merge duplicate rows, keep single verified main per (kpi_id, period); reuse kpi_bep_deriver._is_period_closed; no schema change (data_status/remark only). Task: finance-20260914-autocollect-reviewflow-impl
493 lines
22 KiB
Python
493 lines
22 KiB
Python
"""KPI实际值自动归集采集器 — 管理会计OS (P1-④ 2026-08-28)
|
||
|
||
按 kpi_value_sources 取数映射，从源头表(科目余额/进销存/现金流水)自动汇总写入 kpi_values。
|
||
- 源头: voucher_details(网银凭证明细) / product_inventory(库存汇总) / product_inventory_detail(库存明细) / cash_plans(收付款计划)
|
||
- 严格按 entity_id + period 过滤，避免跨账套/跨期串数
|
||
- 幂等: 同 kpi_id+period 已有 auto_collect 记录则更新；人工 excel/manual 写入不覆盖
|
||
- 调度: 系统 crontab 每日 06:30 (参考 auto_verify_cron.py 模式)
|
||
|
||
用法:
|
||
/usr/bin/python3 scripts/kpi_value_collector.py # 全量采集当月
|
||
/usr/bin/python3 scripts/kpi_value_collector.py 2026-08 # 指定期间
|
||
/usr/bin/python3 scripts/kpi_value_collector.py 2026-08 5 # 指定期间+KPI
|
||
"""
|
||
import os
|
||
import sys
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Optional, Tuple
|
||
|
||
# 复核闸门开关（R2/R3，财务裁定 2026-09-14；KPI_REVIEW_GATE=off 可关闭，仅供测试）
|
||
_REVIEW_GATE_ENABLED = os.environ.get("KPI_REVIEW_GATE", "on").strip().lower() not in ("0", "off", "false", "no")
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
)
|
||
logger = logging.getLogger("cma.kpi_collector")
|
||
|
||
# 保证从 backend 目录直接运行时能 import app
|
||
if __name__ == "__main__":
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from sqlalchemy import and_, func, not_, or_
|
||
from sqlalchemy.orm import aliased
|
||
from app.database import get_session_local
|
||
from app.models import (
|
||
KPIValueSource, KPIValueCollectLog, KPIValue, KPIDefinition,
|
||
VoucherDetail, CashPlan,
|
||
)
|
||
|
||
|
||
# 源头表 → ORM模型映射（动态 import 避免循环依赖）
|
||
def _source_model(table: str):
|
||
if table == "voucher_details":
|
||
return VoucherDetail
|
||
if table == "cash_plans":
|
||
return CashPlan
|
||
# product_inventory / product_inventory_detail 无ORM模型 → SQLAlchemy Table 反射
|
||
from sqlalchemy import Table, MetaData
|
||
from app.database import get_engine
|
||
md = MetaData()
|
||
return Table(table, md, autoload_with=get_engine())
|
||
|
||
|
||
def _field_expression(model, field: str, aggregate: str = "sum"):
|
||
"""聚合表达式: sum/avg/count/max/min"""
|
||
col = getattr(model, field)
|
||
if aggregate == "count":
|
||
return func.count(col)
|
||
if aggregate == "avg":
|
||
return func.avg(col)
|
||
if aggregate == "max":
|
||
return func.max(col)
|
||
if aggregate == "min":
|
||
return func.min(col)
|
||
return func.sum(col)
|
||
|
||
|
||
def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: bool = True) -> Tuple[Optional[float], str]:
|
||
"""执行单条取数映射，返回 (采集值, 说明)。write_kpi=False 时为试跑模式（不写库）。"""
|
||
table = mapping.source_table
|
||
field = mapping.source_field
|
||
aggregate = mapping.aggregate or "sum"
|
||
filter_rule = mapping.filter_rule or {}
|
||
period_field = mapping.period_field or "period"
|
||
unit = mapping.unit_conversion or 1
|
||
|
||
# 现金流两族KPI 激活(2026-09-14): 现金流量表直接法·对手方分类口径
|
||
# filter_rule 形如 {"calc":"cashflow_direct","bucket":"operating","cash_subjects":["1001%","1002%","1012%"],
|
||
# "financing_counterpart":[...],"investing_counterpart":[...],"round":4,
|
||
# "minus_capex":{"counterpart":[...],"side":"debit"}}
|
||
if isinstance(filter_rule, dict) and filter_rule.get("calc") == "cashflow_direct":
|
||
return collect_cashflow_direct(db, mapping, period, write_kpi=write_kpi)
|
||
|
||
# P1③④ 激活(2026-09-14): 科目组合映射 —— 由 kpi_subject_map 推导的多科目加权归集
|
||
# filter_rule 形如 {"subject_codes":["5001%"],"measure":"credit","weights":{"5001%":1.0},
|
||
# "calc":"sum|ratio","denominator":"5001%","scale":100}
|
||
if isinstance(filter_rule, dict) and filter_rule.get("subject_codes"):
|
||
return collect_subject_group(db, mapping, period, write_kpi=write_kpi)
|
||
|
||
model = _source_model(table)
|
||
|
||
# 构建查询
|
||
col = getattr(model, field, None)
|
||
if col is None:
|
||
return None, f"字段 {field} 不存在于表 {table}"
|
||
|
||
q = db.query(_field_expression(model, field, aggregate))
|
||
|
||
# entity_id 过滤（所有源头表都有）
|
||
q = q.filter(model.entity_id == mapping.entity_id)
|
||
|
||
# 期间过滤
|
||
if period_field == "voucher_date":
|
||
# voucher_date 是 DATE 类型 → 按 %Y-%m 前缀匹配
|
||
q = q.filter(func.date_format(model.voucher_date, "%Y-%m") == period)
|
||
else:
|
||
pf = getattr(model, period_field, None)
|
||
if pf is None:
|
||
return None, f"期间字段 {period_field} 不存在于表 {table}"
|
||
q = q.filter(pf == period)
|
||
|
||
# 过滤规则: subject_code / direction / plan_type / carry_forward
|
||
subject_code = filter_rule.get("subject_code")
|
||
if subject_code and hasattr(model, "subject_code"):
|
||
q = q.filter(model.subject_code == subject_code)
|
||
direction = filter_rule.get("direction")
|
||
if direction:
|
||
# direction 覆盖: credit→只算贷方, debit→只算借方
|
||
if direction == "credit" and hasattr(model, "credit_amount"):
|
||
q = db.query(_field_expression(model, "credit_amount", aggregate))
|
||
q = q.filter(model.entity_id == mapping.entity_id)
|
||
if period_field == "voucher_date":
|
||
q = q.filter(func.date_format(model.voucher_date, "%Y-%m") == period)
|
||
else:
|
||
q = q.filter(getattr(model, period_field) == period)
|
||
field = "credit_amount"
|
||
elif direction == "debit" and hasattr(model, "debit_amount"):
|
||
q = db.query(_field_expression(model, "debit_amount", aggregate))
|
||
q = q.filter(model.entity_id == mapping.entity_id)
|
||
if period_field == "voucher_date":
|
||
q = q.filter(func.date_format(model.voucher_date, "%Y-%m") == period)
|
||
else:
|
||
q = q.filter(getattr(model, period_field) == period)
|
||
field = "debit_amount"
|
||
plan_type = filter_rule.get("plan_type")
|
||
if plan_type and hasattr(model, "plan_type"):
|
||
q = q.filter(model.plan_type == plan_type)
|
||
if filter_rule.get("exclude_carry_forward") and hasattr(model, "carry_forward"):
|
||
q = q.filter(model.carry_forward == 0)
|
||
|
||
value = q.scalar()
|
||
if value is None:
|
||
# 源表无记录 → 不写 0，交由 run_collector 记 skipped，避免污染 KPI 实际值
|
||
return None, "源表无记录"
|
||
value = float(value)
|
||
value = round(value * unit, 2)
|
||
|
||
message = f"表{table}.{field} {aggregate}(period={period}) × {unit}"
|
||
if subject_code:
|
||
message += f", 科目{subject_code}"
|
||
if direction:
|
||
message += f", 方向{direction}"
|
||
if plan_type:
|
||
message += f", 类型{plan_type}"
|
||
|
||
if write_kpi:
|
||
_upsert_auto_value(db, mapping, period, value, f"自动归集: {table}")
|
||
|
||
return value, message
|
||
|
||
|
||
def _upsert_auto_value(db, mapping, period: str, value: float, remark: str) -> None:
|
||
"""upsert kpi_values: 同 kpi_id+period 已有 auto_collect 记录则更新。
|
||
|
||
人工来源(excel/manual)行不匹配本查询 → 自动归集永不覆盖人工录入值。
|
||
|
||
复核闸门（财务裁定 2026-09-14，规则 R2/R3）：落 pending 后**同事务**判定是否转正 /
|
||
归并 / 进例外；判定异常一律 fail-safe 保持 pending（N 条转正条件未证成就不转正）。
|
||
"""
|
||
existing = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == mapping.kpi_id,
|
||
KPIValue.period == period,
|
||
KPIValue.source_type == "auto_collect",
|
||
).first()
|
||
if existing:
|
||
existing.actual_value = value
|
||
existing.source_batch = _batch_no()
|
||
existing.remark = remark
|
||
existing.data_status = "pending"
|
||
kv = existing
|
||
else:
|
||
kv = KPIValue(
|
||
entity_id=mapping.entity_id,
|
||
kpi_id=mapping.kpi_id,
|
||
period=period,
|
||
actual_value=value,
|
||
source_type="auto_collect",
|
||
source_batch=_batch_no(),
|
||
data_status="pending",
|
||
remark=remark,
|
||
)
|
||
db.add(kv)
|
||
db.flush() # 取 id（幂等去重与 remark 留痕需要）
|
||
_review_gate(db, mapping, period, kv)
|
||
|
||
|
||
def _review_gate(db, mapping, period: str, kv) -> None:
|
||
"""R2/R3 复核闸门：同事务判定并落状态。任何异常都不阻断采集（保持 pending）。"""
|
||
if not _REVIEW_GATE_ENABLED:
|
||
return
|
||
try:
|
||
try:
|
||
from scripts.kpi_review_gate import review_auto_value
|
||
except ImportError:
|
||
from kpi_review_gate import review_auto_value
|
||
except ImportError as e:
|
||
logger.warning("复核闸门模块不可用（保持 pending）: %s", e)
|
||
return
|
||
try:
|
||
review_auto_value(db, mapping, kv, period)
|
||
except Exception as e:
|
||
logger.warning("复核闸门判定异常 → 保持 pending (kpi=%s period=%s): %s",
|
||
mapping.kpi_id, period, e)
|
||
|
||
|
||
# ── 科目组合归集（P1③④ 激活 · 2026-09-14）─────────────────────────
|
||
# 背景: kpi_subject_map 用新准则科目(6xxx)，voucher_details 是智享通凭证明细(5xxx/1xxx/2xxx)，
|
||
# 编码体系不同 → 由 scripts/kpi_value_source_derive.py 先做「同代码/同名/子目前缀/结转科目」
|
||
# 机械桥接，把**实际存在的科目编码或前缀**写进 filter_rule.subject_codes，再由本函数归集。
|
||
_CREDIT_NORMAL_PREFIXES = ("2", "4", "60", "61", "63")
|
||
|
||
|
||
def _is_credit_normal(subject_code: str) -> bool:
|
||
return str(subject_code).startswith(_CREDIT_NORMAL_PREFIXES)
|
||
|
||
|
||
def _subject_amount(db, entity_id: int, pattern: str, period: str, measure: str):
|
||
"""单科目(支持 LIKE 前缀)期间金额 → (金额, 行数)。measure: credit|debit|net|debit_net"""
|
||
q = db.query(
|
||
func.coalesce(func.sum(VoucherDetail.credit_amount), 0),
|
||
func.coalesce(func.sum(VoucherDetail.debit_amount), 0),
|
||
func.count(VoucherDetail.id),
|
||
).filter(VoucherDetail.entity_id == entity_id)
|
||
if "%" in pattern:
|
||
q = q.filter(VoucherDetail.subject_code.like(pattern))
|
||
else:
|
||
q = q.filter(VoucherDetail.subject_code == pattern)
|
||
if period:
|
||
q = q.filter(VoucherDetail.period == period)
|
||
credit, debit, rows = q.first()
|
||
credit, debit = float(credit or 0), float(debit or 0)
|
||
if measure == "debit":
|
||
amount = debit
|
||
elif measure == "net":
|
||
amount = credit - debit
|
||
elif measure == "debit_net":
|
||
amount = debit - credit
|
||
else: # credit（默认）: 贷方发生额——损益类科目月结"结转"腿与本期发生额同额，取贷方即本期发生额
|
||
amount = credit
|
||
return amount, int(rows or 0)
|
||
|
||
|
||
def collect_subject_group(db, mapping, period: str, write_kpi: bool = True):
|
||
"""多科目加权归集（filter_rule.subject_codes 路径），返回 (采集值, 说明)。"""
|
||
rule = mapping.filter_rule or {}
|
||
codes = list(rule.get("subject_codes") or [])
|
||
weights = rule.get("weights") or {}
|
||
measure = rule.get("measure", "credit")
|
||
calc = rule.get("calc", "sum")
|
||
scale = float(rule.get("scale", 1) or 1)
|
||
unit = mapping.unit_conversion or 1
|
||
|
||
total, rows, parts, matched = 0.0, 0, [], 0
|
||
for code in codes:
|
||
amount, n = _subject_amount(db, mapping.entity_id, code, period, measure)
|
||
if n:
|
||
matched += 1
|
||
rows += n
|
||
weight = float(weights.get(code, 1) or 0)
|
||
total += weight * amount
|
||
parts.append(f"{code}×{weight:g}")
|
||
if rows == 0:
|
||
return None, f"源表无记录(period={period}, {len(codes)}个科目均无发生额)"
|
||
|
||
if calc == "ratio":
|
||
denom_code = rule.get("denominator") or (codes[0] if codes else None)
|
||
if not denom_code:
|
||
return None, "比率映射缺少 denominator"
|
||
denom, _ = _subject_amount(db, mapping.entity_id, str(denom_code), period, measure)
|
||
if not denom:
|
||
return None, f"比率分母 {denom_code} 期间无发生额"
|
||
value = round(float(total) / float(denom) * scale * unit, 4)
|
||
else:
|
||
value = round(float(total) * unit, 2)
|
||
|
||
message = (f"科目组合(period={period}, {measure}, 命中{matched}/{len(codes)}): "
|
||
+ " + ".join(parts))
|
||
if calc == "ratio":
|
||
message += f" ÷ {denom_code} × {scale:g}"
|
||
|
||
if write_kpi:
|
||
_upsert_auto_value(db, mapping, period, value,
|
||
f"自动归集: {mapping.source_table} 科目组合")
|
||
return value, message
|
||
|
||
|
||
# ── 现金流量表直接法·对手方分类归集（2026-09-14 激活）──────────────────
|
||
# 任务: yanxue-budget-cashflow-kpi-activate-20260914（财务Bot 口径说明 §二/§三/§五）
|
||
# 口径: 以**凭证为单元**取含现金类科目(1001%/1002%/1012%)的凭证 → 对手方**行级**分类
|
||
# (筹资类/投资类/其余=经营)；经营净额 = Σ(经营类对手方行净额 × -1)
|
||
# 恒等式: 经营净额 + 筹资净额 + 投资净额 = 现金总净额
|
||
# 关键: 必须行级拆分，不能按凭证级分类 —— 反例 2026-06 凭证 31226 = 还借款 2,550,000
|
||
# + 手续费用 18,700 同单，整单归筹资会把 18,700 手续费误算作经营（高估 18,700）。
|
||
# 纪律: 不硬编码凭证号/常量；对手方集合、bucket、舍位全部来自 filter_rule。
|
||
_DEFAULT_CASH_SUBJECTS = ("1001%", "1002%", "1012%")
|
||
_CASHFLOW_BUCKETS = ("operating", "financing", "investing", "net_cash")
|
||
|
||
|
||
def _cashflow_voucher_subquery(db, entity_id: int, period: Optional[str], cash_subjects):
|
||
"""含现金类科目的凭证键集合（entity_id + voucher_date + voucher_no + period）。"""
|
||
q = db.query(
|
||
VoucherDetail.entity_id.label("entity_id"),
|
||
VoucherDetail.voucher_date.label("voucher_date"),
|
||
VoucherDetail.voucher_no.label("voucher_no"),
|
||
VoucherDetail.period.label("period"),
|
||
).filter(VoucherDetail.entity_id == entity_id)
|
||
if period:
|
||
q = q.filter(VoucherDetail.period == period)
|
||
q = q.filter(or_(*[VoucherDetail.subject_code.like(p) for p in cash_subjects]))
|
||
voucher_cnt = q.distinct().count()
|
||
return q.distinct().subquery(), voucher_cnt
|
||
|
||
|
||
def _cashflow_side_amount(db, entity_id: int, period: Optional[str], cash_subjects,
|
||
include_prefixes=None, exclude_prefixes=None, side: str = "cashflow"):
|
||
"""现金凭证内「对手方行级」金额 → (金额元, 行数, 凭证数)。
|
||
|
||
side="cashflow" : Σ(贷方 − 借方)，即现金流口径（流入为正）
|
||
side="debit_net" : Σ(借方 − 贷方)，即资本支出/借方净额口径（付现为正）
|
||
include_prefixes / exclude_prefixes: 4 位科目前缀，命中即纳入 / 排除（均为 LIKE 前缀，非硬编码）。
|
||
"""
|
||
sub, voucher_cnt = _cashflow_voucher_subquery(db, entity_id, period, cash_subjects)
|
||
c = aliased(VoucherDetail)
|
||
if side == "debit_net":
|
||
expr = func.coalesce(func.sum(c.debit_amount - c.credit_amount), 0)
|
||
else:
|
||
expr = func.coalesce(func.sum(c.credit_amount - c.debit_amount), 0)
|
||
q = db.query(expr, func.count(c.id)).select_from(c).join(
|
||
sub,
|
||
and_(sub.c.entity_id == c.entity_id, sub.c.voucher_date == c.voucher_date,
|
||
sub.c.voucher_no == c.voucher_no, sub.c.period == c.period),
|
||
).filter(c.entity_id == entity_id)
|
||
if period:
|
||
q = q.filter(c.period == period)
|
||
# 只保留对手方行（剔除现金类科目自身）
|
||
q = q.filter(not_(or_(*[c.subject_code.like(p) for p in cash_subjects])))
|
||
if include_prefixes:
|
||
q = q.filter(or_(*[c.subject_code.like(str(p) + "%") for p in include_prefixes]))
|
||
if exclude_prefixes:
|
||
q = q.filter(not_(or_(*[c.subject_code.like(str(p) + "%") for p in exclude_prefixes])))
|
||
row = q.first()
|
||
amount = float(row[0] or 0) if row else 0.0
|
||
rows = int(row[1] or 0) if row else 0
|
||
return amount, rows, voucher_cnt
|
||
|
||
|
||
def collect_cashflow_direct(db, mapping, period: str, write_kpi: bool = True):
|
||
"""现金流量表直接法归集（filter_rule.calc = cashflow_direct），返回 (采集值, 说明)。
|
||
|
||
bucket: operating(经营/默认) | financing(筹资) | investing(投资) | net_cash(现金总净额)
|
||
minus_capex: 付现资本支出扣减项 {"counterpart":[4位前缀...],"side":"debit"}
|
||
"""
|
||
rule = mapping.filter_rule or {}
|
||
cash_subjects = [str(p) for p in (rule.get("cash_subjects") or _DEFAULT_CASH_SUBJECTS)]
|
||
financing = [str(p) for p in (rule.get("financing_counterpart") or [])]
|
||
investing = [str(p) for p in (rule.get("investing_counterpart") or [])]
|
||
bucket = str(rule.get("bucket") or "operating")
|
||
dp = int(rule.get("round", 4) or 4)
|
||
unit = mapping.unit_conversion or 1
|
||
|
||
if bucket not in _CASHFLOW_BUCKETS:
|
||
return None, f"未知 bucket={bucket}（可选 {'/'.join(_CASHFLOW_BUCKETS)}）"
|
||
|
||
if bucket == "net_cash":
|
||
amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects)
|
||
elif bucket == "financing":
|
||
amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects,
|
||
include_prefixes=financing)
|
||
elif bucket == "investing":
|
||
amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects,
|
||
include_prefixes=investing)
|
||
else:
|
||
amount, rows, vouchers = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects,
|
||
exclude_prefixes=financing + investing)
|
||
|
||
if vouchers == 0 or rows == 0:
|
||
return None, f"源表无记录(现金流量直接法 period={period}, bucket={bucket})"
|
||
|
||
value_yuan = amount
|
||
parts = [f"{bucket} {amount:.2f}元"]
|
||
capex_rule = rule.get("minus_capex") or {}
|
||
if capex_rule:
|
||
capex_prefixes = [str(p) for p in (capex_rule.get("counterpart") or [])]
|
||
capex_side = "debit_net" if str(capex_rule.get("side") or "debit").startswith("debit") else "cashflow"
|
||
capex, capex_rows, _ = _cashflow_side_amount(db, mapping.entity_id, period, cash_subjects,
|
||
include_prefixes=capex_prefixes, side=capex_side)
|
||
value_yuan = amount - capex
|
||
parts.append(f"资本支出({capex_side}) {capex:.2f}元/{capex_rows}行")
|
||
|
||
value = round(float(value_yuan) * unit, dp)
|
||
message = (f"现金流量直接法(period={period}, bucket={bucket}, 现金凭证{vouchers}张/对手方{rows}行): "
|
||
+ " − ".join(parts) + f" × {unit}")
|
||
if write_kpi:
|
||
_upsert_auto_value(db, mapping, period, value,
|
||
f"自动归集: 现金流量直接法({bucket})")
|
||
return value, message
|
||
|
||
|
||
def _batch_no() -> str:
|
||
return f"auto-{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
|
||
|
||
def run_collector(db, entity_id: Optional[int] = None, period: Optional[str] = None, kpi_id: Optional[int] = None) -> dict:
|
||
"""运行采集器：遍历 active 映射 → 汇总 → upsert kpi_values → 写采集日志"""
|
||
if period is None:
|
||
period = datetime.now().strftime("%Y-%m")
|
||
|
||
query = db.query(KPIValueSource).filter(KPIValueSource.status == "active")
|
||
if entity_id is not None:
|
||
query = query.filter(KPIValueSource.entity_id == entity_id)
|
||
if kpi_id is not None:
|
||
query = query.filter(KPIValueSource.kpi_id == kpi_id)
|
||
mappings = query.all()
|
||
|
||
if not mappings:
|
||
return {"success": True, "collected": 0, "failed": 0, "message": "无激活取数映射"}
|
||
|
||
collected, failed = 0, 0
|
||
errors = []
|
||
for m in mappings:
|
||
try:
|
||
value, message = collect_for_mapping(db, m, period, write_kpi=True)
|
||
if value is None:
|
||
# 源表无记录 → 跳过，不落 pending=0 的 KPIValue
|
||
db.add(KPIValueCollectLog(
|
||
entity_id=m.entity_id,
|
||
kpi_id=m.kpi_id,
|
||
period=period,
|
||
source_table=m.source_table,
|
||
collected_value=None,
|
||
status="skipped",
|
||
message=message,
|
||
))
|
||
continue
|
||
db.add(KPIValueCollectLog(
|
||
entity_id=m.entity_id,
|
||
kpi_id=m.kpi_id,
|
||
period=period,
|
||
source_table=m.source_table,
|
||
collected_value=value,
|
||
status="success",
|
||
message=message,
|
||
))
|
||
collected += 1
|
||
except Exception as e:
|
||
failed += 1
|
||
errors.append({"kpi_id": m.kpi_id, "source_table": m.source_table, "error": str(e)})
|
||
db.add(KPIValueCollectLog(
|
||
entity_id=m.entity_id,
|
||
kpi_id=m.kpi_id,
|
||
period=period,
|
||
source_table=m.source_table,
|
||
collected_value=None,
|
||
status="failed",
|
||
message=str(e)[:500],
|
||
))
|
||
logger.error("采集失败 kpi=%s table=%s: %s", m.kpi_id, m.source_table, e)
|
||
|
||
db.commit()
|
||
logger.info("采集完成: 成功%s 失败%s (period=%s)", collected, failed, period)
|
||
return {
|
||
"success": failed == 0,
|
||
"collected": collected,
|
||
"failed": failed,
|
||
"period": period,
|
||
"errors": errors[:20],
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
period_arg = sys.argv[1] if len(sys.argv) > 1 else None
|
||
kpi_arg = int(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2].isdigit() else None
|
||
db = get_session_local()()
|
||
try:
|
||
r = run_collector(db, period=period_arg, kpi_id=kpi_arg)
|
||
print(f"实际值自动归集完成: 成功{r['collected']} 失败{r['failed']} (period={r.get('period')})")
|
||
for e in r.get("errors", []):
|
||
print(f" 失败: kpi={e['kpi_id']} table={e['source_table']} -> {e['error']}")
|
||
finally:
|
||
db.close()
|