"""Same read-only query/alignment contract for mock data and approved SQL sources.""" from copy import deepcopy from decimal import Decimal from api import risk_agent_config as catalog from services.assistant_store import uid, digest from services.assistant_materials import NUMERIC, scalar from services.skill_executor import number from api.risk_agent_alignment import MONEY_FIELDS, SCALE def load_catalog(): with catalog._db() as db: return {kind: catalog._all(db, kind) for kind in catalog.SEED} def read_cards(project, components): values, sources = deepcopy(project["values"]), {} for field, refs in project["sources"].items(): sources[field] = [r for r in refs if r["id"] == project["selected_sources"].get(field)] datasets, errors = [], [] if project.get("pending_candidates"): errors.append({"card_id": "materials", "message": "存在尚未人工确认的材料候选,本次仍使用已确认事实"}) samples = {} for (table, rows), mapping in zip(catalog.SAMPLES.items(), catalog.MAPPINGS): row = {key: None for key in rows[0]} row["project_id"] = project["id"] for field, column in mapping.items(): row[column] = values.get(field) row.update(project_id=project["id"]) if table == "project": row["customer_name"] = project["customer"] if table == "finance": row.update(period=project.get("period", "2025"), scope=project.get("scope", "合并")) samples[table] = [row] for card in components.get("legacy_cards", []): connection = next(c for c in components["legacy_connections"] if c["id"] == card["connection"]) if connection.get("data_domain", "demo" if connection.get("type") == "demo" else "production") != project["data_domain"]: raise ValueError("演示数据与正式数据源不可混用") try: rows = catalog.query_card(card, connection, project["id"], demo_samples=samples, exact_decimals=True) records, selected = catalog.align(card, rows, project["id"], project.get("period", "2025")) datasets.append({"card_id": card["id"], "name": card["name"], "version": card["version"], "records": records, "row_hash": digest(rows)}) if selected is None: errors.append({"card_id": card["id"], "message": "多期或多口径记录未选择,未合并计算"}) continue for field, column in card.get("mapping", {}).items(): raw = selected["values"].get(field) if field in NUMERIC: raw = selected["raw"].get(column) unit = card.get("field_units", {}).get(field, card.get("unit") if field in MONEY_FIELDS else None) if raw is not None and unit: raw = number(raw) * SCALE[unit] raw = scalar(field, raw) # Mock SQL uses the actual parsed fixture values; retain their file locators. if connection.get("type") == "demo" and field in sources: same = values.get(field) == raw if field in NUMERIC and values.get(field) is not None and raw is not None: same = number(values[field]) == number(raw) if not same: values[field] = None errors.append({"card_id": card["id"], "message": field + "映射结果与原始材料冲突"}) continue ref = {"id": uid("ev"), "project_id": project["id"], "field": field, "value": raw, "source": card["name"], "locator": "数据卡字段 " + column, "card_id": card["id"], "version": card["version"], "period": selected["period"], "scope": selected["scope"], "unit": selected["units"].get(field), "currency": card.get("currency", "CNY")} same = values.get(field) == raw if field in NUMERIC and values.get(field) is not None and raw is not None: same = number(values[field]) == number(raw) if field in values and not same: values[field] = None errors.append({"card_id": card["id"], "message": field + "多来源冲突,请人工确认"}) else: values[field] = raw sources.setdefault(field, []).append(ref) except Exception as exc: # No connection strings/credentials in project-facing diagnostics. errors.append({"card_id": card["id"], "message": "数据卡读取或口径校验失败", "error_type": type(exc).__name__}) return values, sources, datasets, errors