strepsiades
258 lines
9.5 KiB
Python
258 lines
9.5 KiB
Python
"""
|
||
LAR (移民征地) API Handler
|
||
|
||
职责：
|
||
- 更新移民征地数据（依赖 Project）
|
||
- 支持主数据、搬迁信息、投资信息三种更新
|
||
- 不支持创建和删除（数据随项目自动生成）
|
||
|
||
实现的接口：
|
||
PUT:
|
||
- /lar/main - 更新主数据
|
||
- /lar/resettlement - 更新搬迁信息
|
||
- /lar/investment - 更新投资信息
|
||
"""
|
||
|
||
from typing import List, Dict, Any
|
||
from ...datasource.api.handler import BaseApiHandler
|
||
from ...datasource.task_result import TaskResult
|
||
from ...common.sync_node import SyncNode
|
||
from ...datasource.api.errors import ERROR_VALIDATION_FAILED
|
||
from .sync_node import LarSyncNode
|
||
from schemas.project_extensions.lar import (
|
||
LarDetailSchema,
|
||
LarMainInfoUpdateSchema,
|
||
LarResettlementSchema,
|
||
LarInvestmentSchema
|
||
)
|
||
|
||
|
||
class LarApiHandler(BaseApiHandler):
|
||
"""移民征地 API Handler"""
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._node_type = "lar"
|
||
self._node_class = LarSyncNode
|
||
self._schema = LarDetailSchema
|
||
|
||
async def load(self) -> List[SyncNode]:
|
||
"""
|
||
加载移民征地数据（从 project all_info 中提取）
|
||
|
||
注意：LAR 是可选的，只有特定项目类型（水电、抽蓄）才有
|
||
"""
|
||
from ..project.api_handler import ProjectApiHandler
|
||
|
||
# 获取已加载的项目
|
||
projects = self._collection.filter(node_type="project")
|
||
if not projects:
|
||
print("⚠️ No projects found, skipping LAR load")
|
||
return []
|
||
|
||
nodes = []
|
||
for project in projects:
|
||
project_id = project.data_id
|
||
|
||
try:
|
||
# 从 project context 或缓存获取 all_info
|
||
all_info = project.context.get("all_info")
|
||
if not all_info:
|
||
all_info = await ProjectApiHandler.get_all_info(
|
||
self.api_client,
|
||
project_id
|
||
)
|
||
|
||
# 提取 lar（注意：lar 是单个对象，不是列表）
|
||
lar_data = all_info.get("lar")
|
||
if lar_data:
|
||
try:
|
||
validated = self._schema.model_validate(lar_data)
|
||
node = self._create_node(validated.model_dump())
|
||
node.context["project_id"] = project_id
|
||
nodes.append(node)
|
||
print(f" Project {project_id}: 1 LAR")
|
||
except Exception as e:
|
||
print(f"⚠️ Skipping invalid LAR for project {project_id}: {e}")
|
||
continue
|
||
|
||
except Exception as e:
|
||
print(f"❌ Failed to load LAR for project {project_id}: {e}")
|
||
continue
|
||
|
||
return nodes
|
||
|
||
async def create_all(self, nodes: List[SyncNode]) -> List[TaskResult]:
|
||
"""不支持创建（LAR随项目自动生成）"""
|
||
return [
|
||
TaskResult.failed(node_id=node.node_id, error="LAR creation not supported (auto-generated with project)")
|
||
for node in nodes
|
||
]
|
||
|
||
async def update_all(self, nodes: List[SyncNode]) -> List[TaskResult]:
|
||
"""批量更新移民征地数据"""
|
||
from pydantic import ValidationError
|
||
|
||
results = []
|
||
|
||
# 定义三种更新类型
|
||
update_configs = [
|
||
(LarMainInfoUpdateSchema, api_update_lar_main),
|
||
(LarResettlementSchema, api_update_lar_resettlement),
|
||
(LarInvestmentSchema, api_update_lar_investment),
|
||
]
|
||
|
||
for node in nodes:
|
||
node_data = node.get_data()
|
||
if not node_data:
|
||
error_msg = node.error or "Node data is missing or invalid"
|
||
results.append(TaskResult.failed(node_id=node.node_id, error=f"Cannot update: {error_msg}"))
|
||
continue
|
||
|
||
origin_data = node.get_origin_data() or {}
|
||
node_updated = False
|
||
force_steps_update = self._steps_changed(node)
|
||
steps_payload = self._get_request_steps(node)
|
||
steps_sent = False
|
||
allow_force_steps = True
|
||
|
||
for schema, api_func in update_configs:
|
||
# 获取过滤后的更新数据
|
||
update_data = self.build_update_request_data(
|
||
node_data,
|
||
origin_data,
|
||
schema,
|
||
force_full_schema=force_steps_update and allow_force_steps,
|
||
)
|
||
if update_data:
|
||
# 验证并转换为 Pydantic model
|
||
try:
|
||
update_model = schema.model_validate(update_data)
|
||
except ValidationError as e:
|
||
results.append(TaskResult.failed(node_id=node.node_id, error=f"{ERROR_VALIDATION_FAILED}: {e}"))
|
||
break
|
||
|
||
push_id = self._generate_push_id()
|
||
update_model_data = update_model.model_dump()
|
||
biz_id = node.data_id or update_model_data.get("id")
|
||
if not biz_id:
|
||
results.append(TaskResult.failed(node_id=node.node_id, error="Missing biz_id for update"))
|
||
break
|
||
|
||
try:
|
||
await api_func(
|
||
self.api_client,
|
||
update_model,
|
||
push_id,
|
||
biz_id,
|
||
steps=steps_payload if not steps_sent else [],
|
||
)
|
||
node_updated = True
|
||
steps_sent = steps_sent or bool(steps_payload)
|
||
allow_force_steps = False
|
||
self._mark_approval_snapshot_synced(node)
|
||
except Exception as e:
|
||
results.append(TaskResult.failed(node_id=node.node_id, error=str(e)))
|
||
break
|
||
|
||
if node_updated:
|
||
results.append(TaskResult.success(node_id=node.node_id))
|
||
elif not any(r.node_id == node.node_id for r in results):
|
||
# 如果没有任何更新且没有报错，视为跳过（归一化后无差异）
|
||
reason = f"No physical diff found across all 3 LAR sub-schemas for {node.data_id} despite strategy update request (Normalization Discrepancy)."
|
||
print(f" [WARNING] [lar] {reason}")
|
||
results.append(TaskResult.skipped(node_id=node.node_id, reason=reason))
|
||
|
||
return results
|
||
|
||
async def delete_all(self, nodes: List[SyncNode]) -> List[TaskResult]:
|
||
"""不支持删除（LAR随项目存在）"""
|
||
return [
|
||
TaskResult.failed(node_id=node.node_id, error="LAR deletion not supported")
|
||
for node in nodes
|
||
]
|
||
|
||
async def poll_tasks(self, task_ids: List[str]) -> Dict[str, TaskResult]:
|
||
"""轮询异步任务状态"""
|
||
return await super().poll_tasks(task_ids)
|
||
|
||
def _create_node(self, data: Dict[str, Any]) -> SyncNode:
|
||
node = self._create_basic_node(data, depend_ids=[])
|
||
|
||
if data.get("project_id"):
|
||
node.context["project_id"] = data["project_id"]
|
||
node.context["lar_id"] = node.data_id
|
||
|
||
return node
|
||
|
||
|
||
# ========== 静态 API 函数 ==========
|
||
|
||
async def api_update_lar_main(
|
||
client,
|
||
data: LarMainInfoUpdateSchema,
|
||
push_id: str,
|
||
biz_id: str,
|
||
*,
|
||
steps: List[Dict[str, Any]] | None = None,
|
||
) -> None:
|
||
"""
|
||
PUT /lar/main - 更新主数据
|
||
|
||
Args:
|
||
client: API 客户端
|
||
data: 更新数据（Pydantic model）
|
||
push_id: 推送ID
|
||
biz_id: 业务ID
|
||
"""
|
||
request_data = {"push_id": push_id, "biz_id": biz_id, "data": data.model_dump(exclude_unset=True), "steps": steps or []}
|
||
response = await client.request(method="PUT", endpoint="/lar/main", data=request_data)
|
||
if response.get("code") != 200:
|
||
raise Exception(f"API error: {response.get('message', 'Unknown error')}")
|
||
|
||
|
||
async def api_update_lar_resettlement(
|
||
client,
|
||
data: LarResettlementSchema,
|
||
push_id: str,
|
||
biz_id: str,
|
||
*,
|
||
steps: List[Dict[str, Any]] | None = None,
|
||
) -> None:
|
||
"""
|
||
PUT /lar/resettlement - 更新搬迁信息
|
||
|
||
Args:
|
||
client: API 客户端
|
||
data: 更新数据（Pydantic model）
|
||
push_id: 推送ID
|
||
biz_id: 业务ID
|
||
"""
|
||
request_data = {"push_id": push_id, "biz_id": biz_id, "data": data.model_dump(exclude_unset=True), "steps": steps or []}
|
||
response = await client.request(method="PUT", endpoint="/lar/resettlement", data=request_data)
|
||
if response.get("code") != 200:
|
||
raise Exception(f"API error: {response.get('message', 'Unknown error')}")
|
||
|
||
|
||
async def api_update_lar_investment(
|
||
client,
|
||
data: LarInvestmentSchema,
|
||
push_id: str,
|
||
biz_id: str,
|
||
*,
|
||
steps: List[Dict[str, Any]] | None = None,
|
||
) -> None:
|
||
"""
|
||
PUT /lar/investment - 更新投资信息
|
||
|
||
Args:
|
||
client: API 客户端
|
||
data: 更新数据（Pydantic model）
|
||
push_id: 推送ID
|
||
biz_id: 业务ID
|
||
"""
|
||
request_data = {"push_id": push_id, "biz_id": biz_id, "data": data.model_dump(exclude_unset=True), "steps": steps or []}
|
||
response = await client.request(method="PUT", endpoint="/lar/investment", data=request_data)
|
||
if response.get("code") != 200:
|
||
raise Exception(f"API error: {response.get('message', 'Unknown error')}")
|