```

feat(core): 添加多API密钥支持和配置字段

添加了api_keys、audit_api_keys、approval_api_keys等字段用于支持多个服务密钥，
新增masked_response_fields用于配置响应掩码字段，以及legacy相关配置项。

feat(core): 增强响应数据掩码功能

扩展mask_configured函数支持域名参数，实现更精确的敏感字段掩码控制，
添加自定义掩码字段配置验证器。

feat(scheduler): 添加遗留系统同步调度任务

集成遗留项目和任务同步到定时调度器中，支持通过配置启用或禁用同步功能，
并可设置不同的执行时间计划。

feat(security): 实现多服务密钥认证机制

重构API密钥验证逻辑，支持单个主密钥和多个配置密钥的混合验证模式，
增加服务密钥启用状态检查和角色映射功能。

feat(task_queue): 扩展现有队列任务处理

为日常简报和周报推送任务添加Celery异步处理支持，新增遗留项目和任务同步任务，
统一任务分发接口。

feat(business): 扩展业务模型字段

为工作任务模型添加外部系统标识和外部ID字段，为风险事件模型增加分配、解决、关闭
等相关字段，并创建风险事件操作记录表。

feat(legacy_mysql): 实现遗留任务同步功能

添加遗留任务查询和同步路由，支持从旧MySQL数据库同步任务数据到内部系统，
包括同步结果统计和运行记录。

refactor(dashboard): 更新仪表板统计数据

增加未分配风险和失败推送运行统计，在概览中显示最新的推送和同步运行记录，
完善数据序列化展示。

fix(feishu): 修复审批事件重复处理

实现审批卡片操作事件的唯一性检查，防止重复审批操作，添加事件审计日志记录。
```

app/core/config.py

@@ -24,18 +24,24 @@ class Settings(BaseSettings):
     api_prefix: str = "/api/v1"
     api_key: str | None = None
     api_actor: str = ActorValue.API
+    api_keys: list[dict[str, Any]] = Field(default_factory=list)
     audit_api_key: str | None = None
     audit_api_actor: str = ActorValue.AUDITOR
+    audit_api_keys: list[dict[str, Any]] = Field(default_factory=list)
     approval_api_key: str | None = None
     approval_api_actor: str = ActorValue.APPROVER
+    approval_api_keys: list[dict[str, Any]] = Field(default_factory=list)
     cors_origins: list[str] = Field(default_factory=lambda: ["*"])
     mask_sensitive_responses: bool = True
+    masked_response_fields: list[str] = Field(default_factory=list)
 
     database_url: str = "sqlite:///./company_ai.db"
     legacy_database_url: str | None = None
     legacy_project_query: str | None = None
+    legacy_task_query: str | None = None
     legacy_allowed_queries: dict[str, str] = Field(default_factory=dict)
     legacy_project_code_prefix: str = "LEGACY"
+    legacy_task_code_prefix: str = "LEGACY-TASK"
     redis_url: str = "redis://127.0.0.1:6379/0"
 
     feishu_base_url: str = "https://open.feishu.cn/open-apis"
@@ -66,12 +72,17 @@ class Settings(BaseSettings):
     scheduler_enabled: bool = False
     task_queue_enabled: bool = False
     task_queue_always_eager: bool = False
+    legacy_sync_enabled: bool = False
     celery_result_backend_url: str | None = None
     daily_brief_cron_hour: int = 9
     daily_brief_cron_minute: int = 0
     weekly_project_report_day_of_week: str = "mon"
     weekly_project_report_cron_hour: int = 9
     weekly_project_report_cron_minute: int = 30
+    legacy_project_sync_cron_hour: int = 2
+    legacy_project_sync_cron_minute: int = 0
+    legacy_task_sync_cron_hour: int = 2
+    legacy_task_sync_cron_minute: int = 30
 
     @field_validator("cors_origins", mode="before")
     @classmethod
@@ -95,6 +106,39 @@ class Settings(BaseSettings):
             return value
         return [item.strip() for item in value.split(",") if item.strip()]
 
+    @field_validator("masked_response_fields", mode="before")
+    @classmethod
+    def parse_masked_response_fields(cls, value: Any) -> list[str]:
+        if isinstance(value, list):
+            return [str(item).strip() for item in value if str(item).strip()]
+        if value is None:
+            return []
+        text = str(value).strip()
+        if not text:
+            return []
+        if text.startswith("["):
+            data = json.loads(text)
+            if not isinstance(data, list):
+                raise ValueError(ConfigErrorDetail.CORS_ORIGINS_FORMAT)
+            return [str(item).strip() for item in data if str(item).strip()]
+        return [item.strip() for item in text.split(",") if item.strip()]
+
+    @field_validator("api_keys", "audit_api_keys", "approval_api_keys", mode="before")
+    @classmethod
+    def parse_service_keys(cls, value: Any) -> list[dict[str, Any]]:
+        if value is None or value == "":
+            return []
+        if isinstance(value, list):
+            return [dict(item) for item in value if isinstance(item, dict)]
+        if isinstance(value, str):
+            data = json.loads(value)
+            if not isinstance(data, list):
+                raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT)
+            if not all(isinstance(item, dict) for item in data):
+                raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT)
+            return [dict(item) for item in data]
+        raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT)
+
     @field_validator("legacy_allowed_queries", mode="before")
     @classmethod
     def parse_legacy_allowed_queries(cls, value: Any) -> dict[str, str]:

app/core/constants.py

@@ -37,6 +37,7 @@ class SecurityErrorDetail(StrEnum):
 class ConfigErrorDetail(StrEnum):
     CORS_ORIGINS_FORMAT = "CORS_ORIGINS must be a CSV string or JSON list"
     LEGACY_ALLOWED_QUERIES_FORMAT = "LEGACY_ALLOWED_QUERIES must be a JSON object"
+    SERVICE_KEYS_FORMAT = "Service keys must be a JSON list of objects"
 
 
 BEARER_TOKEN_TEMPLATE = "Bearer {token}"

app/core/masking.py

@@ -28,27 +28,41 @@ SENSITIVE_RESPONSE_KEYS = frozenset(
 )
 
 
-def mask_configured(value: Any) -> Any:
+def mask_configured(value: Any, domain: str | None = None) -> Any:
     """Mask sensitive response fields when response masking is enabled."""
 
     settings = get_settings()
     if not settings.mask_sensitive_responses:
         return value
-    return mask_sensitive(value)
+    configured_fields = {item.lower() for item in settings.masked_response_fields}
+    return mask_sensitive(value, domain=domain, configured_fields=configured_fields)
 
 
-def mask_sensitive(value: Any) -> Any:
+def mask_sensitive(
+    value: Any,
+    domain: str | None = None,
+    configured_fields: set[str] | None = None,
+) -> Any:
     if isinstance(value, dict):
         masked: dict[str, Any] = {}
         for key, item in value.items():
             key_text = str(key)
-            if key_text.lower() in SENSITIVE_RESPONSE_KEYS:
+            if _should_mask(key_text, domain, configured_fields or set()):
                 masked[key_text] = MASKED_VALUE
             else:
-                masked[key_text] = mask_sensitive(item)
+                masked[key_text] = mask_sensitive(item, domain, configured_fields)
         return masked
     if isinstance(value, list):
-        return [mask_sensitive(item) for item in value]
+        return [mask_sensitive(item, domain, configured_fields) for item in value]
     if isinstance(value, tuple):
-        return [mask_sensitive(item) for item in value]
+        return [mask_sensitive(item, domain, configured_fields) for item in value]
     return value
+
+
+def _should_mask(key: str, domain: str | None, configured_fields: set[str]) -> bool:
+    field = key.lower()
+    if field in SENSITIVE_RESPONSE_KEYS or field in configured_fields:
+        return True
+    if f"*.{field}" in configured_fields:
+        return True
+    return bool(domain and f"{domain.lower()}.{field}" in configured_fields)

app/core/scheduler.py

@@ -15,7 +15,12 @@ def attach_scheduler(app: FastAPI) -> None:
     from apscheduler.schedulers.background import BackgroundScheduler
 
     from app.core.database import SessionLocal
-    from app.core.task_queue import enqueue_daily_brief_push, enqueue_project_weekly_push
+    from app.core.task_queue import (
+        enqueue_daily_brief_push,
+        enqueue_legacy_project_sync,
+        enqueue_legacy_task_sync,
+        enqueue_project_weekly_push,
+    )
     from app.modules.reports.service import ReportService
 
     scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
