From db751f03b4790b26ae47cf8c8ab4bd69ac6cf35e Mon Sep 17 00:00:00 2001 From: JiuContinent Date: Wed, 15 Jul 2026 16:36:42 +0800 Subject: [PATCH] =?UTF-8?q?```=20refactor(Dockerfile):=20=E4=BD=BF?= =?UTF-8?q?=E7=94=A8requirements.txt=E6=9B=BF=E4=BB=A3=E7=A1=AC=E7=BC=96?= =?UTF-8?q?=E7=A0=81=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将Dockerfile中的硬编码pip包列表替换为通过requirements.txt文件安装， 提高依赖管理的灵活性和可维护性。 feat(scheduling): 移除内置APScheduler，采用独立调度系统 移除app/core/background/scheduler.py中原来的APScheduler实现， 改为使用新的应用级调度系统app.application.scheduling。 refactor(task_queue): 调整任务队列模块结构和导入路径 将任务队列相关常量从app.core.background.task_queue.constants迁移至 app.tasks.constants，并更新所有相关导入路径和引用。 refactor(events): 将事件服务重构为独立的应用层组件 将事件分发逻辑从核心层迁移到应用层，使用app.application.events.EventDispatchService 替代原有的app.modules.events.services.EventService。 feat(ai_memory): 增强AI记忆自动写入的安全策略 新增ai_memory_blocked_content_terms配置项用于阻止敏感内容， 添加TTL过期机制控制自动写入条目的生命周期。 fix(security): 强化生产环境安全验证机制 增加model_validator确保生产环境中数据库连接、API密钥、CORS设置等 关键安全配置符合要求。 feat(risks): 优化风险事件操作动作的外键约束 为RiskEventAction模型的风险事件ID字段添加外键约束， 防止孤立记录并增强数据完整性。 refactor(audit): 优化审计服务方法命名和事务处理 将AuditService的log方法重命名为record以反映其阶段行为， 并调整事务提交时机以提高性能。 feat(events): 增强领域事件并发处理和响应模型 添加事件锁定机制防止重复处理，更新API响应模型以提供 更准确的数据类型定义。 ``` --- Dockerfile | 20 +- .../202607150001_architecture_hardening.py | 126 +++ app/application/__init__.py | 1 + app/application/delivery/__init__.py | 3 + .../delivery/reports.py} | 29 +- app/application/events/__init__.py | 3 + .../events}/dispatch.py | 89 ++- .../events}/handlers.py | 4 +- app/application/feishu/__init__.py | 4 + app/application/feishu/commands.py | 201 +++++ app/application/feishu/delivery.py | 34 + app/{modules => application}/feishu/events.py | 23 +- app/application/feishu/handlers/__init__.py | 9 + app/application/feishu/handlers/finance.py | 141 ++++ app/application/feishu/handlers/market.py | 243 ++++++ app/application/feishu/handlers/rules.py | 192 +++++ app/application/feishu/results.py | 29 + app/application/pipelines/__init__.py | 4 + .../pipelines/lifecycle.py} | 6 +- .../pipelines/market.py} | 3 +- app/application/scheduling/__init__.py | 3 + .../scheduling}/scheduler.py | 9 +- app/core/background/task_queue/__init__.py | 2 +- app/core/background/task_queue/events.py | 6 +- app/core/background/task_queue/legacy.py | 2 +- app/core/background/task_queue/lifecycle.py | 4 +- app/core/background/task_queue/market.py | 2 +- app/core/background/task_queue/reports.py | 7 +- app/core/background/task_queue/risk.py | 2 +- app/core/config/settings.py | 43 +- app/core/http/responses.py | 12 + app/core/security/__init__.py | 8 + app/core/security/operation_guard.py | 4 +- app/core/security/operation_policy.py | 20 + app/main.py | 4 +- app/modules/ai_memory/constants.py | 1 + app/modules/ai_memory/service.py | 45 +- app/modules/audit/service.py | 11 +- app/modules/business/models/risks.py | 7 +- app/modules/events/constants.py | 1 + app/modules/events/models.py | 2 +- app/modules/events/routes.py | 22 +- app/modules/events/schemas.py | 7 +- app/modules/events/services/query.py | 52 +- app/modules/events/services/service.py | 6 +- app/modules/feishu/commands.py | 732 ------------------ app/modules/feishu/long_connection.py | 2 +- app/modules/feishu/routes.py | 10 +- .../legacy_mysql/services/project_sync.py | 14 +- .../legacy_mysql/services/task_sync.py | 14 +- app/modules/observability/routes.py | 9 +- app/modules/observability/service.py | 6 +- app/modules/reports/routes.py | 13 +- app/modules/reports/services/enterprise.py | 6 +- app/modules/reports/services/push_runs.py | 8 +- app/modules/reports/services/service.py | 2 - app/modules/reports/services/work_reports.py | 10 +- app/modules/risk/services/actions.py | 8 +- app/modules/risk/services/generation.py | 6 +- app/modules/workflows/models.py | 15 +- app/modules/workflows/routes.py | 5 +- app/modules/workflows/schemas.py | 3 +- app/modules/workflows/service.py | 8 +- .../task_queue => tasks}/constants.py | 0 app/tasks/events.py | 4 +- app/tasks/lifecycle.py | 4 +- app/tasks/market.py | 4 +- app/tasks/reports.py | 5 +- app/tools/run_scheduler.py | 2 +- environment.yml | 17 +- requirements.txt | 16 + tests/test_architecture_hardening.py | 75 ++ tests/test_smoke.py | 104 ++- 73 files changed, 1615 insertions(+), 933 deletions(-) create mode 100644 alembic/versions/202607150001_architecture_hardening.py create mode 100644 app/application/__init__.py create mode 100644 app/application/delivery/__init__.py rename app/{modules/reports/services/delivery.py => application/delivery/reports.py} (83%) create mode 100644 app/application/events/__init__.py rename app/{modules/events/services => application/events}/dispatch.py (64%) rename app/{modules/events/services => application/events}/handlers.py (98%) create mode 100644 app/application/feishu/__init__.py create mode 100644 app/application/feishu/commands.py create mode 100644 app/application/feishu/delivery.py rename app/{modules => application}/feishu/events.py (80%) create mode 100644 app/application/feishu/handlers/__init__.py create mode 100644 app/application/feishu/handlers/finance.py create mode 100644 app/application/feishu/handlers/market.py create mode 100644 app/application/feishu/handlers/rules.py create mode 100644 app/application/feishu/results.py create mode 100644 app/application/pipelines/__init__.py rename app/{modules/reports/lifecycle_pipeline.py => application/pipelines/lifecycle.py} (97%) rename app/{modules/market/pipeline.py => application/pipelines/market.py} (98%) create mode 100644 app/application/scheduling/__init__.py rename app/{core/background => application/scheduling}/scheduler.py (97%) create mode 100644 app/core/http/responses.py create mode 100644 app/core/security/operation_policy.py delete mode 100644 app/modules/feishu/commands.py rename app/{core/background/task_queue => tasks}/constants.py (100%) create mode 100644 requirements.txt create mode 100644 tests/test_architecture_hardening.py diff --git a/Dockerfile b/Dockerfile index bce049c..03469c5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,24 +5,8 @@ ENV PYTHONUNBUFFERED=1 WORKDIR /app -COPY environment.yml /app/environment.yml -RUN pip install --no-cache-dir \ - fastapi==0.115.6 \ - "uvicorn[standard]==0.34.0" \ - sqlalchemy==2.0.36 \ - pymysql==1.1.1 \ - "psycopg[binary]==3.2.3" \ - pydantic-settings==2.7.1 \ - python-dotenv==1.0.1 \ - alembic==1.14.0 \ - httpx==0.28.1 \ - lark-oapi==1.6.8 \ - apscheduler==3.10.4 \ - redis==5.2.1 \ - celery==5.4.0 \ - cryptography==44.0.0 \ - pandas==2.2.3 \ - pillow==11.0.0 +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt COPY alembic.ini /app/alembic.ini COPY alembic /app/alembic diff --git a/alembic/versions/202607150001_architecture_hardening.py b/alembic/versions/202607150001_architecture_hardening.py new file mode 100644 index 0000000..8bf9672 --- /dev/null +++ b/alembic/versions/202607150001_architecture_hardening.py @@ -0,0 +1,126 @@ +"""Harden event, workflow, and owned-ledger constraints. + +Revision ID: 202607150001 +Revises: 202607120004 +Create Date: 2026-07-15 +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "202607150001" +down_revision = "202607120004" +branch_labels = None +depends_on = None + + +def _scalar_count(sql: str) -> int: + return int(op.get_bind().execute(sa.text(sql)).scalar() or 0) + + +def _assert_owned_relations_are_consistent() -> None: + orphan_risk_actions = _scalar_count( + """ + SELECT COUNT(*) + FROM risk_event_actions action + LEFT JOIN risk_events event ON event.id = action.risk_event_id + WHERE event.id IS NULL + """ + ) + if orphan_risk_actions: + raise RuntimeError( + "Cannot add risk-event foreign key: orphan risk_event_actions rows exist" + ) + + orphan_workflow_actions = _scalar_count( + """ + SELECT COUNT(*) + FROM workflow_actions action + LEFT JOIN workflow_instances workflow ON workflow.code = action.workflow_code + WHERE workflow.code IS NULL + """ + ) + if orphan_workflow_actions: + raise RuntimeError( + "Cannot add workflow foreign key: orphan workflow_actions rows exist" + ) + + duplicate_workflows = _scalar_count( + """ + SELECT COUNT(*) + FROM ( + SELECT workflow_type, aggregate_type, aggregate_id + FROM workflow_instances + WHERE aggregate_id IS NOT NULL + GROUP BY workflow_type, aggregate_type, aggregate_id + HAVING COUNT(*) > 1 + ) duplicates + """ + ) + if duplicate_workflows: + raise RuntimeError( + "Cannot add workflow uniqueness constraint: duplicate aggregate workflows exist" + ) + + +def upgrade() -> None: + _assert_owned_relations_are_consistent() + op.execute("UPDATE domain_events SET max_attempts = 3 WHERE max_attempts IS NULL") + + with op.batch_alter_table("domain_events") as batch_op: + batch_op.alter_column( + "max_attempts", + existing_type=sa.Integer(), + nullable=False, + server_default=sa.text("3"), + ) + + with op.batch_alter_table("risk_event_actions") as batch_op: + batch_op.create_foreign_key( + "fk_risk_event_actions_risk_event_id", + "risk_events", + ["risk_event_id"], + ["id"], + ondelete="RESTRICT", + ) + + with op.batch_alter_table("workflow_actions") as batch_op: + batch_op.create_foreign_key( + "fk_workflow_actions_workflow_code", + "workflow_instances", + ["workflow_code"], + ["code"], + ondelete="RESTRICT", + ) + + with op.batch_alter_table("workflow_instances") as batch_op: + batch_op.create_unique_constraint( + "uq_workflow_aggregate", + ["workflow_type", "aggregate_type", "aggregate_id"], + ) + + +def downgrade() -> None: + with op.batch_alter_table("workflow_instances") as batch_op: + batch_op.drop_constraint("uq_workflow_aggregate", type_="unique") + + with op.batch_alter_table("workflow_actions") as batch_op: + batch_op.drop_constraint( + "fk_workflow_actions_workflow_code", + type_="foreignkey", + ) + + with op.batch_alter_table("risk_event_actions") as batch_op: + batch_op.drop_constraint( + "fk_risk_event_actions_risk_event_id", + type_="foreignkey", + ) + + with op.batch_alter_table("domain_events") as batch_op: + batch_op.alter_column( + "max_attempts", + existing_type=sa.Integer(), + nullable=True, + server_default=None, + ) diff --git a/app/application/__init__.py b/app/application/__init__.py new file mode 100644 index 0000000..c49a6e5 --- /dev/null +++ b/app/application/__init__.py @@ -0,0 +1 @@ +"""Application-level orchestration across domain modules and adapters.""" diff --git a/app/application/delivery/__init__.py b/app/application/delivery/__init__.py new file mode 100644 index 0000000..b47cd38 --- /dev/null +++ b/app/application/delivery/__init__.py @@ -0,0 +1,3 @@ +from app.application.delivery.reports import ReportDeliveryService + +__all__ = ["ReportDeliveryService"] diff --git a/app/modules/reports/services/delivery.py b/app/application/delivery/reports.py similarity index 83% rename from app/modules/reports/services/delivery.py rename to app/application/delivery/reports.py index 1875097..1cf0c2e 100644 --- a/app/modules/reports/services/delivery.py +++ b/app/application/delivery/reports.py @@ -1,4 +1,4 @@ - +from sqlalchemy.orm import Session from app.modules.audit.constants import AuditAction, AuditSource from app.modules.audit.schemas import AuditLogCreate @@ -16,9 +16,16 @@ from app.modules.reports.constants import ( ReportResponseKey, ) from app.modules.reports.chart import lifecycle_chart_alt, render_lifecycle_chart +from app.modules.reports.services import ReportService -class ReportDeliveryMixin: +class ReportDeliveryService: + """Deliver generated reports without coupling report generation to Feishu.""" + + def __init__(self, db: Session): + self.db = db + self.reports = ReportService(db) + def push_report( self, report: dict, @@ -30,9 +37,9 @@ class ReportDeliveryMixin: report_type = str(report.get(ReportResponseKey.REPORT_TYPE) or report.get("type") or "report") title = report.get(ReportResponseKey.TITLE) push_run = ( - self._get_push_run(push_run_code) + self.reports._get_push_run(push_run_code) if push_run_code - else self.create_push_run( + else self.reports.create_push_run( report_type=report_type, title=title, receive_id=receive_id, @@ -59,12 +66,13 @@ class ReportDeliveryMixin: ) result = feishu.send_card(card, receive_id, receive_id_type, actor) except Exception as exc: - failed_run = self.update_push_run( + failed_run = self.reports.update_push_run( push_run.code, ReportPushStatus.FAILED, error_message=str(exc), + commit=False, ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.REPORT_PUSH_FAILED, source=EventSource.REPORTS, aggregate_type=EventAggregateType.REPORT_PUSH_RUN, @@ -77,14 +85,16 @@ class ReportDeliveryMixin: }, idempotency_key=f"report-push:{failed_run.code}:{failed_run.status}", ) + self.db.commit() raise - success_run = self.update_push_run( + success_run = self.reports.update_push_run( push_run.code, ReportPushStatus.SUCCESS, provider_response=result, sent=True, + commit=False, ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.REPORT_PUSH_SUCCEEDED, source=EventSource.REPORTS, aggregate_type=EventAggregateType.REPORT_PUSH_RUN, @@ -96,7 +106,7 @@ class ReportDeliveryMixin: }, idempotency_key=f"report-push:{success_run.code}:{success_run.status}", ) - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.REPORTS, @@ -105,4 +115,5 @@ class ReportDeliveryMixin: response_payload={"status": ReportPushStatus.SUCCESS}, ) ) + self.db.commit() return result diff --git a/app/application/events/__init__.py b/app/application/events/__init__.py new file mode 100644 index 0000000..a58e352 --- /dev/null +++ b/app/application/events/__init__.py @@ -0,0 +1,3 @@ +from app.application.events.dispatch import EventDispatchService + +__all__ = ["EventDispatchService"] diff --git a/app/modules/events/services/dispatch.py b/app/application/events/dispatch.py similarity index 64% rename from app/modules/events/services/dispatch.py rename to app/application/events/dispatch.py index 51f8924..9f870ab 100644 --- a/app/modules/events/services/dispatch.py +++ b/app/application/events/dispatch.py @@ -5,6 +5,7 @@ from uuid import uuid4 from fastapi import HTTPException, status from sqlalchemy import or_, select +from app.application.events.handlers import EventHandlerMixin from app.core.config import get_settings from app.core.constants import ActorValue from app.core.http.pagination import bounded_limit @@ -23,23 +24,42 @@ from app.modules.events.constants import ( EventStatus, ) from app.modules.events.models import DomainEvent +from app.modules.events.services import EventService from app.modules.events.services.serialization import _serialize_event -class EventDispatchMixin: - def dispatch_event(self, event_id: str, worker_id: str | None = None) -> DomainEvent: - record = self.get_event(event_id) +class EventDispatchService(EventHandlerMixin): + """Claim and dispatch outbox events at the application boundary.""" + + def __init__(self, db: Any): + self.db = db + self.events = EventService(db) + + def get_event(self, event_id: str) -> DomainEvent: + return self.events.get_event(event_id) + + def dispatch_event( + self, + event_id: str, + worker_id: str | None = None, + preclaimed: bool = False, + ) -> DomainEvent: + lock_owner = worker_id or f"api:{uuid4().hex}" + record = ( + self.get_event(event_id) + if preclaimed + else self._claim_event(event_id, lock_owner) + ) if record.status == EventStatus.PROCESSED: return record - if not self._can_attempt(record): + if preclaimed and record.locked_by != lock_owner: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=EventErrorDetail.EVENT_LOCKED, + ) + if not preclaimed and record.locked_by != lock_owner: return record settings = get_settings() - now = utc_now() - lock_owner = worker_id or f"api:{uuid4().hex}" - record.locked_by = lock_owner - record.locked_until = now + timedelta(seconds=settings.event_dispatch_lock_seconds) - record.status = EventStatus.PENDING - record.attempts += 1 try: self._handle_event(record) except Exception as exc: @@ -93,11 +113,26 @@ class EventDispatchMixin: ) .order_by(DomainEvent.id.asc()) .limit(bounded_limit(limit)) + .with_for_update(skip_locked=True) ) records = list(self.db.execute(stmt).scalars()) lock_owner = worker_id or f"worker:{uuid4().hex}" + locked_until = now + timedelta( + seconds=get_settings().event_dispatch_lock_seconds + ) + for record in records: + record.locked_by = lock_owner + record.locked_until = locked_until + record.attempts += 1 + self.db.commit() return [ - _serialize_event(self.dispatch_event(record.event_id, worker_id=lock_owner)) + _serialize_event( + self.dispatch_event( + record.event_id, + worker_id=lock_owner, + preclaimed=True, + ) + ) for record in records ] @@ -140,6 +175,38 @@ class EventDispatchMixin: def _can_attempt(self, record: DomainEvent) -> bool: return record.attempts < self._max_attempts(record) + def _claim_event(self, event_id: str, lock_owner: str) -> DomainEvent: + now = utc_now() + record = self.db.execute( + select(DomainEvent) + .where(DomainEvent.event_id == event_id) + .with_for_update(skip_locked=True) + ).scalar_one_or_none() + if record is None: + self.db.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=EventErrorDetail.EVENT_LOCKED, + ) + if record.status == EventStatus.PROCESSED or not self._can_attempt(record): + self.db.commit() + return record + if record.locked_until is not None and record.locked_until > now: + self.db.commit() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=EventErrorDetail.EVENT_LOCKED, + ) + record.locked_by = lock_owner + record.locked_until = now + timedelta( + seconds=get_settings().event_dispatch_lock_seconds + ) + record.status = EventStatus.PENDING + record.attempts += 1 + self.db.commit() + self.db.refresh(record) + return record + @staticmethod def _max_attempts(record: DomainEvent) -> int: return record.max_attempts or get_settings().event_dispatch_max_attempts diff --git a/app/modules/events/services/handlers.py b/app/application/events/handlers.py similarity index 98% rename from app/modules/events/services/handlers.py rename to app/application/events/handlers.py index e5fa801..54dbea4 100644 --- a/app/modules/events/services/handlers.py +++ b/app/application/events/handlers.py @@ -1,5 +1,3 @@ - - from app.modules.events.constants import ( EventAggregateType, EventPayloadKey, @@ -54,6 +52,7 @@ class EventHandlerMixin: action=action or record.event_type, actor=record.actor, payload=payload, + commit=False, ) def _handle_report_event(self, record: DomainEvent) -> None: @@ -125,4 +124,5 @@ class EventHandlerMixin: action=record.event_type, actor=record.actor, payload=record.payload or {}, + commit=False, ) diff --git a/app/application/feishu/__init__.py b/app/application/feishu/__init__.py new file mode 100644 index 0000000..f3eade6 --- /dev/null +++ b/app/application/feishu/__init__.py @@ -0,0 +1,4 @@ +from app.application.feishu.commands import FeishuCommandService +from app.application.feishu.events import FeishuEventService + +__all__ = ["FeishuCommandService", "FeishuEventService"] diff --git a/app/application/feishu/commands.py b/app/application/feishu/commands.py new file mode 100644 index 0000000..8c86888 --- /dev/null +++ b/app/application/feishu/commands.py @@ -0,0 +1,201 @@ +import json +import re +from typing import Any + +from sqlalchemy.orm import Session + +from app.application.feishu.delivery import send_card_if_configured, send_text_if_configured +from app.application.feishu.handlers import ( + handle_finance_command, + handle_market_command, + handle_rule_command, +) +from app.application.feishu.results import command_result +from app.core.constants import ActorValue +from app.modules.ai_agent.constants import AIResponseKey +from app.modules.ai_agent.service import AIService +from app.modules.audit.constants import AuditSource +from app.modules.feishu.constants import ( + FEISHU_AI_REPLY_TITLE, + FEISHU_MENTION_PATTERN, + FEISHU_ZERO_WIDTH_SPACE, + FeishuCommandKey, + FeishuCommandName, + FeishuPayloadKey, + FeishuReplyType, +) +from app.modules.feishu.service import FeishuService +from app.modules.reports.constants import ReportResponseKey +from app.modules.reports.services import ReportService + +DAILY_REPORT_KEYWORDS = ("日报", "晨报", "经营日报", "经营晨报") +PROJECT_WEEKLY_KEYWORDS = ("周报", "项目周报") +ATTENDANCE_KEYWORDS = ("打卡", "考勤", "attendance") +RISK_KEYWORDS = ("风险", "预警", "risk") +AI_COMMAND_PREFIXES = ("问 ", "ai ", "AI ", "/ask ") +DEFAULT_AI_PROMPT = "请说明你能做什么。" + + +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(FeishuPayloadKey.TEXT) or content.get(FeishuPayloadKey.CONTENT) or "" + ) + if not isinstance(content, str): + return "" + try: + data = json.loads(content) + except json.JSONDecodeError: + return content + if isinstance(data, dict): + 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(FEISHU_MENTION_PATTERN, "", text or "") + text = text.replace(FEISHU_ZERO_WIDTH_SPACE, "") + return text.strip() + + +class FeishuCommandService: + """Route Feishu text commands to focused application handlers.""" + + def __init__(self, db: Session): + self.db = db + self.feishu = FeishuService(db) + + def extract_event_command(self, payload: dict[str, Any]) -> dict[str, Any] | None: + 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(FeishuPayloadKey.CONTENT))) + if not text: + return None + 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), + FeishuCommandKey.ACTOR: actor, + } + + def handle_text( + self, + text: str, + chat_id: str | None = None, + actor: str = ActorValue.FEISHU, + auto_reply: bool = True, + ) -> dict[str, Any]: + command_text = _clean_command_text(text) + lowered = command_text.lower() + + for handler in ( + handle_rule_command, + handle_finance_command, + handle_market_command, + ): + result = handler( + self.db, + self.feishu, + command_text, + chat_id, + actor, + auto_reply, + ) + if result is not None: + return result + + report_result = self._handle_report_command(command_text, chat_id, actor, auto_reply) + if report_result is not None: + return report_result + return self._handle_ai_command(command_text, lowered, chat_id, actor, auto_reply) + + def _handle_report_command( + self, + command_text: str, + chat_id: str | None, + actor: str, + auto_reply: bool, + ) -> dict[str, Any] | None: + service = ReportService(self.db) + if any(keyword in command_text for keyword in DAILY_REPORT_KEYWORDS): + command = FeishuCommandName.DAILY_BRIEF + report = service.daily_brief() + elif any(keyword in command_text for keyword in PROJECT_WEEKLY_KEYWORDS): + command = FeishuCommandName.PROJECT_WEEKLY + report = service.project_weekly() + elif any(keyword in command_text for keyword in ATTENDANCE_KEYWORDS): + command = FeishuCommandName.ATTENDANCE_SUMMARY + report = service.attendance_summary() + elif any(keyword in command_text for keyword in RISK_KEYWORDS): + command = FeishuCommandName.RISK_SUMMARY + report = service.risk_progress() + else: + return None + response = ( + send_card_if_configured( + self.feishu, + chat_id, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], + actor, + ) + if auto_reply + else None + ) + return command_result( + command, + FeishuReplyType.CARD, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.CONTENT], + response, + report[ReportResponseKey.LINES], + ) + + def _handle_ai_command( + self, + command_text: str, + lowered: str, + chat_id: str | None, + actor: str, + auto_reply: bool, + ) -> dict[str, Any]: + prompt = command_text + for prefix in AI_COMMAND_PREFIXES: + if command_text.startswith(prefix): + prompt = command_text[len(prefix) :].strip() + break + if not prompt: + prompt = DEFAULT_AI_PROMPT + ai_result = AIService(self.db).ask( + prompt, + context={}, + actor=actor, + source=AuditSource.FEISHU, + ) + content = ai_result[AIResponseKey.ANSWER] + is_explicit_ai = any( + command_text.startswith(prefix) or lowered.startswith(prefix) + for prefix in AI_COMMAND_PREFIXES + ) + response = ( + send_text_if_configured(self.feishu, chat_id, content, actor) if auto_reply else None + ) + return command_result( + FeishuCommandName.AI_ASK if is_explicit_ai else FeishuCommandName.FALLBACK_AI, + FeishuReplyType.TEXT, + FEISHU_AI_REPLY_TITLE, + content, + response, + ) diff --git a/app/application/feishu/delivery.py b/app/application/feishu/delivery.py new file mode 100644 index 0000000..0815615 --- /dev/null +++ b/app/application/feishu/delivery.py @@ -0,0 +1,34 @@ +from typing import Any + +from app.core.config import get_settings +from app.modules.feishu.service import FeishuService + + +def send_text_if_configured( + feishu: FeishuService, + chat_id: str | None, + text: str, + actor: str, +) -> dict[str, Any] | None: + """Send a text reply only when Feishu credentials are configured.""" + + settings = get_settings() + if not (settings.feishu_app_id and settings.feishu_app_secret): + return None + return feishu.send_text(text, receive_id=chat_id, actor=actor) + + +def send_card_if_configured( + feishu: FeishuService, + chat_id: str | None, + title: str, + lines: list[str], + actor: str, +) -> dict[str, Any] | None: + """Send a basic card only when Feishu credentials are configured.""" + + settings = get_settings() + if not (settings.feishu_app_id and settings.feishu_app_secret): + return None + card = FeishuService.build_basic_card(title, lines) + return feishu.send_card(card, receive_id=chat_id, actor=actor) diff --git a/app/modules/feishu/events.py b/app/application/feishu/events.py similarity index 80% rename from app/modules/feishu/events.py rename to app/application/feishu/events.py index 8934a87..e8438c4 100644 --- a/app/modules/feishu/events.py +++ b/app/application/feishu/events.py @@ -3,10 +3,10 @@ from typing import Any from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session +from app.application.feishu.commands import FeishuCommandService from app.core.constants import ActorValue 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 ( FeishuCommandKey, FeishuEventReceiptKey, @@ -60,7 +60,7 @@ class FeishuEventService: if event_identity else None ), - request_payload=payload, + request_payload=_audit_event_metadata(payload), response_payload={FeishuResponseKey.ACCEPTED: True}, ) ) @@ -95,6 +95,25 @@ class FeishuEventService: return True +def _audit_event_metadata(payload: dict[str, Any]) -> dict[str, Any]: + """Keep webhook audit evidence without storing message content or tokens.""" + + header = payload.get(FeishuPayloadKey.HEADER) or {} + event = payload.get(FeishuPayloadKey.EVENT) or {} + message = event.get(FeishuPayloadKey.MESSAGE) or {} + sender = event.get(FeishuPayloadKey.SENDER) or {} + sender_id = sender.get(FeishuPayloadKey.SENDER_ID) or {} + return { + "schema": payload.get("schema"), + FeishuPayloadKey.EVENT_ID: header.get(FeishuPayloadKey.EVENT_ID), + FeishuPayloadKey.EVENT_TYPE: header.get(FeishuPayloadKey.EVENT_TYPE), + FeishuPayloadKey.MESSAGE_ID: message.get(FeishuPayloadKey.MESSAGE_ID), + FeishuCommandKey.CHAT_ID: message.get(FeishuCommandKey.CHAT_ID), + FeishuPayloadKey.MESSAGE_TYPE: message.get(FeishuPayloadKey.MESSAGE_TYPE), + FeishuPayloadKey.OPEN_ID: sender_id.get(FeishuPayloadKey.OPEN_ID), + } + + def _normalize_source(source: str | FeishuEventSource) -> FeishuEventSource: return FeishuEventSource(source) diff --git a/app/application/feishu/handlers/__init__.py b/app/application/feishu/handlers/__init__.py new file mode 100644 index 0000000..cb17ca9 --- /dev/null +++ b/app/application/feishu/handlers/__init__.py @@ -0,0 +1,9 @@ +from app.application.feishu.handlers.finance import handle_finance_command +from app.application.feishu.handlers.market import handle_market_command +from app.application.feishu.handlers.rules import handle_rule_command + +__all__ = [ + "handle_finance_command", + "handle_market_command", + "handle_rule_command", +] diff --git a/app/application/feishu/handlers/finance.py b/app/application/feishu/handlers/finance.py new file mode 100644 index 0000000..b514839 --- /dev/null +++ b/app/application/feishu/handlers/finance.py @@ -0,0 +1,141 @@ +import re +from typing import Any + +from sqlalchemy.orm import Session + +from app.application.feishu.delivery import send_text_if_configured +from app.application.feishu.results import command_result +from app.core.config import get_settings +from app.modules.ai_agent.constants import AIResponseKey +from app.modules.feishu.constants import FeishuCommandName, FeishuReplyType +from app.modules.feishu.service import FeishuService +from app.modules.reports.chart import lifecycle_chart_alt, render_lifecycle_chart +from app.modules.reports.constants import ReportResponseKey +from app.modules.reports.services import ReportService + +PROJECT_FINANCE_PATTERN = re.compile(r"^项目资金\s+(.+)$") +FINANCE_COMMANDS = {"资金需求", "未来30天资金需求"} + + +def handle_finance_command( + db: Session, + feishu: FeishuService, + command_text: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any] | None: + """Handle project cash-needs commands.""" + + project_match = PROJECT_FINANCE_PATTERN.fullmatch(command_text) + if command_text not in FINANCE_COMMANDS and project_match is None: + return None + command = ( + FeishuCommandName.PROJECT_FINANCE if project_match else FeishuCommandName.FINANCE_NEEDS + ) + if not get_settings().finance_needs_enabled: + return _text_result( + feishu, + command, + "项目资金需求分析", + "项目资金需求分析尚未启用，请先配置并启用财务只读同步。", + chat_id, + actor, + auto_reply, + ) + + project_code = project_match.group(1).strip() if project_match else None + service = ReportService(db) + preview = service.project_finance_needs_report( + project_code=project_code, + include_ai=False, + actor=actor, + ) + if project_code and not preview["items"]: + return _text_result( + feishu, + command, + "项目资金需求分析", + f"未找到项目“{project_code}”，请使用稳定项目编号或展示编号。", + chat_id, + actor, + auto_reply, + ) + if not preview["summary"]["data_available"]: + return _text_result( + feishu, + command, + "项目资金需求分析", + "项目财务数据未接入或无有效记录，暂不生成资金分析报告。", + chat_id, + actor, + auto_reply, + ) + + report = service.project_finance_needs_report( + project_code=project_code, + include_ai=True, + actor=actor, + ) + ai_analysis = report.get("ai_analysis") or {} + if not ai_analysis.get(AIResponseKey.OK): + return _text_result( + feishu, + command, + "AI 暂不可用", + "AI 当前不可用，本次项目资金分析报告未发送。请检查模型服务。", + chat_id, + actor, + auto_reply, + ) + + provider_response = None + if auto_reply: + provider_response = _send_finance_card(feishu, chat_id, report, actor) + return command_result( + command, + FeishuReplyType.CARD, + report[ReportResponseKey.TITLE], + report[ReportResponseKey.CONTENT], + provider_response, + report[ReportResponseKey.LINES], + ) + + +def _text_result( + feishu: FeishuService, + command: FeishuCommandName, + title: str, + content: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any]: + response = send_text_if_configured(feishu, chat_id, content, actor) if auto_reply else None + return command_result(command, FeishuReplyType.TEXT, title, content, response) + + +def _send_finance_card( + feishu: FeishuService, + chat_id: str | None, + report: dict[str, Any], + actor: str, +) -> dict[str, Any] | None: + settings = get_settings() + if not (settings.feishu_app_id and settings.feishu_app_secret): + return None + chart_data = { + "period": report.get("as_of"), + "finance": report.get("finance_chart_data"), + } + image_result = feishu.upload_image(render_lifecycle_chart(chart_data), actor) + image_key = (image_result.get("data") or {}).get("image_key") + if not image_key: + raise ValueError("Feishu image upload did not return image_key") + card = FeishuService.build_basic_card( + report[ReportResponseKey.TITLE], + report[ReportResponseKey.LINES], + image_key=image_key, + image_alt=lifecycle_chart_alt(chart_data), + ) + return feishu.send_card(card, receive_id=chat_id, actor=actor) diff --git a/app/application/feishu/handlers/market.py b/app/application/feishu/handlers/market.py new file mode 100644 index 0000000..19f3a95 --- /dev/null +++ b/app/application/feishu/handlers/market.py @@ -0,0 +1,243 @@ +import re +from typing import Any + +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from app.application.feishu.delivery import send_text_if_configured +from app.application.feishu.results import command_result +from app.core.config import get_settings +from app.modules.feishu.constants import FeishuCommandName, FeishuReplyType +from app.modules.feishu.service import FeishuService +from app.modules.market.chart import render_market_chart +from app.modules.market.service import MarketService + +STOCK_ANALYSIS_PATTERN = re.compile( + r"^(?:股票分析|估值分析|财报分析)\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE +) +WATCHLIST_ADD_PATTERN = re.compile(r"^加入自选\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE) +MARKET_COMMANDS = { + "市场分析", + "今日收盘分析", + "本周市场分析", + "宏观金融分析", + "最新公告", +} +INDUSTRY_ANALYSIS_PATTERN = re.compile(r"^行业分析\s+(.+)$") +STOCK_COMPARE_PATTERN = re.compile( + r"^股票对比\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)\s+" + r"([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", + re.IGNORECASE, +) + + +def handle_market_command( + db: Session, + feishu: FeishuService, + text: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any] | None: + """Handle market analysis and watchlist commands.""" + + stock = STOCK_ANALYSIS_PATTERN.fullmatch(text) + add = WATCHLIST_ADD_PATTERN.fullmatch(text) + industry = INDUSTRY_ANALYSIS_PATTERN.fullmatch(text) + comparison = STOCK_COMPARE_PATTERN.fullmatch(text) + if ( + text not in MARKET_COMMANDS + and text != "查看自选" + and not stock + and not add + and not industry + and not comparison + ): + return None + if not get_settings().market_analysis_enabled: + return _text_result( + feishu, + FeishuCommandName.MARKET_OVERVIEW, + "市场分析", + "市场分析尚未启用，请配置市场数据源后启用。", + chat_id, + actor, + auto_reply, + ) + + service = MarketService(db) + if add: + if get_settings().read_only_mode: + return _text_result( + feishu, + FeishuCommandName.WATCHLIST_ADD, + "自选股", + "当前为只读模式，不能修改自选股。请由管理员启用操作后重试。", + chat_id, + actor, + auto_reply, + ) + item = service.add_watchlist(actor, add.group(1)) + return _text_result( + feishu, + FeishuCommandName.WATCHLIST_ADD, + "自选股", + f"已加入自选：{item['symbol']}", + chat_id, + actor, + auto_reply, + ) + if text == "查看自选": + items = service.watchlist(actor) + content = "自选股：" + ("、".join(item["symbol"] for item in items) or "暂无") + return _text_result( + feishu, + FeishuCommandName.WATCHLIST_LIST, + "自选股", + content, + chat_id, + actor, + auto_reply, + ) + if text == "最新公告": + items = service.announcements(limit=10)["items"] + content = ( + "最新公告：\n" + + "\n".join( + f"- {item['announcement_date']} {item['symbol'] or '市场'}：{item['title']}" + for item in items + ) + if items + else "公告元数据尚未接入。" + ) + return _text_result( + feishu, + FeishuCommandName.MARKET_ANNOUNCEMENTS, + "最新公告", + content, + chat_id, + actor, + auto_reply, + ) + if industry: + try: + data = service.industry_analysis(industry.group(1).strip()) + content = ( + f"{data['industry']} 平均涨跌 {data['average_pct_change']}%\n" + + "\n".join( + f"- {item['name']}（{item['symbol']}）：{item['pct_change']}%" + for item in data["items"][:10] + ) + ) + except HTTPException: + content = "未找到该行业的最新市场数据。" + return _text_result( + feishu, + FeishuCommandName.MARKET_OVERVIEW, + "行业分析", + content, + chat_id, + actor, + auto_reply, + ) + if comparison: + try: + content = service.compare_stocks([comparison.group(1), comparison.group(2)])["content"] + except HTTPException: + content = "至少一只股票缺少可用行情，暂时无法比较。" + return _text_result( + feishu, + FeishuCommandName.STOCK_ANALYSIS, + "股票对比", + content, + chat_id, + actor, + auto_reply, + ) + return _analysis_result(service, feishu, text, stock.group(1) if stock else None, chat_id, actor, auto_reply) + + +def _analysis_result( + service: MarketService, + feishu: FeishuService, + text: str, + symbol: str | None, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any]: + command = FeishuCommandName.STOCK_ANALYSIS if symbol else FeishuCommandName.MARKET_OVERVIEW + if text == "宏观金融分析": + command = FeishuCommandName.MARKET_MACRO + try: + if symbol: + report = service.stock_analysis(symbol, True, actor) + elif text == "本周市场分析": + report = service.weekly_overview(include_ai=True, actor=actor) + elif text == "宏观金融分析": + report = service.macro_analysis(include_ai=True, actor=actor) + else: + report = service.market_overview(include_ai=True, actor=actor) + except HTTPException: + return _text_result( + feishu, + command, + "股票分析", + "未找到该股票的可用行情，请确认代码或先执行行情同步。", + chat_id, + actor, + auto_reply, + ) + ai = report.get("ai_analysis") or {} + if not report.get("data_available"): + content = "市场数据未接入，暂不生成分析报告。" + elif not ai.get("ok"): + content = "AI 当前不可用，本次市场分析报告未发送。" + else: + content = report["content"] + response = _send_analysis(feishu, text, report, content, chat_id, actor) if auto_reply else None + return command_result( + command, + FeishuReplyType.CARD if ai.get("ok") and text != "宏观金融分析" else FeishuReplyType.TEXT, + report["title"], + content, + response, + report["lines"] if ai.get("ok") else None, + ) + + +def _send_analysis( + feishu: FeishuService, + text: str, + report: dict[str, Any], + content: str, + chat_id: str | None, + actor: str, +) -> dict[str, Any] | None: + ai = report.get("ai_analysis") or {} + if not ai.get("ok") or text == "宏观金融分析": + return send_text_if_configured(feishu, chat_id, content, actor) + image = feishu.upload_image(render_market_chart(report), actor) + image_key = (image.get("data") or {}).get("image_key") + if not image_key: + raise ValueError("Feishu image upload did not return image_key") + card = FeishuService.build_basic_card( + report["title"], + report["lines"], + image_key=image_key, + image_alt=report["title"], + ) + return feishu.send_card(card, receive_id=chat_id, actor=actor) + + +def _text_result( + feishu: FeishuService, + command: FeishuCommandName, + title: str, + content: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any]: + response = send_text_if_configured(feishu, chat_id, content, actor) if auto_reply else None + return command_result(command, FeishuReplyType.TEXT, title, content, response) diff --git a/app/application/feishu/handlers/rules.py b/app/application/feishu/handlers/rules.py new file mode 100644 index 0000000..a5d09e7 --- /dev/null +++ b/app/application/feishu/handlers/rules.py @@ -0,0 +1,192 @@ +import re +from typing import Any + +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from app.application.feishu.delivery import send_text_if_configured +from app.application.feishu.results import command_result +from app.core.config import get_settings +from app.modules.ai_memory.constants import AIMemoryStatus +from app.modules.ai_memory.service import AIMemoryService +from app.modules.feishu.constants import FeishuCommandName, FeishuReplyType +from app.modules.feishu.service import FeishuService + +RULE_TITLE = "AI 学习规则" +RULE_CREATE_PATTERN = re.compile(r"^学习规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$") +MARKET_RULE_CREATE_PATTERN = re.compile( + r"^学习市场规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$" +) +RULE_DISABLE_PATTERN = re.compile(r"^停用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) +RULE_ENABLE_PATTERN = re.compile(r"^启用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) +RULE_LIST_COMMANDS = {"查看规则", "规则列表", "查看市场规则"} +RULE_COMMAND_PREFIXES = ( + "学习市场规则", + "学习规则", + "查看市场规则", + "查看规则", + "规则列表", + "停用规则", + "启用规则", +) +RULE_COMMAND_HELP = ( + "规则指令格式：\n" + "学习规则：<规则内容>\n" + "学习规则 80：<规则内容>\n" + "学习市场规则 80：<仅用于市场分析的规则内容>\n" + "查看规则\n" + "停用规则 <规则编号>\n" + "启用规则 <规则编号>" +) + + +def handle_rule_command( + db: Session, + feishu: FeishuService, + command_text: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any] | None: + """Handle persistent AI rule commands.""" + + if not command_text.startswith(RULE_COMMAND_PREFIXES): + return None + command = _command_name(command_text) + if command in { + FeishuCommandName.RULE_CREATE, + FeishuCommandName.RULE_DISABLE, + FeishuCommandName.RULE_ENABLE, + } and get_settings().read_only_mode: + return _result( + feishu, + command, + "当前为只读模式，不能新增或修改学习规则。请由管理员启用操作后重试。", + chat_id, + actor, + auto_reply, + ) + + content = RULE_COMMAND_HELP + try: + command, content = _execute(db, command_text, command, actor) + except HTTPException as exc: + detail = str(exc.detail) + if "secret-like" in detail: + content = "规则疑似包含密码、令牌或其他密钥信息，已拒绝学习。" + elif exc.status_code == 404: + content = "没有找到该规则，请先发送“查看规则”确认规则编号。" + elif "priority" in detail: + content = "规则优先级必须在 1 到 100 之间。" + else: + content = "规则未保存，请检查指令内容后重试。" + return _result(feishu, command, content, chat_id, actor, auto_reply) + + +def _execute( + db: Session, + command_text: str, + command: FeishuCommandName, + actor: str, +) -> tuple[FeishuCommandName, str]: + market_create_match = MARKET_RULE_CREATE_PATTERN.fullmatch(command_text) + create_match = market_create_match or RULE_CREATE_PATTERN.fullmatch(command_text) + disable_match = RULE_DISABLE_PATTERN.fullmatch(command_text) + enable_match = RULE_ENABLE_PATTERN.fullmatch(command_text) + memory = AIMemoryService(db) + if create_match: + return command, _create_rule(memory, create_match, market_create_match is not None, actor) + if command_text in RULE_LIST_COMMANDS: + return FeishuCommandName.RULE_LIST, _list_rules(memory, command_text) + if disable_match or enable_match: + enabled = enable_match is not None + match = enable_match or disable_match + rule = memory.update_rule( + code=match.group(1), + content=None, + priority=None, + tags=None, + enabled=enabled, + actor=actor, + ) + state = "已启用" if enabled else "已停用" + return ( + FeishuCommandName.RULE_ENABLE if enabled else FeishuCommandName.RULE_DISABLE, + f"规则{state}。\n" + f"编号：{rule['code']}\n" + f"优先级：{rule['importance']}\n" + f"范围：{rule['scope']} / {rule['subject']}\n" + f"状态：{state}", + ) + return command, RULE_COMMAND_HELP + + +def _create_rule( + memory: AIMemoryService, + match: re.Match[str], + market_rule: bool, + actor: str, +) -> str: + priority = int(match.group(1) or 50) + content = match.group(2).strip() + if not content: + return f"规则内容不能为空。\n\n{RULE_COMMAND_HELP}" + if not 1 <= priority <= 100: + return "规则优先级必须在 1 到 100 之间。" + rule = memory.create_rule( + content=content, + scope="market" if market_rule else "global", + subject="market" if market_rule else "company", + priority=priority, + tags=["feishu", *(["market"] if market_rule else [])], + actor=actor, + ) + return ( + "规则已学习。\n" + f"编号：{rule['code']}\n" + f"优先级：{rule['importance']}\n" + f"范围：{rule['scope']} / {rule['subject']}\n" + "状态：已启用" + ) + + +def _list_rules(memory: AIMemoryService, command_text: str) -> str: + rules = memory.list_rules( + scope="market" if command_text == "查看市场规则" else None, + status_filter=AIMemoryStatus.ACTIVE, + limit=20, + ) + if not rules: + return "当前没有已启用的学习规则。" + lines = ["当前已启用的学习规则："] + for rule in rules: + rule_text = str(rule["content"]) + if len(rule_text) > 80: + rule_text = f"{rule_text[:80]}…" + lines.append( + f"{rule['code']}｜优先级 {rule['importance']}｜" + f"{rule['scope']}/{rule['subject']}\n{rule_text}" + ) + return "\n\n".join(lines) + + +def _command_name(command_text: str) -> FeishuCommandName: + if command_text.startswith("停用规则"): + return FeishuCommandName.RULE_DISABLE + if command_text.startswith("启用规则"): + return FeishuCommandName.RULE_ENABLE + if command_text.startswith(("查看市场规则", "查看规则", "规则列表")): + return FeishuCommandName.RULE_LIST + return FeishuCommandName.RULE_CREATE + + +def _result( + feishu: FeishuService, + command: FeishuCommandName, + content: str, + chat_id: str | None, + actor: str, + auto_reply: bool, +) -> dict[str, Any]: + response = send_text_if_configured(feishu, chat_id, content, actor) if auto_reply else None + return command_result(command, FeishuReplyType.TEXT, RULE_TITLE, content, response) diff --git a/app/application/feishu/results.py b/app/application/feishu/results.py new file mode 100644 index 0000000..08451d3 --- /dev/null +++ b/app/application/feishu/results.py @@ -0,0 +1,29 @@ +from typing import Any + +from app.modules.feishu.constants import ( + FeishuCommandName, + FeishuCommandResultKey, + FeishuReplyType, +) + + +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]: + """Build the stable command response contract.""" + + 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 diff --git a/app/application/pipelines/__init__.py b/app/application/pipelines/__init__.py new file mode 100644 index 0000000..223a0cd --- /dev/null +++ b/app/application/pipelines/__init__.py @@ -0,0 +1,4 @@ +from app.application.pipelines.lifecycle import LifecyclePipelineService +from app.application.pipelines.market import MarketPipelineService + +__all__ = ["LifecyclePipelineService", "MarketPipelineService"] diff --git a/app/modules/reports/lifecycle_pipeline.py b/app/application/pipelines/lifecycle.py similarity index 97% rename from app/modules/reports/lifecycle_pipeline.py rename to app/application/pipelines/lifecycle.py index c8154f9..265259e 100644 --- a/app/modules/reports/lifecycle_pipeline.py +++ b/app/application/pipelines/lifecycle.py @@ -5,8 +5,10 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app.core.config import get_settings +from app.core.security import business_mutations_enabled from app.core.constants import ActorValue from app.core.utils.time import utc_now +from app.application.delivery import ReportDeliveryService from app.modules.feishu.service import FeishuService from app.modules.legacy_mysql.intasect import IntasectSyncService from app.modules.reports.constants import ReportPushStatus, ReportType @@ -64,7 +66,7 @@ class LifecyclePipelineService: force: bool = False, actor: str = ActorValue.SCHEDULER, ) -> dict[str, Any]: - if get_settings().read_only_mode: + if not business_mutations_enabled(): return { "period_key": self.period_key(report_type), "deduplicated": False, @@ -153,7 +155,7 @@ class LifecyclePipelineService: idempotency_key=idempotency_key, ) if push_run.status != ReportPushStatus.SUCCESS: - report_service.push_report( + ReportDeliveryService(self.db).push_report( report, target_receive_id, receive_id_type, diff --git a/app/modules/market/pipeline.py b/app/application/pipelines/market.py similarity index 98% rename from app/modules/market/pipeline.py rename to app/application/pipelines/market.py index a0e0d04..5634933 100644 --- a/app/modules/market/pipeline.py +++ b/app/application/pipelines/market.py @@ -6,6 +6,7 @@ from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.constants import ActorValue +from app.core.security import business_mutations_enabled from app.core.utils.time import utc_now from app.modules.feishu.service import FeishuService from app.modules.market.chart import render_market_chart @@ -50,7 +51,7 @@ class MarketPipelineService: ) -> dict[str, Any]: target = reference_date or date.today() period_key = self.period_key(report_type, target) - if get_settings().read_only_mode: + if not business_mutations_enabled(): return { "period_key": period_key, "status": "operations_disabled", diff --git a/app/application/scheduling/__init__.py b/app/application/scheduling/__init__.py new file mode 100644 index 0000000..0f7c15a --- /dev/null +++ b/app/application/scheduling/__init__.py @@ -0,0 +1,3 @@ +from app.application.scheduling.scheduler import attach_scheduler, create_scheduler + +__all__ = ["attach_scheduler", "create_scheduler"] diff --git a/app/core/background/scheduler.py b/app/application/scheduling/scheduler.py similarity index 97% rename from app/core/background/scheduler.py rename to app/application/scheduling/scheduler.py index 885e0bc..3150055 100644 --- a/app/core/background/scheduler.py +++ b/app/application/scheduling/scheduler.py @@ -8,6 +8,7 @@ from fastapi import FastAPI from app.core.config import get_settings from app.core.constants import ActorValue from app.modules.feishu.constants import FeishuReceiveIdType +from app.application.delivery import ReportDeliveryService from app.modules.observability.constants import HeartbeatComponent @@ -74,7 +75,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: ) _set_state(app, f"{state_key}_dispatch", dispatch) return - service.push_report( + ReportDeliveryService(db).push_report( report, receive_id=settings.feishu_default_chat_id, receive_id_type=FeishuReceiveIdType.CHAT_ID, @@ -188,7 +189,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: finally: db.close() - if settings.lifecycle_pipeline_enabled: + if settings.lifecycle_pipeline_enabled and not settings.read_only_mode: scheduler.add_job( run_daily_lifecycle, trigger="cron", @@ -265,7 +266,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: id="event_dispatch", replace_existing=True, ) - if settings.market_analysis_enabled: + if settings.market_analysis_enabled and not settings.read_only_mode: scheduler.add_job( run_market_premarket, trigger="cron", @@ -302,6 +303,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: ) if ( not settings.lifecycle_pipeline_enabled + and not settings.read_only_mode and settings.legacy_sync_enabled and settings.legacy_project_query ): @@ -315,6 +317,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: ) if ( not settings.lifecycle_pipeline_enabled + and not settings.read_only_mode and settings.legacy_sync_enabled and settings.legacy_task_query ): diff --git a/app/core/background/task_queue/__init__.py b/app/core/background/task_queue/__init__.py index 2389286..477f221 100644 --- a/app/core/background/task_queue/__init__.py +++ b/app/core/background/task_queue/__init__.py @@ -1,4 +1,4 @@ -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_DISPATCH_PENDING_EVENTS, TASK_GENERATE_RISK_EVENTS, TASK_PUSH_ATTENDANCE_SUMMARY, diff --git a/app/core/background/task_queue/events.py b/app/core/background/task_queue/events.py index 93adb1a..75acee7 100644 --- a/app/core/background/task_queue/events.py +++ b/app/core/background/task_queue/events.py @@ -1,7 +1,8 @@ from typing import Any +from app.application.events import EventDispatchService from app.core.config import get_settings -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_DISPATCH_PENDING_EVENTS, ) from app.core.background.task_queue.dispatcher import dispatch_task @@ -13,11 +14,10 @@ def enqueue_event_dispatch( ) -> dict[str, Any]: def inline() -> Any: from app.core.database import SessionLocal - from app.modules.events.services import EventService db = SessionLocal() try: - return EventService(db).dispatch_pending( + return EventDispatchService(db).dispatch_pending( limit=limit or get_settings().event_dispatch_batch_size, worker_id=actor, ) diff --git a/app/core/background/task_queue/legacy.py b/app/core/background/task_queue/legacy.py index 0173ecd..c424b52 100644 --- a/app/core/background/task_queue/legacy.py +++ b/app/core/background/task_queue/legacy.py @@ -1,6 +1,6 @@ from typing import Any -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_SYNC_LEGACY_PROJECTS, TASK_SYNC_LEGACY_TASKS, ) diff --git a/app/core/background/task_queue/lifecycle.py b/app/core/background/task_queue/lifecycle.py index 279862b..7ec7d7b 100644 --- a/app/core/background/task_queue/lifecycle.py +++ b/app/core/background/task_queue/lifecycle.py @@ -1,10 +1,10 @@ from typing import Any -from app.core.background.task_queue.constants import TASK_RUN_LIFECYCLE +from app.tasks.constants import TASK_RUN_LIFECYCLE from app.core.config import get_settings from app.core.constants import ActorValue from app.core.database import SessionLocal -from app.modules.reports.lifecycle_pipeline import LifecyclePipelineService +from app.application.pipelines import LifecyclePipelineService def enqueue_lifecycle_report( diff --git a/app/core/background/task_queue/market.py b/app/core/background/task_queue/market.py index 6d0d062..9607570 100644 --- a/app/core/background/task_queue/market.py +++ b/app/core/background/task_queue/market.py @@ -1,7 +1,7 @@ from datetime import date from typing import Any -from app.core.background.task_queue.constants import TASK_RUN_MARKET_REPORT +from app.tasks.constants import TASK_RUN_MARKET_REPORT from app.core.config import get_settings diff --git a/app/core/background/task_queue/reports.py b/app/core/background/task_queue/reports.py index 3e8e0af..553d0ad 100644 --- a/app/core/background/task_queue/reports.py +++ b/app/core/background/task_queue/reports.py @@ -1,7 +1,7 @@ from collections.abc import Callable from typing import Any -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_PUSH_ATTENDANCE_SUMMARY, TASK_PUSH_DAILY_BRIEF, TASK_PUSH_PROJECT_WEEKLY, @@ -135,6 +135,7 @@ def _enqueue_report_push( ) def inline() -> Any: + from app.application.delivery import ReportDeliveryService from app.core.database import SessionLocal from app.modules.reports.services import ReportService @@ -142,7 +143,9 @@ def _enqueue_report_push( try: service = ReportService(db) report = build_report(service, actor) - return service.push_report(report, receive_id, receive_id_type, actor) + return ReportDeliveryService(db).push_report( + report, receive_id, receive_id_type, actor + ) finally: db.close() diff --git a/app/core/background/task_queue/risk.py b/app/core/background/task_queue/risk.py index 214c7e4..fb35ad5 100644 --- a/app/core/background/task_queue/risk.py +++ b/app/core/background/task_queue/risk.py @@ -1,6 +1,6 @@ from typing import Any -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_GENERATE_RISK_EVENTS, ) from app.core.background.task_queue.dispatcher import dispatch_task diff --git a/app/core/config/settings.py b/app/core/config/settings.py index ca6dcc4..281fec8 100644 --- a/app/core/config/settings.py +++ b/app/core/config/settings.py @@ -2,7 +2,7 @@ import json from functools import lru_cache from typing import Annotated, Any -from pydantic import Field, field_validator +from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict from app.core.constants import ( @@ -112,6 +112,25 @@ class Settings(BaseSettings): ai_memory_enabled: bool = True ai_memory_auto_write_enabled: bool = True ai_memory_recall_limit: int = 5 + ai_memory_auto_write_ttl_days: int = Field(default=90, ge=1, le=3650) + ai_memory_blocked_content_terms: list[str] = Field( + default_factory=lambda: [ + "bank account", + "budget", + "cash flow", + "financial statement", + "id card", + "payment", + "salary", + "成本", + "付款", + "工资", + "收款", + "现金流", + "财务", + "预算", + ] + ) ai_analysis_max_attempts: int = 3 ai_memory_forbidden_keys: list[str] = Field( default_factory=lambda: [ @@ -151,6 +170,7 @@ class Settings(BaseSettings): "openclaw_allowed_tools", "openclaw_allowed_actions", "ai_memory_forbidden_keys", + "ai_memory_blocked_content_terms", mode="before", ) @classmethod @@ -216,6 +236,27 @@ class Settings(BaseSettings): return {str(key): str(item) for key, item in data.items()} raise ValueError(ConfigErrorDetail.LEGACY_ALLOWED_QUERIES_FORMAT) + @model_validator(mode="after") + def validate_production_safety(self) -> "Settings": + if self.app_env.lower() not in {"prod", "production"}: + return self + errors: list[str] = [] + if self.database_url.startswith("sqlite"): + errors.append("DATABASE_URL must use PostgreSQL in production") + if not self.api_key and not any(item.get("key") for item in self.api_keys): + errors.append("API_KEY or API_KEYS is required in production") + if not self.audit_api_key and not any( + item.get("key") for item in self.audit_api_keys + ): + errors.append("AUDIT_API_KEY or AUDIT_API_KEYS is required in production") + if "*" in self.cors_origins: + errors.append("CORS_ORIGINS cannot contain '*' in production") + if self.debug: + errors.append("DEBUG must be false in production") + if errors: + raise ValueError("; ".join(errors)) + return self + @lru_cache def get_settings() -> Settings: diff --git a/app/core/http/responses.py b/app/core/http/responses.py new file mode 100644 index 0000000..34e21ca --- /dev/null +++ b/app/core/http/responses.py @@ -0,0 +1,12 @@ +from typing import Any + +from fastapi.responses import JSONResponse + +from app.core.http.masking import mask_configured + + +class MaskedJSONResponse(JSONResponse): + """Apply baseline sensitive-field masking to every JSON response.""" + + def render(self, content: Any) -> bytes: + return super().render(mask_configured(content)) diff --git a/app/core/security/__init__.py b/app/core/security/__init__.py index f768bff..3f9cd96 100644 --- a/app/core/security/__init__.py +++ b/app/core/security/__init__.py @@ -3,10 +3,18 @@ from app.core.security.operation_guard import ( READ_ONLY_OPERATION_DISABLED, require_operations_enabled, ) +from app.core.security.operation_policy import ( + OperationsDisabledError, + business_mutations_enabled, + ensure_business_mutations_enabled, +) __all__ = [ "ApiPrincipal", + "OperationsDisabledError", "READ_ONLY_OPERATION_DISABLED", + "business_mutations_enabled", + "ensure_business_mutations_enabled", "require_api_key", "require_audit_api_key", "require_operations_enabled", diff --git a/app/core/security/operation_guard.py b/app/core/security/operation_guard.py index b2073a9..a62bfa8 100644 --- a/app/core/security/operation_guard.py +++ b/app/core/security/operation_guard.py @@ -1,6 +1,6 @@ from fastapi import HTTPException, status -from app.core.config import get_settings +from app.core.security.operation_policy import business_mutations_enabled READ_ONLY_OPERATION_DISABLED = "This service is read-only; data mutation operations are disabled" @@ -9,7 +9,7 @@ READ_ONLY_OPERATION_DISABLED = "This service is read-only; data mutation operati def require_operations_enabled(detail: str = READ_ONLY_OPERATION_DISABLED) -> None: """Reject mutation-oriented endpoints when the product is running read-only.""" - if get_settings().read_only_mode: + if not business_mutations_enabled(): raise HTTPException( status_code=status.HTTP_405_METHOD_NOT_ALLOWED, detail=detail, diff --git a/app/core/security/operation_policy.py b/app/core/security/operation_policy.py new file mode 100644 index 0000000..1c9a09c --- /dev/null +++ b/app/core/security/operation_policy.py @@ -0,0 +1,20 @@ +from app.core.config import get_settings + + +class OperationsDisabledError(RuntimeError): + """Raised when a business mutation is attempted in read-only mode.""" + + +def business_mutations_enabled() -> bool: + """Return whether platform-owned business ledgers may be mutated.""" + + return not get_settings().read_only_mode + + +def ensure_business_mutations_enabled() -> None: + """Enforce read-only policy outside the HTTP transport layer.""" + + if not business_mutations_enabled(): + raise OperationsDisabledError( + "Business mutations are disabled while READ_ONLY_MODE is enabled" + ) diff --git a/app/main.py b/app/main.py index 8fb373c..6ecbd91 100644 --- a/app/main.py +++ b/app/main.py @@ -4,7 +4,8 @@ from fastapi.middleware.cors import CORSMiddleware from app.api.router import api_router from app.core.config import get_settings from app.core.http.middleware import request_id_middleware -from app.core.background.scheduler import attach_scheduler +from app.core.http.responses import MaskedJSONResponse +from app.application.scheduling import attach_scheduler def _allow_cors_credentials(cors_origins: list[str]) -> bool: @@ -23,6 +24,7 @@ def create_app() -> FastAPI: "AI integration layer for company lifecycle management, existing MySQL " "project systems, Feishu, OpenClaw, Hermes, and model providers." ), + default_response_class=MaskedJSONResponse, ) app.middleware("http")(request_id_middleware) diff --git a/app/modules/ai_memory/constants.py b/app/modules/ai_memory/constants.py index 2b2ca2f..9b3a236 100644 --- a/app/modules/ai_memory/constants.py +++ b/app/modules/ai_memory/constants.py @@ -48,6 +48,7 @@ class AIMemoryText(StrEnum): DEFAULT_SUBJECT = "company" AUTO_TAG = "auto" REJECTED_SECRET = "secret-like content rejected" + REJECTED_SENSITIVE_FACT = "sensitive business fact rejected" AI_MEMORY_CODE_PREFIX = "MEM" diff --git a/app/modules/ai_memory/service.py b/app/modules/ai_memory/service.py index 53063a2..3783925 100644 --- a/app/modules/ai_memory/service.py +++ b/app/modules/ai_memory/service.py @@ -1,3 +1,4 @@ +from datetime import datetime, timedelta from typing import Any from uuid import uuid4 @@ -99,9 +100,8 @@ class AIMemoryService: items = items[:limit_value] for item in items: item.last_used_at = now - self.db.commit() result = [serialize_model(item) for item in items] - self.audit.log( + self.audit.record( AuditLogCreate( actor=actor, source=AuditSource.AI_MEMORY, @@ -117,6 +117,7 @@ class AIMemoryService: response_payload={AIMemoryPayloadKey.COUNT: len(result)}, ) ) + self.db.commit() return result def auto_write( @@ -152,8 +153,24 @@ class AIMemoryService: importance=0, status_value=AIMemoryStatus.REJECTED, actor=actor, + expires_at=utc_now() + + timedelta(days=settings.ai_memory_auto_write_ttl_days), ) return record + if _contains_blocked_content(content, settings.ai_memory_blocked_content_terms): + return self._create_entry( + scope=scope, + subject=subject, + content=str(AIMemoryText.REJECTED_SENSITIVE_FACT), + summary=str(AIMemoryText.REJECTED_SENSITIVE_FACT), + tags=[str(AIMemoryText.AUTO_TAG)], + source=AIMemorySource.AUTO, + importance=0, + status_value=AIMemoryStatus.REJECTED, + actor=actor, + expires_at=utc_now() + + timedelta(days=settings.ai_memory_auto_write_ttl_days), + ) summary = _truncate(answer, AI_MEMORY_MAX_SUMMARY_LENGTH) record = self._create_entry( scope=scope, @@ -165,6 +182,7 @@ class AIMemoryService: importance=1, status_value=AIMemoryStatus.ACTIVE, actor=actor, + expires_at=utc_now() + timedelta(days=settings.ai_memory_auto_write_ttl_days), ) return record @@ -275,9 +293,7 @@ class AIMemoryService: record.tags = ["user-rule", *tags] if enabled is not None: record.status = AIMemoryStatus.ACTIVE if enabled else AIMemoryStatus.ARCHIVED - self.db.commit() - self.db.refresh(record) - self.audit.log( + self.audit.record( AuditLogCreate( actor=actor, source=AuditSource.AI_MEMORY, @@ -292,6 +308,8 @@ class AIMemoryService: }, ) ) + self.db.commit() + self.db.refresh(record) return serialize_model(record) def _validate_rule(self, content: str, priority: int) -> None: @@ -324,6 +342,7 @@ class AIMemoryService: status_value: str, actor: str, audit_action: str = AuditAction.AI_MEMORY_WRITE, + expires_at: datetime | None = None, ) -> AIMemoryEntry: record = AIMemoryEntry( code=( @@ -339,11 +358,11 @@ class AIMemoryService: importance=importance, status=status_value, actor=actor, + expires_at=expires_at, ) self.db.add(record) - self.db.commit() - self.db.refresh(record) - self.audit.log( + self.db.flush() + self.audit.record( AuditLogCreate( actor=actor, source=AuditSource.AI_MEMORY, @@ -360,7 +379,7 @@ class AIMemoryService: response_payload={AIMemoryPayloadKey.CODE: record.code}, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.AI_MEMORY_WRITTEN, source=EventSource.AI_MEMORY, aggregate_type=EventAggregateType.AI_MEMORY_ENTRY, @@ -373,8 +392,9 @@ class AIMemoryService: AIMemoryPayloadKey.STATUS: status_value, }, idempotency_key=f"ai-memory:{record.code}", - dispatch=True, ) + self.db.commit() + self.db.refresh(record) return record @@ -402,6 +422,11 @@ def _contains_forbidden_value(value: Any, forbidden_keys: list[str]) -> bool: return False +def _contains_blocked_content(value: str, blocked_terms: list[str]) -> bool: + lowered = value.lower() + return any(term.lower() in lowered for term in blocked_terms if term.strip()) + + def _matches_query(entry: AIMemoryEntry, query: str) -> bool: query_text = query.lower().strip() if not query_text: diff --git a/app/modules/audit/service.py b/app/modules/audit/service.py index bebab68..3ef76c2 100644 --- a/app/modules/audit/service.py +++ b/app/modules/audit/service.py @@ -44,7 +44,9 @@ class AuditService: def __init__(self, db: Session): self.db = db - def log(self, payload: AuditLogCreate) -> AuditLog: + def record(self, payload: AuditLogCreate) -> AuditLog: + """Stage an audit record in the caller's transaction.""" + record = AuditLog( actor=payload.actor, source=payload.source, @@ -58,6 +60,13 @@ class AuditService: request_id=payload.request_id or get_request_id(), ) self.db.add(record) + self.db.flush() + return record + + def log(self, payload: AuditLogCreate) -> AuditLog: + """Persist an audit record as a standalone transaction.""" + + record = self.record(payload) self.db.commit() self.db.refresh(record) return record diff --git a/app/modules/business/models/risks.py b/app/modules/business/models/risks.py index ff4251d..602ef5d 100644 --- a/app/modules/business/models/risks.py +++ b/app/modules/business/models/risks.py @@ -1,6 +1,6 @@ from datetime import date, datetime -from sqlalchemy import JSON, Date, DateTime, Integer, String, Text +from sqlalchemy import JSON, Date, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from app.core.constants import ActorValue @@ -42,7 +42,10 @@ class RiskEventAction(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) code: Mapped[str] = mapped_column(String(64), unique=True, index=True) - risk_event_id: Mapped[int] = mapped_column(Integer, index=True) + risk_event_id: Mapped[int] = mapped_column( + ForeignKey("risk_events.id", ondelete="RESTRICT"), + index=True, + ) action: Mapped[str] = mapped_column(String(64), index=True) actor: Mapped[str] = mapped_column(String(128), default=ActorValue.SYSTEM, index=True) from_status: Mapped[str | None] = mapped_column(String(32), nullable=True) diff --git a/app/modules/events/constants.py b/app/modules/events/constants.py index ac3976f..13789ff 100644 --- a/app/modules/events/constants.py +++ b/app/modules/events/constants.py @@ -60,6 +60,7 @@ class EventPayloadKey(StrEnum): class EventErrorDetail(StrEnum): EVENT_NOT_FOUND = "Domain event not found" EVENT_NOT_RETRYABLE = "Domain event is not retryable" + EVENT_LOCKED = "Domain event is already being processed" EVENT_CODE_PREFIX = "EVT" diff --git a/app/modules/events/models.py b/app/modules/events/models.py index e7ec762..19158ff 100644 --- a/app/modules/events/models.py +++ b/app/modules/events/models.py @@ -36,6 +36,6 @@ class DomainEvent(Base): ) locked_until: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) locked_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) - max_attempts: Mapped[int] = mapped_column(Integer, default=3) + max_attempts: Mapped[int] = mapped_column(Integer, default=3, server_default="3") created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) processed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) diff --git a/app/modules/events/routes.py b/app/modules/events/routes.py index 53b159c..4e34a4f 100644 --- a/app/modules/events/routes.py +++ b/app/modules/events/routes.py @@ -1,15 +1,17 @@ from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session +from app.application.events import EventDispatchService from app.core.database import get_db from app.core.security import ApiPrincipal, require_api_key from app.modules.events.constants import EventResponseKey +from app.modules.events.schemas import DomainEventListRead, DomainEventRead from app.modules.events.services import EventService, _serialize_event router = APIRouter(dependencies=[Depends(require_api_key)]) -@router.get("") +@router.get("", response_model=DomainEventListRead) def list_events( status: str | None = None, event_type: str | None = None, @@ -25,15 +27,19 @@ def list_events( } -@router.post("/{event_id}/dispatch") +@router.post("/{event_id}/dispatch", response_model=dict[str, DomainEventRead]) def dispatch_event( event_id: str, db: Session = Depends(get_db), ) -> dict: - return {EventResponseKey.EVENT: _serialize_event(EventService(db).dispatch_event(event_id))} + return { + EventResponseKey.EVENT: _serialize_event( + EventDispatchService(db).dispatch_event(event_id) + ) + } -@router.post("/{event_id}/retry") +@router.post("/{event_id}/retry", response_model=dict[str, DomainEventRead]) def retry_event( event_id: str, db: Session = Depends(get_db), @@ -41,14 +47,16 @@ def retry_event( ) -> dict: return { EventResponseKey.EVENT: _serialize_event( - EventService(db).retry_event(event_id, actor=principal.actor) + EventDispatchService(db).retry_event(event_id, actor=principal.actor) ) } -@router.post("/dispatch-pending") +@router.post("/dispatch-pending", response_model=DomainEventListRead) def dispatch_pending( limit: int = Query(default=100, ge=1, le=500), db: Session = Depends(get_db), ) -> dict: - return {EventResponseKey.ITEMS: EventService(db).dispatch_pending(limit=limit)} + return { + EventResponseKey.ITEMS: EventDispatchService(db).dispatch_pending(limit=limit) + } diff --git a/app/modules/events/schemas.py b/app/modules/events/schemas.py index c9787bc..85a8946 100644 --- a/app/modules/events/schemas.py +++ b/app/modules/events/schemas.py @@ -1,3 +1,4 @@ +from datetime import datetime from typing import Any from pydantic import BaseModel @@ -15,9 +16,9 @@ class DomainEventRead(BaseModel): attempts: int idempotency_key: str | None last_error: str | None - created_at: str - processed_at: str | None + created_at: datetime + processed_at: datetime | None class DomainEventListRead(BaseModel): - items: list[dict[str, Any]] + items: list[DomainEventRead] diff --git a/app/modules/events/services/query.py b/app/modules/events/services/query.py index 108e6a1..984c694 100644 --- a/app/modules/events/services/query.py +++ b/app/modules/events/services/query.py @@ -10,14 +10,13 @@ from app.core.utils.time import utc_now from app.modules.events.constants import ( EVENT_CODE_PREFIX, EventErrorDetail, - EventStatus, ) from app.modules.events.models import DomainEvent from app.modules.events.services.serialization import _serialize_event class EventQueryMixin: - def emit( + def enqueue( self, event_type: str, source: str, @@ -26,16 +25,12 @@ class EventQueryMixin: actor: str = ActorValue.SYSTEM, payload: dict[str, Any] | None = None, idempotency_key: str | None = None, - dispatch: bool = False, ) -> DomainEvent: - if idempotency_key: - existing = self.db.execute( - select(DomainEvent).where(DomainEvent.idempotency_key == idempotency_key) - ).scalar_one_or_none() - if existing is not None: - if dispatch and existing.status == EventStatus.PENDING: - return self.dispatch_event(existing.event_id) - return existing + """Stage an outbox event in the caller's transaction.""" + + existing = self._find_idempotent_event(idempotency_key) + if existing is not None: + return existing settings = get_settings() now = utc_now() @@ -52,12 +47,43 @@ class EventQueryMixin: max_attempts=settings.event_dispatch_max_attempts, ) self.db.add(record) + self.db.flush() + return record + + def emit( + self, + event_type: str, + source: str, + aggregate_type: str, + aggregate_id: str | int | None, + actor: str = ActorValue.SYSTEM, + payload: dict[str, Any] | None = None, + idempotency_key: str | None = None, + ) -> DomainEvent: + existing = self._find_idempotent_event(idempotency_key) + if existing is not None: + return existing + + record = self.enqueue( + event_type=event_type, + source=source, + aggregate_type=aggregate_type, + aggregate_id=aggregate_id, + actor=actor, + payload=payload, + idempotency_key=idempotency_key, + ) self.db.commit() self.db.refresh(record) - if dispatch: - return self.dispatch_event(record.event_id) return record + def _find_idempotent_event(self, idempotency_key: str | None) -> DomainEvent | None: + if not idempotency_key: + return None + return self.db.execute( + select(DomainEvent).where(DomainEvent.idempotency_key == idempotency_key) + ).scalar_one_or_none() + def list_events( self, status_filter: str | None = None, diff --git a/app/modules/events/services/service.py b/app/modules/events/services/service.py index ed0ac5d..3761aea 100644 --- a/app/modules/events/services/service.py +++ b/app/modules/events/services/service.py @@ -1,16 +1,12 @@ from sqlalchemy.orm import Session -from app.modules.events.services.dispatch import EventDispatchMixin -from app.modules.events.services.handlers import EventHandlerMixin from app.modules.events.services.query import EventQueryMixin class EventService( - EventDispatchMixin, - EventHandlerMixin, EventQueryMixin, ): - """Persist outbox events and dispatch the V3 internal handlers.""" + """Persist and query transactional outbox events.""" def __init__(self, db: Session): self.db = db diff --git a/app/modules/feishu/commands.py b/app/modules/feishu/commands.py deleted file mode 100644 index c7f615b..0000000 --- a/app/modules/feishu/commands.py +++ /dev/null @@ -1,732 +0,0 @@ -import json -import re -from typing import Any - -from fastapi import HTTPException -from sqlalchemy.orm import Session - -from app.core.constants import ActorValue -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.ai_memory.constants import AIMemoryStatus -from app.modules.ai_memory.service import AIMemoryService -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.feishu.service import FeishuService -from app.modules.market.chart import render_market_chart -from app.modules.market.service import MarketService -from app.modules.reports.chart import lifecycle_chart_alt, render_lifecycle_chart -from app.modules.reports.constants import ReportResponseKey -from app.modules.reports.services import ReportService - -DAILY_REPORT_KEYWORDS = ("日报", "晨报", "经营日报", "经营晨报") -PROJECT_WEEKLY_KEYWORDS = ("周报", "项目周报") -ATTENDANCE_KEYWORDS = ("打卡", "考勤", "attendance") -RISK_KEYWORDS = ("风险", "预警", "risk") -AI_COMMAND_PREFIXES = ("问 ", "ai ", "AI ", "/ask ") -DEFAULT_AI_PROMPT = "请说明你能做什么。" -RULE_TITLE = "AI 学习规则" -RULE_CREATE_PATTERN = re.compile(r"^学习规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$") -MARKET_RULE_CREATE_PATTERN = re.compile( - r"^学习市场规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$" -) -RULE_DISABLE_PATTERN = re.compile(r"^停用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) -RULE_ENABLE_PATTERN = re.compile(r"^启用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) -RULE_LIST_COMMANDS = {"查看规则", "规则列表", "查看市场规则"} -RULE_COMMAND_PREFIXES = ( - "学习市场规则", - "学习规则", - "查看市场规则", - "查看规则", - "规则列表", - "停用规则", - "启用规则", -) -RULE_COMMAND_HELP = ( - "规则指令格式：\n" - "学习规则：<规则内容>\n" - "学习规则 80：<规则内容>\n" - "学习市场规则 80：<仅用于市场分析的规则内容>\n" - "查看规则\n" - "停用规则 <规则编号>\n" - "启用规则 <规则编号>" -) -PROJECT_FINANCE_PATTERN = re.compile(r"^项目资金\s+(.+)$") -FINANCE_COMMANDS = {"资金需求", "未来30天资金需求"} -STOCK_ANALYSIS_PATTERN = re.compile( - r"^(?:股票分析|估值分析|财报分析)\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE -) -WATCHLIST_ADD_PATTERN = re.compile(r"^加入自选\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE) -MARKET_COMMANDS = { - "市场分析", - "今日收盘分析", - "本周市场分析", - "宏观金融分析", - "最新公告", -} -INDUSTRY_ANALYSIS_PATTERN = re.compile(r"^行业分析\s+(.+)$") -STOCK_COMPARE_PATTERN = re.compile( - r"^股票对比\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)\s+" r"([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", - re.IGNORECASE, -) - - -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(FeishuPayloadKey.TEXT) or content.get(FeishuPayloadKey.CONTENT) or "" - ) - if not isinstance(content, str): - return "" - try: - data = json.loads(content) - except json.JSONDecodeError: - return content - if isinstance(data, dict): - 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(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.""" - - def __init__(self, db: Session): - self.db = db - self.feishu = FeishuService(db) - - def extract_event_command(self, payload: dict[str, Any]) -> dict[str, Any] | None: - 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(FeishuPayloadKey.CONTENT))) - if not text: - return None - 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), - FeishuCommandKey.ACTOR: actor, - } - - def handle_text( - self, - text: str, - chat_id: str | None = None, - actor: str = ActorValue.FEISHU, - auto_reply: bool = True, - ) -> dict[str, Any]: - command_text = _clean_command_text(text) - lowered = command_text.lower() - provider_response: dict[str, Any] | None = None - - rule_result = self._handle_rule_command( - command_text, - chat_id=chat_id, - actor=actor, - auto_reply=auto_reply, - ) - if rule_result is not None: - return rule_result - - finance_result = self._handle_finance_command( - command_text, - chat_id=chat_id, - actor=actor, - auto_reply=auto_reply, - ) - if finance_result is not None: - return finance_result - - market_result = self._handle_market_command(command_text, chat_id, actor, auto_reply) - if market_result is not None: - return market_result - - if any(keyword in command_text for keyword in DAILY_REPORT_KEYWORDS): - report = ReportService(self.db).daily_brief() - if auto_reply: - provider_response = self._send_card_if_configured( - chat_id, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.LINES], - actor, - ) - 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() - if auto_reply: - provider_response = self._send_card_if_configured( - chat_id, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.LINES], - actor, - ) - 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() - if auto_reply: - provider_response = self._send_card_if_configured( - chat_id, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.LINES], - actor, - ) - 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): - report = ReportService(self.db).risk_progress() - if auto_reply: - provider_response = self._send_card_if_configured( - chat_id, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.LINES], - actor, - ) - return _command_result( - FeishuCommandName.RISK_SUMMARY, - FeishuReplyType.CARD, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.CONTENT], - provider_response, - report[ReportResponseKey.LINES], - ) - - prompt = command_text - for prefix in AI_COMMAND_PREFIXES: - if command_text.startswith(prefix): - prompt = command_text[len(prefix) :].strip() - break - if not prompt: - prompt = DEFAULT_AI_PROMPT - ai_result = AIService(self.db).ask( - prompt, - context={}, - actor=actor, - source=AuditSource.FEISHU, - ) - content = ai_result[AIResponseKey.ANSWER] - is_explicit_ai = any( - command_text.startswith(prefix) or lowered.startswith(prefix) - for prefix in AI_COMMAND_PREFIXES - ) - if auto_reply: - provider_response = self._send_text_if_configured(chat_id, content, actor) - return _command_result( - FeishuCommandName.AI_ASK if is_explicit_ai else FeishuCommandName.FALLBACK_AI, - FeishuReplyType.TEXT, - FEISHU_AI_REPLY_TITLE, - content, - provider_response, - ) - - def _handle_finance_command( - self, - command_text: str, - chat_id: str | None, - actor: str, - auto_reply: bool, - ) -> dict[str, Any] | None: - project_match = PROJECT_FINANCE_PATTERN.fullmatch(command_text) - if command_text not in FINANCE_COMMANDS and project_match is None: - return None - command = ( - FeishuCommandName.PROJECT_FINANCE if project_match else FeishuCommandName.FINANCE_NEEDS - ) - if not get_settings().finance_needs_enabled: - content = "项目资金需求分析尚未启用，请先配置并启用财务只读同步。" - provider_response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - command, - FeishuReplyType.TEXT, - "项目资金需求分析", - content, - provider_response, - ) - - project_code = project_match.group(1).strip() if project_match else None - service = ReportService(self.db) - preview = service.project_finance_needs_report( - project_code=project_code, - include_ai=False, - actor=actor, - ) - if project_code and not preview["items"]: - content = f"未找到项目“{project_code}”，请使用稳定项目编号或展示编号。" - provider_response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - command, - FeishuReplyType.TEXT, - "项目资金需求分析", - content, - provider_response, - ) - if not preview["summary"]["data_available"]: - content = "项目财务数据未接入或无有效记录，暂不生成资金分析报告。" - provider_response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - command, - FeishuReplyType.TEXT, - "项目资金需求分析", - content, - provider_response, - ) - report = service.project_finance_needs_report( - project_code=project_code, - include_ai=True, - actor=actor, - ) - ai_analysis = report.get("ai_analysis") or {} - if not ai_analysis.get(AIResponseKey.OK): - content = "AI 当前不可用，本次项目资金分析报告未发送。请检查模型服务。" - provider_response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - command, - FeishuReplyType.TEXT, - "AI 暂不可用", - content, - provider_response, - ) - provider_response = None - if auto_reply: - provider_response = self._send_finance_card_if_configured(chat_id, report, actor) - return _command_result( - command, - FeishuReplyType.CARD, - report[ReportResponseKey.TITLE], - report[ReportResponseKey.CONTENT], - provider_response, - report[ReportResponseKey.LINES], - ) - - def _handle_market_command( - self, text: str, chat_id: str | None, actor: str, auto_reply: bool - ) -> dict[str, Any] | None: - stock = STOCK_ANALYSIS_PATTERN.fullmatch(text) - add = WATCHLIST_ADD_PATTERN.fullmatch(text) - industry = INDUSTRY_ANALYSIS_PATTERN.fullmatch(text) - comparison = STOCK_COMPARE_PATTERN.fullmatch(text) - if ( - text not in MARKET_COMMANDS - and text != "查看自选" - and not stock - and not add - and not industry - and not comparison - ): - return None - if not get_settings().market_analysis_enabled: - content = "市场分析尚未启用，请配置市场数据源后启用。" - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.MARKET_OVERVIEW, - FeishuReplyType.TEXT, - "市场分析", - content, - response, - ) - service = MarketService(self.db) - if add: - if get_settings().read_only_mode: - content = "当前为只读模式，不能修改自选股。请由管理员启用操作后重试。" - response = ( - self._send_text_if_configured(chat_id, content, actor) - if auto_reply - else None - ) - return _command_result( - FeishuCommandName.WATCHLIST_ADD, - FeishuReplyType.TEXT, - "自选股", - content, - response, - ) - item = service.add_watchlist(actor, add.group(1)) - content = f"已加入自选：{item['symbol']}" - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.WATCHLIST_ADD, FeishuReplyType.TEXT, "自选股", content, response - ) - if text == "查看自选": - items = service.watchlist(actor) - content = "自选股：" + ("、".join(item["symbol"] for item in items) or "暂无") - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.WATCHLIST_LIST, FeishuReplyType.TEXT, "自选股", content, response - ) - if text == "最新公告": - items = service.announcements(limit=10)["items"] - content = ( - "最新公告：\n" - + "\n".join( - f"- {item['announcement_date']} {item['symbol'] or '市场'}：{item['title']}" - for item in items - ) - if items - else "公告元数据尚未接入。" - ) - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.MARKET_ANNOUNCEMENTS, - FeishuReplyType.TEXT, - "最新公告", - content, - response, - ) - if industry: - try: - data = service.industry_analysis(industry.group(1).strip()) - content = ( - f"{data['industry']} 平均涨跌 {data['average_pct_change']}%\n" - + "\n".join( - f"- {item['name']}（{item['symbol']}）：{item['pct_change']}%" - for item in data["items"][:10] - ) - ) - except HTTPException: - content = "未找到该行业的最新市场数据。" - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.MARKET_OVERVIEW, - FeishuReplyType.TEXT, - "行业分析", - content, - response, - ) - if comparison: - try: - content = service.compare_stocks([comparison.group(1), comparison.group(2)])[ - "content" - ] - except HTTPException: - content = "至少一只股票缺少可用行情，暂时无法比较。" - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - FeishuCommandName.STOCK_ANALYSIS, - FeishuReplyType.TEXT, - "股票对比", - content, - response, - ) - command = FeishuCommandName.STOCK_ANALYSIS if stock else FeishuCommandName.MARKET_OVERVIEW - if text == "宏观金融分析": - command = FeishuCommandName.MARKET_MACRO - try: - if stock: - report = service.stock_analysis(stock.group(1), True, actor) - elif text == "本周市场分析": - report = service.weekly_overview(include_ai=True, actor=actor) - elif text == "宏观金融分析": - report = service.macro_analysis(include_ai=True, actor=actor) - else: - report = service.market_overview(include_ai=True, actor=actor) - except HTTPException: - content = "未找到该股票的可用行情，请确认代码或先执行行情同步。" - response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result(command, FeishuReplyType.TEXT, "股票分析", content, response) - ai = report.get("ai_analysis") or {} - if not report.get("data_available"): - content = "市场数据未接入，暂不生成分析报告。" - elif not ai.get("ok"): - content = "AI 当前不可用，本次市场分析报告未发送。" - else: - content = report["content"] - response = None - if auto_reply: - if ai.get("ok"): - if text == "宏观金融分析": - response = self._send_text_if_configured(chat_id, content, actor) - else: - image = self.feishu.upload_image(render_market_chart(report), actor) - image_key = (image.get("data") or {}).get("image_key") - if not image_key: - raise ValueError("Feishu image upload did not return image_key") - card = FeishuService.build_basic_card( - report["title"], - report["lines"], - image_key=image_key, - image_alt=report["title"], - ) - response = self.feishu.send_card(card, receive_id=chat_id, actor=actor) - else: - response = self._send_text_if_configured(chat_id, content, actor) - return _command_result( - command, - ( - FeishuReplyType.CARD - if ai.get("ok") and text != "宏观金融分析" - else FeishuReplyType.TEXT - ), - report["title"], - content, - response, - report["lines"] if ai.get("ok") else None, - ) - - def _send_finance_card_if_configured( - self, - chat_id: str | None, - report: dict[str, Any], - actor: str, - ) -> dict[str, Any] | None: - settings = get_settings() - if not (settings.feishu_app_id and settings.feishu_app_secret): - return None - chart_data = { - "period": report.get("as_of"), - "finance": report.get("finance_chart_data"), - } - image_result = self.feishu.upload_image(render_lifecycle_chart(chart_data), actor) - image_key = (image_result.get("data") or {}).get("image_key") - if not image_key: - raise ValueError("Feishu image upload did not return image_key") - card = FeishuService.build_basic_card( - report[ReportResponseKey.TITLE], - report[ReportResponseKey.LINES], - image_key=image_key, - image_alt=lifecycle_chart_alt(chart_data), - ) - return self.feishu.send_card(card, receive_id=chat_id, actor=actor) - - def _handle_rule_command( - self, - command_text: str, - chat_id: str | None, - actor: str, - auto_reply: bool, - ) -> dict[str, Any] | None: - if not command_text.startswith(RULE_COMMAND_PREFIXES): - return None - - if command_text.startswith("停用规则"): - command = FeishuCommandName.RULE_DISABLE - elif command_text.startswith("启用规则"): - command = FeishuCommandName.RULE_ENABLE - elif command_text.startswith(("查看市场规则", "查看规则", "规则列表")): - command = FeishuCommandName.RULE_LIST - else: - command = FeishuCommandName.RULE_CREATE - if command in { - FeishuCommandName.RULE_CREATE, - FeishuCommandName.RULE_DISABLE, - FeishuCommandName.RULE_ENABLE, - } and get_settings().read_only_mode: - content = "当前为只读模式，不能新增或修改学习规则。请由管理员启用操作后重试。" - provider_response = ( - self._send_text_if_configured(chat_id, content, actor) if auto_reply else None - ) - return _command_result( - command, - FeishuReplyType.TEXT, - RULE_TITLE, - content, - provider_response, - ) - content = RULE_COMMAND_HELP - try: - market_create_match = MARKET_RULE_CREATE_PATTERN.fullmatch(command_text) - create_match = market_create_match or RULE_CREATE_PATTERN.fullmatch(command_text) - disable_match = RULE_DISABLE_PATTERN.fullmatch(command_text) - enable_match = RULE_ENABLE_PATTERN.fullmatch(command_text) - memory = AIMemoryService(self.db) - - if create_match: - priority = int(create_match.group(1) or 50) - rule_content = create_match.group(2).strip() - if not rule_content: - content = f"规则内容不能为空。\n\n{RULE_COMMAND_HELP}" - elif not 1 <= priority <= 100: - content = "规则优先级必须在 1 到 100 之间。" - else: - rule = memory.create_rule( - content=rule_content, - scope="market" if market_create_match else "global", - subject="market" if market_create_match else "company", - priority=priority, - tags=["feishu", *(["market"] if market_create_match else [])], - actor=actor, - ) - content = ( - "规则已学习。\n" - f"编号：{rule['code']}\n" - f"优先级：{rule['importance']}\n" - f"范围：{rule['scope']} / {rule['subject']}\n" - "状态：已启用" - ) - elif command_text in RULE_LIST_COMMANDS: - command = FeishuCommandName.RULE_LIST - rules = memory.list_rules( - scope="market" if command_text == "查看市场规则" else None, - status_filter=AIMemoryStatus.ACTIVE, - limit=20, - ) - if not rules: - content = "当前没有已启用的学习规则。" - else: - lines = ["当前已启用的学习规则："] - for rule in rules: - rule_text = str(rule["content"]) - if len(rule_text) > 80: - rule_text = f"{rule_text[:80]}…" - lines.append( - f"{rule['code']}｜优先级 {rule['importance']}｜" - f"{rule['scope']}/{rule['subject']}\n{rule_text}" - ) - content = "\n\n".join(lines) - elif disable_match or enable_match: - enabled = enable_match is not None - command = ( - FeishuCommandName.RULE_ENABLE if enabled else FeishuCommandName.RULE_DISABLE - ) - match = enable_match or disable_match - rule = memory.update_rule( - code=match.group(1), - content=None, - priority=None, - tags=None, - enabled=enabled, - actor=actor, - ) - state = "已启用" if enabled else "已停用" - content = ( - f"规则{state}。\n" - f"编号：{rule['code']}\n" - f"优先级：{rule['importance']}\n" - f"范围：{rule['scope']} / {rule['subject']}\n" - f"状态：{state}" - ) - except HTTPException as exc: - detail = str(exc.detail) - if "secret-like" in detail: - content = "规则疑似包含密码、令牌或其他密钥信息，已拒绝学习。" - elif exc.status_code == 404: - content = "没有找到该规则，请先发送“查看规则”确认规则编号。" - elif "priority" in detail: - content = "规则优先级必须在 1 到 100 之间。" - else: - content = "规则未保存，请检查指令内容后重试。" - - provider_response = None - if auto_reply: - provider_response = self._send_text_if_configured(chat_id, content, actor) - return _command_result( - command, - FeishuReplyType.TEXT, - RULE_TITLE, - content, - provider_response, - ) - - def _send_card_if_configured( - self, - chat_id: str | None, - title: str, - lines: list[str], - actor: str, - ) -> dict[str, Any] | None: - settings = get_settings() - if not (settings.feishu_app_id and settings.feishu_app_secret): - return None - card = FeishuService.build_basic_card(title, lines) - return self.feishu.send_card(card, receive_id=chat_id, actor=actor) - - def _send_text_if_configured( - self, - chat_id: str | None, - text: str, - actor: str, - ) -> dict[str, Any] | None: - settings = get_settings() - if not (settings.feishu_app_id and settings.feishu_app_secret): - return None - return self.feishu.send_text(text, receive_id=chat_id, actor=actor) diff --git a/app/modules/feishu/long_connection.py b/app/modules/feishu/long_connection.py index c74d98b..0c83e85 100644 --- a/app/modules/feishu/long_connection.py +++ b/app/modules/feishu/long_connection.py @@ -6,7 +6,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 +from app.application.feishu import FeishuEventService logger = logging.getLogger(__name__) diff --git a/app/modules/feishu/routes.py b/app/modules/feishu/routes.py index 327b032..f5ad2af 100644 --- a/app/modules/feishu/routes.py +++ b/app/modules/feishu/routes.py @@ -1,11 +1,12 @@ -from fastapi import APIRouter, Depends, Request +from typing import Any + +from fastapi import APIRouter, Depends 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.application.feishu import FeishuCommandService, FeishuEventService from app.modules.feishu.constants import FeishuEventSource, FeishuPayloadKey, FeishuResponseKey -from app.modules.feishu.events import FeishuEventService from app.modules.feishu.schemas import ( FeishuCardMessage, FeishuCommandRequest, @@ -19,10 +20,9 @@ router = APIRouter() @router.post("/webhook") -async def feishu_webhook(request: Request, db: Session = Depends(get_db)) -> dict: +def feishu_webhook(payload: dict[str, Any], db: Session = Depends(get_db)) -> dict: """Handle Feishu webhook challenge and text command events.""" - payload = await request.json() return FeishuEventService(db).handle_event( payload, source=FeishuEventSource.WEBHOOK, diff --git a/app/modules/legacy_mysql/services/project_sync.py b/app/modules/legacy_mysql/services/project_sync.py index be96e8d..1d0eb91 100644 --- a/app/modules/legacy_mysql/services/project_sync.py +++ b/app/modules/legacy_mysql/services/project_sync.py @@ -4,6 +4,7 @@ from fastapi import HTTPException, status from sqlalchemy import select from app.core.constants import ActorValue +from app.core.security import ensure_business_mutations_enabled from app.core.utils.time import utc_now from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource, AuditStatus from app.modules.audit.schemas import AuditLogCreate @@ -43,6 +44,7 @@ class LegacyProjectSyncMixin: dry_run: bool = True, actor: str = ActorValue.API, ) -> dict[str, Any]: + ensure_business_mutations_enabled() if self.db is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, @@ -120,9 +122,6 @@ class LegacyProjectSyncMixin: } ) - if not dry_run: - self.db.commit() - result = { LegacyResponseKey.DRY_RUN: dry_run, LegacyResponseKey.CREATED: created, @@ -142,11 +141,10 @@ class LegacyProjectSyncMixin: note=LEGACY_PROJECT_SYNC_NOTE, ) self.db.add(sync_run) - self.db.commit() - self.db.refresh(sync_run) + self.db.flush() result[LegacyResponseKey.SYNC_RUN_CODE] = sync_run.code - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.LEGACY_MYSQL, @@ -170,7 +168,7 @@ class LegacyProjectSyncMixin: }, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.LEGACY_SYNC_COMPLETED, source=EventSource.LEGACY_MYSQL, aggregate_type=EventAggregateType.LEGACY_SYNC_RUN, @@ -186,4 +184,6 @@ class LegacyProjectSyncMixin: }, idempotency_key=f"legacy-sync:{sync_run.code}", ) + self.db.commit() + self.db.refresh(sync_run) return result diff --git a/app/modules/legacy_mysql/services/task_sync.py b/app/modules/legacy_mysql/services/task_sync.py index f019b61..6581f5b 100644 --- a/app/modules/legacy_mysql/services/task_sync.py +++ b/app/modules/legacy_mysql/services/task_sync.py @@ -4,6 +4,7 @@ from fastapi import HTTPException, status from sqlalchemy import select from app.core.constants import ActorValue +from app.core.security import ensure_business_mutations_enabled from app.core.utils.time import utc_now from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource, AuditStatus from app.modules.audit.schemas import AuditLogCreate @@ -43,6 +44,7 @@ class LegacyTaskSyncMixin: dry_run: bool = True, actor: str = ActorValue.API, ) -> dict[str, Any]: + ensure_business_mutations_enabled() if self.db is None: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, @@ -120,9 +122,6 @@ class LegacyTaskSyncMixin: } ) - if not dry_run: - self.db.commit() - result = { LegacyResponseKey.DRY_RUN: dry_run, LegacyResponseKey.CREATED: created, @@ -142,11 +141,10 @@ class LegacyTaskSyncMixin: note=LEGACY_TASK_SYNC_NOTE, ) self.db.add(sync_run) - self.db.commit() - self.db.refresh(sync_run) + self.db.flush() result[LegacyResponseKey.SYNC_RUN_CODE] = sync_run.code - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.LEGACY_MYSQL, @@ -170,7 +168,7 @@ class LegacyTaskSyncMixin: }, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.LEGACY_SYNC_COMPLETED, source=EventSource.LEGACY_MYSQL, aggregate_type=EventAggregateType.LEGACY_SYNC_RUN, @@ -186,4 +184,6 @@ class LegacyTaskSyncMixin: }, idempotency_key=f"legacy-sync:{sync_run.code}", ) + self.db.commit() + self.db.refresh(sync_run) return result diff --git a/app/modules/observability/routes.py b/app/modules/observability/routes.py index 56eb628..fe91fc3 100644 --- a/app/modules/observability/routes.py +++ b/app/modules/observability/routes.py @@ -1,8 +1,9 @@ -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Response, status from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import require_api_key +from app.modules.observability.constants import ObservabilityKey, ObservabilityStatus from app.modules.observability.service import ObservabilityService router = APIRouter() @@ -18,9 +19,13 @@ def live( @router.get("/health/ready") def ready( + response: Response, db: Session = Depends(get_db), ) -> dict: - return ObservabilityService(db).ready() + result = ObservabilityService(db).ready() + if result[ObservabilityKey.STATUS] != ObservabilityStatus.OK: + response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE + return result @router.get("/metrics", dependencies=[Depends(require_api_key)]) diff --git a/app/modules/observability/service.py b/app/modules/observability/service.py index ac76378..8dff83a 100644 --- a/app/modules/observability/service.py +++ b/app/modules/observability/service.py @@ -94,9 +94,7 @@ class ObservabilityService: record.status = status_value record.last_seen_at = now record.updated_at = now - self.db.commit() - self.db.refresh(record) - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.OBSERVABILITY, @@ -112,6 +110,8 @@ class ObservabilityService: }, ) ) + self.db.commit() + self.db.refresh(record) return self._serialize_heartbeat(record) def heartbeat_summary(self) -> dict[str, Any]: diff --git a/app/modules/reports/routes.py b/app/modules/reports/routes.py index f97b4e8..94a08dd 100644 --- a/app/modules/reports/routes.py +++ b/app/modules/reports/routes.py @@ -3,6 +3,7 @@ from datetime import date from fastapi import APIRouter, Depends from sqlalchemy.orm import Session +from app.application.delivery import ReportDeliveryService from app.core.background.task_queue import ( enqueue_attendance_summary_push, enqueue_daily_brief_push, @@ -188,7 +189,7 @@ def push_daily_brief( ) -> dict: service = ReportService(db) report = service.daily_brief() - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, @@ -204,7 +205,7 @@ def push_project_weekly( ) -> dict: service = ReportService(db) report = service.project_weekly() - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, @@ -220,7 +221,7 @@ def push_attendance_summary( ) -> dict: service = ReportService(db) report = service.attendance_summary() - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, @@ -236,7 +237,7 @@ def push_risk_progress( ) -> dict: service = ReportService(db) report = service.risk_progress() - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, @@ -252,7 +253,7 @@ def push_work_daily( ) -> dict: service = ReportService(db) report = service.work_daily_report(reporter=principal.actor, actor=principal.actor) - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, @@ -268,7 +269,7 @@ def push_work_weekly( ) -> dict: service = ReportService(db) report = service.work_weekly_report(reporter=principal.actor, actor=principal.actor) - return service.push_report( + return ReportDeliveryService(db).push_report( report, payload.receive_id, payload.receive_id_type, diff --git a/app/modules/reports/services/enterprise.py b/app/modules/reports/services/enterprise.py index e69f414..96983ff 100644 --- a/app/modules/reports/services/enterprise.py +++ b/app/modules/reports/services/enterprise.py @@ -111,7 +111,7 @@ class ReportEnterpriseAnalyticsMixin: EnterpriseAnalyticsKey.CONTENT: "\n".join(lines), } ) - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.REPORTS, @@ -121,7 +121,7 @@ class ReportEnterpriseAnalyticsMixin: response_payload=report, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.ENTERPRISE_ANALYTICS_GENERATED, source=EventSource.ANALYTICS, aggregate_type=EventAggregateType.ENTERPRISE_ANALYTICS, @@ -132,8 +132,8 @@ class ReportEnterpriseAnalyticsMixin: EventPayloadKey.STATUS: ReportStatus.GENERATED, }, idempotency_key=f"enterprise-analytics:{code}", - dispatch=True, ) + self.db.commit() return report def _enterprise_performance_stats(self) -> dict[str, Any]: diff --git a/app/modules/reports/services/push_runs.py b/app/modules/reports/services/push_runs.py index 7731641..08fa8c6 100644 --- a/app/modules/reports/services/push_runs.py +++ b/app/modules/reports/services/push_runs.py @@ -60,6 +60,7 @@ class ReportPushRunMixin: provider_response: dict[str, Any] | None = None, error_message: str | None = None, sent: bool = False, + commit: bool = True, ) -> ReportPushRun: record = self._get_push_run(code) record.status = status @@ -70,8 +71,11 @@ class ReportPushRunMixin: record.error_message = error_message if sent: record.sent_at = utc_now() - self.db.commit() - self.db.refresh(record) + if commit: + self.db.commit() + self.db.refresh(record) + else: + self.db.flush() return record def list_push_runs( diff --git a/app/modules/reports/services/service.py b/app/modules/reports/services/service.py index ceb282b..f298a70 100644 --- a/app/modules/reports/services/service.py +++ b/app/modules/reports/services/service.py @@ -1,7 +1,6 @@ from sqlalchemy.orm import Session from app.modules.reports.services.common import ReportQueryMixin -from app.modules.reports.services.delivery import ReportDeliveryMixin from app.modules.reports.services.enterprise import ReportEnterpriseAnalyticsMixin from app.modules.reports.services.finance_needs import FinanceNeedsReportMixin from app.modules.reports.services.lifecycle import ReportLifecycleMixin @@ -13,7 +12,6 @@ from app.modules.risk.services import RiskService class ReportService( - ReportDeliveryMixin, FinanceNeedsReportMixin, IntasectLifecycleReportMixin, ReportWorkReportMixin, diff --git a/app/modules/reports/services/work_reports.py b/app/modules/reports/services/work_reports.py index 788f335..720ff4b 100644 --- a/app/modules/reports/services/work_reports.py +++ b/app/modules/reports/services/work_reports.py @@ -119,10 +119,9 @@ class ReportWorkReportMixin: risk_summary=risk_summary, ) self.db.add(record) - self.db.commit() - self.db.refresh(record) + self.db.flush() record_data = serialize_model(record) - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.REPORTS, @@ -132,7 +131,7 @@ class ReportWorkReportMixin: response_payload=record_data, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.REPORT_GENERATED, source=EventSource.REPORTS, aggregate_type=EventAggregateType.WORK_REPORT, @@ -143,8 +142,9 @@ class ReportWorkReportMixin: EventPayloadKey.STATUS: record.status, }, idempotency_key=f"report-generated:{record.code}", - dispatch=True, ) + self.db.commit() + self.db.refresh(record) return {ReportResponseKey.REPORT: report, ReportResponseKey.DATA: record_data} diff --git a/app/modules/risk/services/actions.py b/app/modules/risk/services/actions.py index c88904f..f860ee2 100644 --- a/app/modules/risk/services/actions.py +++ b/app/modules/risk/services/actions.py @@ -2,6 +2,7 @@ from typing import Any from app.core.constants import ActorValue +from app.core.security import ensure_business_mutations_enabled from app.core.utils.time import utc_now from app.modules.audit.constants import ( AuditAction, @@ -181,6 +182,7 @@ class RiskActionMixin: comment: str | None, payload: dict[str, Any], ) -> RiskEventAction: + ensure_business_mutations_enabled() action_record = RiskEventAction( code=f"RISK-ACTION-{utc_now():%Y%m%d%H%M%S%f}", risk_event_id=record.id, @@ -193,7 +195,8 @@ class RiskActionMixin: payload=payload, ) self.db.add(action_record) - AuditService(self.db).log( + self.db.flush() + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.RISK, @@ -210,7 +213,7 @@ class RiskActionMixin: }, ) ) - EventService(self.db).emit( + EventService(self.db).enqueue( event_type=EventType.RISK_ACTION_RECORDED, source=EventSource.RISK, aggregate_type=EventAggregateType.RISK_EVENT, @@ -226,7 +229,6 @@ class RiskActionMixin: RiskEventActionKey.PAYLOAD: payload, }, idempotency_key=f"risk:{record.id}:{action_record.code}", - dispatch=True, ) return action_record diff --git a/app/modules/risk/services/generation.py b/app/modules/risk/services/generation.py index 302566d..c225a81 100644 --- a/app/modules/risk/services/generation.py +++ b/app/modules/risk/services/generation.py @@ -3,6 +3,7 @@ from typing import Any from sqlalchemy import select from app.core.constants import ActorValue +from app.core.security import ensure_business_mutations_enabled from app.modules.audit.constants import ( AuditAction, AuditRiskLevel, @@ -29,6 +30,7 @@ class RiskGenerationMixin: def generate_events(self, actor: str = ActorValue.API) -> dict[str, Any]: """Generate or refresh risk-event ledger entries from current signals.""" + ensure_business_mutations_enabled() payloads = self._build_event_payloads() created = 0 updated = 0 @@ -67,8 +69,7 @@ class RiskGenerationMixin: } ) - self.db.commit() - AuditService(self.db).log( + AuditService(self.db).record( AuditLogCreate( actor=actor, source=AuditSource.RISK, @@ -82,6 +83,7 @@ class RiskGenerationMixin: }, ) ) + self.db.commit() return { RiskGenerationResultKey.CREATED: created, RiskGenerationResultKey.UPDATED: updated, diff --git a/app/modules/workflows/models.py b/app/modules/workflows/models.py index 55180d3..8d36712 100644 --- a/app/modules/workflows/models.py +++ b/app/modules/workflows/models.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import JSON, DateTime, Integer, String +from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from app.core.constants import ActorValue @@ -11,6 +11,14 @@ from app.modules.workflows.constants import WorkflowStatus class WorkflowInstance(Base): __tablename__ = "workflow_instances" + __table_args__ = ( + UniqueConstraint( + "workflow_type", + "aggregate_type", + "aggregate_id", + name="uq_workflow_aggregate", + ), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) code: Mapped[str] = mapped_column(String(64), unique=True, index=True) @@ -35,7 +43,10 @@ class WorkflowAction(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) code: Mapped[str] = mapped_column(String(64), unique=True, index=True) - workflow_code: Mapped[str] = mapped_column(String(64), index=True) + workflow_code: Mapped[str] = mapped_column( + ForeignKey("workflow_instances.code", ondelete="RESTRICT"), + index=True, + ) action: Mapped[str] = mapped_column(String(128), index=True) actor: Mapped[str] = mapped_column(String(128), default=ActorValue.SYSTEM, index=True) from_status: Mapped[str | None] = mapped_column(String(32), nullable=True) diff --git a/app/modules/workflows/routes.py b/app/modules/workflows/routes.py index 5a773c9..fb12352 100644 --- a/app/modules/workflows/routes.py +++ b/app/modules/workflows/routes.py @@ -4,12 +4,13 @@ from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import require_api_key from app.modules.workflows.constants import WorkflowResponseKey +from app.modules.workflows.schemas import WorkflowListRead, WorkflowRead from app.modules.workflows.service import WorkflowService router = APIRouter(dependencies=[Depends(require_api_key)]) -@router.get("") +@router.get("", response_model=WorkflowListRead) def list_workflows( status: str | None = None, workflow_type: str | None = None, @@ -25,7 +26,7 @@ def list_workflows( } -@router.get("/{code}") +@router.get("/{code}", response_model=dict[str, WorkflowRead]) def get_workflow( code: str, db: Session = Depends(get_db), diff --git a/app/modules/workflows/schemas.py b/app/modules/workflows/schemas.py index 13a6d8a..61259ff 100644 --- a/app/modules/workflows/schemas.py +++ b/app/modules/workflows/schemas.py @@ -15,7 +15,8 @@ class WorkflowRead(BaseModel): created_at: str updated_at: str completed_at: str | None + actions: list[dict[str, Any]] | None = None class WorkflowListRead(BaseModel): - items: list[dict[str, Any]] + items: list[WorkflowRead] diff --git a/app/modules/workflows/service.py b/app/modules/workflows/service.py index 6c6f8b1..a7ff682 100644 --- a/app/modules/workflows/service.py +++ b/app/modules/workflows/service.py @@ -33,6 +33,7 @@ class WorkflowService: action: str, actor: str = ActorValue.SYSTEM, payload: dict[str, Any] | None = None, + commit: bool = True, ) -> WorkflowInstance: aggregate_id_text = str(aggregate_id) if aggregate_id is not None else None record = self.db.execute( @@ -92,8 +93,11 @@ class WorkflowService: payload=payload or {}, ) ) - self.db.commit() - self.db.refresh(record) + if commit: + self.db.commit() + self.db.refresh(record) + else: + self.db.flush() return record def list_workflows( diff --git a/app/core/background/task_queue/constants.py b/app/tasks/constants.py similarity index 100% rename from app/core/background/task_queue/constants.py rename to app/tasks/constants.py diff --git a/app/tasks/events.py b/app/tasks/events.py index 36ddb47..4a47db9 100644 --- a/app/tasks/events.py +++ b/app/tasks/events.py @@ -1,6 +1,7 @@ from socket import gethostname from typing import Any +from app.application.events import EventDispatchService from app.core.constants import ActorValue from app.core.database import SessionLocal from app.tasks.app import celery_app, settings @@ -11,7 +12,6 @@ def dispatch_pending_events( limit: int | None = None, actor: str = ActorValue.WORKER, ) -> list[dict[str, Any]]: - from app.modules.events.services import EventService from app.modules.observability.constants import HeartbeatComponent from app.modules.observability.service import ObservabilityService @@ -22,7 +22,7 @@ def dispatch_pending_events( instance_id=gethostname(), actor=actor, ) - return EventService(db).dispatch_pending( + return EventDispatchService(db).dispatch_pending( limit=limit or settings.event_dispatch_batch_size, worker_id=f"{actor}:{gethostname()}", ) diff --git a/app/tasks/lifecycle.py b/app/tasks/lifecycle.py index 4b2af92..d9ccddc 100644 --- a/app/tasks/lifecycle.py +++ b/app/tasks/lifecycle.py @@ -1,9 +1,9 @@ from typing import Any -from app.core.background.task_queue.constants import TASK_RUN_LIFECYCLE +from app.tasks.constants import TASK_RUN_LIFECYCLE from app.core.constants import ActorValue from app.core.database import SessionLocal -from app.modules.reports.lifecycle_pipeline import LifecyclePipelineService +from app.application.pipelines import LifecyclePipelineService from app.tasks.app import celery_app diff --git a/app/tasks/market.py b/app/tasks/market.py index f50501b..c3f501c 100644 --- a/app/tasks/market.py +++ b/app/tasks/market.py @@ -1,8 +1,8 @@ from datetime import date -from app.core.background.task_queue.constants import TASK_RUN_MARKET_CLOSE +from app.tasks.constants import TASK_RUN_MARKET_CLOSE from app.core.database import SessionLocal -from app.modules.market.pipeline import MarketPipelineService +from app.application.pipelines import MarketPipelineService from app.tasks.app import celery_app diff --git a/app/tasks/reports.py b/app/tasks/reports.py index 6deea98..bc4934c 100644 --- a/app/tasks/reports.py +++ b/app/tasks/reports.py @@ -1,7 +1,7 @@ from collections.abc import Callable from typing import Any -from app.core.background.task_queue.constants import ( +from app.tasks.constants import ( TASK_PUSH_ATTENDANCE_SUMMARY, TASK_PUSH_DAILY_BRIEF, TASK_PUSH_PROJECT_WEEKLY, @@ -90,13 +90,14 @@ def _push_report( actor: str, push_run_code: str | None, ) -> dict[str, Any]: + from app.application.delivery import ReportDeliveryService from app.modules.reports.services import ReportService db = SessionLocal() try: service = ReportService(db) report = build_report(service, actor) - return service.push_report( + return ReportDeliveryService(db).push_report( report, receive_id, receive_id_type, diff --git a/app/tools/run_scheduler.py b/app/tools/run_scheduler.py index 850ecfd..8107cbf 100644 --- a/app/tools/run_scheduler.py +++ b/app/tools/run_scheduler.py @@ -1,6 +1,6 @@ from time import sleep -from app.core.background.scheduler import create_scheduler +from app.application.scheduling import create_scheduler def main() -> None: diff --git a/environment.yml b/environment.yml index 63f0d95..e3300e3 100644 --- a/environment.yml +++ b/environment.yml @@ -5,21 +5,6 @@ dependencies: - python=3.11 - pip - pip: - - fastapi==0.115.6 - - uvicorn[standard]==0.34.0 - - sqlalchemy==2.0.36 - - pymysql==1.1.1 - - psycopg[binary]==3.2.3 - - pydantic-settings==2.7.1 - - python-dotenv==1.0.1 - - alembic==1.14.0 - - httpx==0.28.1 - - lark-oapi==1.6.8 - - apscheduler==3.10.4 - - redis==5.2.1 - - celery==5.4.0 - - cryptography==44.0.0 - - pandas==2.2.3 - - pillow==11.0.0 + - -r requirements.txt - pytest==8.3.4 - ruff==0.8.4 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..06cda53 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,16 @@ +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +sqlalchemy==2.0.36 +pymysql==1.1.1 +psycopg[binary]==3.2.3 +pydantic-settings==2.7.1 +python-dotenv==1.0.1 +alembic==1.14.0 +httpx==0.28.1 +lark-oapi==1.6.8 +apscheduler==3.10.4 +redis==5.2.1 +celery==5.4.0 +cryptography==44.0.0 +pandas==2.2.3 +pillow==11.0.0 diff --git a/tests/test_architecture_hardening.py b/tests/test_architecture_hardening.py new file mode 100644 index 0000000..01fd9af --- /dev/null +++ b/tests/test_architecture_hardening.py @@ -0,0 +1,75 @@ +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, func, select +from sqlalchemy.orm import Session, sessionmaker + +from app.core.config import get_settings +from app.core.config import Settings +from app.core.database import Base +from app.core.security import OperationsDisabledError +from app.modules.audit.models import AuditLog +from app.modules.audit.schemas import AuditLogCreate +from app.modules.audit.service import AuditService +from app.modules.events.models import DomainEvent +from app.modules.events.services import EventService +from app.modules.risk.services import RiskService + + +def test_audit_and_outbox_rollback_with_business_transaction() -> None: + engine = create_engine("sqlite://") + Base.metadata.create_all(engine) + session_factory = sessionmaker(bind=engine, expire_on_commit=False) + + with session_factory() as db: + AuditService(db).record(AuditLogCreate(action="transaction.rollback")) + EventService(db).enqueue( + event_type="test.rollback", + source="pytest", + aggregate_type="test", + aggregate_id="rollback", + idempotency_key="test-transaction-rollback", + ) + db.rollback() + + with session_factory() as db: + assert db.scalar(select(func.count()).select_from(AuditLog)) == 0 + assert db.scalar(select(func.count()).select_from(DomainEvent)) == 0 + + +def test_business_service_enforces_read_only_policy(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("READ_ONLY_MODE", "true") + get_settings.cache_clear() + try: + with pytest.raises(OperationsDisabledError): + RiskService(Session()).generate_events(actor="pytest") + finally: + get_settings.cache_clear() + + +def test_production_settings_fail_closed() -> None: + with pytest.raises(ValueError, match="PostgreSQL"): + Settings( + app_env="production", + database_url="sqlite:///unsafe.db", + api_key="api", + audit_api_key="audit", + cors_origins=["https://internal.example.com"], + ) + + +def test_migrations_match_sqlalchemy_metadata( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + database_path = tmp_path / "migration-check.db" + monkeypatch.setenv("DATABASE_URL", f"sqlite:///{database_path}") + get_settings.cache_clear() + config = Config("alembic.ini") + try: + command.upgrade(config, "head") + command.check(config) + finally: + get_settings.cache_clear() diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 97b8cbc..4e8df19 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -30,12 +30,12 @@ os.environ["SCHEDULER_ENABLED"] = "false" from fastapi.testclient import TestClient from app.core.config import Settings, get_settings -from app.core.background.scheduler import create_scheduler +from app.application.scheduling import create_scheduler from app.core.database import Base, SessionLocal, engine from app.core.http.pagination import bounded_limit, bounded_offset +from app.core.http.responses import MaskedJSONResponse from app.core.security import require_api_key, require_audit_api_key from app.main import _allow_cors_credentials, app -from app.modules.audit.constants import AUDIT_REDACTED_VALUE from app.modules.ai_memory.constants import ( AIMemoryPayloadKey, AIMemoryResponseKey, @@ -48,7 +48,9 @@ from app.modules.events.constants import ( EventStatus, EventType, ) +from app.modules.events.models import DomainEvent from app.modules.events.services import EventService +from app.application.events import EventDispatchService from app.modules.business.registry import get_domain_model from app.modules.business.models import ( Employee, @@ -101,17 +103,18 @@ from app.modules.reports.constants import ( ReportTitle, ReportType, ) -from app.modules.reports.lifecycle_pipeline import LifecyclePipelineService +from app.application.pipelines import LifecyclePipelineService +from app.application.delivery import ReportDeliveryService from app.modules.reports.chart import render_lifecycle_chart from app.modules.reports.services import ReportService from app.modules.feishu.service import FeishuService -from app.modules.feishu.commands import FeishuCommandService -from app.modules.feishu.events import FeishuEventService +from app.application.feishu import FeishuCommandService +from app.application.feishu.events import FeishuEventService, _audit_event_metadata from app.modules.feishu.constants import FeishuEventSource from app.modules.risk.constants import RiskEventActionValue from app.modules.market.chart import render_market_chart from app.modules.market.service import MarketService, TushareClient, normalize_symbol -from app.modules.market.pipeline import MarketPipelineService +from app.application.pipelines import MarketPipelineService from app.modules.workflows.constants import WorkflowStatus, WorkflowType from app.modules.workflows.models import WorkflowInstance @@ -208,6 +211,9 @@ def test_feishu_webhook_routes_message_event() -> None: data = response.json() assert data["handled"] is True assert data["result"]["command"] == "risk_summary" + audit_metadata = _audit_event_metadata(payload) + assert "content" not in json.dumps(audit_metadata) + assert "test-feishu-token" not in json.dumps(audit_metadata) duplicate_response = client.post("/api/v1/integrations/feishu/webhook", json=payload) assert duplicate_response.status_code == 200 @@ -220,7 +226,7 @@ def test_feishu_webhook_routes_message_event() -> None: assert logs_response.status_code == 200 audit_payload = json.dumps(logs_response.json(), ensure_ascii=False) assert "test-feishu-token" not in audit_payload - assert AUDIT_REDACTED_VALUE in audit_payload + assert "evt-smoke-risk-001" in audit_payload def test_feishu_rule_commands_create_list_disable_and_enable(monkeypatch) -> None: @@ -371,7 +377,10 @@ def test_v3_event_idempotency_and_workflow_dispatch() -> None: EventPayloadKey.STATUS: StatusValue.OPEN, }, idempotency_key="v3-risk-idempotency", - dispatch=True, + ) + event = EventDispatchService(db).dispatch_event( + event.event_id, + worker_id="pytest-idempotency", ) duplicate = service.emit( event_type=EventType.RISK_ACTION_RECORDED, @@ -384,7 +393,10 @@ def test_v3_event_idempotency_and_workflow_dispatch() -> None: EventPayloadKey.STATUS: StatusValue.OPEN, }, idempotency_key="v3-risk-idempotency", - dispatch=True, + ) + duplicate = EventDispatchService(db).dispatch_event( + duplicate.event_id, + worker_id="pytest-idempotency-duplicate", ) assert duplicate.event_id == event.event_id assert event.status == EventStatus.PROCESSED @@ -442,7 +454,7 @@ def test_v3_ai_memory_recall_and_auto_write() -> None: "/api/v1/ai/ask", headers=headers, json={ - "prompt": "Summarize quarterly cash planning for project memory smoke", + "prompt": "Remember that project updates should use concise bullet summaries", "context": { AIMemoryPayloadKey.SCOPE: "project", AIMemoryPayloadKey.SUBJECT: "P-MEM-SMOKE", @@ -467,7 +479,7 @@ def test_v3_ai_memory_recall_and_auto_write() -> None: "/api/v1/ai/memory/recall", headers=headers, json={ - "query": "quarterly cash planning", + "query": "concise bullet summaries", "scope": "project", "subject": "P-MEM-SMOKE", }, @@ -476,6 +488,49 @@ def test_v3_ai_memory_recall_and_auto_write() -> None: assert recall_response.json()[AIMemoryResponseKey.ITEMS] +def test_ai_memory_rejects_financial_facts_and_applies_retention() -> None: + response = client.post( + "/api/v1/ai/ask", + headers=headers, + json={ + "prompt": "Remember the project cash flow and budget details for next quarter", + "context": { + AIMemoryPayloadKey.SCOPE: "project", + AIMemoryPayloadKey.SUBJECT: "P-MEM-FINANCIAL", + }, + }, + ) + assert response.status_code == 200 + memory_write = response.json()[AIResponseKey.RAW][AIResponseKey.MEMORY_WRITE] + assert memory_write[AIMemoryPayloadKey.STATUS] == AIMemoryStatus.REJECTED + + rejected = client.get( + "/api/v1/ai/memory", + headers=headers, + params={ + "scope": "project", + "subject": "P-MEM-FINANCIAL", + "status": AIMemoryStatus.REJECTED, + }, + ).json()[AIMemoryResponseKey.ITEMS] + assert rejected + assert rejected[0]["expires_at"] is not None + assert "cash flow" not in rejected[0][AIMemoryPayloadKey.CONTENT].lower() + + +def test_default_json_response_masks_sensitive_fields() -> None: + response = MaskedJSONResponse( + { + "token": "provider-token", + "nested": {"password": "provider-password", "status": "ok"}, + } + ) + payload = json.loads(response.body) + assert payload["token"] == "[MASKED]" + assert payload["nested"]["password"] == "[MASKED]" + assert payload["nested"]["status"] == "ok" + + def test_v3_risk_action_routes_are_disabled_in_read_only_mode() -> None: response = create_business_record( "risk-events", @@ -776,8 +831,8 @@ def test_new_ledgers_reports_and_risk_events() -> None: def test_independent_feishu_report_schedules_and_tasks_are_registered() -> None: - from app.core.background.scheduler import create_scheduler - from app.core.background.task_queue.constants import ( + from app.application.scheduling import create_scheduler + from app.tasks.constants import ( TASK_PUSH_ATTENDANCE_SUMMARY, TASK_PUSH_DAILY_BRIEF, TASK_PUSH_PROJECT_WEEKLY, @@ -979,6 +1034,17 @@ def test_v3_enterprise_analytics_returns_read_only_sections() -> None: db = SessionLocal() try: + event = db.execute( + select(DomainEvent).where( + DomainEvent.idempotency_key + == f"enterprise-analytics:{data[EnterpriseAnalyticsKey.CODE]}" + ) + ).scalar_one() + assert event.status == EventStatus.PENDING + EventDispatchService(db).dispatch_event( + event.event_id, + worker_id="pytest-enterprise", + ) workflow = db.execute( select(WorkflowInstance).where( WorkflowInstance.workflow_type == WorkflowType.ENTERPRISE_ANALYTICS, @@ -1438,7 +1504,11 @@ def test_lifecycle_pipeline_is_idempotent(monkeypatch) -> None: "ai_analysis": {"ok": True, "answer": "analysis"}, }, ) - monkeypatch.setattr(ReportService, "push_report", lambda self, *args, **kwargs: {"ok": True}) + monkeypatch.setattr( + ReportDeliveryService, + "push_report", + lambda self, *args, **kwargs: {"ok": True}, + ) db = SessionLocal() try: @@ -1506,7 +1576,7 @@ def test_ai_unavailable_sends_notice_without_business_report(monkeypatch) -> Non }, ) monkeypatch.setattr( - ReportService, + ReportDeliveryService, "push_report", lambda self, *args, **kwargs: pytest.fail("business report must not be sent"), ) @@ -1596,7 +1666,7 @@ def test_lifecycle_chart_is_uploaded_and_embedded_in_feishu_card(monkeypatch) -> monkeypatch.setattr(FeishuService, "send_card", fake_send_card) db = SessionLocal() try: - ReportService(db).push_report( + ReportDeliveryService(db).push_report( { "title": "Lifecycle", "report_type": "daily", @@ -2190,6 +2260,7 @@ def test_market_closed_day_skips_quote_collection() -> None: def test_market_scheduler_registers_close_and_weekly_jobs(monkeypatch) -> None: monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + monkeypatch.setenv("READ_ONLY_MODE", "false") get_settings.cache_clear() try: scheduler = create_scheduler() @@ -2199,6 +2270,7 @@ def test_market_scheduler_registers_close_and_weekly_jobs(monkeypatch) -> None: assert "market_weekly_analysis" in job_ids finally: monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + monkeypatch.delenv("READ_ONLY_MODE", raising=False) get_settings.cache_clear()