"""Secure orchestration for one-off annual-report uploads.
The uploaded PDF is written only to a temporary directory because the existing
financial analysis core consumes a filesystem path. The directory is removed
before this function returns; only structured analysis results are retained.
"""
from __future__ import annotations
import hashlib
import copy
import json
import math
import re
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
import pymupdf as fitz
MAX_UPLOAD_BYTES = 50 * 1024 * 1024
MAX_UPLOAD_FILES = 8
MAX_BATCH_UPLOAD_BYTES = 200 * 1024 * 1024
MIN_TEXT_LENGTH = 200
MIN_EXTRACTED_CURRENT_ITEMS = 6
ANALYSIS_SCHEMA_VERSION = "2026-08-07-formal-release-v2"
MANUAL_CORRECTION_METRICS = {
"revenue": "营业收入",
"net_profit": "净利润",
"total_assets": "资产总计",
"total_liabilities": "负债合计",
"total_equity": "所有者权益合计",
"current_assets": "流动资产合计",
"accounts_receivable": "应收账款",
"cost_of_goods_sold": "营业成本",
"cash_flow_operations": "经营活动现金流量净额",
"monetary_funds": "货币资金",
"inventory": "存货",
"short_term_borrowings": "短期借款",
"long_term_borrowings": "长期借款",
}
class UploadedPdfAnalysisError(ValueError):
"""A user-facing upload failure with a stable application error code."""
def __init__(self, code: str, message: str, detail: str = "") -> None:
super().__init__(message)
self.code = code
self.message = message
self.detail = detail
def sanitize_upload_filename(filename: str) -> str:
"""Return a display-safe basename without allowing HTML/path injection."""
basename = Path(str(filename or "uploaded.pdf").replace("\\", "/")).name
safe = re.sub(r"[^\w.\-（）()【】\u4e00-\u9fff ]+", "_", basename, flags=re.UNICODE)
safe = re.sub(r"\s+", " ", safe).strip(" ._")
if not safe:
safe = "uploaded.pdf"
if not safe.lower().endswith(".pdf"):
safe += ".pdf"
return safe[:120]
def uploaded_report_id(pdf_bytes: bytes) -> str:
"""Build a stable, non-identifying report id from the file content."""
return "upload_" + hashlib.sha256(pdf_bytes).hexdigest()[:20]
def extract_report_year(filename: str) -> Optional[str]:
"""Extract the last plausible four-digit reporting year from a filename."""
matches = re.findall(r"(?<!\d)((?:19|20)\d{2})(?!\d)", str(filename or ""))
return matches[-1] if matches else None
def extract_company_hint(filename: str) -> str:
"""Build a conservative company identity key from an upload filename."""
stem = Path(str(filename or "")).stem
stem = re.sub(r"(?<!\d)(?:19|20)\d{2}(?!\d)\s*年?", "", stem)
stem = re.sub(r"(?:年度报告|年报|年度|报告)", "", stem, flags=re.IGNORECASE)
return re.sub(r"[^A-Za-z0-9\u4e00-\u9fff]+", "", stem).lower()
def extract_report_identity(
pages_text: Sequence[Dict[str, Any]],
filename: str,
) -> Dict[str, str]:
"""Extract a best-effort company name/code and retain a filename fallback."""
text = "\n".join(str(page.get("text") or "") for page in list(pages_text or [])[:30])
company_name = ""
stock_code = ""
company_patterns = (
r"(?:公司中文名称|公司的中文名称|公司名称)\s*[:：]?\s*([^\n]{2,60})",
r"中文名称\s*[:：]?\s*([^\n]{2,60})",
)
for pattern in company_patterns:
match = re.search(pattern, text)
if match:
company_name = str(match.group(1)).strip(" ：:|_-")
company_name = re.split(r"(?:英文名称|股票简称|证券简称|股票代码|证券代码)", company_name)[0].strip()
if company_name:
break
code_match = re.search(r"(?:股票代码|证券代码)\s*[:：]?\s*([036689]\d{5})", text)
if code_match:
stock_code = code_match.group(1)
normalized_name = re.sub(r"[^A-Za-z0-9\u4e00-\u9fff]+", "", company_name).lower()
filename_hint = extract_company_hint(filename)
return {
"company_name": company_name,
"stock_code": stock_code,
"company_key": stock_code or normalized_name or filename_hint,
"identity_source": "stock_code" if stock_code else ("report_text" if normalized_name else "filename"),
"filename_hint": filename_hint,
}
def uploaded_batch_report_id(files: Sequence[Tuple[bytes, str]]) -> str:
"""Build an order-independent id for a set of uploaded annual reports."""
fingerprints = sorted(hashlib.sha256(bytes(data or b"")).hexdigest() for data, _ in files)
digest = hashlib.sha256("|".join(fingerprints).encode("ascii")).hexdigest()
return "upload_multi_" + digest[:20]
def _safe_div(numerator: Any, denominator: Any) -> Optional[float]:
if not isinstance(numerator, (int, float)) or not isinstance(denominator, (int, float)) or denominator == 0:
return None
return round(float(numerator) / float(denominator), 6)
def _growth(values: Sequence[Any]) -> List[Optional[float]]:
result: List[Optional[float]] = [0.0] if values else []
for index in range(1, len(values)):
result.append(_safe_div(
(values[index] - values[index - 1]) if isinstance(values[index], (int, float)) and isinstance(values[index - 1], (int, float)) else None,
values[index - 1],
))
return result
def _build_professional_trend(yearly_snapshots: Dict[str, Dict[str, Any]]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""Build both the core trend result and the professional-page chart schema."""
from core.multi_year_trend import MultiYearTrendGenerator
trend_result = MultiYearTrendGenerator().build_from_snapshots(yearly_snapshots)
years = list(trend_result.get("years") or [])
metric_map = {
"revenue": "revenue",
"net_profit": "net_profit",
"cash_flow_ops": "cash_flow_operations",
"accounts_receivable": "accounts_receivable",
"inventory": "inventory",
"monetary_funds": "monetary_funds",
"short_term_borrowings": "short_term_borrowings",
"long_term_borrowings": "long_term_borrowings",
"total_assets": "total_assets",
"total_equity": "total_equity",
"total_liabilities": "total_liabilities",
}
metrics = {target: list(trend_result.get(source) or []) for target, source in metric_map.items()}
revenue = metrics["revenue"]
profit = metrics["net_profit"]
cfo = metrics["cash_flow_ops"]
assets = metrics["total_assets"]
receivables = metrics["accounts_receivable"]
funds = metrics["monetary_funds"]
inventory = metrics["inventory"]
short_debt = metrics["short_term_borrowings"]
long_debt = metrics["long_term_borrowings"]
ratios = {
"cfo_to_net_profit": [_safe_div(cfo[i], profit[i]) for i in range(len(years))],
"ar_to_revenue": [_safe_div(receivables[i], revenue[i]) for i in range(len(years))],
"funds_to_assets": [_safe_div(funds[i], assets[i]) for i in range(len(years))],
"borrowings_to_assets": [
_safe_div(
(short_debt[i] + long_debt[i]) if isinstance(short_debt[i], (int, float)) and isinstance(long_debt[i], (int, float)) else None,
assets[i],
)
for i in range(len(years))
],
"inventory_growth_yoy": _growth(inventory),
"revenue_growth_yoy": _growth(revenue),
}
timeline: List[Dict[str, Any]] = []
first_seen: set[str] = set()
for year in years:
red_flags = (yearly_snapshots.get(str(year), {}).get("red_flags_result") or {})
for flag_id in red_flags.get("triggered_flag_ids") or []:
if flag_id in first_seen:
continue
flag = red_flags.get(flag_id) or {}
detail = flag.get("detail") or {}
trigger_value = next((value for value in detail.values() if isinstance(value, (int, float))), 0)
timeline.append({
"year": int(year) if str(year).isdigit() else str(year),
"flag": flag_id,
"severity": flag.get("severity") or "红旗",
"trigger_value": trigger_value,
"threshold": detail.get("threshold", 0),
"note": flag.get("reason") or "该年度首次触发此红旗因子",
})
first_seen.add(flag_id)
first_warning = timeline[0]["year"] if timeline else None
turning_points = [
{"year": item["year"], "label": f"{item['flag']} 首次触发", "cn": item["note"]}
for item in timeline[:4]
]
professional = {
"company": "用户上传年报",
"source_scope": "user_upload",
"years": [int(year) if str(year).isdigit() else str(year) for year in years],
"explosion_year": None,
"earliest_warning_year": first_warning,
"warning_lead_years": 0,
"metrics": metrics,
"derived_ratios": ratios,
"red_flag_timeline": timeline,
"narrative": {
"hook": f"已对 {len(years)} 个年度的真实上传年报完成逐年模型分析与纵向对比；趋势结论需结合原始披露和人工复核使用。",
"deterioration": "图表按文件名识别的报告年度排序，展示营收、净利润、经营现金流及核心红旗比率的变化。",
"key_turning_points": turning_points,
},
}
return trend_result, professional
def _serializable_preflight(preflight: Dict[str, Any]) -> Dict[str, Any]:
result = dict(preflight or {})
result.pop("pdf_path", None)
order = result.get("column_order_recommendation")
if order is not None:
result["column_order_recommendation"] = getattr(order, "name", str(order))
pairs = result.get("matched_pairs") or []
result["matched_pairs"] = [list(pair) for pair in pairs]
return result
def _validate_pdf_container(data: bytes) -> None:
"""校验 PDF 容器结构与加密状态。
使用 bytes 流打开（``fitz.open(stream=...)``）而非以文件路径打开：
PyMuPDF 1.28 在打开损坏 PDF 失败时，底层句柄在 Windows 上不会及时释放，
会导致上传后临时目录清理抛出 PermissionError（WinError 32）。改用流打开
不持有文件句柄，可同时规避该平台回归，并保持对加密/损坏 PDF 的错误判定。
"""
try:
with fitz.open(stream=data, filetype="pdf") as document:
if document.needs_pass:
raise UploadedPdfAnalysisError(
"E-1004",
"PDF 已加密或受密码保护，当前无法读取文本层。",
)
if document.page_count <= 0:
raise UploadedPdfAnalysisError("E-1002", "PDF 不包含可读取页面。")
except UploadedPdfAnalysisError:
raise
except Exception as exc:
raise UploadedPdfAnalysisError(
"E-1002",
"PDF 文件损坏或不是受支持的标准 PDF。",
type(exc).__name__,
) from exc
def analyze_uploaded_pdf_bytes(
pdf_bytes: bytes,
original_filename: str,
*,
agent: Optional[Any] = None,
phase_callback: Optional[Callable[[str], None]] = None,
) -> Tuple[str, Dict[str, Any]]:
"""Analyze uploaded PDF bytes and return ``(report_id, result)``.
The existing ``FinancialAnalysisAgent.analyze_pdf`` signature is not
changed. Evidence tracing is explicitly enabled for the professional UI,
while full report text is still excluded from the returned result.
"""
data = bytes(pdf_bytes or b"")
if not data:
raise UploadedPdfAnalysisError("E-1002", "上传文件为空。")
if len(data) > MAX_UPLOAD_BYTES:
raise UploadedPdfAnalysisError("E-1001", "PDF 超过 50MB 上传上限。")
if not data.startswith(b"%PDF-"):
raise UploadedPdfAnalysisError("E-1002", "文件头校验失败，不是标准 PDF。")
safe_name = sanitize_upload_filename(original_filename)
report_id = uploaded_report_id(data)
callback = phase_callback or (lambda _message: None)
report_identity = extract_report_identity([], safe_name)
with tempfile.TemporaryDirectory(prefix="auditprobe_upload_") as temp_dir:
pdf_path = Path(temp_dir) / "annual_report.pdf"
pdf_path.write_bytes(data)
callback("正在校验 PDF 文件结构与加密状态…")
_validate_pdf_container(data)
if agent is None:
from core.financial_agent import FinancialAnalysisAgent
agent = FinancialAnalysisAgent()
callback("正在预检文本层与双期财务列…")
preflight = agent.extractor.preflight_check_pdf(str(pdf_path))
text_length = int((preflight.get("detail") or {}).get("text_length") or 0)
if text_length < MIN_TEXT_LENGTH:
raise UploadedPdfAnalysisError(
"E-1003",
"未检测到足够的可解析文本层；纯扫描件请先完成 OCR。",
)
callback("正在提取财务指标并运行 M-Score、红旗、F-Score 与仲裁…")
try:
result = agent.analyze_pdf(str(pdf_path), evidence_enabled=True)
except UploadedPdfAnalysisError:
raise
except Exception as exc:
raise UploadedPdfAnalysisError(
"E-1005",
"PDF 分析过程中发生异常，请尝试其他电子版年报。",
type(exc).__name__,
) from exc
callback("正在核对合并报表、附注及非财务事件披露…")
validation_pages: List[Dict[str, Any]] = []
try:
from core.financial_validation import (
build_data_reliability_gate,
build_financial_validation,
)
validation_pages = agent.extractor.extract_text_with_pages(str(pdf_path))
report_identity = extract_report_identity(validation_pages, safe_name)
result["financial_validation"] = build_financial_validation(
result.get("financial_data") or {}, validation_pages
)
result["data_reliability_gate"] = build_data_reliability_gate(result)
except Exception:
result["financial_validation"] = {
"version": "1.0", "status": "unavailable", "checks": [],
"source_matches": {}, "restatement": {"detected": False, "evidence": []},
"manual_correction_allowed": True,
}
result["data_reliability_gate"] = {
"version": "1.0", "status": "blocked", "score": 0.0,
"formal_hitl_allowed": False, "formal_export_allowed": False,
"draft_export_allowed": True,
"blockers": ["报表/附注来源校验不可用"],
"warnings": [],
"decision_text": "数据可信度硬门禁未通过，仅允许导出待复核草稿。",
}
try:
from core.nonfinancial_signals import extract_nonfinancial_context
result["nonfinancial_context"] = extract_nonfinancial_context(
validation_pages,
report_id=report_id,
financial_validation=result.get("financial_validation") or {},
)
except Exception:
result["nonfinancial_context"] = {
"version": "1.0",
"status": "unavailable",
"policy": "conservative_positive_evidence",
"model_used": False,
"events": [],
"summary": {"total": 0, "confirmed_event": 0, "review_required": 0},
"limitations": ["非财务事件增强层本次不可用，不影响确定性财务结论。"],
}
if not isinstance(result, dict):
raise UploadedPdfAnalysisError("E-1005", "分析器未返回有效结果。")
from core.nonfinancial_signals import build_financial_decision_hash
result["decision_integrity"] = {
"version": "1.0",
"financial_decision_hash": build_financial_decision_hash(result),
"nonfinancial_context_hash": str(
(result.get("nonfinancial_context") or {}).get("nonfinancial_context_hash") or ""
),
"namespace_policy": "independent_content_hashes_shared_append_only_hitl_chain",
"statement": "非财务上下文不参与确定性财务结论计算。",
}
financial_data = result.get("financial_data") or {}
current_count = sum(
1
for value in financial_data.values()
if isinstance(value, (list, tuple)) and value and value[0] is not None
)
previous_count = sum(
1
for value in financial_data.values()
if isinstance(value, (list, tuple)) and len(value) > 1 and value[1] is not None
)
if current_count < MIN_EXTRACTED_CURRENT_ITEMS:
raise UploadedPdfAnalysisError(
"E-1006",
f"仅识别到 {current_count} 项本期财务指标，数据不足以形成可靠分析。",
)
callback("正在整理证据、完整性信息与专业页面数据…")
sha256_hex = hashlib.sha256(data).hexdigest()
result["preflight"] = _serializable_preflight(preflight)
result["_upload_meta"] = {
"analysis_schema_version": ANALYSIS_SCHEMA_VERSION,
"sha256": sha256_hex,
"size_bytes": len(data),
"text_length": text_length,
"extracted_current_items": current_count,
"extracted_previous_items": previous_count,
"extracted_total_items": current_count + previous_count,
"analyzed_at": datetime.now(timezone.utc).isoformat(),
"temporary_pdf_removed": True,
"report_identity": report_identity,
}
result["_demo_meta"] = {
"label": f"📄 {safe_name}（真实上传分析）",
"tagline": "基于本次上传 PDF 的自动抽取结果",
"category": "用户上传年报",
"source_scope": "user_upload",
"cache_file": "",
"pdf_filename": safe_name,
"report_id": report_id,
}
return report_id, result
def analyze_uploaded_pdf_batch(
files: Sequence[Tuple[bytes, str]],
*,
agent: Optional[Any] = None,
phase_callback: Optional[Callable[[str], None]] = None,
) -> Tuple[str, Dict[str, Any]]:
"""Analyze 2-8 annual reports and return the latest year plus a real trend."""
items = [(bytes(data or b""), sanitize_upload_filename(name)) for data, name in files]
if len(items) < 2:
if not items:
raise UploadedPdfAnalysisError("E-1002", "未选择可分析的 PDF 文件。")
return analyze_uploaded_pdf_bytes(
items[0][0], items[0][1], agent=agent, phase_callback=phase_callback
)
if len(items) > MAX_UPLOAD_FILES:
raise UploadedPdfAnalysisError("E-1008", f"一次最多分析 {MAX_UPLOAD_FILES} 份年报。")
if sum(len(data) for data, _ in items) > MAX_BATCH_UPLOAD_BYTES:
raise UploadedPdfAnalysisError("E-1009", "多文件合计大小超过 200MB，请分批处理。")
years: List[str] = []
seen_hashes: set[str] = set()
for data, name in items:
year = extract_report_year(name)
if year is None:
raise UploadedPdfAnalysisError(
"E-1007",
f"无法从文件名“{name}”识别报告年度；多文件分析请在文件名中保留四位年份。",
)
if year in years:
raise UploadedPdfAnalysisError("E-1010", f"检测到重复报告年度 {year}，请每年仅上传一份年报。")
digest = hashlib.sha256(data).hexdigest()
if digest in seen_hashes:
raise UploadedPdfAnalysisError("E-1011", f"检测到重复文件：{name}。")
years.append(year)
seen_hashes.add(digest)
callback = phase_callback or (lambda _message: None)
if agent is None:
from core.financial_agent import FinancialAnalysisAgent
agent = FinancialAnalysisAgent()
yearly_snapshots: Dict[str, Dict[str, Any]] = {}
filenames_by_year: Dict[str, str] = {}
for index, ((data, name), year) in enumerate(sorted(zip(items, years), key=lambda item: item[1]), start=1):
callback(f"正在分析第 {index}/{len(items)} 份：{name}（{year}）…")
_, snapshot = analyze_uploaded_pdf_bytes(
data,
name,
agent=agent,
phase_callback=lambda message, y=year: callback(f"{y}：{message}"),
)
yearly_snapshots[year] = snapshot
filenames_by_year[year] = name
identity_by_year = {}
for year, snapshot in yearly_snapshots.items():
identity = dict(((snapshot.get("_upload_meta") or {}).get("report_identity") or {}))
identity.setdefault("filename_hint", extract_company_hint(filenames_by_year.get(str(year), "")))
identity.setdefault("company_key", identity.get("filename_hint", ""))
identity.setdefault("identity_source", "filename")
identity_by_year[str(year)] = identity
stock_code_values = [str(identity.get("stock_code") or "").strip() for identity in identity_by_year.values()]
company_name_values = [
re.sub(r"[^A-Za-z0-9\u4e00-\u9fff]+", "", str(identity.get("company_name") or "")).lower()
for identity in identity_by_year.values()
]
filename_hint_values = [str(identity.get("filename_hint") or "").strip() for identity in identity_by_year.values()]
if all(stock_code_values):
identity_conflict = len(set(stock_code_values)) > 1
elif all(company_name_values):
identity_conflict = len(set(company_name_values)) > 1
else:
identity_conflict = not all(filename_hint_values) or len(set(filename_hint_values)) > 1
if identity_conflict:
identity_summary = "；".join(
f"{year}:{identity.get('company_name') or identity.get('stock_code') or identity.get('filename_hint') or '未知'}"
for year, identity in sorted(identity_by_year.items())
)
raise UploadedPdfAnalysisError(
"E-1013",
f"检测到多年文件可能不属于同一家公司（{identity_summary}），已阻止合并分析。",
)
callback("正在合并逐年快照并生成多年趋势…")
trend_result, professional_trend = _build_professional_trend(yearly_snapshots)
from core.financial_validation import (
build_batch_data_reliability_gate,
build_cross_period_validation,
)
cross_period_validation = build_cross_period_validation(yearly_snapshots)
yearly_reliability_gates = {
str(year): snapshot.get("data_reliability_gate") or {}
for year, snapshot in yearly_snapshots.items()
}
batch_reliability_gate = build_batch_data_reliability_gate(
yearly_reliability_gates, cross_period_validation
)
successful_years = list(trend_result.get("years") or [])
if len(successful_years) < 2:
raise UploadedPdfAnalysisError("E-1012", "有效年度不足 2 年，无法形成多年趋势。")
latest_year = max(successful_years, key=lambda value: int(value) if str(value).isdigit() else str(value))
result = dict(yearly_snapshots[str(latest_year)])
report_id = uploaded_batch_report_id(items)
batch_sha = hashlib.sha256(
"|".join(sorted(seen_hashes)).encode("ascii")
).hexdigest()
result["trend_result"] = trend_result
result["_multi_year_trend"] = professional_trend
result["cross_period_validation"] = cross_period_validation
result["yearly_reliability_gates"] = yearly_reliability_gates
result["data_reliability_gate"] = batch_reliability_gate
result["multi_year_summary"] = {
"years": [str(year) for year in successful_years],
"latest_year": str(latest_year),
"yearly_risk_levels": {
str(year): ((snapshot.get("hybrid_result") or {}).get("final_risk_level") or "未知")
for year, snapshot in yearly_snapshots.items()
},
"yearly_overview": {
str(year): {
"filename": filenames_by_year.get(str(year), ""),
"final_risk_level": ((snapshot.get("hybrid_result") or {}).get("final_risk_level") or "未知"),
"m_score": (snapshot.get("m_score_result") or {}).get("m_score"),
"red_flag_count": (snapshot.get("red_flags_result") or {}).get("triggered_count", (snapshot.get("red_flags_result") or {}).get("total_red_flags", 0)),
"f_score": (snapshot.get("f_score_result") or {}).get("f_score"),
"completeness_score": snapshot.get("completeness_score"),
"data_gate_status": (snapshot.get("data_reliability_gate") or {}).get("status"),
}
for year, snapshot in yearly_snapshots.items()
},
}
batch_total_seconds = round(sum(
float((snapshot.get("timing") or {}).get("total") or 0)
for snapshot in yearly_snapshots.values()
), 3)
latest_timing = dict(result.get("timing") or {})
latest_timing.update({
"latest_year_total": latest_timing.get("total", 0),
"batch_total": batch_total_seconds,
"total": batch_total_seconds,
})
result["timing"] = latest_timing
upload_meta = dict(result.get("_upload_meta") or {})
upload_meta.update({
"sha256": batch_sha,
"batch_file_count": len(items),
"batch_years": [str(year) for year in successful_years],
"batch_filenames": [filenames_by_year[str(year)] for year in successful_years],
"batch_file_sha256": {
str(year): str(((yearly_snapshots[str(year)].get("_upload_meta") or {}).get("sha256") or ""))
for year in successful_years
},
"report_identity": next(iter(identity_by_year.values()), {}),
"identity_by_year": identity_by_year,
"batch_total_size_bytes": sum(len(data) for data, _ in items),
"temporary_pdf_removed": True,
})
result["_upload_meta"] = upload_meta
result["_demo_meta"] = {
"label": f"📚 多年年报 {successful_years[0]}-{successful_years[-1]}（真实上传分析）",
"tagline": f"基于 {len(successful_years)} 份上传 PDF 的逐年分析与趋势合并",
"category": "用户上传年报",
"source_scope": "user_upload",
"cache_file": "",
"pdf_filename": filenames_by_year[str(latest_year)],
"report_id": report_id,
}
return report_id, result
def _close_amount(left: float, right: float) -> bool:
return abs(left - right) <= max(1.0, 0.001 * max(abs(left), abs(right), 1.0))
def _refresh_latest_trend(result: Dict[str, Any], metric: str, value: float) -> None:
summary = result.get("multi_year_summary") or {}
latest_year = str(summary.get("latest_year") or "")
trend = result.get("trend_result") or {}
years = [str(year) for year in (trend.get("years") or [])]
if latest_year and latest_year in years and isinstance(trend.get(metric), list):
trend[metric][years.index(latest_year)] = value
professional = result.get("_multi_year_trend") or {}
professional_names = {"cash_flow_operations": "cash_flow_ops"}
target = professional_names.get(metric, metric)
professional_years = [str(year) for year in (professional.get("years") or [])]
values = (professional.get("metrics") or {}).get(target)
if latest_year and latest_year in professional_years and isinstance(values, list):
values[professional_years.index(latest_year)] = value
def _refresh_cross_period_after_manual(result: Dict[str, Any], metric: str, period_index: int, value: float) -> None:
if period_index != 1:
return
cross = result.get("cross_period_validation") or {}
latest_year = str((result.get("multi_year_summary") or {}).get("latest_year") or "")
for item in cross.get("comparisons") or []:
if str(item.get("later_year")) != latest_year or item.get("metric") != metric:
continue
item["later_report_comparative"] = value
prior = item.get("prior_report_current")
item["difference"] = None if not isinstance(prior, (int, float)) else value - float(prior)
if not isinstance(prior, (int, float)):
item["status"] = "unavailable"
elif _close_amount(float(prior), value):
item["status"] = "matched"
elif item.get("restatement_evidence_available"):
item["status"] = "restated_comparative"
else:
item["status"] = "mismatch"
comparisons = cross.get("comparisons") or []
cross["summary"] = {status: sum(1 for item in comparisons if item.get("status") == status) for status in (
"matched", "mismatch", "restated_comparative", "unavailable"
)}
def apply_manual_financial_correction(
analyze_result: Dict[str, Any],
metric: str,
period_index: int,
corrected_value: float,
reason: str,
*,
agent: Optional[Any] = None,
operator: str = "当前会话人工复核",
) -> Dict[str, Any]:
"""Apply an auditable session-only correction and recompute all risk models."""
if metric not in MANUAL_CORRECTION_METRICS:
raise ValueError("不支持修复该财务科目。")
if period_index not in (0, 1):
raise ValueError("期间必须为本期或上期。")
if not isinstance(corrected_value, (int, float)) or not math.isfinite(float(corrected_value)):
raise ValueError("修复值必须为有限数字。")
if abs(float(corrected_value)) > 1e16:
raise ValueError("修复值超出允许范围，请确认单位为人民币元。")
clean_reason = str(reason or "").strip()
if len(clean_reason) < 4:
raise ValueError("请填写至少 4 个字的修复依据。")
if agent is None:
from core.financial_agent import FinancialAnalysisAgent
agent = FinancialAnalysisAgent()
result = copy.deepcopy(analyze_result)
financial_data = result.get("financial_data") or {}
original_pair = financial_data.get(metric)
if not isinstance(original_pair, (list, tuple)) or len(original_pair) < 2:
original_pair = (None, None)
updated_pair = [original_pair[0], original_pair[1]]
original_value = updated_pair[period_index]
updated_pair[period_index] = float(corrected_value)
financial_data[metric] = tuple(updated_pair)
result["financial_data"] = financial_data
before_risk = (result.get("hybrid_result") or {}).get("final_risk_level")
filled_data, estimation_flags = agent.extractor.estimate_missing_periods(financial_data)
m_score = agent.m_score_calculator.analyze(filled_data, estimation_flags)
red_flags = agent.red_flag_calculator.evaluate_all(financial_data)
try:
f_score = agent.f_score_calculator.calculate(financial_data)
except Exception:
f_score = None
hybrid = agent.hybrid_arbiter.arbitrate(m_score, red_flags, f_score)
report = agent.generate_report(financial_data, m_score, red_flags, hybrid)
try:
adjustment = agent.adjustment_calculator.calc(
financial_data, hybrid, None, f_score_result=f_score
)
except Exception:
adjustment = None
result.update({
"m_score_result": m_score,
"red_flags_result": red_flags,
"f_score_result": f_score,
"hybrid_result": hybrid,
"report": report,
"audit_adjustment_result": adjustment,
"estimation_flags": estimation_flags,
})
if result.get("evidence_result") is not None:
try:
result["cas_references"] = agent._trace_cas_standards(red_flags, result.get("evidence_result"))
except Exception:
pass
from core.financial_validation import (
build_batch_data_reliability_gate,
build_data_reliability_gate,
refresh_financial_validation,
)
result["financial_validation"] = refresh_financial_validation(result)
result["manual_correction_evidence_review_required"] = True
if period_index == 0:
_refresh_latest_trend(result, metric, float(corrected_value))
_refresh_cross_period_after_manual(result, metric, period_index, float(corrected_value))
overview = ((result.get("multi_year_summary") or {}).get("yearly_overview") or {})
latest_year = str((result.get("multi_year_summary") or {}).get("latest_year") or "")
if latest_year and isinstance(overview.get(latest_year), dict):
overview[latest_year].update({
"final_risk_level": hybrid.get("final_risk_level") or "未知",
"m_score": m_score.get("m_score"),
"red_flag_count": red_flags.get("triggered_count", red_flags.get("total_red_flags", 0)),
"f_score": (f_score or {}).get("f_score"),
})
(result.get("multi_year_summary") or {}).setdefault("yearly_risk_levels", {})[latest_year] = (
hybrid.get("final_risk_level") or "未知"
)
audit_payload = {
"sequence": len(result.get("manual_corrections") or []) + 1,
"metric": metric,
"metric_label": MANUAL_CORRECTION_METRICS[metric],
"period": "current" if period_index == 0 else "previous",
"original_value": original_value,
"corrected_value": float(corrected_value),
"reason": clean_reason,
"operator": str(operator or "当前会话人工复核"),
"corrected_at": datetime.now(timezone.utc).isoformat(),
"before_risk_level": before_risk,
"after_risk_level": hybrid.get("final_risk_level"),
"source_scope": "session_only_manual_override",
}
audit_payload["entry_sha256"] = hashlib.sha256(
json.dumps(audit_payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
).hexdigest()
result.setdefault("manual_corrections", []).append(audit_payload)
single_gate = build_data_reliability_gate(result)
yearly_gates = result.get("yearly_reliability_gates") or {}
if yearly_gates:
latest_gate_year = str((result.get("multi_year_summary") or {}).get("latest_year") or "")
if latest_gate_year:
yearly_gates[latest_gate_year] = single_gate
result["yearly_reliability_gates"] = yearly_gates
result["data_reliability_gate"] = build_batch_data_reliability_gate(
yearly_gates, result.get("cross_period_validation") or {}
)
else:
result["data_reliability_gate"] = single_gate
from core.nonfinancial_signals import build_financial_decision_hash
integrity = dict(result.get("decision_integrity") or {})
integrity["financial_decision_hash"] = build_financial_decision_hash(result)
result["decision_integrity"] = integrity
result.pop("regulatory_review_agent", None)
return result