87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
import json
|
|
from sqlalchemy import select, or_, func, case
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import User
|
|
from app.schemas.user import UserPublic
|
|
|
|
|
|
async def search_directory(
|
|
db: AsyncSession,
|
|
class_id: int,
|
|
search: str | None = None,
|
|
industry: str | None = None,
|
|
company: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
) -> tuple[list[User], int]:
|
|
"""Search approved members in a class."""
|
|
query = select(User).where(
|
|
User.class_id == class_id, User.status == "approved"
|
|
)
|
|
count_query = select(func.count(User.id)).where(
|
|
User.class_id == class_id, User.status == "approved"
|
|
)
|
|
|
|
if search:
|
|
search_term = f"%{search}%"
|
|
query = query.where(
|
|
or_(
|
|
User.name.ilike(search_term),
|
|
User.company.ilike(search_term),
|
|
User.position.ilike(search_term),
|
|
)
|
|
)
|
|
count_query = count_query.where(
|
|
or_(
|
|
User.name.ilike(search_term),
|
|
User.company.ilike(search_term),
|
|
User.position.ilike(search_term),
|
|
)
|
|
)
|
|
|
|
if industry:
|
|
query = query.where(User.industry == industry)
|
|
count_query = count_query.where(User.industry == industry)
|
|
|
|
if company:
|
|
company_term = f"%{company}%"
|
|
query = query.where(User.company.ilike(company_term))
|
|
count_query = count_query.where(User.company.ilike(company_term))
|
|
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
# Committee role priority: 班长(1) > 副班长(2) > other roles(3) > no role(4)
|
|
committee_order = case(
|
|
(User.committee_role == None, 4),
|
|
(User.committee_role == "班长", 1),
|
|
(User.committee_role == "副班长", 2),
|
|
else_=3,
|
|
)
|
|
|
|
result = await db.execute(
|
|
query.order_by(committee_order, User.committee_role, User.name)
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
)
|
|
users = list(result.scalars().all())
|
|
return users, total
|
|
|
|
|
|
def user_to_public(user: User, include_contact: bool = True) -> UserPublic:
|
|
"""Convert User model to public profile, optionally hiding contact info."""
|
|
return UserPublic(
|
|
id=user.id,
|
|
name=user.name,
|
|
student_id=user.student_id,
|
|
industry=user.industry,
|
|
company=user.company,
|
|
position=user.position,
|
|
committee_role=user.committee_role,
|
|
wechat_id=user.wechat_id if include_contact else None,
|
|
phone=user.phone if include_contact else None,
|
|
avatar_url=user.avatar_url,
|
|
bio=user.bio,
|
|
)
|