diff --git a/alembic/versions/202607080001_v2_production_controls.py b/alembic/versions/202607080001_v2_production_controls.py new file mode 100644 index 0000000..377c464 --- /dev/null +++ b/alembic/versions/202607080001_v2_production_controls.py @@ -0,0 +1,167 @@ +"""Add V2 production control tables and fields. + +Revision ID: 202607080001 +Revises: 202607060003 +Create Date: 2026-07-08 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import inspect + + +revision = "202607080001" +down_revision = "202607060003" +branch_labels = None +depends_on = None + +WORK_TASKS_TABLE = "work_tasks" +RISK_EVENTS_TABLE = "risk_events" +RISK_EVENT_ACTIONS_TABLE = "risk_event_actions" +REPORT_PUSH_RUNS_TABLE = "report_push_runs" + + +def _table_exists(table_name: str) -> bool: + inspector = inspect(op.get_bind()) + return table_name in inspector.get_table_names() + + +def _column_names(table_name: str) -> set[str]: + inspector = inspect(op.get_bind()) + if table_name not in inspector.get_table_names(): + return set() + return {column["name"] for column in inspector.get_columns(table_name)} + + +def _add_column_if_missing(table_name: str, column: sa.Column) -> None: + if column.name not in _column_names(table_name): + op.add_column(table_name, column) + + +def upgrade() -> None: + _add_column_if_missing( + WORK_TASKS_TABLE, + sa.Column("source_system", sa.String(length=64), nullable=True), + ) + _add_column_if_missing( + WORK_TASKS_TABLE, + sa.Column("external_id", sa.String(length=128), nullable=True), + ) + op.create_index( + op.f("ix_work_tasks_external_id"), + WORK_TASKS_TABLE, + ["external_id"], + unique=False, + if_not_exists=True, + ) + + for column in [ + sa.Column("assigned_to", sa.String(length=128), nullable=True), + sa.Column("resolved_at", sa.DateTime(), nullable=True), + sa.Column("closed_at", sa.DateTime(), nullable=True), + sa.Column("closed_reason", sa.Text(), nullable=True), + sa.Column("review_summary", sa.Text(), nullable=True), + ]: + _add_column_if_missing(RISK_EVENTS_TABLE, column) + op.create_index( + op.f("ix_risk_events_assigned_to"), + RISK_EVENTS_TABLE, + ["assigned_to"], + unique=False, + if_not_exists=True, + ) + + if not _table_exists(RISK_EVENT_ACTIONS_TABLE): + op.create_table( + RISK_EVENT_ACTIONS_TABLE, + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("code", sa.String(length=64), nullable=False), + sa.Column("risk_event_id", sa.Integer(), nullable=False), + sa.Column("action", sa.String(length=64), nullable=False), + sa.Column("actor", sa.String(length=128), nullable=False), + sa.Column("from_status", sa.String(length=32), nullable=True), + sa.Column("to_status", sa.String(length=32), nullable=True), + sa.Column("assigned_to", sa.String(length=128), nullable=True), + sa.Column("comment", sa.Text(), nullable=True), + sa.Column("payload", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + for column_name, unique in [ + ("code", True), + ("risk_event_id", False), + ("action", False), + ("actor", False), + ("assigned_to", False), + ("created_at", False), + ]: + op.create_index( + op.f(f"ix_{RISK_EVENT_ACTIONS_TABLE}_{column_name}"), + RISK_EVENT_ACTIONS_TABLE, + [column_name], + unique=unique, + ) + + if not _table_exists(REPORT_PUSH_RUNS_TABLE): + op.create_table( + REPORT_PUSH_RUNS_TABLE, + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("code", sa.String(length=64), nullable=False), + sa.Column("report_type", sa.String(length=64), nullable=False), + sa.Column("title", sa.String(length=255), nullable=True), + sa.Column("receive_id", sa.String(length=128), nullable=True), + sa.Column("receive_id_type", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("task_id", sa.String(length=128), nullable=True), + sa.Column("actor", sa.String(length=128), nullable=False), + sa.Column("provider_response", sa.JSON(), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.Column("queued_at", sa.DateTime(), nullable=False), + sa.Column("sent_at", sa.DateTime(), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + for column_name, unique in [ + ("code", True), + ("report_type", False), + ("receive_id", False), + ("receive_id_type", False), + ("status", False), + ("task_id", False), + ("actor", False), + ("queued_at", False), + ]: + op.create_index( + op.f(f"ix_{REPORT_PUSH_RUNS_TABLE}_{column_name}"), + REPORT_PUSH_RUNS_TABLE, + [column_name], + unique=unique, + ) + + +def downgrade() -> None: + if _table_exists(REPORT_PUSH_RUNS_TABLE): + op.drop_table(REPORT_PUSH_RUNS_TABLE) + if _table_exists(RISK_EVENT_ACTIONS_TABLE): + op.drop_table(RISK_EVENT_ACTIONS_TABLE) + + risk_columns = _column_names(RISK_EVENTS_TABLE) + if "assigned_to" in risk_columns: + op.drop_index(op.f("ix_risk_events_assigned_to"), table_name=RISK_EVENTS_TABLE) + for column_name in [ + "review_summary", + "closed_reason", + "closed_at", + "resolved_at", + "assigned_to", + ]: + if column_name in _column_names(RISK_EVENTS_TABLE): + op.drop_column(RISK_EVENTS_TABLE, column_name) + + task_columns = _column_names(WORK_TASKS_TABLE) + if "external_id" in task_columns: + op.drop_index(op.f("ix_work_tasks_external_id"), table_name=WORK_TASKS_TABLE) + op.drop_column(WORK_TASKS_TABLE, "external_id") + if "source_system" in _column_names(WORK_TASKS_TABLE): + op.drop_column(WORK_TASKS_TABLE, "source_system") diff --git a/app/core/config.py b/app/core/config.py index 9ff3cbe..4a18829 100644 --- a/app/core/config.py +++ b/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]: diff --git a/app/core/constants.py b/app/core/constants.py index 8d89f81..78b1329 100644 --- a/app/core/constants.py +++ b/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}" diff --git a/app/core/masking.py b/app/core/masking.py index 158da88..6ec8603 100644 --- a/app/core/masking.py +++ b/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) diff --git a/app/core/scheduler.py b/app/core/scheduler.py index d241d56..c5977ec 100644 --- a/app/core/scheduler.py +++ b/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: diff --git a/app/core/security.py b/app/core/security.py index 2991f1d..67a09c1 100644 --- a/app/core/security.py +++ b/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"} diff --git a/app/core/task_queue.py b/app/core/task_queue.py index 76d3f8f..dbeac2f 100644 --- a/app/core/task_queue.py +++ b/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, + ) diff --git a/app/modules/audit/constants.py b/app/modules/audit/constants.py index 5c4ad96..dde9e0b 100644 --- a/app/modules/audit/constants.py +++ b/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): diff --git a/app/modules/business/models.py b/app/modules/business/models.py index 939d270..9dd9d56 100644 --- a/app/modules/business/models.py +++ b/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) diff --git a/app/modules/business/routes.py b/app/modules/business/routes.py index 074f30d..b5a3383 100644 --- a/app/modules/business/routes.py +++ b/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), } diff --git a/app/modules/dashboard/service.py b/app/modules/dashboard/service.py index 3d5f6ac..9cdcf01 100644 --- a/app/modules/dashboard/service.py +++ b/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], } diff --git a/app/modules/feishu/events.py b/app/modules/feishu/events.py index 7a6b772..2d99a14 100644 --- a/app/modules/feishu/events.py +++ b/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, + } diff --git a/app/modules/legacy_mysql/constants.py b/app/modules/legacy_mysql/constants.py index 6f2dd00..147ea59 100644 --- a/app/modules/legacy_mysql/constants.py +++ b/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 = "未命名任务" diff --git a/app/modules/legacy_mysql/routes.py b/app/modules/legacy_mysql/routes.py index 65973a3..6bcad56 100644 --- a/app/modules/legacy_mysql/routes.py +++ b/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, + ) diff --git a/app/modules/legacy_mysql/schemas.py b/app/modules/legacy_mysql/schemas.py index fe3e397..a8dc98a 100644 --- a/app/modules/legacy_mysql/schemas.py +++ b/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]] diff --git a/app/modules/legacy_mysql/service.py b/app/modules/legacy_mysql/service.py index 5fdf4d0..1c1ee2e 100644 --- a/app/modules/legacy_mysql/service.py +++ b/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 diff --git a/app/modules/reports/constants.py b/app/modules/reports/constants.py index 7d175ed..9cb07a1 100644 --- a/app/modules/reports/constants.py +++ b/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" diff --git a/app/modules/reports/routes.py b/app/modules/reports/routes.py index 6b6456e..e68d593 100644 --- a/app/modules/reports/routes.py +++ b/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, diff --git a/app/modules/reports/service.py b/app/modules/reports/service.py index 17db868..90e8db4 100644 --- a/app/modules/reports/service.py +++ b/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 diff --git a/app/modules/risk/constants.py b/app/modules/risk/constants.py index 5d57c8c..0ffeb5c 100644 --- a/app/modules/risk/constants.py +++ b/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, diff --git a/app/modules/risk/routes.py b/app/modules/risk/routes.py index 36484fe..218b177 100644 --- a/app/modules/risk/routes.py +++ b/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), diff --git a/app/modules/risk/schemas.py b/app/modules/risk/schemas.py new file mode 100644 index 0000000..49102b5 --- /dev/null +++ b/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 diff --git a/app/modules/risk/service.py b/app/modules/risk/service.py index ed3276d..b1a598c 100644 --- a/app/modules/risk/service.py +++ b/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.""" diff --git a/app/tasks.py b/app/tasks.py index a1ee7d8..6708b97 100644 --- a/app/tasks.py +++ b/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() diff --git a/app/tools/init_db.py b/app/tools/init_db.py index b9ccd40..ee6ebb2 100644 --- a/app/tools/init_db.py +++ b/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, ] diff --git a/scripts/sample_requests.http b/scripts/sample_requests.http index 962778e..eb70445 100644 --- a/scripts/sample_requests.http +++ b/scripts/sample_requests.http @@ -27,6 +27,35 @@ X-API-Key: {{apiKey}} GET http://127.0.0.1:8010/api/v1/reports/daily-brief X-API-Key: {{apiKey}} +### Sync legacy tasks +POST http://127.0.0.1:8010/api/v1/integrations/mysql/tasks/sync +Content-Type: application/json +X-API-Key: {{apiKey}} + +{ + "dry_run": true, + "limit": 50, + "field_map": { + "title": "task_name", + "owner": "task_assignee", + "due_date": "task_end_time" + } +} + +### Risk assign +POST http://127.0.0.1:8010/api/v1/risks/events/1/assign +Content-Type: application/json +X-API-Key: {{apiKey}} + +{ + "assigned_to": "risk-owner", + "comment": "请跟进处理" +} + +### Report push runs +GET http://127.0.0.1:8010/api/v1/reports/push-runs +X-API-Key: {{apiKey}} + ### AI ask POST http://127.0.0.1:8010/api/v1/ai/ask Content-Type: application/json diff --git a/security_fix_ignore.yml b/security_fix_ignore.yml index d9f37f0..c7ae0a5 100644 --- a/security_fix_ignore.yml +++ b/security_fix_ignore.yml @@ -6,6 +6,6 @@ ignored_findings: notes: "Do not copy the values into reports or tracked documentation." - id: P0_DEPLOYMENT_DOC_CONTAINS_GATEWAY_CREDENTIALS path: docs/deployment/外部项目接入OpenClawHermes网关说明.md - reason: "User explicitly excluded deployment document credential cleanup from this remediation pass." - status: ignored - notes: "Do not copy the values into reports or additional tracked files." + reason: "Deployment documentation now uses placeholders instead of concrete gateway credentials." + status: remediated + notes: "Keep future examples as placeholders and read real values from secure server-side storage." diff --git a/tests/test_smoke.py b/tests/test_smoke.py index d48aa25..30f6852 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -42,6 +42,7 @@ from app.modules.reports.constants import ( LifecycleResponseKey, LifecycleSection, MetricKey, + ReportPushStatus, ReportTitle, ReportType, ) @@ -179,6 +180,29 @@ def test_api_key_and_feishu_webhook_fail_closed(monkeypatch) -> None: get_settings.cache_clear() +def test_service_key_rotation_config(monkeypatch) -> None: + monkeypatch.setenv("API_KEY", "") + monkeypatch.setenv( + "API_KEYS", + json.dumps( + [ + {"key": "disabled-key", "actor": "disabled", "enabled": False}, + {"key": "rotated-key", "actor": "rotated-api", "enabled": True}, + ] + ), + ) + get_settings.cache_clear() + try: + assert require_api_key("rotated-key").actor == "rotated-api" + with pytest.raises(HTTPException) as exc_info: + require_api_key("disabled-key") + assert exc_info.value.status_code == 401 + finally: + monkeypatch.setenv("API_KEY", "test-key") + monkeypatch.delenv("API_KEYS", raising=False) + get_settings.cache_clear() + + def test_config_and_pagination_guardrails() -> None: settings = Settings(cors_origins='["https://app.example.com", "https://admin.example.com"]') @@ -227,6 +251,37 @@ def test_dashboard_and_response_masking() -> None: assert "metrics" in dashboard_response.json() +def test_configured_domain_response_masking(monkeypatch) -> None: + monkeypatch.setenv("MASKED_RESPONSE_FIELDS", json.dumps(["expenses.amount"])) + get_settings.cache_clear() + try: + response = client.post( + "/api/v1/business/expenses", + headers=headers, + json={ + "data": { + "code": "EXP-MASK-CONFIG-001", + "expense_type": "测试", + "amount": 123, + }, + }, + ) + assert response.status_code == 200 + assert response.json()["data"]["amount"] == "[MASKED]" + + list_response = client.get("/api/v1/business/expenses", headers=headers) + assert list_response.status_code == 200 + item = next( + item + for item in list_response.json()["items"] + if item["code"] == "EXP-MASK-CONFIG-001" + ) + assert item["amount"] == "[MASKED]" + finally: + monkeypatch.delenv("MASKED_RESPONSE_FIELDS", raising=False) + get_settings.cache_clear() + + def test_approval_gate_for_high_risk_update() -> None: create_payload = { "code": "FUND-SMOKE-001", @@ -420,6 +475,24 @@ def test_feishu_approval_card_action_approves_ticket() -> None: assert callback_response.json()["result"]["status"] == "approved" assert callback_response.json()["result"]["approver"] == "ou_card_approver" + duplicate_response = client.post( + "/api/v1/integrations/feishu/approval-card-action", + json={ + "token": "test-feishu-token", + "operator": {"operator_id": {"open_id": "ou_card_approver"}}, + "action": { + "value": { + "ticket_id": ticket_id, + "decision": "approve", + "comment": "approved from card", + } + }, + }, + ) + assert duplicate_response.status_code == 200 + assert duplicate_response.json()["duplicate"] is True + assert duplicate_response.json()["result"]["status"] == "approved" + def test_new_ledgers_reports_and_risk_events() -> None: domains_response = client.get("/api/v1/business/domains", headers=headers) @@ -684,6 +757,148 @@ def test_work_report_counts_pending_approval_backlog_outside_period() -> None: assert metrics["expenses_pending"] == 1 +def test_legacy_task_sync_creates_and_updates_internal_tasks(monkeypatch) -> None: + rows = [ + { + "id": 9001, + "task_name": "Legacy task one", + "project_code": "P-SMOKE-001", + "owner": "legacy-owner", + "status": "待办", + } + ] + + def fake_execute_allowed_query(self, query_name, params=None, limit=100): + return {"columns": list(rows[0]), "rows": rows, "row_count": len(rows)} + + monkeypatch.setattr( + LegacyMySQLService, + "execute_allowed_query", + fake_execute_allowed_query, + ) + + response = client.post( + "/api/v1/integrations/mysql/tasks/sync", + headers=headers, + json={ + "dry_run": False, + "field_map": {"title": "task_name"}, + }, + ) + assert response.status_code == 200 + assert response.json()["created"] == 1 + assert response.json()["items"][0]["task"]["external_id"] == "9001" + + rows[0]["task_name"] = "Legacy task one updated" + second_response = client.post( + "/api/v1/integrations/mysql/tasks/sync", + headers=headers, + json={ + "dry_run": False, + "field_map": {"title": "task_name"}, + }, + ) + assert second_response.status_code == 200 + assert second_response.json()["updated"] == 1 + + tasks_response = client.get("/api/v1/business/tasks", headers=headers) + assert tasks_response.status_code == 200 + item = next( + item for item in tasks_response.json()["items"] if item["external_id"] == "9001" + ) + assert item["title"] == "Legacy task one updated" + + +def test_risk_event_workflow_records_actions() -> None: + create_response = client.post( + "/api/v1/business/risk-events", + headers=headers, + json={ + "data": { + "code": "RISK-FLOW-001", + "title": "Workflow risk", + "risk_type": "manual", + "risk_level": "medium", + "source_domain": "projects", + "source_record_id": "P-SMOKE-001", + "status": "open", + }, + }, + ) + assert create_response.status_code == 200 + event_id = create_response.json()["data"]["id"] + + assign_response = client.post( + f"/api/v1/risks/events/{event_id}/assign", + headers=headers, + json={"assigned_to": "risk-owner", "comment": "please handle"}, + ) + assert assign_response.status_code == 200 + assert assign_response.json()["risk_event"]["assigned_to"] == "risk-owner" + + comment_response = client.post( + f"/api/v1/risks/events/{event_id}/comment", + headers=headers, + json={"comment": "working on it", "payload": {"step": 1}}, + ) + assert comment_response.status_code == 200 + + resolve_response = client.post( + f"/api/v1/risks/events/{event_id}/resolve", + headers=headers, + json={"comment": "resolved"}, + ) + assert resolve_response.status_code == 200 + assert resolve_response.json()["risk_event"]["status"] == "resolved" + + close_response = client.post( + f"/api/v1/risks/events/{event_id}/close", + headers=headers, + json={"closed_reason": "verified", "review_summary": "handled"}, + ) + assert close_response.status_code == 200 + assert close_response.json()["risk_event"]["status"] == "closed" + assert close_response.json()["risk_event"]["closed_reason"] == "verified" + + reopen_response = client.post( + f"/api/v1/risks/events/{event_id}/reopen", + headers=headers, + json={"comment": "recheck"}, + ) + assert reopen_response.status_code == 200 + assert reopen_response.json()["risk_event"]["status"] == "open" + + actions_response = client.get( + f"/api/v1/risks/events/{event_id}/actions", + headers=headers, + ) + assert actions_response.status_code == 200 + assert len(actions_response.json()["items"]) >= 5 + + +def test_report_push_failure_is_recorded() -> None: + response = client.post( + "/api/v1/reports/daily-brief/push", + headers=headers, + json={"receive_id": "oc_missing_config"}, + ) + assert response.status_code == 503 + + runs_response = client.get( + f"/api/v1/reports/push-runs?status={ReportPushStatus.FAILED}", + headers=headers, + ) + assert runs_response.status_code == 200 + assert any( + item["title"] == ReportTitle.DAILY_BRIEF + for item in runs_response.json()["items"] + ) + + dashboard_response = client.get("/api/v1/dashboard/summary", headers=headers) + assert dashboard_response.status_code == 200 + assert dashboard_response.json()["metrics"]["failed_push_runs"] >= 1 + + def test_ai_noop_provider() -> None: response = client.post( "/api/v1/ai/ask",