@@ -72,6 +77,16 @@ def attach_scheduler(app: FastAPI) -> None:
         finally:
             db.close()
 
+    def run_legacy_project_sync() -> None:
+        app.state.last_legacy_project_sync_dispatch = enqueue_legacy_project_sync(
+            actor=ActorValue.SCHEDULER,
+        )
+
+    def run_legacy_task_sync() -> None:
+        app.state.last_legacy_task_sync_dispatch = enqueue_legacy_task_sync(
+            actor=ActorValue.SCHEDULER,
+        )
+
     scheduler.add_job(
         run_daily_brief,
         trigger="cron",
@@ -89,6 +104,24 @@ def attach_scheduler(app: FastAPI) -> None:
         id="project_weekly_push",
         replace_existing=True,
     )
+    if settings.legacy_sync_enabled and settings.legacy_project_query:
+        scheduler.add_job(
+            run_legacy_project_sync,
+            trigger="cron",
+            hour=settings.legacy_project_sync_cron_hour,
+            minute=settings.legacy_project_sync_cron_minute,
+            id="legacy_project_sync",
+            replace_existing=True,
+        )
+    if settings.legacy_sync_enabled and settings.legacy_task_query:
+        scheduler.add_job(
+            run_legacy_task_sync,
+            trigger="cron",
+            hour=settings.legacy_task_sync_cron_hour,
+            minute=settings.legacy_task_sync_cron_minute,
+            id="legacy_task_sync",
+            replace_existing=True,
+        )
 
     @app.on_event("startup")
     def start_scheduler() -> None:

app/core/security.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from secrets import compare_digest
+from typing import Any
 
 from fastapi import Header, HTTPException, status
 
@@ -20,17 +21,23 @@ def require_api_key(
     """Validate the internal API key header and return its service principal."""
 
     settings = get_settings()
-    if not settings.api_key:
+    if not settings.api_key and not _has_enabled_keys(settings.api_keys):
         raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
             detail=SecurityErrorDetail.API_KEY_REQUIRED,
         )
-    if not x_api_key or not compare_digest(x_api_key, settings.api_key):
+    principal = _match_service_key(
+        x_api_key,
+        settings.api_key,
+        settings.api_actor,
+        settings.api_keys,
+    )
+    if principal is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail=SecurityErrorDetail.INVALID_API_KEY,
         )
-    return ApiPrincipal(actor=settings.api_actor)
+    return principal
 
 
 def require_approval_api_key(
@@ -42,20 +49,23 @@ def require_approval_api_key(
     """Validate the approval API key and return the approval principal."""
 
     settings = get_settings()
-    if not settings.approval_api_key:
+    if not settings.approval_api_key and not _has_enabled_keys(settings.approval_api_keys):
         raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
             detail=SecurityErrorDetail.APPROVAL_API_KEY_REQUIRED,
         )
-    if (
-        not x_approval_api_key
-        or not compare_digest(x_approval_api_key, settings.approval_api_key)
-    ):
+    principal = _match_service_key(
+        x_approval_api_key,
+        settings.approval_api_key,
+        settings.approval_api_actor,
+        settings.approval_api_keys,
+    )
+    if principal is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail=SecurityErrorDetail.INVALID_APPROVAL_API_KEY,
         )
-    return ApiPrincipal(actor=settings.approval_api_actor)
+    return principal
 
 
 def require_audit_api_key(
@@ -67,14 +77,50 @@ def require_audit_api_key(
     """Validate the audit API key and return the audit principal."""
 
     settings = get_settings()
-    if not settings.audit_api_key:
+    if not settings.audit_api_key and not _has_enabled_keys(settings.audit_api_keys):
         raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
             detail=SecurityErrorDetail.AUDIT_API_KEY_REQUIRED,
         )
-    if not x_audit_api_key or not compare_digest(x_audit_api_key, settings.audit_api_key):
+    principal = _match_service_key(
+        x_audit_api_key,
+        settings.audit_api_key,
+        settings.audit_api_actor,
+        settings.audit_api_keys,
+    )
+    if principal is None:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail=SecurityErrorDetail.INVALID_AUDIT_API_KEY,
         )
-    return ApiPrincipal(actor=settings.audit_api_actor)
+    return principal
+
+
+def _has_enabled_keys(configured_keys: list[dict[str, Any]]) -> bool:
+    return any(_key_enabled(item) and item.get("key") for item in configured_keys)
+
+
+def _match_service_key(
+    provided_key: str | None,
+    legacy_key: str | None,
+    legacy_actor: str,
+    configured_keys: list[dict[str, Any]],
+) -> ApiPrincipal | None:
+    if not provided_key:
+        return None
+    if legacy_key and compare_digest(provided_key, legacy_key):
+        return ApiPrincipal(actor=legacy_actor)
+    for item in configured_keys:
+        key = item.get("key")
+        if not key or not _key_enabled(item):
+            continue
+        if compare_digest(provided_key, str(key)):
+            return ApiPrincipal(actor=str(item.get("actor") or legacy_actor))
+    return None
+
+
+def _key_enabled(item: dict[str, Any]) -> bool:
+    value = item.get("enabled", True)
+    if isinstance(value, bool):
+        return value
+    return str(value).strip().lower() not in {"0", "false", "no", "off", "disabled"}

app/core/task_queue.py

@@ -7,6 +7,8 @@ from app.modules.feishu.constants import FeishuReceiveIdType
 TASK_PUSH_DAILY_BRIEF = "reports.push_daily_brief"
 TASK_PUSH_PROJECT_WEEKLY = "reports.push_project_weekly"
 TASK_GENERATE_RISK_EVENTS = "risks.generate_events"
+TASK_SYNC_LEGACY_PROJECTS = "legacy.sync_projects"
+TASK_SYNC_LEGACY_TASKS = "legacy.sync_tasks"
 
 
 def dispatch_task(
@@ -40,6 +42,47 @@ def enqueue_daily_brief_push(
     receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
     actor: str = "scheduler",
 ) -> dict[str, Any]:
+    settings = get_settings()
+    if settings.task_queue_enabled:
+        from app.core.database import SessionLocal
+        from app.modules.reports.constants import ReportPushStatus, ReportTitle, ReportType
+        from app.modules.reports.service import ReportService
+        from app.tasks import celery_app
+
+        db = SessionLocal()
+        try:
+            push_run = ReportService(db).create_push_run(
+                report_type=ReportType.DAILY,
+                title=ReportTitle.DAILY_BRIEF,
+                receive_id=receive_id,
+                receive_id_type=receive_id_type,
+                actor=actor,
+                status=ReportPushStatus.QUEUED,
+            )
+            async_result = celery_app.signature(
+                TASK_PUSH_DAILY_BRIEF,
+                kwargs={
+                    "receive_id": receive_id,
+                    "receive_id_type": receive_id_type,
+                    "actor": actor,
+                    "push_run_code": push_run.code,
+                },
+            ).apply_async()
+            ReportService(db).update_push_run(
+                push_run.code,
+                ReportPushStatus.QUEUED,
+                task_id=async_result.id,
+            )
+        finally:
+            db.close()
+        return {
+            "queued": True,
+            "mode": "celery",
+            "task_name": TASK_PUSH_DAILY_BRIEF,
+            "task_id": async_result.id,
+            "push_run_code": push_run.code,
+        }
+
     def inline() -> Any:
         from app.core.database import SessionLocal
         from app.modules.reports.service import ReportService
@@ -67,6 +110,47 @@ def enqueue_project_weekly_push(
     receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
     actor: str = "scheduler",
 ) -> dict[str, Any]:
+    settings = get_settings()
+    if settings.task_queue_enabled:
+        from app.core.database import SessionLocal
+        from app.modules.reports.constants import ReportPushStatus, ReportTitle, ReportType
+        from app.modules.reports.service import ReportService
+        from app.tasks import celery_app
+
+        db = SessionLocal()
+        try:
+            push_run = ReportService(db).create_push_run(
+                report_type=ReportType.WEEKLY,
+                title=ReportTitle.PROJECT_WEEKLY,
+                receive_id=receive_id,
+                receive_id_type=receive_id_type,
+                actor=actor,
+                status=ReportPushStatus.QUEUED,
+            )
+            async_result = celery_app.signature(
+                TASK_PUSH_PROJECT_WEEKLY,
+                kwargs={
+                    "receive_id": receive_id,
+                    "receive_id_type": receive_id_type,
+                    "actor": actor,
+                    "push_run_code": push_run.code,
+                },
+            ).apply_async()
+            ReportService(db).update_push_run(
+                push_run.code,
+                ReportPushStatus.QUEUED,
+                task_id=async_result.id,
+            )
+        finally:
+            db.close()
+        return {
+            "queued": True,
+            "mode": "celery",
+            "task_name": TASK_PUSH_PROJECT_WEEKLY,
+            "task_id": async_result.id,
+            "push_run_code": push_run.code,
+        }
+
     def inline() -> Any:
         from app.core.database import SessionLocal
         from app.modules.reports.service import ReportService
@@ -105,3 +189,81 @@ def enqueue_risk_event_generation(actor: str = "scheduler") -> dict[str, Any]:
         {"actor": actor},
         inline,
     )
