JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from typing import Any
|
|
|
|
from app.core.constants import ActorValue
|
|
from app.core.database import SessionLocal
|
|
from app.modules.ai_agent.constants import AIResponseKey
|
|
from app.modules.ai_agent.service import AIService
|
|
from app.modules.feishu.service import FeishuService
|
|
from app.modules.subscriptions.services import (
|
|
DeliveryGenerationRequest,
|
|
DeliverySendRequest,
|
|
DeliveryService,
|
|
RetryableDeliveryError,
|
|
SubscriptionScanner,
|
|
)
|
|
|
|
|
|
class AISubscriptionGenerator:
|
|
"""Generate side-effect-free subscription content through the configured AI."""
|
|
|
|
def __init__(self, ai: AIService):
|
|
self.ai = ai
|
|
|
|
def generate(self, request: DeliveryGenerationRequest) -> str:
|
|
result = self.ai.generate_scheduled(
|
|
request.prompt,
|
|
owner_id=request.owner_id,
|
|
group=request.use_company_rules and not request.use_personal_context,
|
|
actor=ActorValue.SCHEDULER,
|
|
)
|
|
if not result.get(AIResponseKey.OK):
|
|
raise RetryableDeliveryError("AI provider is unavailable")
|
|
return str(result[AIResponseKey.ANSWER])
|
|
|
|
|
|
class FeishuSubscriptionSender:
|
|
"""Send a delivery with the stable Feishu UUID supplied by durable state."""
|
|
|
|
def __init__(self, feishu: FeishuService):
|
|
self.feishu = feishu
|
|
|
|
def send(self, request: DeliverySendRequest) -> dict[str, Any]:
|
|
return self.feishu.send_text(
|
|
request.text,
|
|
receive_id=request.receive_id,
|
|
receive_id_type=request.receive_id_type,
|
|
actor=ActorValue.SCHEDULER,
|
|
uuid=request.uuid,
|
|
tenant_key=request.tenant_key,
|
|
)
|
|
|
|
|
|
def run_subscription_cycle(
|
|
*,
|
|
actor: str = ActorValue.SCHEDULER,
|
|
) -> dict[str, Any]:
|
|
"""Materialize due windows and process pending/retry deliveries."""
|
|
|
|
_ = actor
|
|
db = SessionLocal()
|
|
try:
|
|
created = SubscriptionScanner(db).scan_due()
|
|
delivery_service = DeliveryService(
|
|
db,
|
|
generator=AISubscriptionGenerator(AIService(db)),
|
|
sender=FeishuSubscriptionSender(FeishuService(db)),
|
|
)
|
|
processed = delivery_service.process_due()
|
|
return {
|
|
"created": [item.code for item in created],
|
|
"processed": [
|
|
{"code": item.code, "status": item.status}
|
|
for item in processed
|
|
],
|
|
}
|
|
finally:
|
|
db.close()
|