232 lines
8.5 KiB
Python
232 lines
8.5 KiB
Python
"""风控证据文件归档 Service。
|
||
|
||
文件只写入项目受控目录；数据库只更新已有 `fin_risk_alert.evidence_snapshot`，
|
||
不假设存在独立的 `evidence_archived` 字段。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import os
|
||
import re
|
||
from hashlib import sha256
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
from zipfile import BadZipFile, ZipFile
|
||
|
||
from fastapi import UploadFile
|
||
from sqlalchemy import ColumnElement, false, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import ConflictAgentError, RecoverableAgentError, ValidationAgentError
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.fund import FundRiskAlert
|
||
from app.service.authorization_service import AuthorizationService
|
||
|
||
ALERT_NO_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
|
||
ALLOWED_EXTENSIONS = {
|
||
".jpg",
|
||
".jpeg",
|
||
".png",
|
||
".webp",
|
||
".pdf",
|
||
".doc",
|
||
".docx",
|
||
".xls",
|
||
".xlsx",
|
||
".ppt",
|
||
".pptx",
|
||
".txt",
|
||
}
|
||
OFFICE_ZIP_ROOTS = {".docx": "word/", ".xlsx": "xl/", ".pptx": "ppt/"}
|
||
OLE_SIGNATURE = bytes.fromhex("D0CF11E0A1B11AE1")
|
||
|
||
|
||
class RiskEvidenceValidationError(ValidationAgentError):
|
||
"""证据文件不符合归档要求。"""
|
||
|
||
|
||
class RiskEvidenceSizeError(RiskEvidenceValidationError):
|
||
"""证据文件超过大小限制。"""
|
||
|
||
status_code = 413
|
||
|
||
|
||
class RiskEvidenceAlreadyArchivedError(ConflictAgentError):
|
||
"""当前预警已经归档证据。"""
|
||
|
||
|
||
class RiskEvidenceStorageError(RecoverableAgentError):
|
||
"""证据文件无法安全写入。"""
|
||
|
||
|
||
class RiskEvidenceArchiveService:
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
root: Path | None = None,
|
||
max_bytes: int | None = None,
|
||
) -> None:
|
||
self.session = session
|
||
self.root = root or _default_root()
|
||
self.max_bytes = max_bytes or _default_max_bytes()
|
||
|
||
async def archive(
|
||
self,
|
||
alert_no: str,
|
||
upload: UploadFile,
|
||
context: RequestContext,
|
||
) -> dict[str, object]:
|
||
await AuthorizationService.require(context, "risk:alert:write")
|
||
statement = (
|
||
select(FundRiskAlert)
|
||
.where(FundRiskAlert.alert_no == alert_no)
|
||
.with_for_update()
|
||
)
|
||
scope = _scope_condition(context)
|
||
if scope is not None:
|
||
statement = statement.where(scope)
|
||
alert = await self.session.scalar(statement)
|
||
if alert is None:
|
||
raise RiskEvidenceValidationError("预警不存在")
|
||
if alert.ack_at is None or alert.ack_status != "已确认":
|
||
raise RiskEvidenceAlreadyArchivedError("请先确认接收预警")
|
||
if alert.status != "调查中":
|
||
raise RiskEvidenceAlreadyArchivedError("只有调查中的预警可以归档证据")
|
||
if not ALERT_NO_PATTERN.fullmatch(alert.alert_no):
|
||
raise RiskEvidenceStorageError("预警编号不符合文件归档要求")
|
||
snapshot = dict(alert.evidence_snapshot or {})
|
||
if snapshot.get("evidence_archive"):
|
||
raise RiskEvidenceAlreadyArchivedError("当前预警已经归档证据，不能重复上传")
|
||
|
||
extension, content = await self._read_and_validate(upload)
|
||
root = self._safe_root()
|
||
target = root / f"{alert.alert_no}{extension}"
|
||
if any(root.glob(f"{alert.alert_no}.*")):
|
||
raise RiskEvidenceAlreadyArchivedError("当前预警证据文件已存在，不能覆盖")
|
||
temporary = root / f".{alert.alert_no}.{uuid4().hex}.tmp"
|
||
committed = False
|
||
try:
|
||
temporary.write_bytes(content)
|
||
temporary.replace(target)
|
||
snapshot["evidence_archive"] = {
|
||
"stored_name": target.name,
|
||
"file_size": len(content),
|
||
"sha256": sha256(content).hexdigest(),
|
||
}
|
||
alert.evidence_snapshot = snapshot
|
||
self.session.add(InteractionAudit(
|
||
actor_type="user",
|
||
actor_id=int(context.user_id),
|
||
target_customer_id=alert.customer_id,
|
||
portal="api",
|
||
action_type="risk_evidence_archived",
|
||
detail={
|
||
"alert_no": alert.alert_no,
|
||
"stored_name": target.name,
|
||
"file_size": len(content),
|
||
"sha256": sha256(content).hexdigest(),
|
||
},
|
||
))
|
||
await self.session.commit()
|
||
committed = True
|
||
except OSError as error:
|
||
await self.session.rollback()
|
||
raise RiskEvidenceStorageError("证据文件写入失败") from error
|
||
except Exception:
|
||
await self.session.rollback()
|
||
raise
|
||
finally:
|
||
temporary.unlink(missing_ok=True)
|
||
if not committed:
|
||
target.unlink(missing_ok=True)
|
||
return {
|
||
"alert_no": alert.alert_no,
|
||
"evidence_archived": True,
|
||
"stored_name": target.name,
|
||
"file_size": len(content),
|
||
}
|
||
|
||
async def _read_and_validate(self, upload: UploadFile) -> tuple[str, bytes]:
|
||
filename = Path(upload.filename or "").name
|
||
extension = Path(filename).suffix.lower()
|
||
if extension not in ALLOWED_EXTENSIONS:
|
||
raise RiskEvidenceValidationError("仅支持图片、PDF、Office 文档和 TXT 文件")
|
||
content = await upload.read(self.max_bytes + 1)
|
||
if not content:
|
||
raise RiskEvidenceValidationError("证据文件不能为空")
|
||
if len(content) > self.max_bytes:
|
||
raise RiskEvidenceSizeError("证据文件超过大小限制")
|
||
self._validate_signature(extension, content)
|
||
return extension, content
|
||
|
||
def _safe_root(self) -> Path:
|
||
project_root = Path(__file__).resolve().parents[2]
|
||
root = self.root if self.root.is_absolute() else project_root / self.root
|
||
root = root.resolve()
|
||
try:
|
||
root.relative_to(project_root.resolve())
|
||
except ValueError as error:
|
||
raise RiskEvidenceStorageError("证据目录必须位于项目目录内") from error
|
||
try:
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
except OSError as error:
|
||
raise RiskEvidenceStorageError("证据目录不可用") from error
|
||
return root
|
||
|
||
@staticmethod
|
||
def _validate_signature(extension: str, content: bytes) -> None:
|
||
signatures = {
|
||
".jpg": b"\xff\xd8\xff",
|
||
".jpeg": b"\xff\xd8\xff",
|
||
".png": b"\x89PNG\r\n\x1a\n",
|
||
".pdf": b"%PDF-",
|
||
}
|
||
if extension in signatures and not content.startswith(signatures[extension]):
|
||
raise RiskEvidenceValidationError("文件内容与扩展名不一致")
|
||
if extension == ".webp" and not (
|
||
content.startswith(b"RIFF") and content[8:12] == b"WEBP"
|
||
):
|
||
raise RiskEvidenceValidationError("文件内容与扩展名不一致")
|
||
if extension in {".doc", ".xls", ".ppt"} and not content.startswith(OLE_SIGNATURE):
|
||
raise RiskEvidenceValidationError("文件内容与扩展名不一致")
|
||
if extension in OFFICE_ZIP_ROOTS:
|
||
try:
|
||
with ZipFile(io.BytesIO(content)) as archive:
|
||
names = archive.namelist()
|
||
except BadZipFile as error:
|
||
raise RiskEvidenceValidationError("Office 文件结构无效") from error
|
||
if (
|
||
"[Content_Types].xml" not in names
|
||
or not any(name.startswith(OFFICE_ZIP_ROOTS[extension]) for name in names)
|
||
):
|
||
raise RiskEvidenceValidationError("文件内容与扩展名不一致")
|
||
if extension == ".txt":
|
||
try:
|
||
text = content.decode("utf-8-sig")
|
||
except UnicodeDecodeError as error:
|
||
raise RiskEvidenceValidationError("TXT 文件必须使用 UTF-8 编码") from error
|
||
if "\x00" in text:
|
||
raise RiskEvidenceValidationError("TXT 文件内容无效")
|
||
|
||
|
||
def _scope_condition(context: RequestContext) -> ColumnElement[bool] | None:
|
||
if context.data_scope == "all":
|
||
return None
|
||
if not context.customer_ids:
|
||
return false()
|
||
return FundRiskAlert.customer_id.in_(
|
||
tuple(int(customer_id) for customer_id in context.customer_ids)
|
||
)
|
||
|
||
|
||
def _default_root() -> Path:
|
||
return Path(os.getenv("RISK_EVIDENCE_DIR", "storage/risk_evidence"))
|
||
|
||
|
||
def _default_max_bytes() -> int:
|
||
megabytes = int(os.getenv("RISK_EVIDENCE_MAX_FILE_SIZE_MB", "10"))
|
||
return max(1, megabytes) * 1024 * 1024
|