"""Отчёт «Выполнение работ»: чек-листы, суточные осмотры, ТО и суточные задания по оборудованию, % и отклонения.""" 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())