220 lines
12 KiB
Python
220 lines
12 KiB
Python
"""Отчёт «Выполнение работ»: чек-листы, суточные осмотры, ТО и суточные задания по оборудованию, % и отклонения."""
|
||
import csv
|
||
import io
|
||
from datetime import datetime, timedelta
|
||
|
||
from fpdf import FontFace
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from . import config
|
||
from .logic import item_done
|
||
from .models import Checklist, Comment, DailyTask, Equipment, Location, Order, OrderTemplate, TaskEvent, User
|
||
from .reports import FONTS, Report, fmt_dt, safe_cell
|
||
|
||
CATS = {"daily": "Суточные осмотры", "inspection": "Чек-листы", "regular": "ТО", "tasks": "Задачи и поручения"}
|
||
MAX_DEVIATIONS = 30 # на одну машину
|
||
|
||
|
||
def _pct(done: int, planned: int) -> int | None:
|
||
return round(done * 100 / planned) if planned else None
|
||
|
||
|
||
def execution(db: Session, start: datetime, end: datetime, equipment_id: int | None = None,
|
||
location_id: int | None = None) -> dict:
|
||
"""Выполнение по оборудованию за период (по дате создания наряда / дню задания).
|
||
|
||
Наряд по чек-листу «регламентное обслуживание» — ТО; наряд из ежедневного расписания — суточный осмотр;
|
||
остальные — чек-листы. Процент = выполнено / запланировано (отменённые считаются невыполненными).
|
||
Отклонения: значения вне нормы, возвраты с проверки, комментарии к нарядам, отмены нарядов, комментарии
|
||
к приостановленным и отменённым заданиям."""
|
||
eqs = {e.id: e for e in db.scalars(select(Equipment)).all()}
|
||
locs = dict(db.execute(select(Location.id, Location.name)).all())
|
||
kinds = dict(db.execute(select(Checklist.id, Checklist.kind)).all())
|
||
daily_tpl = {t for t, in db.execute(select(OrderTemplate.id).where(OrderTemplate.recurrence == "daily"))}
|
||
names = {i: (n or un) for i, n, un in db.execute(select(User.id, User.full_name, User.username))}
|
||
|
||
def wanted(eid):
|
||
if equipment_id and eid != equipment_id:
|
||
return False
|
||
if location_id and (eid is None or eqs[eid].location_id != location_id):
|
||
return False
|
||
return True
|
||
|
||
rows: dict = {}
|
||
|
||
def row(eid):
|
||
if eid not in rows:
|
||
e = eqs.get(eid)
|
||
rows[eid] = {"equipment_id": eid, "equipment": e.name if e else "Без оборудования",
|
||
"inventory_no": e.inventory_no if e else "", "location": locs.get(e.location_id, "") if e else "",
|
||
"cats": {k: {"planned": 0, "done": 0, "percent": None} for k in CATS},
|
||
"items_total": 0, "items_done": 0, "deviations": []}
|
||
return rows[eid]
|
||
|
||
def dev(r, at, kind, text, ref):
|
||
if len(r["deviations"]) < MAX_DEVIATIONS:
|
||
r["deviations"].append({"at": config.iso(at), "kind": kind, "text": text, **ref})
|
||
|
||
orders = db.scalars(select(Order).where(Order.created_at >= start, Order.created_at < end).order_by(Order.created_at)).all()
|
||
comments: dict[int, list] = {}
|
||
if orders:
|
||
for c in db.scalars(select(Comment).where(Comment.order_id.in_([o.id for o in orders])).order_by(Comment.id)).all():
|
||
comments.setdefault(c.order_id, []).append(c)
|
||
for o in orders:
|
||
if not wanted(o.equipment_id):
|
||
continue
|
||
r = row(o.equipment_id)
|
||
cat = "regular" if kinds.get(o.checklist_id) == "regular" else "daily" if o.template_id in daily_tpl else "inspection"
|
||
c = r["cats"][cat]
|
||
c["planned"] += 1
|
||
c["done"] += o.status == "completed"
|
||
if o.status != "cancelled":
|
||
r["items_total"] += len(o.items)
|
||
r["items_done"] += sum(item_done(i) for i in o.items)
|
||
ref = {"order_id": o.id, "title": o.title, "category": CATS[cat]}
|
||
for it in o.items:
|
||
if it.out_of_range:
|
||
lo = "…" if it.min_value is None else f"{it.min_value:g}"
|
||
hi = "…" if it.max_value is None else f"{it.max_value:g}"
|
||
dev(r, it.answered_at or o.created_at, "out_of_range",
|
||
f"{it.text}: {it.value} {it.unit} (норма {lo}–{hi} {it.unit})".replace(" ", " ").strip(), ref)
|
||
if o.review_comment:
|
||
dev(r, o.submitted_at or o.created_at, "review", "Проверка: " + o.review_comment, ref)
|
||
for cm in comments.get(o.id, []):
|
||
dev(r, cm.created_at, "comment", f"{names.get(cm.user_id, '—')}: {cm.text}", ref)
|
||
if o.status == "cancelled":
|
||
dev(r, o.updated_at or o.created_at, "cancelled", "Наряд отменён", ref)
|
||
|
||
last_day = (end - timedelta(seconds=1)).date()
|
||
tasks = db.scalars(select(DailyTask).where(DailyTask.day >= start.date(), DailyTask.day <= last_day)).all()
|
||
tev: dict[int, list] = {}
|
||
if tasks:
|
||
for e in db.scalars(select(TaskEvent).where(TaskEvent.task_id.in_([t.id for t in tasks]),
|
||
TaskEvent.to_status.in_(("paused", "cancelled")))
|
||
.order_by(TaskEvent.at)).all():
|
||
tev.setdefault(e.task_id, []).append(e)
|
||
for t in tasks:
|
||
if not wanted(t.equipment_id):
|
||
continue
|
||
r = row(t.equipment_id)
|
||
c = r["cats"]["tasks"]
|
||
c["planned"] += 1
|
||
c["done"] += t.status == "done"
|
||
for e in tev.get(t.id, []):
|
||
what = "приостановлено" if e.to_status == "paused" else "отменено"
|
||
dev(r, e.at, "task_" + e.to_status, f"Задание {what}: {e.comment}",
|
||
{"task_id": t.id, "title": t.title, "category": CATS["tasks"]})
|
||
|
||
out = []
|
||
summary = {k: {"planned": 0, "done": 0} for k in CATS}
|
||
for r in sorted(rows.values(), key=lambda x: (x["equipment_id"] is None, x["equipment"])):
|
||
planned = done = 0
|
||
for k, c in r["cats"].items():
|
||
c["percent"] = _pct(c["done"], c["planned"])
|
||
planned += c["planned"]
|
||
done += c["done"]
|
||
summary[k]["planned"] += c["planned"]
|
||
summary[k]["done"] += c["done"]
|
||
r["total"] = {"planned": planned, "done": done, "percent": _pct(done, planned)}
|
||
r["items_percent"] = _pct(r["items_done"], r["items_total"])
|
||
r["deviations"].sort(key=lambda d: d["at"] or "")
|
||
out.append(r)
|
||
for v in summary.values():
|
||
v["percent"] = _pct(v["done"], v["planned"])
|
||
tp = sum(v["planned"] for v in summary.values())
|
||
td = sum(v["done"] for v in summary.values())
|
||
return {"rows": out, "summary": summary, "total": {"planned": tp, "done": td, "percent": _pct(td, tp)},
|
||
"categories": CATS}
|
||
|
||
|
||
def _p(c: dict) -> str:
|
||
return "—" if not c["planned"] else f"{c['done']}/{c['planned']} ({c['percent']}%)"
|
||
|
||
|
||
def _ref(d: dict) -> str:
|
||
return (f"наряд №{d['order_id']}" if d.get("order_id") else f"задание №{d.get('task_id')}") + f": {d['title']}"
|
||
|
||
|
||
def to_csv(data: dict) -> bytes:
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=";")
|
||
w.writerow(["Машина", "Инв. №", "Место", *CATS.values(), "Итого", "Пункты чек-листов, %", "Отклонений"])
|
||
for r in data["rows"]:
|
||
w.writerow([safe_cell(x) for x in (r["equipment"], r["inventory_no"], r["location"],
|
||
*(_p(r["cats"][k]) for k in CATS), _p(r["total"]),
|
||
"—" if r["items_percent"] is None else r["items_percent"], len(r["deviations"]))])
|
||
w.writerow([])
|
||
w.writerow(["Отклонения"])
|
||
w.writerow(["Машина", "Когда", "Категория", "Наряд/задание", "Описание"])
|
||
for r in data["rows"]:
|
||
for d in r["deviations"]:
|
||
w.writerow([safe_cell(x) for x in (r["equipment"], (d["at"] or "")[:16].replace("T", " "),
|
||
d["category"], _ref(d), d["text"])])
|
||
return ("" + buf.getvalue()).encode("utf-8") # BOM — чтобы Excel понял кодировку
|
||
|
||
|
||
def _fill(p: int | None):
|
||
if p is None:
|
||
return (240, 240, 240)
|
||
return (220, 252, 231) if p >= 90 else (254, 243, 199) if p >= 60 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")
|
||
pdf.cell(0, 5, f"Сформировано: {fmt_dt(config.now())}, {who}. Всего выполнено: {_p(data['total'])}",
|
||
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=(62, 32, 32, 32, 32, 32, 25, 20), line_height=5, text_align=("LEFT",) + ("CENTER",) * 7,
|
||
headings_style=head, padding=1.2, repeat_headings=1) as tb:
|
||
hd = tb.row()
|
||
for c in ("Машина", *CATS.values(), "Итого", "Пункты, %", "Откл."):
|
||
hd.cell(c)
|
||
for r in data["rows"]:
|
||
row = tb.row()
|
||
row.cell(r["equipment"] + (f" ({r['inventory_no']})" if r["inventory_no"] else ""))
|
||
for k in CATS:
|
||
c = r["cats"][k]
|
||
row.cell(_p(c), style=FontFace(fill_color=_fill(c["percent"])))
|
||
row.cell(_p(r["total"]), style=FontFace(emphasis="BOLD", fill_color=_fill(r["total"]["percent"])))
|
||
row.cell("—" if r["items_percent"] is None else f"{r['items_percent']}%")
|
||
row.cell(str(len(r["deviations"])) if r["deviations"] else "—")
|
||
devs = [(r, d) for r in data["rows"] for d in r["deviations"]]
|
||
if devs:
|
||
pdf.ln(4)
|
||
pdf.set_font("DejaVu", "B", 12)
|
||
pdf.cell(0, 7, "Отклонения и комментарии", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("DejaVu", "", 8.5)
|
||
with pdf.table(col_widths=(50, 28, 30, 55, 114), line_height=5, text_align=("LEFT",) * 5,
|
||
headings_style=head, padding=1.2, repeat_headings=1) as tb:
|
||
hd = tb.row()
|
||
for c in ("Машина", "Когда", "Категория", "Наряд / задание", "Описание"):
|
||
hd.cell(c)
|
||
for r, d in devs:
|
||
row = tb.row()
|
||
row.cell(r["equipment"])
|
||
row.cell(fmt_dt(datetime.fromisoformat(d["at"]).replace(tzinfo=None)) if d["at"] else "—")
|
||
row.cell(d["category"])
|
||
row.cell(_ref(d))
|
||
row.cell(d["text"][:400])
|
||
return bytes(pdf.output())
|