From 4d09d8e2e35aed9a9c10b93c0076459e35175587 Mon Sep 17 00:00:00 2001 From: JiuContinent Date: Tue, 7 Jul 2026 18:33:07 +0800 Subject: [PATCH] =?UTF-8?q?```=20feat:=20=E6=B7=BB=E5=8A=A0=E4=BB=AA?= =?UTF-8?q?=E8=A1=A8=E6=9D=BF=E8=B7=AF=E7=94=B1=E5=92=8C=E5=93=8D=E5=BA=94?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E8=84=B1=E6=95=8F=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加了仪表板模块路由并集成到主路由器中 - 实现了敏感数据响应脱敏配置和功能 - 增加了 Feishu 审批卡片操作处理功能 - 支持通过任务队列异步推送日常简报和项目周报 - 添加了风险事件生成的任务队列支持 - 在 smoke 测试中增加了相关功能验证 refactor: 格式化模型注册模块导入列表 - 将单行导入列表改为多行格式以提高可读性 ``` --- alembic/env.py | 7 +- .../versions/202607060001_initial_schema.py | 7 +- app/api/router.py | 2 + app/core/config.py | 4 + app/core/masking.py | 54 +++++++++ app/core/scheduler.py | 15 +++ app/core/task_queue.py | 107 ++++++++++++++++++ app/modules/approvals/routes.py | 34 +++++- app/modules/approvals/schemas.py | 9 ++ app/modules/business/routes.py | 23 +++- app/modules/dashboard/__init__.py | 1 + app/modules/dashboard/routes.py | 18 +++ app/modules/dashboard/service.py | 61 ++++++++++ app/modules/feishu/events.py | 61 ++++++++++ app/modules/feishu/routes.py | 16 ++- app/modules/feishu/service.py | 34 ++++++ app/modules/legacy_mysql/routes.py | 21 +++- app/modules/reports/routes.py | 33 +++++- app/modules/risk/routes.py | 32 +++++- app/tasks.py | 60 ++++++++++ tests/test_smoke.py | 67 +++++++++++ 21 files changed, 641 insertions(+), 25 deletions(-) create mode 100644 app/core/masking.py create mode 100644 app/core/task_queue.py create mode 100644 app/modules/dashboard/__init__.py create mode 100644 app/modules/dashboard/routes.py create mode 100644 app/modules/dashboard/service.py create mode 100644 app/tasks.py diff --git a/alembic/env.py b/alembic/env.py index 52b6ba1..d1e61f4 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -19,7 +19,12 @@ target_metadata = Base.metadata settings = get_settings() # Keep imports referenced so SQLAlchemy model classes register with Base.metadata. -_REGISTERED_MODEL_MODULES = (approval_models, audit_models, business_models, feishu_models) +_REGISTERED_MODEL_MODULES = ( + approval_models, + audit_models, + business_models, + feishu_models, +) def run_migrations_offline() -> None: diff --git a/alembic/versions/202607060001_initial_schema.py b/alembic/versions/202607060001_initial_schema.py index b2a7cbe..ee234e2 100644 --- a/alembic/versions/202607060001_initial_schema.py +++ b/alembic/versions/202607060001_initial_schema.py @@ -19,7 +19,12 @@ branch_labels = None depends_on = None # Keep imports referenced so SQLAlchemy model classes register with Base.metadata. -_REGISTERED_MODEL_MODULES = (approval_models, audit_models, business_models, feishu_models) +_REGISTERED_MODEL_MODULES = ( + approval_models, + audit_models, + business_models, + feishu_models, +) def upgrade() -> None: diff --git a/app/api/router.py b/app/api/router.py index 3c7b0d7..e6b80ff 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -5,6 +5,7 @@ from app.modules.ai_agent.routes import router as ai_router from app.modules.approvals.routes import router as approvals_router from app.modules.audit.routes import router as audit_router from app.modules.business.routes import router as business_router +from app.modules.dashboard.routes import router as dashboard_router from app.modules.feishu.routes import router as feishu_router from app.modules.legacy_mysql.routes import router as legacy_mysql_router from app.modules.reports.routes import router as reports_router @@ -21,6 +22,7 @@ def health_check() -> dict[str, str]: api_router.include_router(business_router, prefix="/business", tags=["business"]) +api_router.include_router(dashboard_router, prefix="/dashboard", tags=["dashboard"]) api_router.include_router(legacy_mysql_router, prefix="/integrations/mysql", tags=["mysql"]) api_router.include_router(feishu_router, prefix="/integrations/feishu", tags=["feishu"]) api_router.include_router(ai_router, prefix="/ai", tags=["ai"]) diff --git a/app/core/config.py b/app/core/config.py index bf929cd..9ff3cbe 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -29,6 +29,7 @@ class Settings(BaseSettings): approval_api_key: str | None = None approval_api_actor: str = ActorValue.APPROVER cors_origins: list[str] = Field(default_factory=lambda: ["*"]) + mask_sensitive_responses: bool = True database_url: str = "sqlite:///./company_ai.db" legacy_database_url: str | None = None @@ -63,6 +64,9 @@ class Settings(BaseSettings): direct_llm_model: str = "gpt-4.1-mini" scheduler_enabled: bool = False + task_queue_enabled: bool = False + task_queue_always_eager: 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" diff --git a/app/core/masking.py b/app/core/masking.py new file mode 100644 index 0000000..158da88 --- /dev/null +++ b/app/core/masking.py @@ -0,0 +1,54 @@ +from typing import Any + +from app.core.config import get_settings + + +MASKED_VALUE = "[MASKED]" +SENSITIVE_RESPONSE_KEYS = frozenset( + { + "account_number", + "api_key", + "bank_account", + "card_no", + "direct_llm_api_key", + "email", + "feishu_app_secret", + "feishu_verification_token", + "hermes_api_key", + "id_card", + "mobile", + "openclaw_gateway_token", + "password", + "payment_account", + "phone", + "secret", + "tenant_access_token", + "token", + } +) + + +def mask_configured(value: Any) -> 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) + + +def mask_sensitive(value: Any) -> 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: + masked[key_text] = MASKED_VALUE + else: + masked[key_text] = mask_sensitive(item) + return masked + if isinstance(value, list): + return [mask_sensitive(item) for item in value] + if isinstance(value, tuple): + return [mask_sensitive(item) for item in value] + return value diff --git a/app/core/scheduler.py b/app/core/scheduler.py index 4fb2ec0..d241d56 100644 --- a/app/core/scheduler.py +++ b/app/core/scheduler.py @@ -15,6 +15,7 @@ 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.modules.reports.service import ReportService scheduler = BackgroundScheduler(timezone="Asia/Shanghai") @@ -29,6 +30,13 @@ def attach_scheduler(app: FastAPI) -> None: and settings.feishu_app_secret and settings.feishu_default_chat_id ): + if settings.task_queue_enabled: + app.state.last_daily_brief_dispatch = enqueue_daily_brief_push( + receive_id=settings.feishu_default_chat_id, + receive_id_type=FeishuReceiveIdType.CHAT_ID, + actor=ActorValue.SCHEDULER, + ) + return ReportService(db).push_report( report, receive_id=settings.feishu_default_chat_id, @@ -48,6 +56,13 @@ def attach_scheduler(app: FastAPI) -> None: and settings.feishu_app_secret and settings.feishu_default_chat_id ): + if settings.task_queue_enabled: + app.state.last_project_weekly_dispatch = enqueue_project_weekly_push( + receive_id=settings.feishu_default_chat_id, + receive_id_type=FeishuReceiveIdType.CHAT_ID, + actor=ActorValue.SCHEDULER, + ) + return ReportService(db).push_report( report, receive_id=settings.feishu_default_chat_id, diff --git a/app/core/task_queue.py b/app/core/task_queue.py new file mode 100644 index 0000000..76d3f8f --- /dev/null +++ b/app/core/task_queue.py @@ -0,0 +1,107 @@ +from collections.abc import Callable +from typing import Any + +from app.core.config import get_settings +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" + + +def dispatch_task( + task_name: str, + kwargs: dict[str, Any], + inline: Callable[[], Any], +) -> dict[str, Any]: + """Dispatch a task through Celery when enabled, otherwise run inline.""" + + settings = get_settings() + if settings.task_queue_enabled: + from app.tasks import celery_app + + async_result = celery_app.signature(task_name, kwargs=kwargs).apply_async() + return { + "queued": True, + "mode": "celery", + "task_name": task_name, + "task_id": async_result.id, + } + return { + "queued": False, + "mode": "inline", + "task_name": task_name, + "result": inline(), + } + + +def enqueue_daily_brief_push( + receive_id: str | None = None, + receive_id_type: str = FeishuReceiveIdType.CHAT_ID, + actor: str = "scheduler", +) -> dict[str, Any]: + def inline() -> Any: + from app.core.database import SessionLocal + 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) + finally: + db.close() + + return dispatch_task( + TASK_PUSH_DAILY_BRIEF, + { + "receive_id": receive_id, + "receive_id_type": receive_id_type, + "actor": actor, + }, + inline, + ) + + +def enqueue_project_weekly_push( + receive_id: str | None = None, + receive_id_type: str = FeishuReceiveIdType.CHAT_ID, + actor: str = "scheduler", +) -> dict[str, Any]: + def inline() -> Any: + from app.core.database import SessionLocal + 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) + finally: + db.close() + + return dispatch_task( + TASK_PUSH_PROJECT_WEEKLY, + { + "receive_id": receive_id, + "receive_id_type": receive_id_type, + "actor": actor, + }, + inline, + ) + + +def enqueue_risk_event_generation(actor: str = "scheduler") -> dict[str, Any]: + def inline() -> Any: + from app.core.database import SessionLocal + from app.modules.risk.service import RiskService + + db = SessionLocal() + try: + return RiskService(db).generate_events(actor=actor) + finally: + db.close() + + return dispatch_task( + TASK_GENERATE_RISK_EVENTS, + {"actor": actor}, + inline, + ) diff --git a/app/modules/approvals/routes.py b/app/modules/approvals/routes.py index e73f7f2..eec750a 100644 --- a/app/modules/approvals/routes.py +++ b/app/modules/approvals/routes.py @@ -3,8 +3,15 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import ApiPrincipal, require_api_key, require_approval_api_key -from app.modules.approvals.schemas import ApprovalCreate, ApprovalDecision, ApprovalRead +from app.modules.approvals.schemas import ( + ApprovalCreate, + ApprovalDecision, + ApprovalRead, + PushApprovalCardRequest, +) from app.modules.approvals.service import ApprovalService +from app.modules.feishu.constants import FeishuResponseKey +from app.modules.feishu.service import FeishuService router = APIRouter(dependencies=[Depends(require_api_key)]) @@ -57,3 +64,28 @@ def reject( principal: ApiPrincipal = Depends(require_approval_api_key), ): return ApprovalService(db).decide(ticket_id, principal.actor, False, payload.comment) + + +@router.post("/{ticket_id}/push-feishu-card") +def push_feishu_approval_card( + ticket_id: str, + payload: PushApprovalCardRequest, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + ticket = ApprovalService(db).get_by_ticket(ticket_id) + lines = [ + f"- 单号：{ticket.ticket_id}", + f"- 领域：{ticket.domain}", + f"- 动作：{ticket.action}", + f"- 申请人：{ticket.applicant}", + f"- 理由：{ticket.reason or '-'}", + ] + card = FeishuService.build_approval_card("审批请求", lines, ticket.ticket_id) + result = FeishuService(db).send_card( + card, + receive_id=payload.receive_id, + receive_id_type=payload.receive_id_type, + actor=principal.actor, + ) + return {FeishuResponseKey.OK: True, FeishuResponseKey.PROVIDER_RESPONSE: result} diff --git a/app/modules/approvals/schemas.py b/app/modules/approvals/schemas.py index b8b427b..5f78515 100644 --- a/app/modules/approvals/schemas.py +++ b/app/modules/approvals/schemas.py @@ -4,6 +4,7 @@ from typing import Any from pydantic import BaseModel, ConfigDict, Field from app.core.constants import ActorValue +from app.modules.feishu.constants import FeishuReceiveIdType class ApprovalCreate(BaseModel): @@ -19,6 +20,14 @@ class ApprovalDecision(BaseModel): comment: str | None = None +class PushApprovalCardRequest(BaseModel): + receive_id: str | None = Field( + default=None, + description="chat_id or open_id depending on receive_id_type.", + ) + receive_id_type: str = FeishuReceiveIdType.CHAT_ID + + class ApprovalRead(BaseModel): model_config = ConfigDict(from_attributes=True) diff --git a/app/modules/business/routes.py b/app/modules/business/routes.py index 55a7590..074f30d 100644 --- a/app/modules/business/routes.py +++ b/app/modules/business/routes.py @@ -3,6 +3,7 @@ from fastapi import status as http_status 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.modules.business.constants import BusinessField, BusinessResponseKey from app.modules.business.registry import supported_domain_values @@ -32,16 +33,22 @@ def list_records( return { BusinessResponseKey.DOMAIN: domain, BusinessResponseKey.TOTAL: total, - BusinessResponseKey.ITEMS: items, + BusinessResponseKey.ITEMS: mask_configured(items), } @router.get("/{domain}/{record_id}") -def get_record(domain: str, record_id: int, db: Session = Depends(get_db)) -> dict: +def get_record( + domain: str, + record_id: int, + db: Session = Depends(get_db), +) -> dict: try: return { BusinessResponseKey.DOMAIN: domain, - BusinessResponseKey.DATA: BusinessService(db).get_record(domain, record_id), + BusinessResponseKey.DATA: mask_configured( + BusinessService(db).get_record(domain, record_id) + ), } except KeyError as exc: raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -63,7 +70,10 @@ def create_record( ) except KeyError as exc: raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc - return {BusinessResponseKey.DOMAIN: domain, BusinessResponseKey.DATA: data} + return { + BusinessResponseKey.DOMAIN: domain, + BusinessResponseKey.DATA: mask_configured(data), + } @router.patch("/{domain}/{record_id}") @@ -84,4 +94,7 @@ def update_record( ) except KeyError as exc: raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc - return {BusinessResponseKey.DOMAIN: domain, BusinessResponseKey.DATA: data} + return { + BusinessResponseKey.DOMAIN: domain, + BusinessResponseKey.DATA: mask_configured(data), + } diff --git a/app/modules/dashboard/__init__.py b/app/modules/dashboard/__init__.py new file mode 100644 index 0000000..a95baca --- /dev/null +++ b/app/modules/dashboard/__init__.py @@ -0,0 +1 @@ +"""Basic operations dashboard module.""" diff --git a/app/modules/dashboard/routes.py b/app/modules/dashboard/routes.py new file mode 100644 index 0000000..8c15c75 --- /dev/null +++ b/app/modules/dashboard/routes.py @@ -0,0 +1,18 @@ +from fastapi import APIRouter, Depends +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.modules.dashboard.service import DashboardService + +router = APIRouter() + + +@router.get("/summary") +def dashboard_summary( + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + _ = principal + return mask_configured(DashboardService(db).summary()) diff --git a/app/modules/dashboard/service.py b/app/modules/dashboard/service.py new file mode 100644 index 0000000..3d5f6ac --- /dev/null +++ b/app/modules/dashboard/service.py @@ -0,0 +1,61 @@ +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +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.service import serialize_model +from app.modules.risk.service import RiskService + + +class DashboardService: + """Build lightweight operational dashboard data for V2.""" + + def __init__(self, db: Session): + self.db = db + self.risks = RiskService(db) + + def summary(self) -> dict[str, Any]: + active_projects = self._count(Project, Project.status.notin_(PROJECT_CLOSED_STATUSES)) + open_tasks = self._count(WorkTask, WorkTask.status.notin_(DONE_STATUSES)) + pending_approvals = self._count( + ApprovalRequest, + ApprovalRequest.status == ApprovalStatus.PENDING, + ) + open_risk_events = self._count(RiskEvent, RiskEvent.status == StatusValue.OPEN) + latest_reports = self.db.execute( + select(WorkReport).order_by(WorkReport.id.desc()).limit(5) + ).scalars() + latest_audit_logs = self.db.execute( + select(AuditLog).order_by(AuditLog.id.desc()).limit(10) + ).scalars() + risk_summary = self.risks.summary() + return { + "metrics": { + "active_projects": active_projects, + "open_tasks": open_tasks, + "pending_approvals": pending_approvals, + "open_risk_events": open_risk_events, + "risk_level": risk_summary["risk_level"], + "risk_score": float(risk_summary["risk_score"]), + }, + "risk_counts": { + "overdue_tasks": len(risk_summary["overdue_tasks"]), + "delayed_projects": len(risk_summary["delayed_projects"]), + "over_budget_projects": len(risk_summary["over_budget_projects"]), + "fund_risks": len(risk_summary["fund_risks"]), + "supplier_risks": len(risk_summary["supplier_risks"]), + }, + "latest_reports": [serialize_model(item) for item in latest_reports], + "latest_audit_logs": [serialize_model(item) for item in latest_audit_logs], + } + + def _count(self, model: type, *conditions: Any) -> int: + stmt = select(func.count()).select_from(model) + for condition in conditions: + stmt = stmt.where(condition) + return int(self.db.execute(stmt).scalar() or 0) diff --git a/app/modules/feishu/events.py b/app/modules/feishu/events.py index 8934a87..7a6b772 100644 --- a/app/modules/feishu/events.py +++ b/app/modules/feishu/events.py @@ -1,9 +1,12 @@ +import json from typing import Any +from fastapi import HTTPException, status from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.constants import ActorValue +from app.modules.approvals.service import ApprovalService from app.modules.audit.constants import AuditAction, AuditSource from app.modules.audit.schemas import AuditLogCreate from app.modules.feishu.commands import FeishuCommandService @@ -79,6 +82,36 @@ class FeishuEventService: FeishuResponseKey.RESULT: result, } + def handle_approval_card_action(self, payload: dict[str, Any]) -> dict[str, Any]: + """Handle Feishu interactive-card approval button callbacks.""" + + self.feishu.verify_event(payload) + value = _approval_action_value(payload) + ticket_id = str(value.get("ticket_id") or "").strip() + decision = str(value.get("decision") or value.get("action") or "").lower() + if not ticket_id or decision not in {"approve", "reject"}: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Invalid Feishu approval action payload", + ) + comment = value.get("comment") + actor = _approval_operator(payload) + ticket = ApprovalService(self.db).decide( + ticket_id, + actor, + approved=decision == "approve", + comment=str(comment) if comment is not None else None, + ) + return { + FeishuResponseKey.OK: True, + FeishuResponseKey.HANDLED: True, + FeishuResponseKey.RESULT: { + "ticket_id": ticket.ticket_id, + "status": ticket.status, + "approver": ticket.approver, + }, + } + def _register_event(self, event_identity: dict[str, str | None]) -> bool: receipt = FeishuEventReceipt( event_key=str(event_identity[FeishuEventReceiptKey.EVENT_KEY]), @@ -123,3 +156,31 @@ def _event_identity( FeishuEventReceiptKey.EVENT_ID: str(event_id) if event_id else None, FeishuEventReceiptKey.MESSAGE_ID: str(message_id) if message_id else None, } + + +def _approval_action_value(payload: dict[str, Any]) -> dict[str, Any]: + action = payload.get("action") or {} + event = payload.get(FeishuPayloadKey.EVENT) or {} + event_action = event.get("action") or {} + value = action.get("value") or event_action.get("value") or payload.get("value") or {} + if isinstance(value, str): + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return {} + return parsed if isinstance(parsed, dict) else {} + return value if isinstance(value, dict) else {} + + +def _approval_operator(payload: dict[str, Any]) -> str: + operator = payload.get("operator") or (payload.get(FeishuPayloadKey.EVENT) or {}).get( + "operator" + ) or {} + operator_id = operator.get("operator_id") or {} + return ( + operator_id.get(FeishuPayloadKey.OPEN_ID) + or operator_id.get(FeishuPayloadKey.USER_ID) + or operator.get(FeishuPayloadKey.OPEN_ID) + or operator.get(FeishuPayloadKey.USER_ID) + or ActorValue.FEISHU + ) diff --git a/app/modules/feishu/routes.py b/app/modules/feishu/routes.py index 492e5de..19761ad 100644 --- a/app/modules/feishu/routes.py +++ b/app/modules/feishu/routes.py @@ -30,7 +30,18 @@ async def feishu_webhook(request: Request, db: Session = Depends(get_db)) -> dic ) -@router.post("/send-text", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) +@router.post("/approval-card-action") +async def feishu_approval_card_action( + request: Request, + db: Session = Depends(get_db), +) -> dict: + """Handle Feishu interactive-card approval actions.""" + + payload = await request.json() + return FeishuEventService(db).handle_approval_card_action(payload) + + +@router.post("/send-text", response_model=FeishuSendResult) def send_text( payload: FeishuTextMessage, db: Session = Depends(get_db), @@ -48,7 +59,7 @@ def send_text( } -@router.post("/send-card", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) +@router.post("/send-card", response_model=FeishuSendResult) def send_card( payload: FeishuCardMessage, db: Session = Depends(get_db), @@ -69,7 +80,6 @@ def send_card( @router.post( "/commands/preview", response_model=FeishuCommandResult, - dependencies=[Depends(require_api_key)], ) def preview_command( payload: FeishuCommandRequest, diff --git a/app/modules/feishu/service.py b/app/modules/feishu/service.py index b93b41c..3ccbdb7 100644 --- a/app/modules/feishu/service.py +++ b/app/modules/feishu/service.py @@ -109,3 +109,37 @@ class FeishuService: } ], } + + @staticmethod + def build_approval_card( + title: str, + lines: list[str], + ticket_id: str, + ) -> dict[str, Any]: + card = FeishuService.build_basic_card(title, lines) + card[FeishuPayloadKey.ELEMENTS].append( + { + FeishuPayloadKey.TAG: "action", + "actions": [ + { + FeishuPayloadKey.TAG: "button", + FeishuPayloadKey.TEXT: { + FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT, + FeishuPayloadKey.CONTENT: "批准", + }, + "type": "primary", + "value": {"ticket_id": ticket_id, "decision": "approve"}, + }, + { + FeishuPayloadKey.TAG: "button", + FeishuPayloadKey.TEXT: { + FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT, + FeishuPayloadKey.CONTENT: "拒绝", + }, + "type": "danger", + "value": {"ticket_id": ticket_id, "decision": "reject"}, + }, + ], + } + ) + return card diff --git a/app/modules/legacy_mysql/routes.py b/app/modules/legacy_mysql/routes.py index 7081f7a..65973a3 100644 --- a/app/modules/legacy_mysql/routes.py +++ b/app/modules/legacy_mysql/routes.py @@ -2,6 +2,7 @@ from fastapi import APIRouter, Depends, Query 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.modules.legacy_mysql.schemas import ( LegacyProjectSyncRequest, @@ -15,16 +16,23 @@ router = APIRouter(dependencies=[Depends(require_api_key)]) @router.get("/health") -def mysql_health(db: Session = Depends(get_db)) -> dict[str, str]: +def mysql_health( + db: Session = Depends(get_db), +) -> dict[str, str]: return LegacyMySQLService(db).health() @router.post("/query", response_model=QueryResult) -def readonly_query(payload: ReadonlyQueryRequest, db: Session = Depends(get_db)) -> dict: +def readonly_query( + payload: ReadonlyQueryRequest, + db: Session = Depends(get_db), +) -> dict: service = LegacyMySQLService(db) if payload.sql: - return service.execute_readonly(payload.sql, payload.params, payload.limit) - return service.execute_allowed_query(payload.query_name, payload.params, payload.limit) + result = service.execute_readonly(payload.sql, payload.params, payload.limit) + else: + result = service.execute_allowed_query(payload.query_name, payload.params, payload.limit) + return mask_configured(result) @router.get("/projects", response_model=QueryResult) @@ -32,7 +40,7 @@ def default_project_query( limit: int = Query(default=100, ge=1, le=500), db: Session = Depends(get_db), ) -> dict: - return LegacyMySQLService(db).fetch_default_projects(limit=limit) + return mask_configured(LegacyMySQLService(db).fetch_default_projects(limit=limit)) @router.post("/projects/sync", response_model=LegacyProjectSyncResult) @@ -41,7 +49,7 @@ def sync_projects( db: Session = Depends(get_db), principal: ApiPrincipal = Depends(require_api_key), ) -> dict: - return LegacyMySQLService(db).sync_projects( + result = LegacyMySQLService(db).sync_projects( source_query=payload.source_query, source_query_name=payload.source_query_name, field_map=payload.field_map, @@ -49,3 +57,4 @@ def sync_projects( dry_run=payload.dry_run, actor=principal.actor, ) + return mask_configured(result) diff --git a/app/modules/reports/routes.py b/app/modules/reports/routes.py index c84ae1b..6b6456e 100644 --- a/app/modules/reports/routes.py +++ b/app/modules/reports/routes.py @@ -5,6 +5,7 @@ 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_daily_brief_push, enqueue_project_weekly_push from app.modules.reports.schemas import ( PushReportRequest, ReportResponse, @@ -16,12 +17,16 @@ router = APIRouter(dependencies=[Depends(require_api_key)]) @router.get("/daily-brief", response_model=ReportResponse) -def daily_brief(db: Session = Depends(get_db)) -> dict: +def daily_brief( + db: Session = Depends(get_db), +) -> dict: return ReportService(db).daily_brief() @router.get("/project-weekly", response_model=ReportResponse) -def project_weekly(db: Session = Depends(get_db)) -> dict: +def project_weekly( + db: Session = Depends(get_db), +) -> dict: return ReportService(db).project_weekly() @@ -99,3 +104,27 @@ def push_project_weekly( payload.receive_id_type, principal.actor, ) + + +@router.post("/daily-brief/enqueue") +def enqueue_daily_brief( + payload: PushReportRequest, + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return enqueue_daily_brief_push( + receive_id=payload.receive_id, + receive_id_type=payload.receive_id_type, + actor=principal.actor, + ) + + +@router.post("/project-weekly/enqueue") +def enqueue_project_weekly( + payload: PushReportRequest, + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return enqueue_project_weekly_push( + receive_id=payload.receive_id, + receive_id_type=payload.receive_id_type, + actor=principal.actor, + ) diff --git a/app/modules/risk/routes.py b/app/modules/risk/routes.py index 7134c05..36484fe 100644 --- a/app/modules/risk/routes.py +++ b/app/modules/risk/routes.py @@ -3,6 +3,7 @@ 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.service import RiskService @@ -10,32 +11,44 @@ router = APIRouter(dependencies=[Depends(require_api_key)]) @router.get("/summary") -def risk_summary(db: Session = Depends(get_db)) -> dict: +def risk_summary( + db: Session = Depends(get_db), +) -> dict: return RiskService(db).summary() @router.get("/overdue-tasks") -def overdue_tasks(db: Session = Depends(get_db)) -> dict: +def overdue_tasks( + db: Session = Depends(get_db), +) -> dict: return {RiskGenerationResultKey.ITEMS: RiskService(db).overdue_tasks()} @router.get("/delayed-projects") -def delayed_projects(db: Session = Depends(get_db)) -> dict: +def delayed_projects( + db: Session = Depends(get_db), +) -> dict: return {RiskGenerationResultKey.ITEMS: RiskService(db).delayed_projects()} @router.get("/over-budget-projects") -def over_budget_projects(db: Session = Depends(get_db)) -> dict: +def over_budget_projects( + db: Session = Depends(get_db), +) -> dict: return {RiskGenerationResultKey.ITEMS: RiskService(db).over_budget_projects()} @router.get("/funds") -def fund_risks(db: Session = Depends(get_db)) -> dict: +def fund_risks( + db: Session = Depends(get_db), +) -> dict: return {RiskGenerationResultKey.ITEMS: RiskService(db).fund_risks()} @router.get("/suppliers") -def supplier_risks(db: Session = Depends(get_db)) -> dict: +def supplier_risks( + db: Session = Depends(get_db), +) -> dict: return {RiskGenerationResultKey.ITEMS: RiskService(db).supplier_risks()} @@ -59,3 +72,10 @@ def generate_risk_events( principal: ApiPrincipal = Depends(require_api_key), ) -> dict: return RiskService(db).generate_events(actor=principal.actor) + + +@router.post("/events/enqueue") +def enqueue_risk_events( + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return enqueue_risk_event_generation(actor=principal.actor) diff --git a/app/tasks.py b/app/tasks.py new file mode 100644 index 0000000..a1ee7d8 --- /dev/null +++ b/app/tasks.py @@ -0,0 +1,60 @@ +from typing import Any + +from celery import Celery + +from app.core.config import get_settings +from app.core.constants import ActorValue +from app.core.database import SessionLocal +from app.modules.feishu.constants import FeishuReceiveIdType + +settings = get_settings() + +celery_app = Celery( + "company_ai_platform", + broker=settings.redis_url, + backend=settings.celery_result_backend_url or settings.redis_url, +) +celery_app.conf.task_always_eager = settings.task_queue_always_eager + + +@celery_app.task(name="reports.push_daily_brief") +def push_daily_brief( + receive_id: str | None = None, + receive_id_type: str = FeishuReceiveIdType.CHAT_ID, + actor: str = ActorValue.SCHEDULER, +) -> 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) + finally: + db.close() + + +@celery_app.task(name="reports.push_project_weekly") +def push_project_weekly( + receive_id: str | None = None, + receive_id_type: str = FeishuReceiveIdType.CHAT_ID, + actor: str = ActorValue.SCHEDULER, +) -> 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) + finally: + db.close() + + +@celery_app.task(name="risks.generate_events") +def generate_risk_events(actor: str = ActorValue.SCHEDULER) -> dict[str, Any]: + from app.modules.risk.service import RiskService + + db = SessionLocal() + try: + return RiskService(db).generate_events(actor=actor) + finally: + db.close() diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 5f77732..d48aa25 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -199,6 +199,34 @@ def test_config_and_pagination_guardrails() -> None: assert negative_offset_response.status_code == 422 +def test_dashboard_and_response_masking() -> None: + expense_response = client.post( + "/api/v1/business/expenses", + headers=headers, + json={ + "data": { + "code": "EXP-MASK-001", + "expense_type": "办公", + "amount": 20, + "payment_account": "6222000000000000", + }, + }, + ) + assert expense_response.status_code == 200 + + masked_response = client.get("/api/v1/business/expenses", headers=headers) + assert masked_response.status_code == 200 + masked_items = masked_response.json()["items"] + assert any(item["code"] == "EXP-MASK-001" for item in masked_items) + assert next( + item["payment_account"] for item in masked_items if item["code"] == "EXP-MASK-001" + ) == "[MASKED]" + + dashboard_response = client.get("/api/v1/dashboard/summary", headers=headers) + assert dashboard_response.status_code == 200 + assert "metrics" in dashboard_response.json() + + def test_approval_gate_for_high_risk_update() -> None: create_payload = { "code": "FUND-SMOKE-001", @@ -359,6 +387,40 @@ def test_approval_gate_for_high_risk_update() -> None: assert reuse_update_response.status_code == 403 +def test_feishu_approval_card_action_approves_ticket() -> None: + approval_response = client.post( + "/api/v1/approvals", + headers=headers, + json={ + "domain": "fund-accounts", + "record_id": "feishu-card-test", + "action": "update:fund-accounts", + "reason": "Card action smoke test", + "payload": {"current_balance": 300}, + }, + ) + assert approval_response.status_code == 200 + ticket_id = approval_response.json()["ticket_id"] + + callback_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 callback_response.status_code == 200 + assert callback_response.json()["result"]["status"] == "approved" + assert callback_response.json()["result"]["approver"] == "ou_card_approver" + + def test_new_ledgers_reports_and_risk_events() -> None: domains_response = client.get("/api/v1/business/domains", headers=headers) assert domains_response.status_code == 200 @@ -419,6 +481,11 @@ def test_new_ledgers_reports_and_risk_events() -> None: assert risk_response.status_code == 200 assert risk_response.json()["created"] >= 1 + enqueue_response = client.post("/api/v1/risks/events/enqueue", headers=headers) + assert enqueue_response.status_code == 200 + assert enqueue_response.json()["queued"] is False + assert "result" in enqueue_response.json() + events_response = client.get("/api/v1/risks/events?status=open", headers=headers) assert events_response.status_code == 200 assert any(item["risk_type"] == "overdue_task" for item in events_response.json()["items"])