You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
# backend/app/services/risk.py
|
|
from sqlmodel import Session, select
|
|
|
|
from ..config import get_settings
|
|
from ..models import Person, Project, Task
|
|
from ..schemas import RiskOut
|
|
|
|
STATUS_LABEL = {"todo": "시작 전", "doing": "진행 중", "waiting": "대기 중", "review": "검토 중"}
|
|
DEPS = [
|
|
{"blocker": "예산 섹션 작성", "blocked": "경영진 검토 요청 메일"},
|
|
{"blocker": "데이터 전처리 파이프라인", "blocked": "사용자 인터뷰 5건 정리"},
|
|
]
|
|
|
|
|
|
def _due_day(t: Task) -> int:
|
|
return t.due.day if t.due else 99
|
|
|
|
|
|
def _due_txt(t: Task) -> str:
|
|
return f"6/{t.due.day}" if t.due else ""
|
|
|
|
|
|
def _status_value(t: Task) -> str:
|
|
return t.status.value if hasattr(t.status, "value") else t.status
|
|
|
|
|
|
def compute_risks(s: Session, area: str = "work") -> list[RiskOut]:
|
|
today = get_settings().risk_today # 8
|
|
all_tasks = s.exec(select(Task)).all()
|
|
people = {p.id: p for p in s.exec(select(Person)).all()}
|
|
|
|
# area 필터: 작업이 속한 project 의 folder 로 판정
|
|
proj_folder = {p.id: p.folder_id for p in s.exec(select(Project)).all()}
|
|
|
|
def in_area(t: Task) -> bool:
|
|
return proj_folder.get(t.project_id) == area
|
|
|
|
scoped = [t for t in all_tasks if in_area(t)]
|
|
roots = [t for t in scoped if t.parent_id is None]
|
|
|
|
risks: list[RiskOut] = []
|
|
|
|
# ① 지연 위험 (최상위 작업, 첫 1건)
|
|
late = [
|
|
t for t in sorted(roots, key=lambda x: x.sort_order)
|
|
if _status_value(t) != "done" and t.due and _due_day(t) <= today
|
|
]
|
|
if late:
|
|
t = late[0]
|
|
risks.append(
|
|
RiskOut(
|
|
kind="지연 위험", icon="clock", tone="coral", task_id=t.id, cta="작업 열기",
|
|
text=f"**{t.title}** — 오늘({_due_txt(t)}) 마감인데 아직 {STATUS_LABEL.get(_status_value(t), '')}이에요", # noqa: E501
|
|
)
|
|
)
|
|
|
|
# ② 업무 쏠림 (전체 노드 카운트)
|
|
counts: dict[str, int] = {}
|
|
for n in scoped:
|
|
if _status_value(n) != "done" and n.assignee_id:
|
|
counts[n.assignee_id] = counts.get(n.assignee_id, 0) + 1
|
|
entries = [(w, c) for w, c in counts.items() if w in people]
|
|
if len(entries) > 1:
|
|
entries.sort(key=lambda x: -x[1])
|
|
top_who, top_n = entries[0]
|
|
avg = sum(c for _, c in entries) / len(entries)
|
|
if top_n >= avg * 1.5 and top_n >= 4:
|
|
mult = round(top_n / avg, 1)
|
|
p = people[top_who]
|
|
if p.is_me:
|
|
text = (
|
|
f"미완료 작업 **{top_n}건**이 내게 몰려 있어요 — 팀 평균의 **{mult}배**. "
|
|
f"벅찬 작업은 위임 & 추적으로 넘겨보세요"
|
|
)
|
|
else:
|
|
text = (
|
|
f"**{p.name}님**에게 미완료 작업 **{top_n}건**이 몰려 있어요 — "
|
|
f"평균의 **{mult}배**, 일부 재배분을 추천드려요"
|
|
)
|
|
risks.append(RiskOut(kind="업무 쏠림", icon="scale", tone="amber", text=text))
|
|
|
|
# ③ 의존성 (가장 임박한 1건)
|
|
def find_by_title(title: str) -> Task | None:
|
|
for n in scoped:
|
|
if n.title == title:
|
|
return n
|
|
return None
|
|
|
|
for d in DEPS:
|
|
a, b = find_by_title(d["blocker"]), find_by_title(d["blocked"])
|
|
if a and b and _status_value(a) != "done" and _status_value(b) != "done":
|
|
suffix = f"({_due_txt(b)})" if b.due else ""
|
|
risks.append(
|
|
RiskOut(
|
|
kind="의존성", icon="link", tone="violet", task_id=b.id, cta="후속 작업 보기",
|
|
text=f"**{d['blocker']}**이(가) 늦어지면 **{d['blocked']}**{suffix}까지 함께 밀려요", # noqa: E501
|
|
)
|
|
)
|
|
break
|
|
|
|
return risks[:3]
|