""" PMFlow 人员能力图谱 API 端点 - 技能标签 CRUD - 用户技能绑定 - 团队负载看板 - 用户个人负载查询 - 人员推荐 """ from typing import List, Optional from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, status, Query from sqlalchemy.ext.asyncio import AsyncSession from ...database import get_db from ...dependencies import get_current_user, require_roles, get_tenant_id from ...models import User from ...schemas.skill import ( SkillTagCreate, SkillTagUpdate, SkillTagResponse, UserSkillBindingCreate, UserSkillBindingUpdate, UserSkillBindingBatchUpdate, UserSkillBindingResponse, UserWorkloadResponse, PersonnelRecommendationResponse, PersonnelRecommendationItem, SkillTaskHistoryResponse, SkillTaskHistoryItem, ) from ...services import skill_service from ...services import workload_service from ...services.auth import has_any_role router = APIRouter(prefix="/skills", tags=["人员能力图谱"]) def _skill_tag_to_response(tag) -> SkillTagResponse: """将 SkillTag ORM 转为响应模型""" return SkillTagResponse( id=str(tag.id), tenant_id=tag.tenant_id, name=tag.name, slug=tag.slug, category=tag.category, description=tag.description, is_system=tag.is_system, sort_order=tag.sort_order, created_at=tag.created_at.isoformat() if tag.created_at else None, updated_at=tag.updated_at.isoformat() if tag.updated_at else None, ) # ============================================================================ # 技能标签 CRUD # ============================================================================ @router.get("/tags", response_model=List[SkillTagResponse]) async def list_skill_tags( category: Optional[str] = Query(None, description="按分类过滤"), include_system: bool = Query(True, description="是否包含系统标签"), tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 列出技能标签 支持按分类过滤，系统标签可选包含。 """ tags = await skill_service.list_skill_tags( db, tenant_id=tenant_id, category=category, include_system=include_system ) return [_skill_tag_to_response(t) for t in tags] @router.post("/tags", response_model=SkillTagResponse) async def create_skill_tag( body: SkillTagCreate, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(require_roles("BOSS", "ADMIN")), db: AsyncSession = Depends(get_db), ): """ 创建技能标签 仅 BOSS、ADMIN 可创建。 """ try: tag = await skill_service.create_skill_tag( db, tenant_id=tenant_id, name=body.name, slug=body.slug, category=body.category, description=body.description, sort_order=body.sort_order, ) return _skill_tag_to_response(tag) except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) @router.get("/tags/{tag_id}", response_model=SkillTagResponse) async def get_skill_tag( tag_id: UUID, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """获取单个技能标签""" tag = await skill_service.get_skill_tag(db, tag_id, tenant_id) if not tag: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="技能标签不存在") return _skill_tag_to_response(tag) @router.patch("/tags/{tag_id}", response_model=SkillTagResponse) async def update_skill_tag( tag_id: UUID, body: SkillTagUpdate, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(require_roles("BOSS", "ADMIN")), db: AsyncSession = Depends(get_db), ): """ 更新技能标签 系统预设标签不可修改。 """ try: tag = await skill_service.update_skill_tag( db, tag_id=tag_id, tenant_id=tenant_id, name=body.name, category=body.category, description=body.description, sort_order=body.sort_order, ) if not tag: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="技能标签不存在") return _skill_tag_to_response(tag) except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) @router.delete("/tags/{tag_id}") async def delete_skill_tag( tag_id: UUID, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(require_roles("BOSS", "ADMIN")), db: AsyncSession = Depends(get_db), ): """ 删除技能标签 系统预设标签不可删除。 """ try: ok = await skill_service.delete_skill_tag(db, tag_id, tenant_id) if not ok: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="技能标签不存在") return {"message": "已删除"} except ValueError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) # ============================================================================ # 用户技能绑定 # ============================================================================ @router.get("/users/{user_id}/bindings", response_model=List[UserSkillBindingResponse]) async def get_user_skill_bindings( user_id: int, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """获取用户的技能绑定列表""" bindings = await skill_service.get_user_skill_bindings(db, tenant_id, user_id) return [ UserSkillBindingResponse( id=str(b.id), user_id=str(b.user_id), skill_tag_id=str(b.skill_tag_id), skill_name=b.skill_tag.name if b.skill_tag else "", proficiency=b.proficiency or 50, level=b.level or b.proficiency_level or "intermediate", years_experience=b.years_experience, quality_score=getattr(b, "quality_score", None), efficiency_score=getattr(b, "efficiency_score", None), last_used_at=b.last_used_at.isoformat() if b.last_used_at else None, created_at=b.created_at.isoformat() if b.created_at else None, updated_at=b.updated_at.isoformat() if b.updated_at else None, ) for b in bindings ] @router.put("/users/{user_id}/bindings/batch") async def batch_update_user_skill_bindings( user_id: int, body: UserSkillBindingBatchUpdate, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 批量更新用户技能绑定 请求体 bindings 每项格式: { "skill_tag_id": "uuid", "proficiency": 80, "level": "advanced", "years_experience": 3 } 使用路径中的 user_id，body.user_id 可省略。 """ bindings = await skill_service.batch_update_user_skill_bindings( db, tenant_id, user_id, body.bindings ) return {"updated_count": len(bindings), "message": "已更新"} @router.get("/tags/{tag_id}/users") async def get_skill_tag_users( tag_id: int, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """按技能标签查询拥有该技能的用户""" bindings = await skill_service.get_skill_tag_user_bindings(db, tenant_id, tag_id) return [ { "user_id": str(b.user_id), "user_name": b.user.name if b.user else "", "proficiency": b.proficiency, "level": b.level, } for b in bindings ] @router.get("/users/{user_id}/tasks", response_model=SkillTaskHistoryResponse) async def get_user_skill_task_history( user_id: int, skill_id: int = Query(..., description="技能标签 ID"), limit: int = Query(50, ge=1, le=200, description="返回条数上限"), tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 查询用户某技能下的历史任务。 权限： - 用户可查询自己 - BOSS / ADMIN 可查询任意用户 """ from enum import Enum current_role = ( current_user.role.value if isinstance(current_user.role, Enum) else current_user.role ) if current_user.id != user_id and not has_any_role(str(current_role), ["BOSS", "ADMIN"]): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="仅可查询自己的技能历史任务", ) items = await skill_service.get_user_skill_task_history( db=db, tenant_id=tenant_id, user_id=user_id, skill_tag_id=skill_id, limit=limit, ) return SkillTaskHistoryResponse( items=[SkillTaskHistoryItem(**item) for item in items], total=len(items), ) # ============================================================================ # 团队负载看板 # ============================================================================ @router.get("/workload/team") async def get_team_workload_board( tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 获取团队负载看板数据 返回当前周期内所有用户的工作量汇总。 """ data = await workload_service.get_team_workload_board(db, tenant_id) return {"items": data, "total": len(data)} # ============================================================================ # 用户个人负载 # ============================================================================ @router.get("/workload/users/{user_id}", response_model=dict) async def get_user_personal_workload( user_id: int, tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 获取用户个人负载详情 用户只能查询自己的负载，BOSS/ADMIN 可查询任意用户。 """ from enum import Enum current_role = ( current_user.role.value if isinstance(current_user.role, Enum) else current_user.role ) if current_user.id != user_id and not has_any_role(str(current_role), ["BOSS", "ADMIN"]): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="仅可查询自己的负载", ) data = await workload_service.get_user_personal_workload(db, tenant_id, user_id) if not data: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="无负载数据") return data # ============================================================================ # 人员推荐 # ============================================================================ @router.get("/recommend", response_model=PersonnelRecommendationResponse) async def recommend_personnel( skills: str = Query(..., description="所需技能，逗号分隔，如: python,react,fastapi"), task_description: Optional[str] = Query(None), task_id: Optional[str] = Query(None), limit: int = Query(5, ge=1, le=20), tenant_id: int = Depends(get_tenant_id), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """ 根据所需技能推荐匹配人员 返回匹配度评分、推荐理由等。 """ required_skills = [s.strip() for s in skills.split(",") if s.strip()] result = await skill_service.recommend_personnel( db, tenant_id=tenant_id, required_skills=required_skills, task_description=task_description, task_id=task_id, limit=limit, ) return PersonnelRecommendationResponse( task_id=result.get("task_id"), task_description=result.get("task_description"), required_skills=result.get("required_skills", []), recommendations=[ PersonnelRecommendationItem(**r) for r in result.get("recommendations", []) ], total_count=result.get("total_count", 0), )