diff --git a/app/api/router.py b/app/api/router.py index 8920b5a..3c7b0d7 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -1,5 +1,6 @@ from fastapi import APIRouter +from app.core.constants import ApiResponseKey, ApiStatus 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 @@ -16,7 +17,7 @@ api_router = APIRouter() def health_check() -> dict[str, str]: """Return basic API health status.""" - return {"status": "ok"} + return {ApiResponseKey.STATUS: ApiStatus.OK} api_router.include_router(business_router, prefix="/business", tags=["business"]) diff --git a/app/core/config.py b/app/core/config.py index 0a89432..bf929cd 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -5,9 +5,12 @@ from typing import Any from pydantic import Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict -from app.core.constants import ActorValue - -DEFAULT_MODEL_PROVIDER = "noop" +from app.core.constants import ( + ActorValue, + ConfigErrorDetail, + DEFAULT_MODEL_PROVIDER, + DEFAULT_OPENCLAW_ACTION_JSON, +) class Settings(BaseSettings): @@ -48,7 +51,9 @@ class Settings(BaseSettings): openclaw_api_key: str | None = None openclaw_gateway_token: str | None = None openclaw_allowed_tools: list[str] = Field(default_factory=list) - openclaw_allowed_actions: list[str] = Field(default_factory=lambda: ["json"]) + openclaw_allowed_actions: list[str] = Field( + default_factory=lambda: [DEFAULT_OPENCLAW_ACTION_JSON] + ) hermes_base_url: str = "http://127.0.0.1:2073/v1" hermes_api_key: str | None = None hermes_model: str = "hermes-agent" @@ -75,7 +80,7 @@ class Settings(BaseSettings): if text.startswith("["): data = json.loads(text) if not isinstance(data, list): - raise ValueError("CORS_ORIGINS must be a CSV string or JSON 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()] @@ -96,9 +101,9 @@ class Settings(BaseSettings): if isinstance(value, str): data = json.loads(value) if not isinstance(data, dict): - raise ValueError("LEGACY_ALLOWED_QUERIES must be a JSON object") + raise ValueError(ConfigErrorDetail.LEGACY_ALLOWED_QUERIES_FORMAT) return {str(key): str(item) for key, item in data.items()} - raise ValueError("LEGACY_ALLOWED_QUERIES must be a JSON object") + raise ValueError(ConfigErrorDetail.LEGACY_ALLOWED_QUERIES_FORMAT) @lru_cache diff --git a/app/core/constants.py b/app/core/constants.py index b56aea4..8d89f81 100644 --- a/app/core/constants.py +++ b/app/core/constants.py @@ -17,4 +17,28 @@ class HttpHeader(StrEnum): X_APPROVAL_API_KEY = "X-Approval-API-Key" +class ApiResponseKey(StrEnum): + STATUS = "status" + + +class ApiStatus(StrEnum): + OK = "ok" + + +class SecurityErrorDetail(StrEnum): + API_KEY_REQUIRED = "API_KEY is required" + INVALID_API_KEY = "Invalid API key" + APPROVAL_API_KEY_REQUIRED = "APPROVAL_API_KEY is required" + INVALID_APPROVAL_API_KEY = "Invalid approval API key" + AUDIT_API_KEY_REQUIRED = "AUDIT_API_KEY is required" + INVALID_AUDIT_API_KEY = "Invalid audit API key" + + +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" + + BEARER_TOKEN_TEMPLATE = "Bearer {token}" +DEFAULT_MODEL_PROVIDER = "noop" +DEFAULT_OPENCLAW_ACTION_JSON = "json" diff --git a/app/core/security.py b/app/core/security.py index b41f98e..2991f1d 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -4,7 +4,7 @@ from secrets import compare_digest from fastapi import Header, HTTPException, status from app.core.config import get_settings -from app.core.constants import HttpHeader +from app.core.constants import HttpHeader, SecurityErrorDetail @dataclass(frozen=True) @@ -23,10 +23,13 @@ def require_api_key( if not settings.api_key: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="API_KEY is required", + detail=SecurityErrorDetail.API_KEY_REQUIRED, ) if not x_api_key or not compare_digest(x_api_key, settings.api_key): - raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key") + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=SecurityErrorDetail.INVALID_API_KEY, + ) return ApiPrincipal(actor=settings.api_actor) @@ -42,7 +45,7 @@ def require_approval_api_key( if not settings.approval_api_key: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="APPROVAL_API_KEY is required", + detail=SecurityErrorDetail.APPROVAL_API_KEY_REQUIRED, ) if ( not x_approval_api_key @@ -50,7 +53,7 @@ def require_approval_api_key( ): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid approval API key", + detail=SecurityErrorDetail.INVALID_APPROVAL_API_KEY, ) return ApiPrincipal(actor=settings.approval_api_actor) @@ -67,11 +70,11 @@ def require_audit_api_key( if not settings.audit_api_key: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="AUDIT_API_KEY is required", + detail=SecurityErrorDetail.AUDIT_API_KEY_REQUIRED, ) if not x_audit_api_key or not compare_digest(x_audit_api_key, settings.audit_api_key): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid audit API key", + detail=SecurityErrorDetail.INVALID_AUDIT_API_KEY, ) return ApiPrincipal(actor=settings.audit_api_actor) diff --git a/app/modules/ai_agent/adapters.py b/app/modules/ai_agent/adapters.py index d886404..cbe1a1d 100644 --- a/app/modules/ai_agent/adapters.py +++ b/app/modules/ai_agent/adapters.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod from typing import Any import httpx -from fastapi import HTTPException +from fastapi import HTTPException, status from app.core.config import Settings, get_settings from app.modules.ai_agent.constants import ( @@ -11,7 +11,10 @@ from app.modules.ai_agent.constants import ( COMPANY_MANAGEMENT_SYSTEM_INSTRUCTIONS, DIRECT_LLM_API_KEY_MISSING, NOOP_PROVIDER_ANSWER, + OPENCLAW_ACTION_NOT_ALLOWED, + OPENCLAW_CHAT_PROVIDER_REQUIRED, OPENCLAW_HERMES_PIPELINE, + OPENCLAW_TOOL_NOT_ALLOWED, OPENCLAW_TOOL_COMPLETED_ANSWER, UNEXPECTED_HERMES_RESPONSE, AIDefault, @@ -67,12 +70,8 @@ class OpenClawAdapter(AIAdapter): tool = context.get(AIContextKey.OPENCLAW_TOOL) if not tool: raise HTTPException( - status_code=400, - detail=( - "OpenClaw Gateway is not configured as a chat provider. " - "Provide context.openclaw_tool for /tools/invoke, or use " - "MODEL_PROVIDER=hermes/openclaw_hermes for AI answers." - ), + status_code=status.HTTP_400_BAD_REQUEST, + detail=OPENCLAW_CHAT_PROVIDER_REQUIRED, ) result = self.invoke_tool( tool=str(tool), @@ -96,7 +95,10 @@ class OpenClawAdapter(AIAdapter): healthz = client.get(f"{base_url}{AIHttpPath.HEALTHZ}", headers=headers) readyz = client.get(f"{base_url}{AIHttpPath.READYZ}", headers=headers) return { - AIResponseKey.OK: healthz.status_code < 400 and readyz.status_code < 400, + AIResponseKey.OK: ( + healthz.status_code < status.HTTP_400_BAD_REQUEST + and readyz.status_code < status.HTTP_400_BAD_REQUEST + ), AIResponseKey.BASE_URL: base_url, AIResponseKey.HEALTHZ: _response_payload(healthz), AIResponseKey.READYZ: _response_payload(readyz), @@ -121,8 +123,11 @@ class OpenClawAdapter(AIAdapter): url = f"{self._base_url()}{AIHttpPath.TOOLS_INVOKE}" with httpx.Client(timeout=120, trust_env=False) as client: response = client.post(url, json=payload, headers=self._headers()) - if response.status_code >= 400: - raise HTTPException(status_code=502, detail={AIErrorKey.OPENCLAW: response.text}) + if response.status_code >= status.HTTP_400_BAD_REQUEST: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail={AIErrorKey.OPENCLAW: response.text}, + ) return _response_payload(response) def _base_url(self) -> str: @@ -138,9 +143,15 @@ class OpenClawAdapter(AIAdapter): def _ensure_tool_allowed(self, tool: str, action: str) -> None: if tool not in set(self.settings.openclaw_allowed_tools): - raise HTTPException(status_code=403, detail="OpenClaw tool is not allowed") + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=OPENCLAW_TOOL_NOT_ALLOWED, + ) if action not in set(self.settings.openclaw_allowed_actions): - raise HTTPException(status_code=403, detail="OpenClaw action is not allowed") + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=OPENCLAW_ACTION_NOT_ALLOWED, + ) class HermesAdapter(AIAdapter): @@ -167,8 +178,11 @@ class HermesAdapter(AIAdapter): } with httpx.Client(timeout=300, trust_env=False) as client: response = client.post(url, json=payload, headers=headers) - if response.status_code >= 400: - raise HTTPException(status_code=502, detail={AIErrorKey.HERMES: response.text}) + if response.status_code >= status.HTTP_400_BAD_REQUEST: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail={AIErrorKey.HERMES: response.text}, + ) data = _chat_completion_payload(response, AIErrorKey.HERMES) try: answer = data[AIHttpPayloadKey.CHOICES][0][AIHttpPayloadKey.MESSAGE][ @@ -176,7 +190,7 @@ class HermesAdapter(AIAdapter): ] except (KeyError, IndexError, TypeError) as exc: raise HTTPException( - status_code=502, + status_code=status.HTTP_502_BAD_GATEWAY, detail={ AIErrorKey.HERMES: UNEXPECTED_HERMES_RESPONSE, AIResponseKey.RAW: data, @@ -196,7 +210,7 @@ class HermesAdapter(AIAdapter): with httpx.Client(timeout=5, trust_env=False) as client: response = client.get(url, headers=headers) return { - AIResponseKey.OK: response.status_code < 400, + AIResponseKey.OK: response.status_code < status.HTTP_400_BAD_REQUEST, AIResponseKey.BASE_URL: self.settings.hermes_base_url.rstrip("/"), AIResponseKey.HEALTH: _response_payload(response), } @@ -267,7 +281,7 @@ class OpenClawHermesAdapter(AIAdapter): raise except Exception as exc: raise HTTPException( - status_code=502, + status_code=status.HTTP_502_BAD_GATEWAY, detail={AIErrorKey.OPENCLAW: _error_detail(exc)}, ) from exc return result @@ -334,7 +348,10 @@ class DirectLLMAdapter(AIAdapter): def ask(self, prompt: str, context: dict[str, Any] | None = None) -> dict[str, Any]: if not self.settings.direct_llm_api_key: - raise HTTPException(status_code=503, detail=DIRECT_LLM_API_KEY_MISSING) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=DIRECT_LLM_API_KEY_MISSING, + ) url = f"{self.settings.direct_llm_base_url.rstrip('/')}{AIHttpPath.CHAT_COMPLETIONS}" headers = { AIHttpHeader.AUTHORIZATION: AUTHORIZATION_BEARER_TEMPLATE.format( @@ -347,8 +364,11 @@ class DirectLLMAdapter(AIAdapter): } with httpx.Client(timeout=60, trust_env=False) as client: response = client.post(url, json=payload, headers=headers) - if response.status_code >= 400: - raise HTTPException(status_code=502, detail={AIErrorKey.DIRECT_LLM: response.text}) + if response.status_code >= status.HTTP_400_BAD_REQUEST: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail={AIErrorKey.DIRECT_LLM: response.text}, + ) data = _chat_completion_payload(response, AIErrorKey.DIRECT_LLM) try: answer = data[AIHttpPayloadKey.CHOICES][0][AIHttpPayloadKey.MESSAGE][ @@ -356,7 +376,7 @@ class DirectLLMAdapter(AIAdapter): ] except (KeyError, IndexError, TypeError) as exc: raise HTTPException( - status_code=502, + status_code=status.HTTP_502_BAD_GATEWAY, detail={ AIErrorKey.DIRECT_LLM: UNEXPECTED_HERMES_RESPONSE, AIResponseKey.RAW: data, @@ -404,7 +424,7 @@ def _chat_completion_payload(response: httpx.Response, error_key: AIErrorKey) -> return response.json() except ValueError as exc: raise HTTPException( - status_code=502, + status_code=status.HTTP_502_BAD_GATEWAY, detail={ error_key: UNEXPECTED_HERMES_RESPONSE, AIResponseKey.RAW: {AIResponseKey.TEXT: response.text}, diff --git a/app/modules/ai_agent/constants.py b/app/modules/ai_agent/constants.py index 060813c..bf54fe3 100644 --- a/app/modules/ai_agent/constants.py +++ b/app/modules/ai_agent/constants.py @@ -89,6 +89,13 @@ class AIHttpPayloadKey(StrEnum): MESSAGE = "message" +class AIToolAuditKey(StrEnum): + TOOL = "tool" + ACTION = "action" + ARGS = "args" + SESSION_KEY = "session_key" + + class AIChatRole(StrEnum): SYSTEM = "system" USER = "user" @@ -126,6 +133,14 @@ NOOP_PROVIDER_ANSWER = ( OPENCLAW_TOOL_COMPLETED_ANSWER = "OpenClaw tool invocation completed." DIRECT_LLM_API_KEY_MISSING = "DIRECT_LLM_API_KEY is not configured" UNEXPECTED_HERMES_RESPONSE = "Unexpected chat completion response" +OPENCLAW_CHAT_PROVIDER_REQUIRED = ( + "OpenClaw Gateway is not configured as a chat provider. " + "Provide context.openclaw_tool for /tools/invoke, or use " + "MODEL_PROVIDER=hermes/openclaw_hermes for AI answers." +) +OPENCLAW_TOOL_NOT_ALLOWED = "OpenClaw tool is not allowed" +OPENCLAW_ACTION_NOT_ALLOWED = "OpenClaw action is not allowed" +UNSUPPORTED_AI_SKILL_TEMPLATE = "Unsupported AI skill: {skill_id}" AI_AUDIT_REDACTED_VALUE = "[REDACTED]" AI_AUDIT_TRUNCATED_VALUE = "[TRUNCATED]" diff --git a/app/modules/ai_agent/service.py b/app/modules/ai_agent/service.py index 4c89194..54aa2f6 100644 --- a/app/modules/ai_agent/service.py +++ b/app/modules/ai_agent/service.py @@ -13,6 +13,7 @@ from app.modules.ai_agent.constants import ( AI_AUDIT_REDACTED_VALUE, AI_AUDIT_SENSITIVE_KEYS, AI_AUDIT_TRUNCATED_VALUE, + AIToolAuditKey, AIProviderName, AIRequestKey, AIResponseKey, @@ -120,10 +121,10 @@ class AIService: target_id=tool, risk_level=AuditRiskLevel.HIGH, request_payload=_audit_safe_payload({ - "tool": tool, - "action": action, - "args": args or {}, - "session_key": session_key, + AIToolAuditKey.TOOL: tool, + AIToolAuditKey.ACTION: action, + AIToolAuditKey.ARGS: args or {}, + AIToolAuditKey.SESSION_KEY: session_key, }), response_payload=_audit_safe_payload(result), ) diff --git a/app/modules/ai_agent/skills.py b/app/modules/ai_agent/skills.py index 1833cb5..f713b5a 100644 --- a/app/modules/ai_agent/skills.py +++ b/app/modules/ai_agent/skills.py @@ -4,6 +4,8 @@ from typing import Any from fastapi import HTTPException, status +from app.modules.ai_agent.constants import UNSUPPORTED_AI_SKILL_TEMPLATE + class AISkillId(StrEnum): """Stable identifiers for AI capabilities exposed to business modules.""" @@ -96,6 +98,6 @@ def get_ai_skill(skill_id: AISkillId | str) -> AISkill: except ValueError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, - detail=f"Unsupported AI skill: {skill_id}", + detail=UNSUPPORTED_AI_SKILL_TEMPLATE.format(skill_id=skill_id), ) from exc return AI_SKILLS[normalized_id] diff --git a/app/modules/approvals/constants.py b/app/modules/approvals/constants.py index 61c46cc..68f3eaa 100644 --- a/app/modules/approvals/constants.py +++ b/app/modules/approvals/constants.py @@ -19,3 +19,19 @@ class ApprovalErrorDetail(StrEnum): SELF_APPROVAL = "Approval applicant cannot approve their own ticket" NOT_APPROVED = "Approval ticket is not approved for this change" PAYLOAD_MISMATCH = "Approval ticket payload does not match this change" + + +class ApprovalPayloadKey(StrEnum): + TICKET_ID = "ticket_id" + STATUS = "status" + COMMENT = "comment" + RECORD_ID = "record_id" + USED_BY = "used_by" + USED_AT = "used_at" + + +APPROVAL_ACTION_SEPARATOR = ":" + + +def approval_action(action: ApprovalActionValue | str, domain: str) -> str: + return f"{action}{APPROVAL_ACTION_SEPARATOR}{domain}" diff --git a/app/modules/approvals/service.py b/app/modules/approvals/service.py index a05e490..2b3d352 100644 --- a/app/modules/approvals/service.py +++ b/app/modules/approvals/service.py @@ -13,7 +13,9 @@ from app.core.time import utc_now from app.modules.approvals.constants import ( ApprovalActionValue, ApprovalErrorDetail, + ApprovalPayloadKey, ApprovalStatus, + approval_action, ) from app.modules.approvals.models import ApprovalRequest from app.modules.approvals.schemas import ApprovalCreate @@ -51,7 +53,10 @@ class ApprovalService: target_id=payload.record_id, risk_level=AuditRiskLevel.MEDIUM, request_payload=payload.model_dump(), - response_payload={"ticket_id": ticket.ticket_id, "status": ticket.status}, + response_payload={ + ApprovalPayloadKey.TICKET_ID: ticket.ticket_id, + ApprovalPayloadKey.STATUS: ticket.status, + }, ) ) return ticket @@ -109,8 +114,11 @@ class ApprovalService: target_type=ticket.domain, target_id=ticket.record_id, risk_level=AuditRiskLevel.HIGH, - request_payload={"ticket_id": ticket_id, "comment": comment}, - response_payload={"status": ticket.status}, + request_payload={ + ApprovalPayloadKey.TICKET_ID: ticket_id, + ApprovalPayloadKey.COMMENT: comment, + }, + response_payload={ApprovalPayloadKey.STATUS: ticket.status}, ) ) return ticket @@ -137,12 +145,12 @@ class ApprovalService: ) used_at = utc_now() values: dict[str, Any] = { - "status": ApprovalStatus.USED, - "used_by": actor, - "used_at": used_at, + ApprovalPayloadKey.STATUS: ApprovalStatus.USED, + ApprovalPayloadKey.USED_BY: actor, + ApprovalPayloadKey.USED_AT: used_at, } if record_id is not None and not ticket.record_id: - values["record_id"] = str(record_id) + values[ApprovalPayloadKey.RECORD_ID] = str(record_id) result = self.db.execute( update(ApprovalRequest) .where( @@ -188,7 +196,7 @@ class ApprovalService: return ticket.action in { action, ApprovalActionValue.UPDATE, - f"{ApprovalActionValue.UPDATE}:{domain}", + approval_action(ApprovalActionValue.UPDATE, domain), } diff --git a/app/modules/audit/constants.py b/app/modules/audit/constants.py index da5840f..5c4ad96 100644 --- a/app/modules/audit/constants.py +++ b/app/modules/audit/constants.py @@ -6,6 +6,8 @@ class AuditAction(StrEnum): AI_PROVIDER_HEALTH = "ai.provider_health" OPENCLAW_TOOLS_INVOKE = "openclaw.tools.invoke" GENERATE_EVENTS = "generate_events" + FEISHU_WEBHOOK_EVENT = "webhook_event" + FEISHU_LONG_CONNECTION_EVENT = "long_connection_event" FEISHU_SEND_TEXT = "send_text" FEISHU_SEND_CARD = "send_card" APPROVAL_CREATE = "approval.create" diff --git a/app/modules/business/constants.py b/app/modules/business/constants.py index 5fda7b5..a58d408 100644 --- a/app/modules/business/constants.py +++ b/app/modules/business/constants.py @@ -35,6 +35,7 @@ class StatusValue(StrEnum): ABNORMAL = "异常" OPEN = "open" RUNNING = "running" + RUNNING_CN = "执行中" DRY_RUN = "dry_run" @@ -62,10 +63,46 @@ class VersionValue(StrEnum): class BusinessDomain(StrEnum): - TASKS = "tasks" PROJECTS = "projects" + TASKS = "tasks" + PROCUREMENTS = "procurements" + EXPENSES = "expenses" FUND_ACCOUNTS = "fund-accounts" + POLICIES = "policies" + STANDARDS = "standards" + PERFORMANCE_METRICS = "performance-metrics" SUPPLIERS = "suppliers" + ATTENDANCE_RECORDS = "attendance-records" + WORK_REPORTS = "work-reports" + RISK_EVENTS = "risk-events" + LEGACY_SYNC_RUNS = "legacy-sync-runs" + + +class BusinessResponseKey(StrEnum): + DOMAINS = "domains" + DOMAIN = "domain" + TOTAL = "total" + ITEMS = "items" + DATA = "data" + + +class BusinessPayloadKey(StrEnum): + DATA = "data" + APPROVAL_TICKET_ID = "approval_ticket_id" + + +class BusinessField(StrEnum): + ID = "id" + STATUS = "status" + + +class BusinessErrorDetail(StrEnum): + RECORD_NOT_FOUND = "Record not found" + HIGH_RISK_APPROVAL_REQUIRED = "High-risk domain change requires approval_ticket_id" + + +UNKNOWN_FIELD_TEMPLATE = "Unknown field '{field}'" +INVALID_FIELD_VALUE_TEMPLATE = "Invalid value for field '{field}'" class RiskEventType(StrEnum): diff --git a/app/modules/business/registry.py b/app/modules/business/registry.py index 0fc48df..c7fb923 100644 --- a/app/modules/business/registry.py +++ b/app/modules/business/registry.py @@ -1,42 +1,49 @@ from sqlalchemy.orm import DeclarativeMeta from app.modules.business import models +from app.modules.business.constants import BusinessDomain -DOMAIN_MODELS: dict[str, type[DeclarativeMeta]] = { - "projects": models.Project, - "tasks": models.WorkTask, - "procurements": models.Procurement, - "expenses": models.Expense, - "fund-accounts": models.FundAccount, - "policies": models.Policy, - "standards": models.Standard, - "performance-metrics": models.PerformanceMetric, - "suppliers": models.Supplier, - "attendance-records": models.AttendanceRecord, - "work-reports": models.WorkReport, - "risk-events": models.RiskEvent, - "legacy-sync-runs": models.LegacySyncRun, +DOMAIN_MODELS: dict[BusinessDomain, type[DeclarativeMeta]] = { + BusinessDomain.PROJECTS: models.Project, + BusinessDomain.TASKS: models.WorkTask, + BusinessDomain.PROCUREMENTS: models.Procurement, + BusinessDomain.EXPENSES: models.Expense, + BusinessDomain.FUND_ACCOUNTS: models.FundAccount, + BusinessDomain.POLICIES: models.Policy, + BusinessDomain.STANDARDS: models.Standard, + BusinessDomain.PERFORMANCE_METRICS: models.PerformanceMetric, + BusinessDomain.SUPPLIERS: models.Supplier, + BusinessDomain.ATTENDANCE_RECORDS: models.AttendanceRecord, + BusinessDomain.WORK_REPORTS: models.WorkReport, + BusinessDomain.RISK_EVENTS: models.RiskEvent, + BusinessDomain.LEGACY_SYNC_RUNS: models.LegacySyncRun, } -LOW_RISK_DOMAINS = { - "projects", - "tasks", - "procurements", - "expenses", - "policies", - "standards", - "suppliers", - "attendance-records", - "work-reports", - "risk-events", - "legacy-sync-runs", -} -HIGH_RISK_DOMAINS = {"fund-accounts", "performance-metrics"} +HIGH_RISK_DOMAINS = frozenset( + { + BusinessDomain.FUND_ACCOUNTS, + BusinessDomain.PERFORMANCE_METRICS, + } +) +LOW_RISK_DOMAINS = frozenset(set(DOMAIN_MODELS) - HIGH_RISK_DOMAINS) -def get_domain_model(domain: str) -> type[DeclarativeMeta]: - if domain not in DOMAIN_MODELS: - supported = ", ".join(sorted(DOMAIN_MODELS)) - raise KeyError(f"Unsupported domain '{domain}'. Supported: {supported}") - return DOMAIN_MODELS[domain] +def normalize_domain(domain: str | BusinessDomain) -> BusinessDomain: + try: + return BusinessDomain(domain) + except ValueError as exc: + supported = ", ".join(sorted(item.value for item in DOMAIN_MODELS)) + raise KeyError(f"Unsupported domain '{domain}'. Supported: {supported}") from exc + + +def supported_domain_values() -> list[str]: + return sorted(item.value for item in DOMAIN_MODELS) + + +def is_high_risk_domain(domain: str | BusinessDomain) -> bool: + return normalize_domain(domain) in HIGH_RISK_DOMAINS + + +def get_domain_model(domain: str | BusinessDomain) -> type[DeclarativeMeta]: + return DOMAIN_MODELS[normalize_domain(domain)] diff --git a/app/modules/business/routes.py b/app/modules/business/routes.py index 8a78b81..55a7590 100644 --- a/app/modules/business/routes.py +++ b/app/modules/business/routes.py @@ -1,9 +1,11 @@ from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import status as http_status from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import ApiPrincipal, require_api_key -from app.modules.business.registry import DOMAIN_MODELS +from app.modules.business.constants import BusinessField, BusinessResponseKey +from app.modules.business.registry import supported_domain_values from app.modules.business.schemas import DomainListRead, DomainRecordCreate, DomainRecordUpdate from app.modules.business.service import BusinessService @@ -12,7 +14,7 @@ router = APIRouter(dependencies=[Depends(require_api_key)]) @router.get("/domains") def list_domains() -> dict[str, list[str]]: - return {"domains": sorted(DOMAIN_MODELS)} + return {BusinessResponseKey.DOMAINS: supported_domain_values()} @router.get("/{domain}", response_model=DomainListRead) @@ -20,22 +22,29 @@ def list_records( domain: str, limit: int = Query(default=50, ge=1, le=500), offset: int = Query(default=0, ge=0), - status: str | None = None, + status_filter: str | None = Query(default=None, alias=BusinessField.STATUS), db: Session = Depends(get_db), ) -> dict: try: - total, items = BusinessService(db).list_records(domain, limit, offset, status) + total, items = BusinessService(db).list_records(domain, limit, offset, status_filter) except KeyError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc - return {"domain": domain, "total": total, "items": items} + raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + return { + BusinessResponseKey.DOMAIN: domain, + BusinessResponseKey.TOTAL: total, + BusinessResponseKey.ITEMS: items, + } @router.get("/{domain}/{record_id}") def get_record(domain: str, record_id: int, db: Session = Depends(get_db)) -> dict: try: - return {"domain": domain, "data": BusinessService(db).get_record(domain, record_id)} + return { + BusinessResponseKey.DOMAIN: domain, + BusinessResponseKey.DATA: BusinessService(db).get_record(domain, record_id), + } except KeyError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.post("/{domain}") @@ -53,8 +62,8 @@ def create_record( payload.approval_ticket_id, ) except KeyError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc - return {"domain": domain, "data": data} + raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + return {BusinessResponseKey.DOMAIN: domain, BusinessResponseKey.DATA: data} @router.patch("/{domain}/{record_id}") @@ -74,5 +83,5 @@ def update_record( approval_ticket_id=payload.approval_ticket_id, ) except KeyError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc - return {"domain": domain, "data": data} + raise HTTPException(status_code=http_status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + return {BusinessResponseKey.DOMAIN: domain, BusinessResponseKey.DATA: data} diff --git a/app/modules/business/service.py b/app/modules/business/service.py index 244fe71..aca62a8 100644 --- a/app/modules/business/service.py +++ b/app/modules/business/service.py @@ -15,8 +15,16 @@ from app.core.pagination import bounded_limit, bounded_offset from app.modules.audit.constants import AuditRiskLevel, AuditSource from app.modules.audit.schemas import AuditLogCreate from app.modules.audit.service import AuditService +from app.modules.approvals.constants import ApprovalActionValue, approval_action from app.modules.approvals.service import ApprovalService -from app.modules.business.registry import HIGH_RISK_DOMAINS, get_domain_model +from app.modules.business.registry import get_domain_model, is_high_risk_domain +from app.modules.business.constants import ( + INVALID_FIELD_VALUE_TEMPLATE, + UNKNOWN_FIELD_TEMPLATE, + BusinessErrorDetail, + BusinessField, + BusinessPayloadKey, +) def serialize_model(record: Any) -> dict[str, Any]: @@ -51,21 +59,25 @@ def _coerce_column_value(column: Column, value: Any) -> Any: def _model_payload(model: Any, data: dict[str, Any]) -> dict[str, Any]: """Validate keys and coerce values according to model column types.""" - columns = {column.name: column for column in model.__table__.columns if column.name != "id"} + columns = { + column.name: column + for column in model.__table__.columns + if column.name != BusinessField.ID + } payload: dict[str, Any] = {} for key, value in data.items(): column = columns.get(key) if column is None: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=f"Unknown field '{key}'", + detail=UNKNOWN_FIELD_TEMPLATE.format(field=key), ) try: payload[key] = _coerce_column_value(column, value) except (ValueError, TypeError, InvalidOperation) as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=f"Invalid value for field '{key}'", + detail=INVALID_FIELD_VALUE_TEMPLATE.format(field=key), ) from exc return payload @@ -87,7 +99,7 @@ class BusinessService: model = get_domain_model(domain) stmt: Select = select(model) count_stmt = select(func.count()).select_from(model) - if status_filter and hasattr(model, "status"): + if status_filter and hasattr(model, BusinessField.STATUS): stmt = stmt.where(model.status == status_filter) count_stmt = count_stmt.where(model.status == status_filter) stmt = stmt.order_by(model.id.desc()).limit(bounded_limit(limit)).offset( @@ -100,7 +112,10 @@ class BusinessService: model = get_domain_model(domain) record = self.db.get(model, record_id) if record is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Record not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=BusinessErrorDetail.RECORD_NOT_FOUND, + ) return serialize_model(record) def create_record( @@ -111,16 +126,17 @@ class BusinessService: approval_ticket_id: str | None = None, ) -> dict[str, Any]: model = get_domain_model(domain) + high_risk = is_high_risk_domain(domain) payload = _model_payload(model, data) record = model(**payload) self.db.add(record) - if domain in HIGH_RISK_DOMAINS: + if high_risk: self.db.flush() self._consume_approval( approval_ticket_id, domain, record.id, - f"create:{domain}", + approval_action(ApprovalActionValue.CREATE, domain), data, actor, ) @@ -131,13 +147,14 @@ class BusinessService: AuditLogCreate( actor=actor, source=AuditSource.API, - action=f"create:{domain}", + action=approval_action(ApprovalActionValue.CREATE, domain), target_type=domain, target_id=str(record.id), - risk_level=( - AuditRiskLevel.HIGH if domain in HIGH_RISK_DOMAINS else AuditRiskLevel.LOW - ), - request_payload={"data": data, "approval_ticket_id": approval_ticket_id}, + risk_level=AuditRiskLevel.HIGH if high_risk else AuditRiskLevel.LOW, + request_payload={ + BusinessPayloadKey.DATA: data, + BusinessPayloadKey.APPROVAL_TICKET_ID: approval_ticket_id, + }, response_payload=result, ) ) @@ -152,19 +169,20 @@ class BusinessService: approval_ticket_id: str | None = None, ) -> dict[str, Any]: model = get_domain_model(domain) + high_risk = is_high_risk_domain(domain) record = self.db.get(model, record_id) if record is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, - detail="Record not found", + detail=BusinessErrorDetail.RECORD_NOT_FOUND, ) payload = _model_payload(model, data) - if domain in HIGH_RISK_DOMAINS: + if high_risk: self._consume_approval( approval_ticket_id, domain, record_id, - f"update:{domain}", + approval_action(ApprovalActionValue.UPDATE, domain), data, actor, ) @@ -177,13 +195,14 @@ class BusinessService: AuditLogCreate( actor=actor, source=AuditSource.API, - action=f"update:{domain}", + action=approval_action(ApprovalActionValue.UPDATE, domain), target_type=domain, target_id=str(record.id), - risk_level=( - AuditRiskLevel.HIGH if domain in HIGH_RISK_DOMAINS else AuditRiskLevel.LOW - ), - request_payload={"data": data, "approval_ticket_id": approval_ticket_id}, + risk_level=AuditRiskLevel.HIGH if high_risk else AuditRiskLevel.LOW, + request_payload={ + BusinessPayloadKey.DATA: data, + BusinessPayloadKey.APPROVAL_TICKET_ID: approval_ticket_id, + }, response_payload=result, ) ) @@ -201,7 +220,7 @@ class BusinessService: if not approval_ticket_id: raise HTTPException( status_code=status.HTTP_409_CONFLICT, - detail="High-risk domain change requires approval_ticket_id", + detail=BusinessErrorDetail.HIGH_RISK_APPROVAL_REQUIRED, ) ApprovalService(self.db).consume_for( approval_ticket_id, diff --git a/app/modules/feishu/client.py b/app/modules/feishu/client.py index 3c3984f..5f0c6e3 100644 --- a/app/modules/feishu/client.py +++ b/app/modules/feishu/client.py @@ -3,7 +3,7 @@ import time from typing import Any import httpx -from fastapi import HTTPException +from fastapi import HTTPException, status from app.core.constants import BEARER_TOKEN_TEMPLATE, HttpHeader from app.core.config import get_settings @@ -34,7 +34,10 @@ class FeishuClient: def _get_tenant_access_token(self) -> str: if not self._is_configured(): - raise HTTPException(status_code=503, detail=FEISHU_AUTH_MISSING) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=FEISHU_AUTH_MISSING, + ) if self._tenant_access_token and time.time() < self._token_expires_at: return self._tenant_access_token @@ -48,7 +51,10 @@ class FeishuClient: response.raise_for_status() data = response.json() if data.get(FeishuPayloadKey.CODE) != FEISHU_SUCCESS_CODE: - raise HTTPException(status_code=502, detail={FeishuPayloadKey.FEISHU_ERROR: data}) + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail={FeishuPayloadKey.FEISHU_ERROR: data}, + ) self._tenant_access_token = data[FeishuPayloadKey.TENANT_ACCESS_TOKEN] expire_seconds = int( data.get(FeishuPayloadKey.EXPIRE, FEISHU_DEFAULT_TOKEN_EXPIRE_SECONDS) @@ -87,7 +93,7 @@ class FeishuClient: chat_id = receive_id or self.settings.feishu_default_chat_id if not chat_id: raise HTTPException( - status_code=400, + status_code=status.HTTP_400_BAD_REQUEST, detail=FEISHU_RECEIVE_ID_MISSING, ) return self.send_message( @@ -106,7 +112,7 @@ class FeishuClient: chat_id = receive_id or self.settings.feishu_default_chat_id if not chat_id: raise HTTPException( - status_code=400, + status_code=status.HTTP_400_BAD_REQUEST, detail=FEISHU_RECEIVE_ID_MISSING, ) return self.send_message(chat_id, receive_id_type, FeishuMessageType.INTERACTIVE, card) diff --git a/app/modules/feishu/commands.py b/app/modules/feishu/commands.py index 7acb0a6..8149661 100644 --- a/app/modules/feishu/commands.py +++ b/app/modules/feishu/commands.py @@ -9,9 +9,20 @@ from app.core.config import get_settings from app.modules.ai_agent.service import AIService from app.modules.ai_agent.constants import AIResponseKey from app.modules.audit.constants import AuditSource -from app.modules.feishu.constants import FeishuCommandKey +from app.modules.feishu.constants import ( + FEISHU_AI_REPLY_TITLE, + FEISHU_MENTION_PATTERN, + FEISHU_ZERO_WIDTH_SPACE, + FeishuCommandKey, + FeishuCommandName, + FeishuCommandResultKey, + FeishuPayloadKey, + FeishuReplyType, +) +from app.modules.reports.constants import ReportResponseKey from app.modules.feishu.service import FeishuService from app.modules.reports.service import ReportService +from app.modules.risk.constants import RiskSummaryKey from app.modules.risk.service import RiskService DAILY_REPORT_KEYWORDS = ("日报", "晨报", "经营日报", "经营晨报") @@ -27,7 +38,11 @@ def _parse_content_text(content: Any) -> str: """Extract plain command text from a Feishu message content payload.""" if isinstance(content, dict): - return str(content.get("text") or content.get("content") or "") + return str( + content.get(FeishuPayloadKey.TEXT) + or content.get(FeishuPayloadKey.CONTENT) + or "" + ) if not isinstance(content, str): return "" try: @@ -35,18 +50,42 @@ def _parse_content_text(content: Any) -> str: except json.JSONDecodeError: return content if isinstance(data, dict): - return str(data.get("text") or data.get("content") or "") + return str( + data.get(FeishuPayloadKey.TEXT) + or data.get(FeishuPayloadKey.CONTENT) + or "" + ) return content def _clean_command_text(text: str) -> str: """Remove mentions and invisible characters from Feishu command text.""" - text = re.sub(r"@\S+", "", text or "") - text = text.replace("\u200b", "") + text = re.sub(FEISHU_MENTION_PATTERN, "", text or "") + text = text.replace(FEISHU_ZERO_WIDTH_SPACE, "") return text.strip() +def _command_result( + command: FeishuCommandName, + reply_type: FeishuReplyType, + title: str, + content: str, + provider_response: dict[str, Any] | None = None, + lines: list[str] | None = None, +) -> dict[str, Any]: + result: dict[str, Any] = { + FeishuCommandResultKey.COMMAND: command, + FeishuCommandResultKey.REPLY_TYPE: reply_type, + FeishuCommandResultKey.TITLE: title, + FeishuCommandResultKey.CONTENT: content, + FeishuCommandResultKey.PROVIDER_RESPONSE: provider_response, + } + if lines is not None: + result[FeishuCommandResultKey.LINES] = lines + return result + + class FeishuCommandService: """Route Feishu text commands to reports, risk summaries, or AI replies.""" @@ -55,16 +94,20 @@ class FeishuCommandService: self.feishu = FeishuService(db) def extract_event_command(self, payload: dict[str, Any]) -> dict[str, Any] | None: - event = payload.get("event") or {} - message = event.get("message") or {} + event = payload.get(FeishuPayloadKey.EVENT) or {} + message = event.get(FeishuPayloadKey.MESSAGE) or {} if not message: return None - text = _clean_command_text(_parse_content_text(message.get("content"))) + text = _clean_command_text(_parse_content_text(message.get(FeishuPayloadKey.CONTENT))) if not text: return None - sender = event.get("sender") or {} - sender_id = sender.get("sender_id") or {} - actor = sender_id.get("open_id") or sender_id.get("user_id") or ActorValue.FEISHU + sender = event.get(FeishuPayloadKey.SENDER) or {} + sender_id = sender.get(FeishuPayloadKey.SENDER_ID) or {} + actor = ( + sender_id.get(FeishuPayloadKey.OPEN_ID) + or sender_id.get(FeishuPayloadKey.USER_ID) + or ActorValue.FEISHU + ) return { FeishuCommandKey.TEXT: text, FeishuCommandKey.CHAT_ID: message.get(FeishuCommandKey.CHAT_ID), @@ -84,80 +127,70 @@ class FeishuCommandService: if any(keyword in command_text for keyword in DAILY_REPORT_KEYWORDS): report = ReportService(self.db).daily_brief() - result = { - "command": "daily_brief", - "reply_type": "card", - "title": report["title"], - "content": report["content"], - "lines": report["lines"], - } if auto_reply: provider_response = self._send_card_if_configured( chat_id, - report["title"], - report["lines"], + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], actor, ) - result["provider_response"] = provider_response - return result + return _command_result( + FeishuCommandName.DAILY_BRIEF, + FeishuReplyType.CARD, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.CONTENT], + provider_response, + report[ReportResponseKey.LINES], + ) if any(keyword in command_text for keyword in PROJECT_WEEKLY_KEYWORDS): report = ReportService(self.db).project_weekly() - result = { - "command": "project_weekly", - "reply_type": "card", - "title": report["title"], - "content": report["content"], - "lines": report["lines"], - } if auto_reply: provider_response = self._send_card_if_configured( chat_id, - report["title"], - report["lines"], + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], actor, ) - result["provider_response"] = provider_response - return result + return _command_result( + FeishuCommandName.PROJECT_WEEKLY, + FeishuReplyType.CARD, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.CONTENT], + provider_response, + report[ReportResponseKey.LINES], + ) if any(keyword in command_text for keyword in ATTENDANCE_KEYWORDS): report = ReportService(self.db).attendance_summary() - result = { - "command": "attendance_summary", - "reply_type": "card", - "title": report["title"], - "content": report["content"], - "lines": report["lines"], - } if auto_reply: provider_response = self._send_card_if_configured( chat_id, - report["title"], - report["lines"], + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], actor, ) - result["provider_response"] = provider_response - return result + return _command_result( + FeishuCommandName.ATTENDANCE_SUMMARY, + FeishuReplyType.CARD, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.CONTENT], + provider_response, + report[ReportResponseKey.LINES], + ) if any(keyword in command_text for keyword in RISK_KEYWORDS): summary = RiskService(self.db).summary() lines = [ - f"- 综合风险等级：{summary['risk_level']}", - f"- 风险分：{summary['risk_score']}", - f"- 逾期任务：{len(summary['overdue_tasks'])}", - f"- 延期项目：{len(summary['delayed_projects'])}", - f"- 超预算项目：{len(summary['over_budget_projects'])}", - f"- 资金风险账户：{len(summary['fund_risks'])}", - f"- 供应商风险：{len(summary['supplier_risks'])}", - f"- 打开风险事件：{len(summary['open_events'])}", + f"- 综合风险等级：{summary[RiskSummaryKey.RISK_LEVEL]}", + f"- 风险分：{summary[RiskSummaryKey.RISK_SCORE]}", + f"- 逾期任务：{len(summary[RiskSummaryKey.OVERDUE_TASKS])}", + f"- 延期项目：{len(summary[RiskSummaryKey.DELAYED_PROJECTS])}", + f"- 超预算项目：{len(summary[RiskSummaryKey.OVER_BUDGET_PROJECTS])}", + f"- 资金风险账户：{len(summary[RiskSummaryKey.FUND_RISKS])}", + f"- 供应商风险：{len(summary[RiskSummaryKey.SUPPLIER_RISKS])}", + f"- 打开风险事件：{len(summary[RiskSummaryKey.OPEN_EVENTS])}", ] - result = { - "command": "risk_summary", - "reply_type": "card", - "title": RISK_TITLE, - "content": "\n".join(lines), - "lines": lines, - } if auto_reply: provider_response = self._send_card_if_configured( chat_id, @@ -165,8 +198,14 @@ class FeishuCommandService: lines, actor, ) - result["provider_response"] = provider_response - return result + return _command_result( + FeishuCommandName.RISK_SUMMARY, + FeishuReplyType.CARD, + RISK_TITLE, + "\n".join(lines), + provider_response, + lines, + ) prompt = command_text for prefix in AI_COMMAND_PREFIXES: @@ -186,16 +225,15 @@ class FeishuCommandService: command_text.startswith(prefix) or lowered.startswith(prefix) for prefix in AI_COMMAND_PREFIXES ) - result = { - "command": "ai_ask" if is_explicit_ai else "fallback_ai", - "reply_type": "text", - "title": "AI 回复", - "content": content, - } if auto_reply: provider_response = self._send_text_if_configured(chat_id, content, actor) - result["provider_response"] = provider_response - return result + return _command_result( + FeishuCommandName.AI_ASK if is_explicit_ai else FeishuCommandName.FALLBACK_AI, + FeishuReplyType.TEXT, + FEISHU_AI_REPLY_TITLE, + content, + provider_response, + ) def _send_card_if_configured( self, diff --git a/app/modules/feishu/constants.py b/app/modules/feishu/constants.py index 1e3514f..a10c2f5 100644 --- a/app/modules/feishu/constants.py +++ b/app/modules/feishu/constants.py @@ -10,25 +10,44 @@ class FeishuMessageType(StrEnum): INTERACTIVE = "interactive" +class FeishuEventSource(StrEnum): + WEBHOOK = "webhook" + LONG_CONNECTION = "long_connection" + + class FeishuPayloadKey(StrEnum): + APP_ID = "app_id" + APP_SECRET = "app_secret" + CARD = "card" + CHALLENGE = "challenge" + CODE = "code" + CONFIG = "config" + CONTENT = "content" + DIV = "div" + ELEMENTS = "elements" + EXPIRE = "expire" + FEISHU_ERROR = "feishu_error" HEADER = "header" EVENT = "event" EVENT_ID = "event_id" EVENT_TYPE = "event_type" + LARK_MARKDOWN = "lark_md" MESSAGE = "message" MESSAGE_ID = "message_id" + MESSAGE_TYPE = "msg_type" + OPEN_ID = "open_id" + PLAIN_TEXT = "plain_text" RECEIVE_ID = "receive_id" RECEIVE_ID_TYPE = "receive_id_type" - MESSAGE_TYPE = "msg_type" - CONTENT = "content" - TEXT = "text" - CODE = "code" + SENDER = "sender" + SENDER_ID = "sender_id" + TAG = "tag" TENANT_ACCESS_TOKEN = "tenant_access_token" - EXPIRE = "expire" - APP_ID = "app_id" - APP_SECRET = "app_secret" - FEISHU_ERROR = "feishu_error" - CARD = "card" + TEXT = "text" + TITLE = "title" + TOKEN = "token" + USER_ID = "user_id" + WIDE_SCREEN_MODE = "wide_screen_mode" class FeishuCommandKey(StrEnum): @@ -37,12 +56,57 @@ class FeishuCommandKey(StrEnum): ACTOR = "actor" +class FeishuResponseKey(StrEnum): + OK = "ok" + ACCEPTED = "accepted" + HANDLED = "handled" + DUPLICATE = "duplicate" + RESULT = "result" + CHALLENGE = "challenge" + PROVIDER_RESPONSE = "provider_response" + + +class FeishuCommandResultKey(StrEnum): + COMMAND = "command" + REPLY_TYPE = "reply_type" + TITLE = "title" + CONTENT = "content" + LINES = "lines" + PROVIDER_RESPONSE = "provider_response" + + +class FeishuCommandName(StrEnum): + DAILY_BRIEF = "daily_brief" + PROJECT_WEEKLY = "project_weekly" + ATTENDANCE_SUMMARY = "attendance_summary" + RISK_SUMMARY = "risk_summary" + AI_ASK = "ai_ask" + FALLBACK_AI = "fallback_ai" + + +class FeishuReplyType(StrEnum): + CARD = "card" + TEXT = "text" + + +class FeishuEventReceiptKey(StrEnum): + EVENT_KEY = "event_key" + SOURCE = "source" + EVENT_ID = "event_id" + MESSAGE_ID = "message_id" + + FEISHU_TENANT_TOKEN_PATH = "/auth/v3/tenant_access_token/internal" FEISHU_MESSAGE_PATH = "/im/v1/messages" +FEISHU_DEFAULT_OPEN_API_DOMAIN = "https://open.feishu.cn" FEISHU_AUTH_MISSING = "Feishu app credentials are not configured" +FEISHU_VERIFICATION_TOKEN_REQUIRED = "FEISHU_VERIFICATION_TOKEN is required" +FEISHU_INVALID_TOKEN = "Invalid Feishu token" FEISHU_RECEIVE_ID_MISSING = "receive_id or FEISHU_DEFAULT_CHAT_ID is required" FEISHU_SUCCESS_CODE = 0 FEISHU_DEFAULT_TOKEN_EXPIRE_SECONDS = 7200 FEISHU_TOKEN_EXPIRE_SAFETY_SECONDS = 300 -FEISHU_WEBHOOK_EVENT_ACTION = "webhook_event" -FEISHU_LONG_CONNECTION_EVENT_ACTION = "long_connection_event" +FEISHU_AI_REPLY_TITLE = "AI 回复" +FEISHU_EMPTY_CARD_TEXT = "暂无数据" +FEISHU_MENTION_PATTERN = r"@\S+" +FEISHU_ZERO_WIDTH_SPACE = "\u200b" diff --git a/app/modules/feishu/events.py b/app/modules/feishu/events.py index 99f4a7b..8934a87 100644 --- a/app/modules/feishu/events.py +++ b/app/modules/feishu/events.py @@ -4,21 +4,22 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.constants import ActorValue -from app.modules.audit.constants import AuditSource +from app.modules.audit.constants import AuditAction, AuditSource from app.modules.audit.schemas import AuditLogCreate from app.modules.feishu.commands import FeishuCommandService from app.modules.feishu.constants import ( - FEISHU_LONG_CONNECTION_EVENT_ACTION, - FEISHU_WEBHOOK_EVENT_ACTION, FeishuCommandKey, + FeishuEventReceiptKey, + FeishuEventSource, FeishuPayloadKey, + FeishuResponseKey, ) from app.modules.feishu.models import FeishuEventReceipt from app.modules.feishu.service import FeishuService FEISHU_EVENT_ACTIONS = { - "webhook": FEISHU_WEBHOOK_EVENT_ACTION, - "long_connection": FEISHU_LONG_CONNECTION_EVENT_ACTION, + FeishuEventSource.WEBHOOK: AuditAction.FEISHU_WEBHOOK_EVENT, + FeishuEventSource.LONG_CONNECTION: AuditAction.FEISHU_LONG_CONNECTION_EVENT, } @@ -33,41 +34,57 @@ class FeishuEventService: def handle_event( self, payload: dict[str, Any], - source: str, + source: str | FeishuEventSource, auto_reply: bool = True, ) -> dict[str, Any]: self.feishu.verify_event(payload) + challenge = payload.get(FeishuPayloadKey.CHALLENGE) + if challenge: + return {FeishuResponseKey.CHALLENGE: challenge} + source_value = _normalize_source(source) event_identity = _event_identity(payload, source) if event_identity and not self._register_event(event_identity): - return {"ok": True, "handled": False, "duplicate": True} + return { + FeishuResponseKey.OK: True, + FeishuResponseKey.HANDLED: False, + FeishuResponseKey.DUPLICATE: True, + } self.feishu.audit.log( AuditLogCreate( actor=ActorValue.FEISHU, source=AuditSource.FEISHU, - action=FEISHU_EVENT_ACTIONS.get(source, FEISHU_WEBHOOK_EVENT_ACTION), - target_type=source, - target_id=event_identity.get("event_key") if event_identity else None, + action=FEISHU_EVENT_ACTIONS[source_value], + target_type=source_value, + target_id=( + event_identity.get(FeishuEventReceiptKey.EVENT_KEY) + if event_identity + else None + ), request_payload=payload, - response_payload={"accepted": True}, + response_payload={FeishuResponseKey.ACCEPTED: True}, ) ) command = self.commands.extract_event_command(payload) if not command: - return {"ok": True, "handled": False} + return {FeishuResponseKey.OK: True, FeishuResponseKey.HANDLED: False} result = self.commands.handle_text( command[FeishuCommandKey.TEXT], chat_id=command[FeishuCommandKey.CHAT_ID], actor=command[FeishuCommandKey.ACTOR], auto_reply=auto_reply, ) - return {"ok": True, "handled": True, "result": result} + return { + FeishuResponseKey.OK: True, + FeishuResponseKey.HANDLED: True, + FeishuResponseKey.RESULT: result, + } def _register_event(self, event_identity: dict[str, str | None]) -> bool: receipt = FeishuEventReceipt( - event_key=str(event_identity["event_key"]), - source=str(event_identity["source"]), - event_id=event_identity.get("event_id"), - message_id=event_identity.get("message_id"), + event_key=str(event_identity[FeishuEventReceiptKey.EVENT_KEY]), + source=str(event_identity[FeishuEventReceiptKey.SOURCE]), + event_id=event_identity.get(FeishuEventReceiptKey.EVENT_ID), + message_id=event_identity.get(FeishuEventReceiptKey.MESSAGE_ID), ) self.db.add(receipt) try: @@ -78,7 +95,15 @@ class FeishuEventService: return True -def _event_identity(payload: dict[str, Any], source: str) -> dict[str, str | None] | None: +def _normalize_source(source: str | FeishuEventSource) -> FeishuEventSource: + return FeishuEventSource(source) + + +def _event_identity( + payload: dict[str, Any], + source: str | FeishuEventSource, +) -> dict[str, str | None] | None: + source_value = _normalize_source(source) header = payload.get(FeishuPayloadKey.HEADER) or {} event = payload.get(FeishuPayloadKey.EVENT) or {} message = event.get(FeishuPayloadKey.MESSAGE) or {} @@ -90,11 +115,11 @@ def _event_identity(payload: dict[str, Any], source: str) -> dict[str, str | Non event_type = header.get(FeishuPayloadKey.EVENT_TYPE) event_key = ":".join( str(part) - for part in (source, event_type or FeishuPayloadKey.EVENT, stable_id) + for part in (source_value, event_type or FeishuPayloadKey.EVENT, stable_id) ) return { - "event_key": event_key, - "source": source, - "event_id": str(event_id) if event_id else None, - "message_id": str(message_id) if message_id else None, + FeishuEventReceiptKey.EVENT_KEY: event_key, + FeishuEventReceiptKey.SOURCE: source_value, + FeishuEventReceiptKey.EVENT_ID: str(event_id) if event_id else None, + FeishuEventReceiptKey.MESSAGE_ID: str(message_id) if message_id else None, } diff --git a/app/modules/feishu/long_connection.py b/app/modules/feishu/long_connection.py index 55c4aa6..c74d98b 100644 --- a/app/modules/feishu/long_connection.py +++ b/app/modules/feishu/long_connection.py @@ -5,6 +5,7 @@ from urllib.parse import urlsplit from app.core.config import get_settings from app.core.database import SessionLocal +from app.modules.feishu.constants import FEISHU_DEFAULT_OPEN_API_DOMAIN, FeishuEventSource from app.modules.feishu.events import FeishuEventService logger = logging.getLogger(__name__) @@ -13,7 +14,7 @@ logger = logging.getLogger(__name__) def _sdk_domain(base_url: str) -> str: parsed = urlsplit(base_url) if not parsed.scheme or not parsed.netloc: - return "https://open.feishu.cn" + return FEISHU_DEFAULT_OPEN_API_DOMAIN return f"{parsed.scheme}://{parsed.netloc}" @@ -35,7 +36,7 @@ def _handle_message_event(event: Any) -> None: try: result = FeishuEventService(db).handle_event( payload, - source="long_connection", + source=FeishuEventSource.LONG_CONNECTION, auto_reply=True, ) logger.info("Handled Feishu long connection event: %s", result) diff --git a/app/modules/feishu/routes.py b/app/modules/feishu/routes.py index 820308c..492e5de 100644 --- a/app/modules/feishu/routes.py +++ b/app/modules/feishu/routes.py @@ -4,6 +4,7 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import ApiPrincipal, require_api_key from app.modules.feishu.commands import FeishuCommandService +from app.modules.feishu.constants import FeishuEventSource, FeishuPayloadKey, FeishuResponseKey from app.modules.feishu.events import FeishuEventService from app.modules.feishu.schemas import ( FeishuCardMessage, @@ -22,11 +23,11 @@ async def feishu_webhook(request: Request, db: Session = Depends(get_db)) -> dic """Handle Feishu webhook challenge and text command events.""" payload = await request.json() - service = FeishuService(db) - service.verify_event(payload) - if payload.get("challenge"): - return {"challenge": payload["challenge"]} - return FeishuEventService(db).handle_event(payload, source="webhook", auto_reply=True) + return FeishuEventService(db).handle_event( + payload, + source=FeishuEventSource.WEBHOOK, + auto_reply=True, + ) @router.post("/send-text", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) @@ -41,7 +42,10 @@ def send_text( receive_id_type=payload.receive_id_type, actor=principal.actor, ) - return {"ok": result.get("code") == 0, "provider_response": result} + return { + FeishuResponseKey.OK: result.get(FeishuPayloadKey.CODE) == 0, + FeishuResponseKey.PROVIDER_RESPONSE: result, + } @router.post("/send-card", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) @@ -56,7 +60,10 @@ def send_card( receive_id_type=payload.receive_id_type, actor=principal.actor, ) - return {"ok": result.get("code") == 0, "provider_response": result} + return { + FeishuResponseKey.OK: result.get(FeishuPayloadKey.CODE) == 0, + FeishuResponseKey.PROVIDER_RESPONSE: result, + } @router.post( diff --git a/app/modules/feishu/service.py b/app/modules/feishu/service.py index bd443ed..b93b41c 100644 --- a/app/modules/feishu/service.py +++ b/app/modules/feishu/service.py @@ -10,7 +10,13 @@ from app.modules.audit.constants import AuditAction, AuditSource from app.modules.audit.schemas import AuditLogCreate from app.modules.audit.service import AuditService from app.modules.feishu.client import FeishuClient -from app.modules.feishu.constants import FeishuPayloadKey, FeishuReceiveIdType +from app.modules.feishu.constants import ( + FEISHU_EMPTY_CARD_TEXT, + FEISHU_INVALID_TOKEN, + FEISHU_VERIFICATION_TOKEN_REQUIRED, + FeishuPayloadKey, + FeishuReceiveIdType, +) class FeishuService: @@ -24,17 +30,17 @@ class FeishuService: def verify_event(self, payload: dict[str, Any]) -> None: settings = get_settings() expected = settings.feishu_verification_token - header = payload.get("header") or {} - token = payload.get("token") or header.get("token") + header = payload.get(FeishuPayloadKey.HEADER) or {} + token = payload.get(FeishuPayloadKey.TOKEN) or header.get(FeishuPayloadKey.TOKEN) if not expected: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="FEISHU_VERIFICATION_TOKEN is required", + detail=FEISHU_VERIFICATION_TOKEN_REQUIRED, ) if not token or not compare_digest(str(token), expected): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid Feishu token", + detail=FEISHU_INVALID_TOKEN, ) def send_text( @@ -86,14 +92,19 @@ class FeishuService: @staticmethod def build_basic_card(title: str, lines: list[str]) -> dict[str, Any]: return { - "config": {"wide_screen_mode": True}, - "header": {"title": {"tag": "plain_text", "content": title}}, - "elements": [ + FeishuPayloadKey.CONFIG: {FeishuPayloadKey.WIDE_SCREEN_MODE: True}, + FeishuPayloadKey.HEADER: { + FeishuPayloadKey.TITLE: { + FeishuPayloadKey.TAG: FeishuPayloadKey.PLAIN_TEXT, + FeishuPayloadKey.CONTENT: title, + } + }, + FeishuPayloadKey.ELEMENTS: [ { - "tag": "div", - "text": { - "tag": "lark_md", - "content": "\n".join(lines) or "暂无数据", + FeishuPayloadKey.TAG: FeishuPayloadKey.DIV, + FeishuPayloadKey.TEXT: { + FeishuPayloadKey.TAG: FeishuPayloadKey.LARK_MARKDOWN, + FeishuPayloadKey.CONTENT: "\n".join(lines) or FEISHU_EMPTY_CARD_TEXT, }, } ], diff --git a/app/modules/legacy_mysql/constants.py b/app/modules/legacy_mysql/constants.py index 38134b9..6f2dd00 100644 --- a/app/modules/legacy_mysql/constants.py +++ b/app/modules/legacy_mysql/constants.py @@ -5,6 +5,71 @@ class LegacyQueryName(StrEnum): PROJECTS = "projects" +class LegacyResponseKey(StrEnum): + STATUS = "status" + COLUMNS = "columns" + ROWS = "rows" + ROW_COUNT = "row_count" + DRY_RUN = "dry_run" + CREATED = "created" + UPDATED = "updated" + SKIPPED = "skipped" + ITEMS = "items" + ACTION = "action" + REASON = "reason" + SOURCE = "source" + PROJECT = "project" + SYNC_RUN_CODE = "sync_run_code" + SOURCE_QUERY = "source_query" + FIELD_MAP = "field_map" + LIMIT = "limit" + + +class LegacyProjectField(StrEnum): + ID = "id" + CODE = "code" + EXTERNAL_ID = "external_id" + SOURCE_SYSTEM = "source_system" + NAME = "name" + OWNER = "owner" + STATUS = "status" + PROGRESS = "progress" + PROGRESS_PERCENT = "progress_percent" + START_DATE = "start_date" + DUE_DATE = "due_date" + BUDGET = "budget" + BUDGET_AMOUNT = "budget_amount" + ACTUAL_COST = "actual_cost" + ACTUAL_AMOUNT = "actual_amount" + DESCRIPTION = "description" + + +class LegacySyncAction(StrEnum): + CREATE = "create" + UPDATE = "update" + SKIPPED = "skipped" + ALLOWLISTED_INLINE_SQL = "allowlisted_inline_sql" + + 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." + ONLY_SELECT_ALLOWED = "Only SELECT statements are allowed" + FORBIDDEN_SQL_TOKEN = "Forbidden SQL token in readonly query" + INVALID_LIMIT = "Invalid readonly query limit" + APP_DB_UNAVAILABLE = "Application database session is not available" + + +LEGACY_SYNC_RUN_CODE_PREFIX = "SYNC-PROJECTS" +LEGACY_PROJECT_QUERY_SOURCE = "LEGACY_PROJECT_QUERY" +LEGACY_SYNC_MISSING_ID_REASON = "missing external_id/code" +LEGACY_PROJECT_SYNC_NOTE = "Project sync from readonly legacy MySQL" +LEGACY_MYSQL_CONNECTION_FAILED_TEMPLATE = "MySQL connection failed: {error}" +LEGACY_HEALTH_SQL = "SELECT 1" +LEGACY_SELECT_PREFIX = "select" +LEGACY_SQL_TRAILING_TERMINATOR = ";" +LEGACY_LIMIT_MARKER = " limit " +LEGACY_LIMIT_CLAUSE = " LIMIT :limit" +LEGACY_PROJECT_CODE_TEMPLATE = "{prefix}-{external_id}" +LEGACY_UNNAMED_PROJECT = "未命名项目" diff --git a/app/modules/legacy_mysql/routes.py b/app/modules/legacy_mysql/routes.py index b128359..7081f7a 100644 --- a/app/modules/legacy_mysql/routes.py +++ b/app/modules/legacy_mysql/routes.py @@ -19,16 +19,6 @@ def mysql_health(db: Session = Depends(get_db)) -> dict[str, str]: return LegacyMySQLService(db).health() -@router.get("/tables") -def list_tables(db: Session = Depends(get_db)) -> dict[str, list[str]]: - return {"tables": LegacyMySQLService(db).list_tables()} - - -@router.get("/tables/{table_name}") -def describe_table(table_name: str, db: Session = Depends(get_db)) -> dict: - return {"table": table_name, "columns": LegacyMySQLService(db).describe_table(table_name)} - - @router.post("/query", response_model=QueryResult) def readonly_query(payload: ReadonlyQueryRequest, db: Session = Depends(get_db)) -> dict: service = LegacyMySQLService(db) diff --git a/app/modules/legacy_mysql/service.py b/app/modules/legacy_mysql/service.py index 75a9446..5fdf4d0 100644 --- a/app/modules/legacy_mysql/service.py +++ b/app/modules/legacy_mysql/service.py @@ -3,13 +3,12 @@ from decimal import Decimal from typing import Any from fastapi import HTTPException, status -from sqlalchemy import inspect, text +from sqlalchemy import select, text from sqlalchemy.engine import Engine, RowMapping from sqlalchemy.exc import SQLAlchemyError -from sqlalchemy import select from sqlalchemy.orm import Session -from app.core.constants import ActorValue +from app.core.constants import ActorValue, ApiStatus from app.core.config import get_settings from app.core.database import legacy_engine from app.core.pagination import bounded_limit @@ -20,7 +19,25 @@ 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.service import serialize_model -from app.modules.legacy_mysql.constants import LegacyQueryError, LegacyQueryName +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_HEALTH_SQL, + LEGACY_LIMIT_CLAUSE, + LEGACY_LIMIT_MARKER, + LEGACY_MYSQL_CONNECTION_FAILED_TEMPLATE, + LEGACY_PROJECT_CODE_TEMPLATE, + LEGACY_SELECT_PREFIX, + LEGACY_SQL_TRAILING_TERMINATOR, + LEGACY_UNNAMED_PROJECT, + LegacyProjectField, + LegacyQueryError, + LegacyQueryName, + LegacyResponseKey, + LegacySyncAction, +) FORBIDDEN_SQL_TOKENS = { "insert", @@ -53,7 +70,7 @@ def _row_to_dict(row: RowMapping) -> dict[str, Any]: def _normalize_sql(sql: str) -> str: - return " ".join(sql.strip().rstrip(";").split()).lower() + return " ".join(sql.strip().rstrip(LEGACY_SQL_TRAILING_TERMINATOR).split()).lower() def _query_name_text(query_name: str | LegacyQueryName | None) -> str: @@ -75,18 +92,24 @@ class LegacyMySQLService: if legacy_engine is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="LEGACY_DATABASE_URL is not configured", + detail=LegacyQueryError.DATABASE_NOT_CONFIGURED, ) return legacy_engine @staticmethod def _ensure_readonly(sql: str) -> None: stripped = sql.strip().lower() - if not stripped.startswith("select"): - raise HTTPException(status_code=400, detail="Only SELECT statements are allowed") + if not stripped.startswith(LEGACY_SELECT_PREFIX): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=LegacyQueryError.ONLY_SELECT_ALLOWED, + ) tokens = {token.strip(" ;,\n\t") for token in stripped.replace("(", " ").split()} if tokens & FORBIDDEN_SQL_TOKENS: - raise HTTPException(status_code=400, detail="Forbidden SQL token in readonly query") + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=LegacyQueryError.FORBIDDEN_SQL_TOKEN, + ) @staticmethod def _allowed_queries() -> dict[str, str]: @@ -103,35 +126,13 @@ class LegacyMySQLService: engine = self._ensure_engine() try: with engine.connect() as conn: - conn.execute(text("SELECT 1")) + conn.execute(text(LEGACY_HEALTH_SQL)) except SQLAlchemyError as exc: - raise HTTPException(status_code=503, detail=f"MySQL connection failed: {exc}") from exc - return {"status": "ok"} - - def list_tables(self) -> list[str]: - engine = self._ensure_engine() - return sorted(inspect(engine).get_table_names()) - - def describe_table(self, table_name: str) -> list[dict[str, Any]]: - engine = self._ensure_engine() - inspector = inspect(engine) - if table_name not in inspector.get_table_names(): - raise HTTPException(status_code=404, detail="Table not found") - columns = [] - for column in inspector.get_columns(table_name): - columns.append( - { - "name": column["name"], - "type": str(column["type"]), - "nullable": column.get("nullable", True), - "default": ( - str(column.get("default")) - if column.get("default") is not None - else None - ), - } - ) - return columns + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=LEGACY_MYSQL_CONNECTION_FAILED_TEMPLATE.format(error=exc), + ) from exc + return {LegacyResponseKey.STATUS: ApiStatus.OK} def execute_readonly( self, @@ -176,29 +177,39 @@ class LegacyMySQLService: engine = self._ensure_engine() params = dict(params or {}) try: - params["limit"] = bounded_limit(params.get("limit", limit)) + params[LegacyResponseKey.LIMIT] = bounded_limit( + params.get(LegacyResponseKey.LIMIT, limit) + ) except (TypeError, ValueError) as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail="Invalid readonly query limit", + detail=LegacyQueryError.INVALID_LIMIT, ) from exc limited_sql = sql - if " limit " not in sql.lower(): - limited_sql = f"{sql.rstrip(';')} LIMIT :limit" + if LEGACY_LIMIT_MARKER not in sql.lower(): + limited_sql = f"{sql.rstrip(LEGACY_SQL_TRAILING_TERMINATOR)}{LEGACY_LIMIT_CLAUSE}" with engine.connect() as conn: result = conn.execute(text(limited_sql), params) rows = [_row_to_dict(row) for row in result.mappings().all()] columns = list(rows[0].keys()) if rows else [] - return {"columns": columns, "rows": rows, "row_count": len(rows)} + return { + LegacyResponseKey.COLUMNS: columns, + LegacyResponseKey.ROWS: rows, + LegacyResponseKey.ROW_COUNT: len(rows), + } def fetch_default_projects(self, limit: int = 100) -> dict[str, Any]: settings = get_settings() if not settings.legacy_project_query: raise HTTPException( - status_code=400, + status_code=status.HTTP_400_BAD_REQUEST, detail=LegacyQueryError.PROJECT_QUERY_NOT_CONFIGURED, ) - return self.execute_allowed_query(LegacyQueryName.PROJECTS, {"limit": limit}, limit=limit) + return self.execute_allowed_query( + LegacyQueryName.PROJECTS, + {LegacyResponseKey.LIMIT: limit}, + limit=limit, + ) @staticmethod def _value( @@ -214,38 +225,51 @@ class LegacyMySQLService: def _project_payload(self, row: dict[str, Any], field_map: dict[str, str]) -> dict[str, Any]: settings = get_settings() - external_id = self._value(row, field_map, "external_id", row.get("id")) - raw_code = self._value(row, field_map, "code", None) + external_id = self._value( + row, + field_map, + LegacyProjectField.EXTERNAL_ID, + row.get(LegacyProjectField.ID), + ) + raw_code = self._value(row, field_map, LegacyProjectField.CODE, None) code = None if raw_code: code = str(raw_code) elif external_id is not None: - code = f"{settings.legacy_project_code_prefix}-{external_id}" + code = LEGACY_PROJECT_CODE_TEMPLATE.format( + prefix=settings.legacy_project_code_prefix, + external_id=external_id, + ) return { - "code": code, - "external_id": str(external_id) if external_id is not None else code, - "source_system": SourceSystem.LEGACY_MYSQL, - "name": self._value(row, field_map, "name", "未命名项目"), - "owner": self._value(row, field_map, "owner", None), - "status": self._value(row, field_map, "status", StatusValue.UNKNOWN), - "progress_percent": int( + LegacyProjectField.CODE: code, + LegacyProjectField.EXTERNAL_ID: str(external_id) if external_id is not None else code, + LegacyProjectField.SOURCE_SYSTEM: SourceSystem.LEGACY_MYSQL, + LegacyProjectField.NAME: self._value( + row, + field_map, + LegacyProjectField.NAME, + LEGACY_UNNAMED_PROJECT, + ), + LegacyProjectField.OWNER: self._value(row, field_map, LegacyProjectField.OWNER, None), + LegacyProjectField.STATUS: self._value(row, field_map, LegacyProjectField.STATUS, StatusValue.UNKNOWN), + LegacyProjectField.PROGRESS_PERCENT: int( self._value( row, field_map, - "progress_percent", - row.get("progress") or 0, + LegacyProjectField.PROGRESS_PERCENT, + row.get(LegacyProjectField.PROGRESS) or 0, ) or 0 ), - "start_date": self._value(row, field_map, "start_date", None), - "due_date": self._value(row, field_map, "due_date", None), - "budget_amount": ( - self._value(row, field_map, "budget_amount", row.get("budget") or 0) or 0 + LegacyProjectField.START_DATE: self._value(row, field_map, LegacyProjectField.START_DATE, None), + LegacyProjectField.DUE_DATE: self._value(row, field_map, LegacyProjectField.DUE_DATE, None), + LegacyProjectField.BUDGET_AMOUNT: ( + self._value(row, field_map, LegacyProjectField.BUDGET_AMOUNT, row.get(LegacyProjectField.BUDGET) or 0) or 0 ), - "actual_amount": ( - self._value(row, field_map, "actual_amount", row.get("actual_cost") or 0) or 0 + LegacyProjectField.ACTUAL_AMOUNT: ( + self._value(row, field_map, LegacyProjectField.ACTUAL_AMOUNT, row.get(LegacyProjectField.ACTUAL_COST) or 0) or 0 ), - "description": self._value(row, field_map, "description", None), + LegacyProjectField.DESCRIPTION: self._value(row, field_map, LegacyProjectField.DESCRIPTION, None), } def sync_projects( @@ -259,16 +283,24 @@ class LegacyMySQLService: ) -> dict[str, Any]: if self.db is None: raise HTTPException( - status_code=503, - detail="Application database session is not available", + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=LegacyQueryError.APP_DB_UNAVAILABLE, ) query_name = source_query_name or LegacyQueryName.PROJECTS if source_query: - rows = self.execute_readonly(source_query, {"limit": limit}, limit=limit)["rows"] - query_ref = "allowlisted_inline_sql" + 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, {"limit": limit}, limit=limit)["rows"] + 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 @@ -278,30 +310,30 @@ class LegacyMySQLService: for row in rows: payload = self._project_payload(row, field_map) - if not payload["external_id"] and not payload["code"]: + if not payload[LegacyProjectField.EXTERNAL_ID] and not payload[LegacyProjectField.CODE]: skipped += 1 items.append( { - "action": "skipped", - "reason": "missing external_id/code", - "source": row, + LegacyResponseKey.ACTION: LegacySyncAction.SKIPPED, + LegacyResponseKey.REASON: LEGACY_SYNC_MISSING_ID_REASON, + LegacyResponseKey.SOURCE: row, } ) continue stmt = select(Project).where( Project.source_system == SourceSystem.LEGACY_MYSQL, - Project.external_id == payload["external_id"], + Project.external_id == payload[LegacyProjectField.EXTERNAL_ID], ) record = self.db.execute(stmt).scalar_one_or_none() if record is None: record = self.db.execute( - select(Project).where(Project.code == payload["code"]) + select(Project).where(Project.code == payload[LegacyProjectField.CODE]) ).scalar_one_or_none() if record is None: created += 1 - action = "create" + action = LegacySyncAction.CREATE result = payload if not dry_run: record = Project(**payload) @@ -310,7 +342,7 @@ class LegacyMySQLService: result = serialize_model(record) else: updated += 1 - action = "update" + action = LegacySyncAction.UPDATE if not dry_run: for key, value in payload.items(): setattr(record, key, value) @@ -318,33 +350,39 @@ class LegacyMySQLService: result = serialize_model(record) else: result = payload - items.append({"action": action, "project": result, "source": row}) + items.append( + { + LegacyResponseKey.ACTION: action, + LegacyResponseKey.PROJECT: result, + LegacyResponseKey.SOURCE: row, + } + ) if not dry_run: self.db.commit() result = { - "dry_run": dry_run, - "created": created, - "updated": updated, - "skipped": skipped, - "items": items, + LegacyResponseKey.DRY_RUN: dry_run, + LegacyResponseKey.CREATED: created, + LegacyResponseKey.UPDATED: updated, + LegacyResponseKey.SKIPPED: skipped, + LegacyResponseKey.ITEMS: items, } sync_run = LegacySyncRun( - code=f"SYNC-PROJECTS-{utc_now():%Y%m%d%H%M%S%f}", + code=f"{LEGACY_SYNC_RUN_CODE_PREFIX}-{utc_now():%Y%m%d%H%M%S%f}", domain=BusinessDomain.PROJECTS, - source_table="LEGACY_PROJECT_QUERY", + source_table=LEGACY_PROJECT_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="Project sync from readonly legacy MySQL", + note=LEGACY_PROJECT_SYNC_NOTE, ) self.db.add(sync_run) self.db.commit() self.db.refresh(sync_run) - result["sync_run_code"] = sync_run.code + result[LegacyResponseKey.SYNC_RUN_CODE] = sync_run.code AuditService(self.db).log( AuditLogCreate( @@ -354,13 +392,19 @@ class LegacyMySQLService: target_type=BusinessDomain.PROJECTS, risk_level=AuditRiskLevel.MEDIUM, request_payload={ - "source_query": query_ref, - "field_map": field_map, - "limit": limit, - "dry_run": dry_run, + 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 ["dry_run", "created", "updated", "skipped"] + key: result[key] + for key in [ + LegacyResponseKey.DRY_RUN, + LegacyResponseKey.CREATED, + LegacyResponseKey.UPDATED, + LegacyResponseKey.SKIPPED, + ] }, ) ) diff --git a/app/modules/reports/constants.py b/app/modules/reports/constants.py index f3346a5..7d175ed 100644 --- a/app/modules/reports/constants.py +++ b/app/modules/reports/constants.py @@ -52,6 +52,23 @@ class LifecycleResponseKey(StrEnum): PROJECT_LIFECYCLE_REPORT = "project_lifecycle_report" +class ReportResponseKey(StrEnum): + REPORT = "report" + DATA = "data" + TITLE = "title" + REPORT_TYPE = "report_type" + PERIOD_START = "period_start" + PERIOD_END = "period_end" + WORK_DATE = "work_date" + TOTAL = "total" + ABNORMAL_TOTAL = "abnormal_total" + STATUS_COUNTS = "status_counts" + LINES = "lines" + CONTENT = "content" + METRICS = "metrics" + RISK_SUMMARY = "risk_summary" + + class LifecycleAttentionKey(StrEnum): DELAYED_PROJECTS = "delayed_projects" OVER_BUDGET_PROJECTS = "over_budget_projects" @@ -109,6 +126,18 @@ class MetricKey(StrEnum): LEVEL = "level" +class WorkReportMetricKey(StrEnum): + PROJECTS_TOTAL = "projects_total" + ACTIVE_PROJECTS = "active_projects" + TASKS_TOTAL = "tasks_total" + TASKS_COMPLETED = "tasks_completed" + TASKS_OVERDUE = "tasks_overdue" + PROCUREMENTS_PENDING = "procurements_pending" + EXPENSES_PENDING = "expenses_pending" + ATTENDANCE_TOTAL = "attendance_total" + OPEN_RISK_EVENTS = "open_risk_events" + + class HealthLevel(StrEnum): HEALTHY = "healthy" ATTENTION = "attention" @@ -133,3 +162,25 @@ class ReportText(StrEnum): RECOMMEND_STABLE = ( "当前生命周期指标稳定，建议继续保持周度复盘和风险事件归档。" ) + + +LIFECYCLE_RISK_SCORE_WEIGHTS = { + MetricKey.OVERDUE_TASKS: 1, + MetricKey.DELAYED_PROJECTS: 3, + MetricKey.OVER_BUDGET_PROJECTS: 4, + MetricKey.EXTERNAL_OPEN_EVENTS: 2, + MetricKey.EXTERNAL_HIGH_EVENTS: 3, +} +HEALTH_PENALTY_WEIGHTS = { + MetricKey.OVERDUE_TASKS: 3, + MetricKey.DELAYED_PROJECTS: 8, + MetricKey.OVER_BUDGET_PROJECTS: 10, + MetricKey.EXTERNAL_HIGH_EVENTS: 8, + MetricKey.BUDGET_USAGE_RATE: 0.4, + MetricKey.COMPLETION_RATE: 0.1, + MetricKey.BLACKLISTED: 10, +} +HEALTH_SCORE_MAX = 100 +HEALTH_SCORE_MIN = 0 +HEALTHY_SCORE_THRESHOLD = 80 +ATTENTION_SCORE_THRESHOLD = 60 diff --git a/app/modules/reports/service.py b/app/modules/reports/service.py index 75be158..17db868 100644 --- a/app/modules/reports/service.py +++ b/app/modules/reports/service.py @@ -35,17 +35,26 @@ from app.modules.business.models import ( from app.modules.business.service import serialize_model from app.modules.feishu.service import FeishuService from app.modules.reports.constants import ( + ATTENTION_SCORE_THRESHOLD, + HEALTH_PENALTY_WEIGHTS, + HEALTH_SCORE_MAX, + HEALTH_SCORE_MIN, + HEALTHY_SCORE_THRESHOLD, + LIFECYCLE_RISK_SCORE_WEIGHTS, HealthLevel, LifecycleAttentionKey, LifecycleFilterKey, LifecycleResponseKey, LifecycleSection, MetricKey, + ReportResponseKey, ReportStatus, ReportText, ReportTitle, ReportType, + WorkReportMetricKey, ) +from app.modules.risk.constants import RiskSummaryKey, risk_level_for_score from app.modules.risk.service import RiskService @@ -164,18 +173,22 @@ class ReportService: f"- 待处理费用：{expense_pending}", f"- 当前账户总余额：{_money(fund_total)}", ( - f"- 今日打卡记录：{attendance['total']}，" - f"异常：{attendance['abnormal_total']}" + f"- 今日打卡记录：{attendance[ReportResponseKey.TOTAL]}，" + f"异常：{attendance[ReportResponseKey.ABNORMAL_TOTAL]}" ), - f"- 逾期任务：{len(risk_summary['overdue_tasks'])}", - f"- 延期项目：{len(risk_summary['delayed_projects'])}", - f"- 超预算项目：{len(risk_summary['over_budget_projects'])}", - f"- 资金风险账户：{len(risk_summary['fund_risks'])}", - f"- 供应商风险：{len(risk_summary['supplier_risks'])}", - f"- 打开风险事件：{len(risk_summary['open_events'])}", - f"- 综合风险等级：{risk_summary['risk_level']}", + f"- 逾期任务：{len(risk_summary[RiskSummaryKey.OVERDUE_TASKS])}", + f"- 延期项目：{len(risk_summary[RiskSummaryKey.DELAYED_PROJECTS])}", + f"- 超预算项目：{len(risk_summary[RiskSummaryKey.OVER_BUDGET_PROJECTS])}", + f"- 资金风险账户：{len(risk_summary[RiskSummaryKey.FUND_RISKS])}", + f"- 供应商风险：{len(risk_summary[RiskSummaryKey.SUPPLIER_RISKS])}", + f"- 打开风险事件：{len(risk_summary[RiskSummaryKey.OPEN_EVENTS])}", + f"- 综合风险等级：{risk_summary[RiskSummaryKey.RISK_LEVEL]}", ] - return {"title": "每日经营晨报", "lines": lines, "content": "\n".join(lines)} + return { + ReportResponseKey.TITLE: ReportTitle.DAILY_BRIEF, + ReportResponseKey.LINES: lines, + ReportResponseKey.CONTENT: "\n".join(lines), + } def project_weekly(self) -> dict: active = self._count( @@ -199,7 +212,11 @@ class ReportService: ) for item in over_budget[:10]: lines.append(f" - 超预算：{item.get('code')} {item.get('name')}") - return {"title": "项目周报", "lines": lines, "content": "\n".join(lines)} + return { + ReportResponseKey.TITLE: ReportTitle.PROJECT_WEEKLY, + ReportResponseKey.LINES: lines, + ReportResponseKey.CONTENT: "\n".join(lines), + } def project_lifecycle_report( self, @@ -559,20 +576,14 @@ class ReportService: *risk_conditions, ) risk_score = ( - overdue_tasks * 1 - + delayed_projects * 3 - + over_budget_projects * 4 - + external_open_events * 2 - + external_high_events * 3 + overdue_tasks * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.OVERDUE_TASKS] + + delayed_projects * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.DELAYED_PROJECTS] + + over_budget_projects * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.OVER_BUDGET_PROJECTS] + + external_open_events * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.EXTERNAL_OPEN_EVENTS] + + external_high_events * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.EXTERNAL_HIGH_EVENTS] ) - if risk_score >= 15: - level = RiskLevel.HIGH - elif risk_score >= 5: - level = RiskLevel.MEDIUM - else: - level = RiskLevel.LOW return { - MetricKey.RISK_LEVEL: level, + MetricKey.RISK_LEVEL: risk_level_for_score(risk_score), MetricKey.RISK_SCORE: risk_score, MetricKey.OVERDUE_TASKS: overdue_tasks, MetricKey.DELAYED_PROJECTS: delayed_projects, @@ -602,19 +613,32 @@ class ReportService: include_global_risk: bool, ) -> dict[str, Any]: penalty = ( - risks[MetricKey.OVERDUE_TASKS] * 3 - + risks[MetricKey.DELAYED_PROJECTS] * 8 - + risks[MetricKey.OVER_BUDGET_PROJECTS] * 10 - + risks[MetricKey.EXTERNAL_HIGH_EVENTS] * 8 - + max(0, projects[MetricKey.BUDGET_USAGE_RATE] - 100) * 0.4 - + (100 - tasks[MetricKey.COMPLETION_RATE]) * 0.1 + risks[MetricKey.OVERDUE_TASKS] * HEALTH_PENALTY_WEIGHTS[MetricKey.OVERDUE_TASKS] + + risks[MetricKey.DELAYED_PROJECTS] + * HEALTH_PENALTY_WEIGHTS[MetricKey.DELAYED_PROJECTS] + + risks[MetricKey.OVER_BUDGET_PROJECTS] + * HEALTH_PENALTY_WEIGHTS[MetricKey.OVER_BUDGET_PROJECTS] + + risks[MetricKey.EXTERNAL_HIGH_EVENTS] + * HEALTH_PENALTY_WEIGHTS[MetricKey.EXTERNAL_HIGH_EVENTS] + + max( + HEALTH_SCORE_MIN, + projects[MetricKey.BUDGET_USAGE_RATE] - HEALTH_SCORE_MAX, + ) + * HEALTH_PENALTY_WEIGHTS[MetricKey.BUDGET_USAGE_RATE] + + (HEALTH_SCORE_MAX - tasks[MetricKey.COMPLETION_RATE]) + * HEALTH_PENALTY_WEIGHTS[MetricKey.COMPLETION_RATE] ) if include_global_risk: - penalty += suppliers[MetricKey.BLACKLISTED] * 10 - score = max(0, min(100, round(100 - penalty, 2))) - if score >= 80: + penalty += suppliers[MetricKey.BLACKLISTED] * HEALTH_PENALTY_WEIGHTS[ + MetricKey.BLACKLISTED + ] + score = max( + HEALTH_SCORE_MIN, + min(HEALTH_SCORE_MAX, round(HEALTH_SCORE_MAX - penalty, 2)), + ) + if score >= HEALTHY_SCORE_THRESHOLD: level = HealthLevel.HEALTHY - elif score >= 60: + elif score >= ATTENTION_SCORE_THRESHOLD: level = HealthLevel.ATTENTION else: level = HealthLevel.CRITICAL @@ -801,13 +825,13 @@ class ReportService: for status, count in sorted(status_counts.items()): lines.append(f"- {status}：{count}") return { - "title": "打卡汇总", - "work_date": target_date.isoformat(), - "total": total, - "abnormal_total": abnormal_total, - "status_counts": status_counts, - "lines": lines, - "content": "\n".join(lines), + ReportResponseKey.TITLE: ReportTitle.ATTENDANCE_SUMMARY, + ReportResponseKey.WORK_DATE: target_date.isoformat(), + ReportResponseKey.TOTAL: total, + ReportResponseKey.ABNORMAL_TOTAL: abnormal_total, + ReportResponseKey.STATUS_COUNTS: status_counts, + ReportResponseKey.LINES: lines, + ReportResponseKey.CONTENT: "\n".join(lines), } def generate_work_report( @@ -833,14 +857,14 @@ class ReportService: ) lines = self._work_report_lines(title, start, end, metrics, risk_summary) report = { - "title": title, - "report_type": report_type, - "period_start": start.isoformat(), - "period_end": end.isoformat(), - "lines": lines, - "content": "\n".join(lines), - "metrics": metrics, - "risk_summary": risk_summary, + ReportResponseKey.TITLE: title, + ReportResponseKey.REPORT_TYPE: report_type, + ReportResponseKey.PERIOD_START: start.isoformat(), + ReportResponseKey.PERIOD_END: end.isoformat(), + ReportResponseKey.LINES: lines, + ReportResponseKey.CONTENT: "\n".join(lines), + ReportResponseKey.METRICS: metrics, + ReportResponseKey.RISK_SUMMARY: risk_summary, } record_data = None @@ -854,7 +878,7 @@ class ReportService: project_code=project_code, period_start=start, period_end=end, - content=report["content"], + content=report[ReportResponseKey.CONTENT], metrics=metrics, risk_summary=risk_summary, ) @@ -873,7 +897,7 @@ class ReportService: ) ) - return {"report": report, "data": record_data} + return {ReportResponseKey.REPORT: report, ReportResponseKey.DATA: record_data} def _resolve_period( self, @@ -932,19 +956,25 @@ class ReportService: *task_filters, ) return { - "projects_total": self._count(Project, *project_filters), - "active_projects": self._count( + WorkReportMetricKey.PROJECTS_TOTAL: self._count(Project, *project_filters), + WorkReportMetricKey.ACTIVE_PROJECTS: self._count( Project, Project.status.notin_(PROJECT_CLOSED_STATUSES), *project_filters, ), - "tasks_total": self._count(WorkTask, *task_filters), - "tasks_completed": completed_tasks, - "tasks_overdue": overdue_tasks, - "procurements_pending": self._count(Procurement, *procurement_filters), - "expenses_pending": self._count(Expense, *expense_filters), - "attendance_total": self._count(AttendanceRecord, *attendance_filters), - "open_risk_events": self._count(RiskEvent, *risk_filters), + WorkReportMetricKey.TASKS_TOTAL: self._count(WorkTask, *task_filters), + WorkReportMetricKey.TASKS_COMPLETED: completed_tasks, + WorkReportMetricKey.TASKS_OVERDUE: overdue_tasks, + WorkReportMetricKey.PROCUREMENTS_PENDING: self._count( + Procurement, + *procurement_filters, + ), + WorkReportMetricKey.EXPENSES_PENDING: self._count(Expense, *expense_filters), + WorkReportMetricKey.ATTENDANCE_TOTAL: self._count( + AttendanceRecord, + *attendance_filters, + ), + WorkReportMetricKey.OPEN_RISK_EVENTS: self._count(RiskEvent, *risk_filters), } def _work_report_lines( @@ -958,14 +988,20 @@ class ReportService: return [ f"- 报告：{title}", f"- 周期：{start.isoformat()} 至 {end.isoformat()}", - f"- 项目：总数 {metrics['projects_total']}，活跃 {metrics['active_projects']}", - f"- 任务：总数 {metrics['tasks_total']}，完成 {metrics['tasks_completed']}", - f"- 逾期任务：{metrics['tasks_overdue']}", - f"- 待处理采购：{metrics['procurements_pending']}", - f"- 待处理费用：{metrics['expenses_pending']}", - f"- 打卡记录：{metrics['attendance_total']}", - f"- 打开风险事件：{metrics['open_risk_events']}", - f"- 综合风险等级：{risk_summary['risk_level']}", + ( + f"- 项目：总数 {metrics[WorkReportMetricKey.PROJECTS_TOTAL]}，" + f"活跃 {metrics[WorkReportMetricKey.ACTIVE_PROJECTS]}" + ), + ( + f"- 任务：总数 {metrics[WorkReportMetricKey.TASKS_TOTAL]}，" + f"完成 {metrics[WorkReportMetricKey.TASKS_COMPLETED]}" + ), + f"- 逾期任务：{metrics[WorkReportMetricKey.TASKS_OVERDUE]}", + f"- 待处理采购：{metrics[WorkReportMetricKey.PROCUREMENTS_PENDING]}", + f"- 待处理费用：{metrics[WorkReportMetricKey.EXPENSES_PENDING]}", + f"- 打卡记录：{metrics[WorkReportMetricKey.ATTENDANCE_TOTAL]}", + f"- 打开风险事件：{metrics[WorkReportMetricKey.OPEN_RISK_EVENTS]}", + f"- 综合风险等级：{risk_summary[RiskSummaryKey.RISK_LEVEL]}", ] def push_report( @@ -975,5 +1011,8 @@ class ReportService: receive_id_type: str, actor: str, ) -> dict: - card = FeishuService.build_basic_card(report["title"], report["lines"]) + card = FeishuService.build_basic_card( + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], + ) return FeishuService(self.db).send_card(card, receive_id, receive_id_type, actor) diff --git a/app/modules/risk/constants.py b/app/modules/risk/constants.py new file mode 100644 index 0000000..5d57c8c --- /dev/null +++ b/app/modules/risk/constants.py @@ -0,0 +1,66 @@ +from enum import StrEnum + +from app.modules.business.constants import RiskLevel + + +class RiskSummaryKey(StrEnum): + RISK_LEVEL = "risk_level" + RISK_SCORE = "risk_score" + OVERDUE_TASKS = "overdue_tasks" + DELAYED_PROJECTS = "delayed_projects" + OVER_BUDGET_PROJECTS = "over_budget_projects" + FUND_RISKS = "fund_risks" + SUPPLIER_RISKS = "supplier_risks" + OPEN_EVENTS = "open_events" + + +class RiskGenerationResultKey(StrEnum): + CREATED = "created" + UPDATED = "updated" + SKIPPED = "skipped" + ITEMS = "items" + ACTION = "action" + RISK_EVENT = "risk_event" + + +class RiskGenerationAction(StrEnum): + CREATED = "created" + UPDATED = "updated" + SKIPPED = "skipped" + + +class RiskEventPayloadKey(StrEnum): + CODE = "code" + TITLE = "title" + RISK_TYPE = "risk_type" + RISK_LEVEL = "risk_level" + STATUS = "status" + SOURCE_DOMAIN = "source_domain" + SOURCE_RECORD_ID = "source_record_id" + PROJECT_CODE = "project_code" + OWNER = "owner" + DUE_DATE = "due_date" + DETECTED_AT = "detected_at" + DESCRIPTION = "description" + MITIGATION = "mitigation" + EVIDENCE = "evidence" + + +RISK_SCORE_WEIGHTS = { + RiskSummaryKey.OVERDUE_TASKS: 1, + RiskSummaryKey.DELAYED_PROJECTS: 3, + RiskSummaryKey.OVER_BUDGET_PROJECTS: 4, + RiskSummaryKey.FUND_RISKS: 5, + RiskSummaryKey.SUPPLIER_RISKS: 3, + RiskSummaryKey.OPEN_EVENTS: 2, +} +RISK_LEVEL_HIGH_THRESHOLD = 15 +RISK_LEVEL_MEDIUM_THRESHOLD = 5 + + +def risk_level_for_score(score: int) -> RiskLevel: + if score >= RISK_LEVEL_HIGH_THRESHOLD: + return RiskLevel.HIGH + if score >= RISK_LEVEL_MEDIUM_THRESHOLD: + return RiskLevel.MEDIUM + return RiskLevel.LOW diff --git a/app/modules/risk/routes.py b/app/modules/risk/routes.py index 7f68c29..7134c05 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.modules.risk.constants import RiskGenerationResultKey from app.modules.risk.service import RiskService router = APIRouter(dependencies=[Depends(require_api_key)]) @@ -15,27 +16,27 @@ def risk_summary(db: Session = Depends(get_db)) -> dict: @router.get("/overdue-tasks") def overdue_tasks(db: Session = Depends(get_db)) -> dict: - return {"items": RiskService(db).overdue_tasks()} + return {RiskGenerationResultKey.ITEMS: RiskService(db).overdue_tasks()} @router.get("/delayed-projects") def delayed_projects(db: Session = Depends(get_db)) -> dict: - return {"items": RiskService(db).delayed_projects()} + return {RiskGenerationResultKey.ITEMS: RiskService(db).delayed_projects()} @router.get("/over-budget-projects") def over_budget_projects(db: Session = Depends(get_db)) -> dict: - return {"items": RiskService(db).over_budget_projects()} + return {RiskGenerationResultKey.ITEMS: RiskService(db).over_budget_projects()} @router.get("/funds") def fund_risks(db: Session = Depends(get_db)) -> dict: - return {"items": RiskService(db).fund_risks()} + return {RiskGenerationResultKey.ITEMS: RiskService(db).fund_risks()} @router.get("/suppliers") def supplier_risks(db: Session = Depends(get_db)) -> dict: - return {"items": RiskService(db).supplier_risks()} + return {RiskGenerationResultKey.ITEMS: RiskService(db).supplier_risks()} @router.get("/events") @@ -44,7 +45,12 @@ def risk_events( status: str | None = None, db: Session = Depends(get_db), ) -> dict: - return {"items": RiskService(db).list_events(limit=limit, status_filter=status)} + return { + RiskGenerationResultKey.ITEMS: RiskService(db).list_events( + limit=limit, + status_filter=status, + ) + } @router.post("/events/generate") diff --git a/app/modules/risk/service.py b/app/modules/risk/service.py index 42f4c84..ed3276d 100644 --- a/app/modules/risk/service.py +++ b/app/modules/risk/service.py @@ -29,6 +29,14 @@ from app.modules.business.constants import ( ) from app.modules.business.models import FundAccount, Project, RiskEvent, Supplier, WorkTask from app.modules.business.service import serialize_model +from app.modules.risk.constants import ( + RISK_SCORE_WEIGHTS, + RiskGenerationAction, + RiskGenerationResultKey, + RiskEventPayloadKey, + RiskSummaryKey, + risk_level_for_score, +) class RiskService: @@ -97,31 +105,26 @@ class RiskService: external_open_events = [ item for item in open_events - if item.get("risk_type") not in GENERATED_RISK_EVENT_TYPES + if item.get(RiskEventPayloadKey.RISK_TYPE) not in GENERATED_RISK_EVENT_TYPES ] risk_score = ( - len(overdue_tasks) * 1 - + len(delayed_projects) * 3 - + len(over_budget_projects) * 4 - + len(fund_risks) * 5 - + len(supplier_risks) * 3 - + len(external_open_events) * 2 + len(overdue_tasks) * RISK_SCORE_WEIGHTS[RiskSummaryKey.OVERDUE_TASKS] + + len(delayed_projects) * RISK_SCORE_WEIGHTS[RiskSummaryKey.DELAYED_PROJECTS] + + len(over_budget_projects) + * RISK_SCORE_WEIGHTS[RiskSummaryKey.OVER_BUDGET_PROJECTS] + + len(fund_risks) * RISK_SCORE_WEIGHTS[RiskSummaryKey.FUND_RISKS] + + len(supplier_risks) * RISK_SCORE_WEIGHTS[RiskSummaryKey.SUPPLIER_RISKS] + + len(external_open_events) * RISK_SCORE_WEIGHTS[RiskSummaryKey.OPEN_EVENTS] ) - if risk_score >= 15: - level = RiskLevel.HIGH - elif risk_score >= 5: - level = RiskLevel.MEDIUM - else: - level = RiskLevel.LOW return { - "risk_level": level, - "risk_score": Decimal(risk_score), - "overdue_tasks": overdue_tasks, - "delayed_projects": delayed_projects, - "over_budget_projects": over_budget_projects, - "fund_risks": fund_risks, - "supplier_risks": supplier_risks, - "open_events": open_events, + RiskSummaryKey.RISK_LEVEL: risk_level_for_score(risk_score), + RiskSummaryKey.RISK_SCORE: Decimal(risk_score), + RiskSummaryKey.OVERDUE_TASKS: overdue_tasks, + RiskSummaryKey.DELAYED_PROJECTS: delayed_projects, + RiskSummaryKey.OVER_BUDGET_PROJECTS: over_budget_projects, + RiskSummaryKey.FUND_RISKS: fund_risks, + RiskSummaryKey.SUPPLIER_RISKS: supplier_risks, + RiskSummaryKey.OPEN_EVENTS: open_events, } def generate_events(self, actor: str = ActorValue.API) -> dict[str, Any]: @@ -135,25 +138,35 @@ class RiskService: for payload in payloads: record = self.db.execute( - select(RiskEvent).where(RiskEvent.code == payload["code"]) + select(RiskEvent).where(RiskEvent.code == payload[RiskEventPayloadKey.CODE]) ).scalar_one_or_none() if record is None: record = RiskEvent(**payload) self.db.add(record) self.db.flush() created += 1 - action = "created" + action = RiskGenerationAction.CREATED elif record.status in CLOSED_RISK_STATUSES: skipped += 1 - items.append({"action": "skipped", "risk_event": serialize_model(record)}) + items.append( + { + RiskGenerationResultKey.ACTION: RiskGenerationAction.SKIPPED, + RiskGenerationResultKey.RISK_EVENT: serialize_model(record), + } + ) continue else: for key, value in payload.items(): - if key != "code": + if key != RiskEventPayloadKey.CODE: setattr(record, key, value) updated += 1 - action = "updated" - items.append({"action": action, "risk_event": serialize_model(record)}) + action = RiskGenerationAction.UPDATED + items.append( + { + RiskGenerationResultKey.ACTION: action, + RiskGenerationResultKey.RISK_EVENT: serialize_model(record), + } + ) self.db.commit() AuditService(self.db).log( @@ -164,13 +177,18 @@ class RiskService: target_type=AuditTargetType.RISK_EVENTS, risk_level=AuditRiskLevel.MEDIUM, response_payload={ - "created": created, - "updated": updated, - "skipped": skipped, + RiskGenerationResultKey.CREATED: created, + RiskGenerationResultKey.UPDATED: updated, + RiskGenerationResultKey.SKIPPED: skipped, }, ) ) - return {"created": created, "updated": updated, "skipped": skipped, "items": items} + return { + RiskGenerationResultKey.CREATED: created, + RiskGenerationResultKey.UPDATED: updated, + RiskGenerationResultKey.SKIPPED: skipped, + RiskGenerationResultKey.ITEMS: items, + } def _build_event_payloads(self) -> list[dict[str, Any]]: payloads: list[dict[str, Any]] = [] @@ -191,22 +209,22 @@ class RiskService: for task in self.db.execute(stmt).scalars(): payloads.append( { - "code": f"RISK-TASK-OVERDUE-{task.id}", - "title": f"任务逾期：{task.title}", - "risk_type": RiskEventType.OVERDUE_TASK, - "risk_level": RiskLevel.MEDIUM, - "status": StatusValue.OPEN, - "source_domain": BusinessDomain.TASKS, - "source_record_id": str(task.id), - "project_code": task.project_code, - "owner": task.owner, - "due_date": task.due_date, - "detected_at": utc_now(), - "description": "任务已超过截止日期且未完成。", - "mitigation": ( + RiskEventPayloadKey.CODE: f"RISK-TASK-OVERDUE-{task.id}", + RiskEventPayloadKey.TITLE: f"任务逾期：{task.title}", + RiskEventPayloadKey.RISK_TYPE: RiskEventType.OVERDUE_TASK, + RiskEventPayloadKey.RISK_LEVEL: RiskLevel.MEDIUM, + RiskEventPayloadKey.STATUS: StatusValue.OPEN, + RiskEventPayloadKey.SOURCE_DOMAIN: BusinessDomain.TASKS, + RiskEventPayloadKey.SOURCE_RECORD_ID: str(task.id), + RiskEventPayloadKey.PROJECT_CODE: task.project_code, + RiskEventPayloadKey.OWNER: task.owner, + RiskEventPayloadKey.DUE_DATE: task.due_date, + RiskEventPayloadKey.DETECTED_AT: utc_now(), + RiskEventPayloadKey.DESCRIPTION: "任务已超过截止日期且未完成。", + RiskEventPayloadKey.MITIGATION: ( "请负责人更新进度、明确阻塞项并给出新的完成时间。" ), - "evidence": serialize_model(task), + RiskEventPayloadKey.EVIDENCE: serialize_model(task), } ) return payloads @@ -222,22 +240,22 @@ class RiskService: level = RiskLevel.HIGH if project.progress_percent < 80 else RiskLevel.MEDIUM payloads.append( { - "code": f"RISK-PROJECT-DELAY-{project.id}", - "title": f"项目延期：{project.name}", - "risk_type": RiskEventType.DELAYED_PROJECT, - "risk_level": level, - "status": StatusValue.OPEN, - "source_domain": BusinessDomain.PROJECTS, - "source_record_id": str(project.id), - "project_code": project.code, - "owner": project.owner, - "due_date": project.due_date, - "detected_at": utc_now(), - "description": "项目已超过计划截止日期且未进入完成状态。", - "mitigation": ( + RiskEventPayloadKey.CODE: f"RISK-PROJECT-DELAY-{project.id}", + RiskEventPayloadKey.TITLE: f"项目延期：{project.name}", + RiskEventPayloadKey.RISK_TYPE: RiskEventType.DELAYED_PROJECT, + RiskEventPayloadKey.RISK_LEVEL: level, + RiskEventPayloadKey.STATUS: StatusValue.OPEN, + RiskEventPayloadKey.SOURCE_DOMAIN: BusinessDomain.PROJECTS, + RiskEventPayloadKey.SOURCE_RECORD_ID: str(project.id), + RiskEventPayloadKey.PROJECT_CODE: project.code, + RiskEventPayloadKey.OWNER: project.owner, + RiskEventPayloadKey.DUE_DATE: project.due_date, + RiskEventPayloadKey.DETECTED_AT: utc_now(), + RiskEventPayloadKey.DESCRIPTION: "项目已超过计划截止日期且未进入完成状态。", + RiskEventPayloadKey.MITIGATION: ( "请项目负责人提交延期原因、资源需求和纠偏计划。" ), - "evidence": serialize_model(project), + RiskEventPayloadKey.EVIDENCE: serialize_model(project), } ) return payloads @@ -251,22 +269,22 @@ class RiskService: for project in self.db.execute(stmt).scalars(): payloads.append( { - "code": f"RISK-PROJECT-BUDGET-{project.id}", - "title": f"项目超预算：{project.name}", - "risk_type": RiskEventType.OVER_BUDGET_PROJECT, - "risk_level": RiskLevel.HIGH, - "status": StatusValue.OPEN, - "source_domain": BusinessDomain.PROJECTS, - "source_record_id": str(project.id), - "project_code": project.code, - "owner": project.owner, - "due_date": project.due_date, - "detected_at": utc_now(), - "description": "项目实际成本已超过预算。", - "mitigation": ( + RiskEventPayloadKey.CODE: f"RISK-PROJECT-BUDGET-{project.id}", + RiskEventPayloadKey.TITLE: f"项目超预算：{project.name}", + RiskEventPayloadKey.RISK_TYPE: RiskEventType.OVER_BUDGET_PROJECT, + RiskEventPayloadKey.RISK_LEVEL: RiskLevel.HIGH, + RiskEventPayloadKey.STATUS: StatusValue.OPEN, + RiskEventPayloadKey.SOURCE_DOMAIN: BusinessDomain.PROJECTS, + RiskEventPayloadKey.SOURCE_RECORD_ID: str(project.id), + RiskEventPayloadKey.PROJECT_CODE: project.code, + RiskEventPayloadKey.OWNER: project.owner, + RiskEventPayloadKey.DUE_DATE: project.due_date, + RiskEventPayloadKey.DETECTED_AT: utc_now(), + RiskEventPayloadKey.DESCRIPTION: "项目实际成本已超过预算。", + RiskEventPayloadKey.MITIGATION: ( "请复核预算科目、冻结非必要采购并补充审批依据。" ), - "evidence": serialize_model(project), + RiskEventPayloadKey.EVIDENCE: serialize_model(project), } ) return payloads @@ -277,21 +295,21 @@ class RiskService: for account in self.db.execute(stmt).scalars(): payloads.append( { - "code": f"RISK-FUND-{account.id}", - "title": f"资金低于安全线：{account.name}", - "risk_type": RiskEventType.FUND_SAFETY_LINE, - "risk_level": RiskLevel.HIGH, - "status": StatusValue.OPEN, - "source_domain": BusinessDomain.FUND_ACCOUNTS, - "source_record_id": str(account.id), - "owner": None, - "detected_at": utc_now(), - "description": "账户当前余额低于设置的安全线。", - "mitigation": ( + RiskEventPayloadKey.CODE: f"RISK-FUND-{account.id}", + RiskEventPayloadKey.TITLE: f"资金低于安全线：{account.name}", + RiskEventPayloadKey.RISK_TYPE: RiskEventType.FUND_SAFETY_LINE, + RiskEventPayloadKey.RISK_LEVEL: RiskLevel.HIGH, + RiskEventPayloadKey.STATUS: StatusValue.OPEN, + RiskEventPayloadKey.SOURCE_DOMAIN: BusinessDomain.FUND_ACCOUNTS, + RiskEventPayloadKey.SOURCE_RECORD_ID: str(account.id), + RiskEventPayloadKey.OWNER: None, + RiskEventPayloadKey.DETECTED_AT: utc_now(), + RiskEventPayloadKey.DESCRIPTION: "账户当前余额低于设置的安全线。", + RiskEventPayloadKey.MITIGATION: ( "请财务确认收付款计划，" "并优先处理关键项目资金安排。" ), - "evidence": serialize_model(account), + RiskEventPayloadKey.EVIDENCE: serialize_model(account), } ) return payloads @@ -310,20 +328,20 @@ class RiskService: ) payloads.append( { - "code": f"RISK-SUPPLIER-{supplier.id}", - "title": f"供应商风险：{supplier.name}", - "risk_type": RiskEventType.SUPPLIER_RISK, - "risk_level": level, - "status": StatusValue.OPEN, - "source_domain": BusinessDomain.SUPPLIERS, - "source_record_id": str(supplier.id), - "owner": supplier.contact, - "detected_at": utc_now(), - "description": "供应商风险等级或黑名单状态需要关注。", - "mitigation": ( + RiskEventPayloadKey.CODE: f"RISK-SUPPLIER-{supplier.id}", + RiskEventPayloadKey.TITLE: f"供应商风险：{supplier.name}", + RiskEventPayloadKey.RISK_TYPE: RiskEventType.SUPPLIER_RISK, + RiskEventPayloadKey.RISK_LEVEL: level, + RiskEventPayloadKey.STATUS: StatusValue.OPEN, + RiskEventPayloadKey.SOURCE_DOMAIN: BusinessDomain.SUPPLIERS, + RiskEventPayloadKey.SOURCE_RECORD_ID: str(supplier.id), + RiskEventPayloadKey.OWNER: supplier.contact, + RiskEventPayloadKey.DETECTED_AT: utc_now(), + RiskEventPayloadKey.DESCRIPTION: "供应商风险等级或黑名单状态需要关注。", + RiskEventPayloadKey.MITIGATION: ( "请采购负责人复核供应商准入、履约和替代方案。" ), - "evidence": serialize_model(supplier), + RiskEventPayloadKey.EVIDENCE: serialize_model(supplier), } ) return payloads diff --git a/scripts/sample_requests.http b/scripts/sample_requests.http index ac8c121..962778e 100644 --- a/scripts/sample_requests.http +++ b/scripts/sample_requests.http @@ -1,11 +1,13 @@ +@apiKey = {{$dotenv API_KEY}} + ### Health GET http://127.0.0.1:8010/api/v1/health -X-API-Key: change-me +X-API-Key: {{apiKey}} ### Create project POST http://127.0.0.1:8010/api/v1/business/projects Content-Type: application/json -X-API-Key: change-me +X-API-Key: {{apiKey}} { "actor": "demo", @@ -23,12 +25,12 @@ X-API-Key: change-me ### Daily brief GET http://127.0.0.1:8010/api/v1/reports/daily-brief -X-API-Key: change-me +X-API-Key: {{apiKey}} ### AI ask POST http://127.0.0.1:8010/api/v1/ai/ask Content-Type: application/json -X-API-Key: change-me +X-API-Key: {{apiKey}} { "actor": "demo", diff --git a/scripts/verify_smoke.py b/scripts/verify_smoke.py index 91ac3e2..d51aa67 100644 --- a/scripts/verify_smoke.py +++ b/scripts/verify_smoke.py @@ -19,14 +19,16 @@ os.environ["SCHEDULER_ENABLED"] = "false" from fastapi.testclient import TestClient +from app.core.constants import HttpHeader from app.core.database import Base, engine from app.main import app +from app.modules.business.constants import BusinessField, StatusValue def request(method: str, url: str, **kwargs): client = TestClient(app) headers = kwargs.pop("headers", {}) - headers.setdefault("X-API-Key", "test-key") + headers.setdefault(HttpHeader.X_API_KEY, "test-key") response = getattr(client, method)(url, headers=headers, **kwargs) response.raise_for_status() return response @@ -44,7 +46,7 @@ try: "code": "P-VERIFY-001", "name": "Verify Project", "owner": "tester", - "status": "执行中", + BusinessField.STATUS: StatusValue.RUNNING_CN, "budget_amount": 1000, "actual_amount": 200, }, diff --git a/security_fix_ignore.yml b/security_fix_ignore.yml new file mode 100644 index 0000000..d9f37f0 --- /dev/null +++ b/security_fix_ignore.yml @@ -0,0 +1,11 @@ +ignored_findings: + - id: P0_LOCAL_ENV_CONTAINS_RUNTIME_SECRETS + path: .env + reason: "User explicitly excluded local .env secret cleanup from this remediation pass." + status: ignored + 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." diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 736ec47..5f77732 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -131,6 +131,19 @@ def test_feishu_webhook_routes_message_event() -> None: assert AUDIT_REDACTED_VALUE in audit_payload +def test_feishu_webhook_challenge_uses_event_service_verification() -> None: + response = client.post( + "/api/v1/integrations/feishu/webhook", + json={ + "challenge": "challenge-token", + "token": "test-feishu-token", + }, + ) + + assert response.status_code == 200 + assert response.json()["challenge"] == "challenge-token" + + def test_api_key_and_feishu_webhook_fail_closed(monkeypatch) -> None: monkeypatch.setenv("API_KEY", "") get_settings.cache_clear()