from fastapi import APIRouter, Depends, Request from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import ApiPrincipal, require_api_key from app.modules.feishu.commands import FeishuCommandService from app.modules.feishu.constants import FeishuEventSource, FeishuPayloadKey, FeishuResponseKey from app.modules.feishu.events import FeishuEventService from app.modules.feishu.schemas import ( FeishuCardMessage, FeishuCommandRequest, FeishuCommandResult, FeishuSendResult, FeishuTextMessage, ) from app.modules.feishu.service import FeishuService router = APIRouter() @router.post("/webhook") async def feishu_webhook(request: Request, 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, auto_reply=True, ) @router.post("/send-text", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) def send_text( payload: FeishuTextMessage, db: Session = Depends(get_db), principal: ApiPrincipal = Depends(require_api_key), ) -> dict: result = FeishuService(db).send_text( payload.text, receive_id=payload.receive_id, receive_id_type=payload.receive_id_type, actor=principal.actor, ) return { FeishuResponseKey.OK: result.get(FeishuPayloadKey.CODE) == 0, FeishuResponseKey.PROVIDER_RESPONSE: result, } @router.post("/send-card", response_model=FeishuSendResult, dependencies=[Depends(require_api_key)]) def send_card( payload: FeishuCardMessage, db: Session = Depends(get_db), principal: ApiPrincipal = Depends(require_api_key), ) -> dict: result = FeishuService(db).send_card( payload.card, receive_id=payload.receive_id, receive_id_type=payload.receive_id_type, actor=principal.actor, ) return { FeishuResponseKey.OK: result.get(FeishuPayloadKey.CODE) == 0, FeishuResponseKey.PROVIDER_RESPONSE: result, } @router.post( "/commands/preview", response_model=FeishuCommandResult, dependencies=[Depends(require_api_key)], ) def preview_command( payload: FeishuCommandRequest, db: Session = Depends(get_db), principal: ApiPrincipal = Depends(require_api_key), ) -> dict: """Preview local Feishu command routing without requiring webhook delivery.""" return FeishuCommandService(db).handle_text( payload.text, chat_id=payload.chat_id, actor=principal.actor, auto_reply=payload.auto_reply, )