+
+
+def enqueue_legacy_project_sync(
+    source_query: str | None = None,
+    source_query_name: str | None = None,
+    field_map: dict[str, str] | None = None,
+    limit: int = 100,
+    dry_run: bool = False,
+    actor: str = "scheduler",
+) -> dict[str, Any]:
+    def inline() -> Any:
+        from app.core.database import SessionLocal
+        from app.modules.legacy_mysql.service import LegacyMySQLService
+
+        db = SessionLocal()
+        try:
+            return LegacyMySQLService(db).sync_projects(
+                source_query=source_query,
+                source_query_name=source_query_name,
+                field_map=field_map or {},
+                limit=limit,
+                dry_run=dry_run,
+                actor=actor,
+            )
+        finally:
+            db.close()
+
+    return dispatch_task(
+        TASK_SYNC_LEGACY_PROJECTS,
+        {
+            "source_query": source_query,
+            "source_query_name": source_query_name,
+            "field_map": field_map or {},
+            "limit": limit,
+            "dry_run": dry_run,
+            "actor": actor,
+        },
+        inline,
+    )
+
+
+def enqueue_legacy_task_sync(
+    source_query: str | None = None,
+    source_query_name: str | None = None,
+    field_map: dict[str, str] | None = None,
+    limit: int = 100,
+    dry_run: bool = False,
+    actor: str = "scheduler",
+) -> dict[str, Any]:
+    def inline() -> Any:
+        from app.core.database import SessionLocal
+        from app.modules.legacy_mysql.service import LegacyMySQLService
+
+        db = SessionLocal()
+        try:
+            return LegacyMySQLService(db).sync_tasks(
+                source_query=source_query,
+                source_query_name=source_query_name,
+                field_map=field_map or {},
+                limit=limit,
+                dry_run=dry_run,
+                actor=actor,
+            )
+        finally:
+            db.close()
+
+    return dispatch_task(
+        TASK_SYNC_LEGACY_TASKS,
+        {
+            "source_query": source_query,
+            "source_query_name": source_query_name,
+            "field_map": field_map or {},
+            "limit": limit,
+            "dry_run": dry_run,
+            "actor": actor,
+        },
+        inline,
+    )

app/modules/audit/constants.py

@@ -14,6 +14,9 @@ class AuditAction(StrEnum):
     APPROVAL_APPROVE = "approval.approve"
     APPROVAL_REJECT = "approval.reject"
     LEGACY_SYNC_PROJECTS = "sync_projects"
+    LEGACY_SYNC_TASKS = "sync_tasks"
+    RISK_EVENT_ACTION = "risk_event_action"
+    REPORT_PUSH = "report_push"
 
 
 class AuditRiskLevel(StrEnum):

app/modules/business/models.py

@@ -58,6 +58,8 @@ class WorkTask(Base, TimestampMixin):
     completed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     blocker: Mapped[str | None] = mapped_column(Text, nullable=True)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
+    source_system: Mapped[str] = mapped_column(String(64), default=SourceSystem.INTERNAL)
+    external_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
 
 
 class Procurement(Base, TimestampMixin):
@@ -233,11 +235,32 @@ class RiskEvent(Base, TimestampMixin):
     owner: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
     detected_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True)
     due_date: Mapped[date | None] = mapped_column(Date, nullable=True, index=True)
+    assigned_to: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    resolved_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    closed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    closed_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
+    review_summary: Mapped[str | None] = mapped_column(Text, nullable=True)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
     mitigation: Mapped[str | None] = mapped_column(Text, nullable=True)
     evidence: Mapped[dict | None] = mapped_column(JSON, nullable=True)
 
 
+class RiskEventAction(Base):
+    __tablename__ = "risk_event_actions"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    code: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    risk_event_id: Mapped[int] = mapped_column(Integer, index=True)
+    action: Mapped[str] = mapped_column(String(64), index=True)
+    actor: Mapped[str] = mapped_column(String(128), default=ActorValue.SYSTEM, index=True)
+    from_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    to_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    assigned_to: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    comment: Mapped[str | None] = mapped_column(Text, nullable=True)
+    payload: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True)
+
+
 class LegacySyncRun(Base, TimestampMixin):
     __tablename__ = "legacy_sync_runs"
 
@@ -253,3 +276,21 @@ class LegacySyncRun(Base, TimestampMixin):
     skipped_count: Mapped[int] = mapped_column(Integer, default=0)
     error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
     note: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+
+class ReportPushRun(Base, TimestampMixin):
+    __tablename__ = "report_push_runs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    code: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    report_type: Mapped[str] = mapped_column(String(64), index=True)
+    title: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    receive_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    receive_id_type: Mapped[str] = mapped_column(String(64), index=True)
+    status: Mapped[str] = mapped_column(String(32), index=True)
+    task_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    actor: Mapped[str] = mapped_column(String(128), default=ActorValue.SYSTEM, index=True)
+    provider_response: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    queued_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True)
+    sent_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

app/modules/business/routes.py

@@ -33,7 +33,7 @@ def list_records(
     return {
         BusinessResponseKey.DOMAIN: domain,
         BusinessResponseKey.TOTAL: total,
-        BusinessResponseKey.ITEMS: mask_configured(items),
+        BusinessResponseKey.ITEMS: mask_configured(items, domain=domain),
     }
 
 
@@ -47,7 +47,8 @@ def get_record(
         return {
             BusinessResponseKey.DOMAIN: domain,
             BusinessResponseKey.DATA: mask_configured(
-                BusinessService(db).get_record(domain, record_id)
+                BusinessService(db).get_record(domain, record_id),
+                domain=domain,
             ),
         }
     except KeyError as exc:
@@ -72,7 +73,7 @@ def create_record(
         raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     return {
         BusinessResponseKey.DOMAIN: domain,
-        BusinessResponseKey.DATA: mask_configured(data),
+        BusinessResponseKey.DATA: mask_configured(data, domain=domain),
     }
 
 
@@ -96,5 +97,5 @@ def update_record(
         raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     return {
         BusinessResponseKey.DOMAIN: domain,
-        BusinessResponseKey.DATA: mask_configured(data),
+        BusinessResponseKey.DATA: mask_configured(data, domain=domain),
     }

app/modules/dashboard/service.py

@@ -7,8 +7,16 @@ from app.modules.approvals.constants import ApprovalStatus
 from app.modules.approvals.models import ApprovalRequest
 from app.modules.audit.models import AuditLog
 from app.modules.business.constants import DONE_STATUSES, PROJECT_CLOSED_STATUSES, StatusValue
-from app.modules.business.models import Project, RiskEvent, WorkReport, WorkTask
+from app.modules.business.models import (
+    LegacySyncRun,
+    Project,
+    ReportPushRun,
+    RiskEvent,
+    WorkReport,
+    WorkTask,
+)
 from app.modules.business.service import serialize_model
+from app.modules.reports.constants import ReportPushStatus
 from app.modules.risk.service import RiskService
 
 
@@ -27,9 +35,21 @@ class DashboardService:
             ApprovalRequest.status == ApprovalStatus.PENDING,
         )
         open_risk_events = self._count(RiskEvent, RiskEvent.status == StatusValue.OPEN)
+        unassigned_open_risks = self._count(
+            RiskEvent,
+            RiskEvent.status == StatusValue.OPEN,
+            RiskEvent.assigned_to.is_(None),
+        )
+        failed_push_runs = self._count(ReportPushRun, ReportPushRun.status == ReportPushStatus.FAILED)
         latest_reports = self.db.execute(
             select(WorkReport).order_by(WorkReport.id.desc()).limit(5)
         ).scalars()
+        latest_push_runs = self.db.execute(
+            select(ReportPushRun).order_by(ReportPushRun.id.desc()).limit(10)
+        ).scalars()
+        latest_sync_runs = self.db.execute(
+            select(LegacySyncRun).order_by(LegacySyncRun.id.desc()).limit(10)
+        ).scalars()
         latest_audit_logs = self.db.execute(
             select(AuditLog).order_by(AuditLog.id.desc()).limit(10)
         ).scalars()
@@ -40,6 +60,8 @@ class DashboardService:
                 "open_tasks": open_tasks,
                 "pending_approvals": pending_approvals,
                 "open_risk_events": open_risk_events,
+                "unassigned_open_risks": unassigned_open_risks,
+                "failed_push_runs": failed_push_runs,
                 "risk_level": risk_summary["risk_level"],
                 "risk_score": float(risk_summary["risk_score"]),
             },
@@ -51,6 +73,8 @@ class DashboardService:
                 "supplier_risks": len(risk_summary["supplier_risks"]),
             },
             "latest_reports": [serialize_model(item) for item in latest_reports],
+            "latest_push_runs": [serialize_model(item) for item in latest_push_runs],
+            "latest_sync_runs": [serialize_model(item) for item in latest_sync_runs],
             "latest_audit_logs": [serialize_model(item) for item in latest_audit_logs],
         }
 

