68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
"""Shared helper functions for signal source APIs."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ...models import Project
|
||
|
||
RECIPE_PROJECT_KEYWORDS = ("食谱", "recipe", "sstc")
|
||
|
||
|
||
async def resolve_target_project_id(
|
||
*,
|
||
db: AsyncSession,
|
||
tenant_id: int,
|
||
project_id: Optional[int],
|
||
) -> int:
|
||
"""解析目标项目：优先显式 project_id，其次按关键字自动匹配。"""
|
||
if project_id is not None:
|
||
result = await db.execute(
|
||
select(Project.id).where(
|
||
Project.tenant_id == tenant_id,
|
||
Project.id == int(project_id),
|
||
).limit(1)
|
||
)
|
||
found = result.scalar_one_or_none()
|
||
if found is None:
|
||
raise HTTPException(status_code=404, detail=f"project_id={project_id} 不存在或无权限")
|
||
return int(found)
|
||
|
||
keyword_conditions = []
|
||
for keyword in RECIPE_PROJECT_KEYWORDS:
|
||
keyword_conditions.extend(
|
||
[
|
||
Project.title.ilike(f"%{keyword}%"),
|
||
Project.name.ilike(f"%{keyword}%"),
|
||
]
|
||
)
|
||
|
||
match_result = await db.execute(
|
||
select(Project.id).where(
|
||
Project.tenant_id == tenant_id,
|
||
or_(*keyword_conditions),
|
||
).limit(1)
|
||
)
|
||
matched = match_result.scalar_one_or_none()
|
||
if matched is not None:
|
||
return int(matched)
|
||
|
||
active_result = await db.execute(
|
||
select(Project.id).where(
|
||
Project.tenant_id == tenant_id,
|
||
Project.status == "active",
|
||
).limit(2)
|
||
)
|
||
active_ids = list(active_result.scalars().all())
|
||
if len(active_ids) == 1:
|
||
return int(active_ids[0])
|
||
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail="未找到目标项目，请在请求中显式传入 project_id（或确保项目名包含 食谱/recipe/sstc）",
|
||
)
|