220 lines
12 KiB
Python
220 lines
12 KiB
Python
"""Отчёт «По сотрудникам»: количество и качество выполненных нарядов и задач за период, трудоёмкость; список и календарь.
|
||
|
||
Количество: выполненные наряды (исполнитель, дата завершения в периоде) и задачи/поручения (исполнитель, день задачи в периоде).
|
||
Качество: доля выполненного в срок, просрочки, возвраты нарядов с проверки, отклонения от нормы в чек-листах, отменённые
|
||
и приостановленные задачи. Трудоёмкость, ч: у задачи — норма, если указана, иначе фактическое время в статусе «в работе»;
|
||
у наряда — фактическое время от начала до сдачи (не больше 24 ч на наряд)."""
|
||
import csv
|
||
import io
|
||
from collections import defaultdict
|
||
from datetime import date, datetime, timedelta
|
||
|
||
from fpdf import FontFace
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from . import config
|
||
from .models import DailyTask, Department, Order, OrderItem, Position, TaskEvent, User
|
||
from .reports import FONTS, Report, fmt_dt, safe_cell
|
||
|
||
MAX_ORDER_HOURS = 24.0
|
||
|
||
|
||
def _hours(a: datetime | None, b: datetime | None, cap: float = MAX_ORDER_HOURS) -> float:
|
||
if not a or not b or b <= a:
|
||
return 0.0
|
||
return min(cap, (b - a).total_seconds() / 3600)
|
||
|
||
|
||
def _task_fact_hours(events: list[TaskEvent], end: datetime) -> float:
|
||
"""Сумма интервалов, когда задача была «в работе»."""
|
||
total, started = 0.0, None
|
||
for e in sorted(events, key=lambda x: (x.at, x.id)):
|
||
if e.to_status == "in_progress" and e.kind != "transfer":
|
||
started = started or e.at
|
||
elif started and e.to_status != "in_progress":
|
||
total += _hours(started, e.at, cap=1e9)
|
||
started = None
|
||
if started:
|
||
total += _hours(started, min(end, config.now()), cap=1e9)
|
||
return min(total, 24.0 * 7)
|
||
|
||
|
||
def people(db: Session, start: datetime, end: datetime, department_id: int | None = None, position_id: int | None = None,
|
||
user_id: int | None = None) -> dict:
|
||
deps = {d.id: d for d in db.scalars(select(Department)).all()}
|
||
|
||
def dep_tree(root: int) -> set[int]:
|
||
out, grown = {root}, True
|
||
while grown:
|
||
grown = False
|
||
for d in deps.values():
|
||
if d.id not in out and d.parent_id in out:
|
||
out.add(d.id)
|
||
grown = True
|
||
return out
|
||
allowed_deps = dep_tree(department_id) if department_id else None
|
||
users = {u.id: u for u in db.scalars(select(User).order_by(User.full_name, User.username)).all()
|
||
if (not user_id or u.id == user_id) and (not position_id or u.position_id == position_id)
|
||
and (allowed_deps is None or u.department_id in allowed_deps)}
|
||
first, last = start.date(), (end - timedelta(seconds=1)).date()
|
||
days = [first + timedelta(days=i) for i in range((last - first).days + 1)]
|
||
|
||
def blank():
|
||
return {"orders_done": 0, "orders_on_time": 0, "orders_overdue": 0, "returns": 0, "deviations": 0, "order_hours": 0.0,
|
||
"tasks_total": 0, "tasks_done": 0, "tasks_on_time": 0, "tasks_cancelled": 0, "tasks_paused": 0, "tasks_open": 0,
|
||
"task_hours": 0.0, "norm_hours": 0.0, "fact_hours": 0.0, "calendar": defaultdict(lambda: {"n": 0, "h": 0.0})}
|
||
stat: dict[int, dict] = {uid: blank() for uid in users}
|
||
|
||
# наряды: выполнены в периоде
|
||
orders = db.scalars(select(Order).where(Order.status == "completed", Order.completed_at >= start, Order.completed_at < end,
|
||
Order.executor_id.in_(list(users) or [-1]))).all()
|
||
for o in orders:
|
||
s = stat[o.executor_id]
|
||
s["orders_done"] += 1
|
||
late = bool(o.due_at and (o.submitted_at or o.completed_at) > o.due_at)
|
||
s["orders_overdue" if late else "orders_on_time"] += 1
|
||
s["returns"] += o.returns or 0
|
||
s["deviations"] += sum(1 for it in o.items if it.out_of_range)
|
||
hrs = _hours(o.started_at or o.created_at, o.submitted_at or o.completed_at)
|
||
s["order_hours"] += hrs
|
||
cell = s["calendar"][o.completed_at.date().isoformat()]
|
||
cell["n"] += 1
|
||
cell["h"] += hrs
|
||
|
||
# задачи и поручения: день задачи в периоде
|
||
tasks = db.scalars(select(DailyTask).where(DailyTask.day >= first, DailyTask.day <= last,
|
||
DailyTask.assignee_id.in_(list(users) or [-1]))).all()
|
||
evs: dict[int, list] = defaultdict(list)
|
||
if tasks:
|
||
for e in db.scalars(select(TaskEvent).where(TaskEvent.task_id.in_([t.id for t in tasks]))).all():
|
||
evs[e.task_id].append(e)
|
||
for t in tasks:
|
||
s = stat[t.assignee_id]
|
||
s["tasks_total"] += 1
|
||
s["tasks_paused"] += sum(1 for e in evs[t.id] if e.to_status == "paused" and e.kind != "transfer")
|
||
if t.status == "cancelled":
|
||
s["tasks_cancelled"] += 1
|
||
continue
|
||
if t.status != "done":
|
||
s["tasks_open"] += 1
|
||
continue
|
||
s["tasks_done"] += 1
|
||
if t.finished_at and t.finished_at.date() <= t.day:
|
||
s["tasks_on_time"] += 1
|
||
fact = _task_fact_hours(evs[t.id], t.finished_at or end)
|
||
hrs = t.labor_hours if t.labor_hours is not None else fact
|
||
s["task_hours"] += hrs
|
||
s["norm_hours" if t.labor_hours is not None else "fact_hours"] += hrs
|
||
cell = s["calendar"][t.day.isoformat()]
|
||
cell["n"] += 1
|
||
cell["h"] += hrs
|
||
|
||
rows = []
|
||
for uid, u in users.items():
|
||
s = stat[uid]
|
||
done = s["orders_done"] + s["tasks_done"]
|
||
if not done and not s["tasks_total"] and user_id is None:
|
||
continue # без активности за период — не показываем (если не выбран конкретный человек)
|
||
on_time = s["orders_on_time"] + s["tasks_on_time"]
|
||
hours = s["order_hours"] + s["task_hours"]
|
||
rows.append({
|
||
"user_id": uid, "name": u.full_name or u.username, "position": u.position.name if u.position else "",
|
||
"department": deps[u.department_id].name if u.department_id in deps else "",
|
||
"done": done, "orders_done": s["orders_done"], "tasks_done": s["tasks_done"], "tasks_total": s["tasks_total"],
|
||
"tasks_open": s["tasks_open"], "on_time_pct": round(on_time * 100 / done) if done else None,
|
||
"overdue": s["orders_overdue"] + (s["tasks_done"] - s["tasks_on_time"]), "returns": s["returns"], "deviations": s["deviations"],
|
||
"tasks_cancelled": s["tasks_cancelled"], "tasks_paused": s["tasks_paused"], "hours": round(hours, 1),
|
||
"calendar": {k: {"n": v["n"], "h": round(v["h"], 1)} for k, v in sorted(s["calendar"].items())},
|
||
})
|
||
rows.sort(key=lambda r: (-r["done"], r["name"]))
|
||
totals = {k: sum(r[k] for r in rows) for k in ("done", "orders_done", "tasks_done", "overdue", "returns", "deviations", "tasks_cancelled")}
|
||
totals["hours"] = round(sum(r["hours"] for r in rows), 1)
|
||
return {"days": [d.isoformat() for d in days], "rows": rows, "totals": totals}
|
||
|
||
|
||
def to_csv(data: dict, calendar: bool = False) -> bytes:
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=";")
|
||
if calendar:
|
||
w.writerow(["Сотрудник", *(d[8:10] + "." + d[5:7] for d in data["days"]), "Итого задач", "Итого ч"])
|
||
for r in data["rows"]:
|
||
cells = [f"{r['calendar'][d]['n']} / {r['calendar'][d]['h']:g} ч" if d in r["calendar"] else "" for d in data["days"]]
|
||
w.writerow([safe_cell(r["name"]), *cells, r["done"], r["hours"]])
|
||
else:
|
||
w.writerow(["Сотрудник", "Должность", "Отдел", "Выполнено всего", "Нарядов", "Задач", "В срок, %", "Просрочено", "Возвраты с проверки",
|
||
"Отклонения от нормы", "Отменено задач", "Приостановок", "Трудоёмкость, ч"])
|
||
for r in data["rows"]:
|
||
w.writerow([safe_cell(x) for x in (r["name"], r["position"], r["department"], r["done"], r["orders_done"], r["tasks_done"],
|
||
"—" if r["on_time_pct"] is None else r["on_time_pct"], r["overdue"], r["returns"], r["deviations"],
|
||
r["tasks_cancelled"], r["tasks_paused"], r["hours"])])
|
||
return ("" + buf.getvalue()).encode("utf-8")
|
||
|
||
|
||
def _fill(p):
|
||
if p is None:
|
||
return (240, 240, 240)
|
||
return (220, 252, 231) if p >= 90 else (254, 243, 199) if p >= 70 else (254, 226, 226)
|
||
|
||
|
||
def to_pdf(data: dict, start: datetime, end: datetime, filters_text: list[str], who: str) -> bytes:
|
||
pdf = Report(orientation="L", unit="mm", format="A4")
|
||
pdf.add_font("DejaVu", "", str(FONTS / "DejaVuSans.ttf"))
|
||
pdf.add_font("DejaVu", "B", str(FONTS / "DejaVuSans-Bold.ttf"))
|
||
pdf.set_margins(10, 10, 10)
|
||
pdf.set_auto_page_break(True, 14)
|
||
pdf.alias_nb_pages()
|
||
pdf.add_page()
|
||
pdf.set_font("DejaVu", "B", 15)
|
||
pdf.cell(0, 8, "Отчёт: работа сотрудников", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("DejaVu", "", 9)
|
||
pdf.set_text_color(70)
|
||
pdf.cell(0, 5, f"Период: {fmt_dt(start)} — {fmt_dt(end)}", new_x="LMARGIN", new_y="NEXT")
|
||
if filters_text:
|
||
pdf.cell(0, 5, "Фильтры: " + "; ".join(filters_text), new_x="LMARGIN", new_y="NEXT")
|
||
t = data["totals"]
|
||
pdf.cell(0, 5, f"Сформировано: {fmt_dt(config.now())}, {who}. Выполнено: {t['done']} (нарядов {t['orders_done']}, задач {t['tasks_done']}), "
|
||
f"трудоёмкость {t['hours']:g} ч", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(3)
|
||
pdf.set_text_color(0)
|
||
pdf.set_font("DejaVu", "", 8.5)
|
||
if not data["rows"]:
|
||
pdf.cell(0, 8, "За период выполненных работ нет.", new_x="LMARGIN", new_y="NEXT")
|
||
return bytes(pdf.output())
|
||
head = FontFace(emphasis="BOLD", fill_color=(230, 233, 240))
|
||
with pdf.table(col_widths=(52, 36, 34, 17, 15, 15, 17, 17, 17, 17, 16, 18), line_height=5, headings_style=head, padding=1.2, repeat_headings=1,
|
||
text_align=("LEFT", "LEFT", "LEFT") + ("CENTER",) * 9) as tb:
|
||
hd = tb.row()
|
||
for c in ("Сотрудник", "Должность", "Отдел", "Выполн.", "Наряды", "Задачи", "В срок", "Просроч.", "Возвраты", "Отклон.", "Отмены", "Часы"):
|
||
hd.cell(c)
|
||
for r in data["rows"]:
|
||
row = tb.row()
|
||
for v in (r["name"], r["position"], r["department"], str(r["done"]), str(r["orders_done"]), str(r["tasks_done"])):
|
||
row.cell(v)
|
||
row.cell("—" if r["on_time_pct"] is None else f"{r['on_time_pct']}%", style=FontFace(fill_color=_fill(r["on_time_pct"])))
|
||
for v in (r["overdue"], r["returns"], r["deviations"], r["tasks_cancelled"]):
|
||
row.cell(str(v) if v else "—")
|
||
row.cell(f"{r['hours']:g}")
|
||
days = data["days"]
|
||
if len(days) <= 31:
|
||
pdf.add_page()
|
||
pdf.set_font("DejaVu", "B", 12)
|
||
pdf.cell(0, 7, "Календарь: выполнено задач и нарядов / часов по дням", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("DejaVu", "", 6.5)
|
||
w_day = (277 - 45 - 16) / max(1, len(days))
|
||
with pdf.table(col_widths=(45, *([w_day] * len(days)), 16), line_height=4, headings_style=head, padding=0.6, repeat_headings=1,
|
||
text_align=("LEFT",) + ("CENTER",) * (len(days) + 1)) as tb:
|
||
hd = tb.row()
|
||
hd.cell("Сотрудник")
|
||
for d in days:
|
||
hd.cell(d[8:10])
|
||
hd.cell("Итого")
|
||
for r in data["rows"]:
|
||
row = tb.row()
|
||
row.cell(r["name"])
|
||
for d in days:
|
||
c = r["calendar"].get(d)
|
||
row.cell(f"{c['n']}\n{c['h']:g}ч" if c else "", style=FontFace(fill_color=(219, 234, 254)) if c else None)
|
||
row.cell(f"{r['done']}\n{r['hours']:g}ч")
|
||
return bytes(pdf.output())
|