app/modules/feishu/events.py

@@ -96,12 +96,36 @@ class FeishuEventService:
             )
         comment = value.get("comment")
         actor = _approval_operator(payload)
+        event_identity = _approval_event_identity(payload, ticket_id, decision, actor)
+        if not self._register_event(event_identity):
+            ticket = ApprovalService(self.db).get_by_ticket(ticket_id)
+            return {
+                FeishuResponseKey.OK: True,
+                FeishuResponseKey.HANDLED: True,
+                FeishuResponseKey.DUPLICATE: True,
+                FeishuResponseKey.RESULT: {
+                    "ticket_id": ticket.ticket_id,
+                    "status": ticket.status,
+                    "approver": ticket.approver,
+                },
+            }
         ticket = ApprovalService(self.db).decide(
             ticket_id,
             actor,
             approved=decision == "approve",
             comment=str(comment) if comment is not None else None,
         )
+        self.feishu.audit.log(
+            AuditLogCreate(
+                actor=actor,
+                source=AuditSource.FEISHU,
+                action=AuditAction.FEISHU_WEBHOOK_EVENT,
+                target_type="approval_card_action",
+                target_id=ticket_id,
+                request_payload=payload,
+                response_payload={"status": ticket.status, "decision": decision},
+            )
+        )
         return {
             FeishuResponseKey.OK: True,
             FeishuResponseKey.HANDLED: True,
@@ -184,3 +208,20 @@ def _approval_operator(payload: dict[str, Any]) -> str:
         or operator.get(FeishuPayloadKey.USER_ID)
         or ActorValue.FEISHU
     )
+
+
+def _approval_event_identity(
+    payload: dict[str, Any],
+    ticket_id: str,
+    decision: str,
+    actor: str,
+) -> dict[str, str | None]:
+    header = payload.get(FeishuPayloadKey.HEADER) or {}
+    event_id = header.get(FeishuPayloadKey.EVENT_ID)
+    stable_id = event_id or f"{ticket_id}:{decision}:{actor}"
+    return {
+        FeishuEventReceiptKey.EVENT_KEY: f"{FeishuEventSource.WEBHOOK}:approval:{stable_id}",
+        FeishuEventReceiptKey.SOURCE: FeishuEventSource.WEBHOOK,
+        FeishuEventReceiptKey.EVENT_ID: str(event_id) if event_id else None,
+        FeishuEventReceiptKey.MESSAGE_ID: None,
+    }

app/modules/legacy_mysql/constants.py

@@ -3,6 +3,7 @@ from enum import StrEnum
 
 class LegacyQueryName(StrEnum):
     PROJECTS = "projects"
+    TASKS = "tasks"
 
 
 class LegacyResponseKey(StrEnum):
@@ -19,6 +20,7 @@ class LegacyResponseKey(StrEnum):
     REASON = "reason"
     SOURCE = "source"
     PROJECT = "project"
+    TASK = "task"
     SYNC_RUN_CODE = "sync_run_code"
     SOURCE_QUERY = "source_query"
     FIELD_MAP = "field_map"
@@ -44,6 +46,22 @@ class LegacyProjectField(StrEnum):
     DESCRIPTION = "description"
 
 
+class LegacyTaskField(StrEnum):
+    ID = "id"
+    CODE = "code"
+    EXTERNAL_ID = "external_id"
+    SOURCE_SYSTEM = "source_system"
+    TITLE = "title"
+    PROJECT_CODE = "project_code"
+    OWNER = "owner"
+    STATUS = "status"
+    PRIORITY = "priority"
+    DUE_DATE = "due_date"
+    COMPLETED_AT = "completed_at"
+    BLOCKER = "blocker"
+    DESCRIPTION = "description"
+
+
 class LegacySyncAction(StrEnum):
     CREATE = "create"
     UPDATE = "update"
@@ -55,6 +73,7 @@ class LegacyQueryError(StrEnum):
     DATABASE_NOT_CONFIGURED = "LEGACY_DATABASE_URL is not configured"
     QUERY_NOT_ALLOWED = "Legacy query is not in the configured allowlist"
     PROJECT_QUERY_NOT_CONFIGURED = "LEGACY_PROJECT_QUERY is not configured. Configure it first."
+    TASK_QUERY_NOT_CONFIGURED = "LEGACY_TASK_QUERY is not configured. Configure it first."
     ONLY_SELECT_ALLOWED = "Only SELECT statements are allowed"
     FORBIDDEN_SQL_TOKEN = "Forbidden SQL token in readonly query"
     INVALID_LIMIT = "Invalid readonly query limit"
@@ -62,9 +81,12 @@ class LegacyQueryError(StrEnum):
 
 
 LEGACY_SYNC_RUN_CODE_PREFIX = "SYNC-PROJECTS"
+LEGACY_TASK_SYNC_RUN_CODE_PREFIX = "SYNC-TASKS"
 LEGACY_PROJECT_QUERY_SOURCE = "LEGACY_PROJECT_QUERY"
+LEGACY_TASK_QUERY_SOURCE = "LEGACY_TASK_QUERY"
 LEGACY_SYNC_MISSING_ID_REASON = "missing external_id/code"
 LEGACY_PROJECT_SYNC_NOTE = "Project sync from readonly legacy MySQL"
+LEGACY_TASK_SYNC_NOTE = "Task sync from readonly legacy MySQL"
 LEGACY_MYSQL_CONNECTION_FAILED_TEMPLATE = "MySQL connection failed: {error}"
 LEGACY_HEALTH_SQL = "SELECT 1"
 LEGACY_SELECT_PREFIX = "select"
@@ -72,4 +94,6 @@ LEGACY_SQL_TRAILING_TERMINATOR = ";"
 LEGACY_LIMIT_MARKER = " limit "
 LEGACY_LIMIT_CLAUSE = " LIMIT :limit"
 LEGACY_PROJECT_CODE_TEMPLATE = "{prefix}-{external_id}"
+LEGACY_TASK_CODE_TEMPLATE = "{prefix}-{external_id}"
 LEGACY_UNNAMED_PROJECT = "未命名项目"
+LEGACY_UNNAMED_TASK = "未命名任务"

app/modules/legacy_mysql/routes.py

@@ -4,9 +4,12 @@ from sqlalchemy.orm import Session
 from app.core.database import get_db
 from app.core.masking import mask_configured
 from app.core.security import ApiPrincipal, require_api_key
+from app.core.task_queue import enqueue_legacy_project_sync, enqueue_legacy_task_sync
 from app.modules.legacy_mysql.schemas import (
     LegacyProjectSyncRequest,
     LegacyProjectSyncResult,
+    LegacyTaskSyncRequest,
+    LegacyTaskSyncResult,
     QueryResult,
     ReadonlyQueryRequest,
 )
@@ -43,6 +46,14 @@ def default_project_query(
     return mask_configured(LegacyMySQLService(db).fetch_default_projects(limit=limit))
 
 
+@router.get("/tasks", response_model=QueryResult)
+def default_task_query(
+    limit: int = Query(default=100, ge=1, le=500),
+    db: Session = Depends(get_db),
+) -> dict:
+    return mask_configured(LegacyMySQLService(db).fetch_default_tasks(limit=limit))
+
+
 @router.post("/projects/sync", response_model=LegacyProjectSyncResult)
 def sync_projects(
     payload: LegacyProjectSyncRequest,
@@ -58,3 +69,50 @@ def sync_projects(
         actor=principal.actor,
     )
     return mask_configured(result)
+
+
+@router.post("/projects/sync/enqueue")
+def enqueue_sync_projects(
+    payload: LegacyProjectSyncRequest,
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return enqueue_legacy_project_sync(
+        source_query=payload.source_query,
+        source_query_name=payload.source_query_name,
+        field_map=payload.field_map,
+        limit=payload.limit,
+        dry_run=payload.dry_run,
+        actor=principal.actor,
+    )
+
+
+@router.post("/tasks/sync", response_model=LegacyTaskSyncResult)
+def sync_tasks(
+    payload: LegacyTaskSyncRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    result = LegacyMySQLService(db).sync_tasks(
+        source_query=payload.source_query,
+        source_query_name=payload.source_query_name,
+        field_map=payload.field_map,
+        limit=payload.limit,
+        dry_run=payload.dry_run,
+        actor=principal.actor,
+    )
+    return mask_configured(result)
+
+
+@router.post("/tasks/sync/enqueue")
+def enqueue_sync_tasks(
+    payload: LegacyTaskSyncRequest,
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return enqueue_legacy_task_sync(
+        source_query=payload.source_query,
+        source_query_name=payload.source_query_name,
+        field_map=payload.field_map,
+        limit=payload.limit,
+        dry_run=payload.dry_run,
+        actor=principal.actor,
+    )

app/modules/legacy_mysql/schemas.py

@@ -67,3 +67,34 @@ class LegacyProjectSyncResult(BaseModel):
     skipped: int
     sync_run_code: str | None = None
     items: list[dict[str, Any]]
+
+
+class LegacyTaskSyncRequest(BaseModel):
+    """Request body for syncing legacy tasks into the internal ledger."""
+
+    source_query: str | None = Field(
+        default=None,
+        description="Deprecated: must exactly match a configured readonly query.",
+    )
+    source_query_name: str | None = Field(
+        default=LegacyQueryName.TASKS,
+        description="Configured readonly query name for task sync.",
+    )
+    field_map: dict[str, str] = Field(
+        default_factory=dict,
+        description="Map internal task fields to legacy row fields.",
+    )
+    limit: int = Field(default=100, ge=1, le=500)
+    dry_run: bool = True
+    actor: str = ActorValue.API
+
+
+class LegacyTaskSyncResult(BaseModel):
+    """Summary of a legacy task sync operation."""
+
+    dry_run: bool
+    created: int
+    updated: int
+    skipped: int
+    sync_run_code: str | None = None
+    items: list[dict[str, Any]]

app/modules/legacy_mysql/service.py

@@ -17,13 +17,17 @@ from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource
 from app.modules.audit.schemas import AuditLogCreate
 from app.modules.audit.service import AuditService
 from app.modules.business.constants import BusinessDomain, SourceSystem, StatusValue
-from app.modules.business.models import LegacySyncRun, Project
+from app.modules.business.models import LegacySyncRun, Project, WorkTask
 from app.modules.business.service import serialize_model
 from app.modules.legacy_mysql.constants import (
     LEGACY_PROJECT_QUERY_SOURCE,
     LEGACY_PROJECT_SYNC_NOTE,
     LEGACY_SYNC_MISSING_ID_REASON,
     LEGACY_SYNC_RUN_CODE_PREFIX,
+    LEGACY_TASK_CODE_TEMPLATE,
+    LEGACY_TASK_QUERY_SOURCE,
+    LEGACY_TASK_SYNC_NOTE,
+    LEGACY_TASK_SYNC_RUN_CODE_PREFIX,
     LEGACY_HEALTH_SQL,
     LEGACY_LIMIT_CLAUSE,
     LEGACY_LIMIT_MARKER,
@@ -31,12 +35,14 @@ from app.modules.legacy_mysql.constants import (
     LEGACY_PROJECT_CODE_TEMPLATE,
     LEGACY_SELECT_PREFIX,
     LEGACY_SQL_TRAILING_TERMINATOR,
+    LEGACY_UNNAMED_TASK,
     LEGACY_UNNAMED_PROJECT,
     LegacyProjectField,
     LegacyQueryError,
     LegacyQueryName,
     LegacyResponseKey,
     LegacySyncAction,
+    LegacyTaskField,
 )
 
 FORBIDDEN_SQL_TOKENS = {
@@ -120,6 +126,8 @@ class LegacyMySQLService:
         }
         if settings.legacy_project_query:
             queries.setdefault(LegacyQueryName.PROJECTS.value, settings.legacy_project_query)
+        if settings.legacy_task_query:
+            queries.setdefault(LegacyQueryName.TASKS.value, settings.legacy_task_query)
         return queries
 
     def health(self) -> dict[str, str]:
@@ -211,6 +219,19 @@ class LegacyMySQLService:
             limit=limit,
         )
 
+    def fetch_default_tasks(self, limit: int = 100) -> dict[str, Any]:
+        settings = get_settings()
+        if not settings.legacy_task_query:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=LegacyQueryError.TASK_QUERY_NOT_CONFIGURED,
+            )
+        return self.execute_allowed_query(
+            LegacyQueryName.TASKS,
+            {LegacyResponseKey.LIMIT: limit},
+            limit=limit,
+        )
+
     @staticmethod
     def _value(
         row: dict[str, Any],
@@ -272,6 +293,68 @@ class LegacyMySQLService:
             LegacyProjectField.DESCRIPTION: self._value(row, field_map, LegacyProjectField.DESCRIPTION, None),
         }
 
+    def _task_payload(self, row: dict[str, Any], field_map: dict[str, str]) -> dict[str, Any]:
+        settings = get_settings()
+        external_id = self._value(
+            row,
+            field_map,
+            LegacyTaskField.EXTERNAL_ID,
+            row.get(LegacyTaskField.ID),
+        )
+        raw_code = self._value(row, field_map, LegacyTaskField.CODE, None)
+        code = None
+        if raw_code:
+            code = str(raw_code)
+        elif external_id is not None:
+            code = LEGACY_TASK_CODE_TEMPLATE.format(
+                prefix=settings.legacy_task_code_prefix,
+                external_id=external_id,
+            )
+        return {
+            LegacyTaskField.CODE: code,
+            LegacyTaskField.EXTERNAL_ID: str(external_id) if external_id is not None else code,
+            LegacyTaskField.SOURCE_SYSTEM: SourceSystem.LEGACY_MYSQL,
+            LegacyTaskField.TITLE: self._value(
+                row,
+                field_map,
+                LegacyTaskField.TITLE,
+                LEGACY_UNNAMED_TASK,
+            ),
+            LegacyTaskField.PROJECT_CODE: self._value(
+                row,
+                field_map,
+                LegacyTaskField.PROJECT_CODE,
+                None,
+            ),
+            LegacyTaskField.OWNER: self._value(row, field_map, LegacyTaskField.OWNER, None),
+            LegacyTaskField.STATUS: self._value(
+                row,
+                field_map,
+                LegacyTaskField.STATUS,
+                StatusValue.TODO,
+            ),
+            LegacyTaskField.PRIORITY: self._value(
+                row,
+                field_map,
+                LegacyTaskField.PRIORITY,
+                "P2",
+            ),
+            LegacyTaskField.DUE_DATE: self._value(row, field_map, LegacyTaskField.DUE_DATE, None),
+            LegacyTaskField.COMPLETED_AT: self._value(
+                row,
+                field_map,
+                LegacyTaskField.COMPLETED_AT,
+                None,
+            ),
+            LegacyTaskField.BLOCKER: self._value(row, field_map, LegacyTaskField.BLOCKER, None),
+            LegacyTaskField.DESCRIPTION: self._value(
+                row,
+                field_map,
+                LegacyTaskField.DESCRIPTION,
+                None,
+            ),
+        }
+
     def sync_projects(
         self,
         source_query: str | None = None,
@@ -409,3 +492,141 @@ class LegacyMySQLService:
             )
         )
         return result
+
+    def sync_tasks(
+        self,
+        source_query: str | None = None,
+        source_query_name: str | None = None,
+        field_map: dict[str, str] | None = None,
+        limit: int = 100,
+        dry_run: bool = True,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        if self.db is None:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=LegacyQueryError.APP_DB_UNAVAILABLE,
+            )
+
+        query_name = source_query_name or LegacyQueryName.TASKS
+        if source_query:
+            rows = self.execute_readonly(
+                source_query,
+                {LegacyResponseKey.LIMIT: limit},
+                limit=limit,
+            )[LegacyResponseKey.ROWS]
+            query_ref = LegacySyncAction.ALLOWLISTED_INLINE_SQL
+        else:
+            rows = self.execute_allowed_query(
+                query_name,
+                {LegacyResponseKey.LIMIT: limit},
+                limit=limit,
+            )[LegacyResponseKey.ROWS]
+            query_ref = _query_name_text(query_name)
+        field_map = field_map or {}
+        created = 0
+        updated = 0
+        skipped = 0
+        items: list[dict[str, Any]] = []
+
+        for row in rows:
+            payload = self._task_payload(row, field_map)
+            if not payload[LegacyTaskField.EXTERNAL_ID] and not payload[LegacyTaskField.CODE]:
+                skipped += 1
+                items.append(
+                    {
+                        LegacyResponseKey.ACTION: LegacySyncAction.SKIPPED,
+                        LegacyResponseKey.REASON: LEGACY_SYNC_MISSING_ID_REASON,
+                        LegacyResponseKey.SOURCE: row,
+                    }
+                )
+                continue
+
+            stmt = select(WorkTask).where(
+                WorkTask.source_system == SourceSystem.LEGACY_MYSQL,
+                WorkTask.external_id == payload[LegacyTaskField.EXTERNAL_ID],
+            )
+            record = self.db.execute(stmt).scalar_one_or_none()
+            if record is None:
+                record = self.db.execute(
+                    select(WorkTask).where(WorkTask.code == payload[LegacyTaskField.CODE])
+                ).scalar_one_or_none()
+
+            if record is None:
+                created += 1
+                action = LegacySyncAction.CREATE
+                result = payload
+                if not dry_run:
+                    record = WorkTask(**payload)
+                    self.db.add(record)
+                    self.db.flush()
+                    result = serialize_model(record)
+            else:
+                updated += 1
+                action = LegacySyncAction.UPDATE
+                if not dry_run:
+                    for key, value in payload.items():
+                        setattr(record, key, value)
+                    self.db.flush()
+                    result = serialize_model(record)
+                else:
+                    result = payload
+            items.append(
+                {
+                    LegacyResponseKey.ACTION: action,
+                    LegacyResponseKey.TASK: result,
+                    LegacyResponseKey.SOURCE: row,
+                }
+            )
+
+        if not dry_run:
+            self.db.commit()
+
+        result = {
+            LegacyResponseKey.DRY_RUN: dry_run,
+            LegacyResponseKey.CREATED: created,
+            LegacyResponseKey.UPDATED: updated,
+            LegacyResponseKey.SKIPPED: skipped,
+            LegacyResponseKey.ITEMS: items,
+        }
+        sync_run = LegacySyncRun(
+            code=f"{LEGACY_TASK_SYNC_RUN_CODE_PREFIX}-{utc_now():%Y%m%d%H%M%S%f}",
+            domain=BusinessDomain.TASKS,
+            source_table=LEGACY_TASK_QUERY_SOURCE,
+            status=StatusValue.DRY_RUN if dry_run else AuditStatus.SUCCESS,
+            finished_at=utc_now(),
+            created_count=created,
+            updated_count=updated,
+            skipped_count=skipped,
+            note=LEGACY_TASK_SYNC_NOTE,
+        )
+        self.db.add(sync_run)
+        self.db.commit()
+        self.db.refresh(sync_run)
+        result[LegacyResponseKey.SYNC_RUN_CODE] = sync_run.code
+
+        AuditService(self.db).log(
+            AuditLogCreate(
+                actor=actor,
+                source=AuditSource.LEGACY_MYSQL,
+                action=AuditAction.LEGACY_SYNC_TASKS,
+                target_type=BusinessDomain.TASKS,
+                risk_level=AuditRiskLevel.MEDIUM,
+                request_payload={
+                    LegacyResponseKey.SOURCE_QUERY: query_ref,
+                    LegacyResponseKey.FIELD_MAP: field_map,
+                    LegacyResponseKey.LIMIT: limit,
+                    LegacyResponseKey.DRY_RUN: dry_run,
+                },
+                response_payload={
+                    key: result[key]
+                    for key in [
+                        LegacyResponseKey.DRY_RUN,
+                        LegacyResponseKey.CREATED,
+                        LegacyResponseKey.UPDATED,
+                        LegacyResponseKey.SKIPPED,
+                    ]
+                },
+            )
+        )
+        return result

app/modules/reports/constants.py

@@ -20,6 +20,20 @@ class ReportStatus(StrEnum):
     GENERATED = "已生成"
 
 
+class ReportPushStatus(StrEnum):
+    PENDING = "pending"
+    QUEUED = "queued"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+class ReportPushKey(StrEnum):
+    ITEMS = "items"
+    CODE = "code"
+    TASK_ID = "task_id"
+    STATUS = "status"
+
+
 class LifecycleSection(StrEnum):
     HEALTH = "health"
     PROJECTS = "projects"

app/modules/reports/routes.py

@@ -58,6 +58,23 @@ def attendance_summary(
     return ReportService(db).attendance_summary(work_date)
 
 
+@router.get("/push-runs")
+def list_push_runs(
+    status: str | None = None,
+    limit: int = 100,
+    db: Session = Depends(get_db),
+) -> dict:
+    return {"items": ReportService(db).list_push_runs(status_filter=status, limit=limit)}
+
+
+@router.get("/push-runs/{code}")
+def get_push_run(
+    code: str,
+    db: Session = Depends(get_db),
+) -> dict:
+    return ReportService(db).get_push_run(code)
+
+
 @router.post("/work-reports/generate")
 def generate_work_report(
     payload: WorkReportGenerateRequest,

app/modules/reports/service.py

@@ -8,7 +8,7 @@ from sqlalchemy.orm import Session
 from app.core.constants import ActorValue
 from app.core.pagination import bounded_limit
 from app.core.time import utc_now
-from app.modules.audit.constants import AuditSource, AuditTargetType
+from app.modules.audit.constants import AuditAction, AuditSource, AuditTargetType
 from app.modules.audit.schemas import AuditLogCreate
 from app.modules.audit.service import AuditService
 from app.modules.business.constants import (
@@ -27,6 +27,7 @@ from app.modules.business.models import (
     FundAccount,
     Procurement,
     Project,
+    ReportPushRun,
     RiskEvent,
     Supplier,
     WorkReport,
@@ -48,6 +49,7 @@ from app.modules.reports.constants import (
     LifecycleSection,
     MetricKey,
     ReportResponseKey,
+    ReportPushStatus,
     ReportStatus,
     ReportText,
     ReportTitle,
@@ -149,6 +151,85 @@ class ReportService:
             stmt = stmt.order_by(model.id.desc())
         return list(self.db.execute(stmt.limit(bounded_limit(limit))).scalars())
 
+    def create_push_run(
+        self,
+        report_type: str,
+        title: str | None,
+        receive_id: str | None,
+        receive_id_type: str,
+        actor: str,
+        status: str = ReportPushStatus.PENDING,
+    ) -> ReportPushRun:
+        record = ReportPushRun(
+            code=_next_code("PUSH"),
+            report_type=report_type,
+            title=title,
+            receive_id=receive_id,
+            receive_id_type=receive_id_type,
+            status=status,
+            actor=actor,
+            queued_at=utc_now(),
+        )
+        self.db.add(record)
+        self.db.commit()
+        self.db.refresh(record)
+        return record
+
+    def update_push_run(
+        self,
+        code: str,
+        status: str,
+        task_id: str | None = None,
+        provider_response: dict[str, Any] | None = None,
+        error_message: str | None = None,
+        sent: bool = False,
+    ) -> ReportPushRun:
+        record = self._get_push_run(code)
+        record.status = status
+        if task_id is not None:
+            record.task_id = task_id
+        if provider_response is not None:
+            record.provider_response = _json_safe(provider_response)
+        record.error_message = error_message
+        if sent:
+            record.sent_at = utc_now()
+        self.db.commit()
+        self.db.refresh(record)
+        return record
+
+    def list_push_runs(
+        self,
+        status_filter: str | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        stmt = select(ReportPushRun).order_by(ReportPushRun.id.desc()).limit(
+            bounded_limit(limit)
+        )
+        if status_filter:
+            stmt = (
+                select(ReportPushRun)
+                .where(ReportPushRun.status == status_filter)
+                .order_by(ReportPushRun.id.desc())
+                .limit(bounded_limit(limit))
+            )
+        return [serialize_model(item) for item in self.db.execute(stmt).scalars()]
+
+    def get_push_run(self, code: str) -> dict[str, Any]:
+        return serialize_model(self._get_push_run(code))
+
+    def _get_push_run(self, code: str) -> ReportPushRun:
+        from fastapi import HTTPException, status
+
+        record = self.db.execute(
+            select(ReportPushRun).where(ReportPushRun.code == code)
+        ).scalar_one_or_none()
+        if record is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Report push run not found",
+            )
+        return record
+
     def daily_brief(self) -> dict:
         project_count = self._count(Project)
         task_count = self._count(WorkTask)
@@ -1010,9 +1091,47 @@ class ReportService:
         receive_id: str | None,
         receive_id_type: str,
         actor: str,
+        push_run_code: str | None = None,
     ) -> dict:
+        report_type = str(report.get(ReportResponseKey.REPORT_TYPE) or report.get("type") or "report")
+        title = report.get(ReportResponseKey.TITLE)
+        push_run = (
+            self._get_push_run(push_run_code)
+            if push_run_code
+            else self.create_push_run(
+                report_type=report_type,
+                title=title,
+                receive_id=receive_id,
+                receive_id_type=receive_id_type,
+                actor=actor,
+            )
+        )
         card = FeishuService.build_basic_card(
             report[ReportResponseKey.TITLE],
             report[ReportResponseKey.LINES],
         )
-        return FeishuService(self.db).send_card(card, receive_id, receive_id_type, actor)
+        try:
+            result = FeishuService(self.db).send_card(card, receive_id, receive_id_type, actor)
+        except Exception as exc:
+            self.update_push_run(
+                push_run.code,
+                ReportPushStatus.FAILED,
+                error_message=str(exc),
+            )
+            raise
+        self.update_push_run(
+            push_run.code,
+            ReportPushStatus.SUCCESS,
+            provider_response=result,
+            sent=True,
+        )
+        AuditService(self.db).log(
+            AuditLogCreate(
+                actor=actor,
+                source=AuditSource.REPORTS,
+                action=AuditAction.REPORT_PUSH,
+                target_id=push_run.code,
+                response_payload={"status": ReportPushStatus.SUCCESS},
+            )
+        )
+        return result

app/modules/risk/constants.py

@@ -46,6 +46,21 @@ class RiskEventPayloadKey(StrEnum):
     EVIDENCE = "evidence"
 
 
+class RiskEventActionValue(StrEnum):
+    ASSIGN = "assign"
+    COMMENT = "comment"
+    RESOLVE = "resolve"
+    CLOSE = "close"
+    REOPEN = "reopen"
+
+
+class RiskEventActionKey(StrEnum):
+    ACTION = "action"
+    RISK_EVENT = "risk_event"
+    ACTION_RECORD = "action_record"
+    ITEMS = "items"
+
+
 RISK_SCORE_WEIGHTS = {
     RiskSummaryKey.OVERDUE_TASKS: 1,
     RiskSummaryKey.DELAYED_PROJECTS: 3,

app/modules/risk/routes.py

@@ -4,7 +4,14 @@ from sqlalchemy.orm import Session
 from app.core.database import get_db
 from app.core.security import ApiPrincipal, require_api_key
 from app.core.task_queue import enqueue_risk_event_generation
-from app.modules.risk.constants import RiskGenerationResultKey
+from app.modules.risk.constants import RiskEventActionKey, RiskGenerationResultKey
+from app.modules.risk.schemas import (
+    RiskAssignRequest,
+    RiskCloseRequest,
+    RiskCommentRequest,
+    RiskReopenRequest,
+    RiskResolveRequest,
+)
 from app.modules.risk.service import RiskService
 
 router = APIRouter(dependencies=[Depends(require_api_key)])
@@ -66,6 +73,94 @@ def risk_events(
     }
 
 
+@router.get("/events/{event_id}/actions")
+def risk_event_actions(
+    event_id: int,
+    limit: int = Query(default=100, ge=1, le=500),
+    db: Session = Depends(get_db),
+) -> dict:
+    return {
+        RiskEventActionKey.ITEMS: RiskService(db).list_actions(
+            risk_event_id=event_id,
+            limit=limit,
+        )
+    }
+
+
+@router.post("/events/{event_id}/assign")
+def assign_risk_event(
+    event_id: int,
+    payload: RiskAssignRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return RiskService(db).assign_event(
+        event_id,
+        assigned_to=payload.assigned_to,
+        comment=payload.comment,
+        actor=principal.actor,
+    )
+
+
+@router.post("/events/{event_id}/comment")
+def comment_risk_event(
+    event_id: int,
+    payload: RiskCommentRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return RiskService(db).comment_event(
+        event_id,
+        comment=payload.comment,
+        payload=payload.payload,
+        actor=principal.actor,
+    )
+
+
+@router.post("/events/{event_id}/resolve")
+def resolve_risk_event(
+    event_id: int,
+    payload: RiskResolveRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return RiskService(db).resolve_event(
+        event_id,
+        comment=payload.comment,
+        payload=payload.payload,
+        actor=principal.actor,
+    )
+
+
+@router.post("/events/{event_id}/close")
+def close_risk_event(
+    event_id: int,
+    payload: RiskCloseRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return RiskService(db).close_event(
+        event_id,
+        closed_reason=payload.closed_reason,
+        review_summary=payload.review_summary,
+        actor=principal.actor,
+    )
+
+
+@router.post("/events/{event_id}/reopen")
+def reopen_risk_event(
+    event_id: int,
+    payload: RiskReopenRequest,
+    db: Session = Depends(get_db),
+    principal: ApiPrincipal = Depends(require_api_key),
+) -> dict:
+    return RiskService(db).reopen_event(
+        event_id,
+        comment=payload.comment,
+        actor=principal.actor,
+    )
+
+
 @router.post("/events/generate")
 def generate_risk_events(
     db: Session = Depends(get_db),

app/modules/risk/schemas.py

@@ -0,0 +1,27 @@
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class RiskAssignRequest(BaseModel):
+    assigned_to: str = Field(..., min_length=1)
+    comment: str | None = None
+
+
+class RiskCommentRequest(BaseModel):
+    comment: str = Field(..., min_length=1)
+    payload: dict[str, Any] = Field(default_factory=dict)
+
+
+class RiskResolveRequest(BaseModel):
+    comment: str | None = None
+    payload: dict[str, Any] = Field(default_factory=dict)
+
+
+class RiskCloseRequest(BaseModel):
+    closed_reason: str = Field(..., min_length=1)
+    review_summary: str | None = None
+
+
+class RiskReopenRequest(BaseModel):
+    comment: str | None = None

app/modules/risk/service.py

@@ -27,10 +27,19 @@ from app.modules.business.constants import (
     RiskLevel,
     StatusValue,
 )
-from app.modules.business.models import FundAccount, Project, RiskEvent, Supplier, WorkTask
+from app.modules.business.models import (
+    FundAccount,
+    Project,
+    RiskEvent,
+    RiskEventAction,
+    Supplier,
+    WorkTask,
+)
 from app.modules.business.service import serialize_model
 from app.modules.risk.constants import (
     RISK_SCORE_WEIGHTS,
+    RiskEventActionKey,
+    RiskEventActionValue,
     RiskGenerationAction,
     RiskGenerationResultKey,
     RiskEventPayloadKey,
@@ -95,6 +104,142 @@ class RiskService:
             )
         return [serialize_model(item) for item in self.db.execute(stmt).scalars()]
 
+    def list_actions(self, risk_event_id: int, limit: int = 100) -> list[dict[str, Any]]:
+        self._get_event(risk_event_id)
+        stmt = (
+            select(RiskEventAction)
+            .where(RiskEventAction.risk_event_id == risk_event_id)
+            .order_by(RiskEventAction.id.desc())
+            .limit(bounded_limit(limit))
+        )
+        return [serialize_model(item) for item in self.db.execute(stmt).scalars()]
+
+    def assign_event(
+        self,
+        risk_event_id: int,
+        assigned_to: str,
+        comment: str | None = None,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        record = self._get_event(risk_event_id)
+        from_status = record.status
+        record.assigned_to = assigned_to
+        action = self._record_action(
+            record,
+            RiskEventActionValue.ASSIGN,
+            actor,
+            from_status,
+            record.status,
+            comment,
+            {"assigned_to": assigned_to},
+        )
+        self.db.commit()
+        self.db.refresh(record)
+        self.db.refresh(action)
+        return self._action_response(record, action)
+
+    def comment_event(
+        self,
+        risk_event_id: int,
+        comment: str,
+        payload: dict[str, Any] | None = None,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        record = self._get_event(risk_event_id)
+        action = self._record_action(
+            record,
+            RiskEventActionValue.COMMENT,
+            actor,
+            record.status,
+            record.status,
+            comment,
+            payload or {},
+        )
+        self.db.commit()
+        self.db.refresh(record)
+        self.db.refresh(action)
+        return self._action_response(record, action)
+
+    def resolve_event(
+        self,
+        risk_event_id: int,
+        comment: str | None = None,
+        payload: dict[str, Any] | None = None,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        record = self._get_event(risk_event_id)
+        from_status = record.status
+        record.status = StatusValue.RESOLVED
+        record.resolved_at = utc_now()
+        action = self._record_action(
+            record,
+            RiskEventActionValue.RESOLVE,
+            actor,
+            from_status,
+            record.status,
+            comment,
+            payload or {},
+        )
+        self.db.commit()
+        self.db.refresh(record)
+        self.db.refresh(action)
+        return self._action_response(record, action)
+
+    def close_event(
+        self,
+        risk_event_id: int,
+        closed_reason: str,
+        review_summary: str | None = None,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        record = self._get_event(risk_event_id)
+        from_status = record.status
+        now = utc_now()
+        record.status = StatusValue.CLOSED
+        record.closed_reason = closed_reason
+        record.review_summary = review_summary
+        record.closed_at = now
+        if record.resolved_at is None:
+            record.resolved_at = now
+        action = self._record_action(
+            record,
+            RiskEventActionValue.CLOSE,
+            actor,
+            from_status,
+            record.status,
+            closed_reason,
+            {"review_summary": review_summary},
+        )
+        self.db.commit()
+        self.db.refresh(record)
+        self.db.refresh(action)
+        return self._action_response(record, action)
+
+    def reopen_event(
+        self,
+        risk_event_id: int,
+        comment: str | None = None,
+        actor: str = ActorValue.API,
+    ) -> dict[str, Any]:
+        record = self._get_event(risk_event_id)
+        from_status = record.status
+        record.status = StatusValue.OPEN
+        record.resolved_at = None
+        record.closed_at = None
+        action = self._record_action(
+            record,
+            RiskEventActionValue.REOPEN,
+            actor,
+            from_status,
+            record.status,
+            comment,
+            {},
+        )
+        self.db.commit()
+        self.db.refresh(record)
+        self.db.refresh(action)
+        return self._action_response(record, action)
+
     def summary(self) -> dict[str, Any]:
         overdue_tasks = self.overdue_tasks()
         delayed_projects = self.delayed_projects()
@@ -127,6 +272,62 @@ class RiskService:
             RiskSummaryKey.OPEN_EVENTS: open_events,
         }
 
+    def _get_event(self, risk_event_id: int) -> RiskEvent:
+        record = self.db.get(RiskEvent, risk_event_id)
+        if record is None:
+            from fastapi import HTTPException, status
+
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Risk event not found")
+        return record
+
+    def _record_action(
+        self,
+        record: RiskEvent,
+        action: str,
+        actor: str,
+        from_status: str | None,
+        to_status: str | None,
+        comment: str | None,
+        payload: dict[str, Any],
+    ) -> RiskEventAction:
+        action_record = RiskEventAction(
+            code=f"RISK-ACTION-{utc_now():%Y%m%d%H%M%S%f}",
+            risk_event_id=record.id,
+            action=action,
+            actor=actor,
+            from_status=from_status,
+            to_status=to_status,
+            assigned_to=record.assigned_to,
+            comment=comment,
+            payload=payload,
+        )
+        self.db.add(action_record)
+        AuditService(self.db).log(
+            AuditLogCreate(
+                actor=actor,
+                source=AuditSource.RISK,
+                action=AuditAction.RISK_EVENT_ACTION,
+                target_type=BusinessDomain.RISK_EVENTS,
+                target_id=str(record.id),
+                risk_level=AuditRiskLevel.MEDIUM,
+                request_payload={
+                    RiskEventActionKey.ACTION: action,
+                    "from_status": from_status,
+                    "to_status": to_status,
+                    "comment": comment,
+                    "payload": payload,
+                },
+            )
+        )
+        return action_record
+
+    @staticmethod
+    def _action_response(record: RiskEvent, action: RiskEventAction) -> dict[str, Any]:
+        return {
+            RiskEventActionKey.RISK_EVENT: serialize_model(record),
+            RiskEventActionKey.ACTION_RECORD: serialize_model(action),
+        }
+
     def generate_events(self, actor: str = ActorValue.API) -> dict[str, Any]:
         """Generate or refresh risk-event ledger entries from current signals."""
 

app/tasks.py

@@ -22,13 +22,20 @@ def push_daily_brief(
     receive_id: str | None = None,
     receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
     actor: str = ActorValue.SCHEDULER,
+    push_run_code: str | None = None,
 ) -> dict[str, Any]:
     from app.modules.reports.service import ReportService
 
     db = SessionLocal()
     try:
         report = ReportService(db).daily_brief()
-        return ReportService(db).push_report(report, receive_id, receive_id_type, actor)
+        return ReportService(db).push_report(
+            report,
+            receive_id,
+            receive_id_type,
+            actor,
+            push_run_code=push_run_code,
+        )
     finally:
         db.close()
 
@@ -38,13 +45,20 @@ def push_project_weekly(
     receive_id: str | None = None,
     receive_id_type: str = FeishuReceiveIdType.CHAT_ID,
     actor: str = ActorValue.SCHEDULER,
+    push_run_code: str | None = None,
 ) -> dict[str, Any]:
     from app.modules.reports.service import ReportService
 
     db = SessionLocal()
     try:
         report = ReportService(db).project_weekly()
-        return ReportService(db).push_report(report, receive_id, receive_id_type, actor)
+        return ReportService(db).push_report(
+            report,
+            receive_id,
+            receive_id_type,
+            actor,
+            push_run_code=push_run_code,
+        )
     finally:
         db.close()
 
@@ -58,3 +72,53 @@ def generate_risk_events(actor: str = ActorValue.SCHEDULER) -> dict[str, Any]:
         return RiskService(db).generate_events(actor=actor)
     finally:
         db.close()
+
+
+@celery_app.task(name="legacy.sync_projects")
+def sync_legacy_projects(
+    source_query: str | None = None,
+    source_query_name: str | None = None,
+    field_map: dict[str, str] | None = None,
+    limit: int = 100,
+    dry_run: bool = False,
+    actor: str = ActorValue.SCHEDULER,
+) -> dict[str, Any]:
+    from app.modules.legacy_mysql.service import LegacyMySQLService
+
+    db = SessionLocal()
+    try:
+        return LegacyMySQLService(db).sync_projects(
+            source_query=source_query,
+            source_query_name=source_query_name,
+            field_map=field_map or {},
+            limit=limit,
+            dry_run=dry_run,
+            actor=actor,
+        )
+    finally:
+        db.close()
+
+
+@celery_app.task(name="legacy.sync_tasks")
+def sync_legacy_tasks(
+    source_query: str | None = None,
+    source_query_name: str | None = None,
+    field_map: dict[str, str] | None = None,
+    limit: int = 100,
+    dry_run: bool = False,
+    actor: str = ActorValue.SCHEDULER,
+) -> dict[str, Any]:
+    from app.modules.legacy_mysql.service import LegacyMySQLService
+
+    db = SessionLocal()
+    try:
+        return LegacyMySQLService(db).sync_tasks(
+            source_query=source_query,
+            source_query_name=source_query_name,
+            field_map=field_map or {},
+            limit=limit,
+            dry_run=dry_run,
+            actor=actor,
+        )
+    finally:
+        db.close()

app/tools/init_db.py

@@ -10,7 +10,9 @@ from app.modules.business.models import (
     Policy,
     Procurement,
     Project,
+    ReportPushRun,
     RiskEvent,
+    RiskEventAction,
     Standard,
     Supplier,
     WorkReport,
@@ -34,7 +36,9 @@ _MODELS = [
     AttendanceRecord,
     WorkReport,
     RiskEvent,
+    RiskEventAction,
     LegacySyncRun,
+    ReportPushRun,
 ]