273 lines
13 KiB
Python
273 lines
13 KiB
Python
"""Отчёты: выборка данных и экспорт (PDF / CSV)."""
|
||
import csv
|
||
import io
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from fpdf import FPDF, FontFace
|
||
from fpdf.enums import Align
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.orm import Session
|
||
|
||
from . import config, service as svc
|
||
from .models import Order, User, Position, Equipment, Location, Checklist, ServiceUnit
|
||
|
||
FONTS = Path(__file__).parent / "fonts"
|
||
MAX_ROWS = 20000
|
||
STATUS_RU = {"completed": "Выполнен", "review": "На проверке"}
|
||
|
||
|
||
@dataclass
|
||
class Filters:
|
||
start: datetime
|
||
end: datetime
|
||
equipment_id: int | None = None
|
||
position_id: int | None = None
|
||
user_id: int | None = None
|
||
include_review: bool = False
|
||
|
||
|
||
def fmt_dt(dt: datetime | None) -> str:
|
||
return dt.strftime("%d.%m.%Y %H:%M") if dt else "—"
|
||
|
||
|
||
def completed_orders(db: Session, f: Filters) -> list[dict]:
|
||
"""Выполненные наряды за период: машина, чек-лист, кто выполнил, должность, когда."""
|
||
when = func.coalesce(Order.completed_at, Order.submitted_at)
|
||
statuses = ["completed", "review"] if f.include_review else ["completed"]
|
||
q = select(Order).where(Order.status.in_(statuses), when >= f.start, when <= f.end)
|
||
if f.equipment_id:
|
||
q = q.where(Order.equipment_id == f.equipment_id)
|
||
if f.user_id:
|
||
q = q.where(Order.executor_id == f.user_id)
|
||
if f.position_id:
|
||
q = q.where(Order.executor_position_id == f.position_id)
|
||
orders = db.scalars(q.order_by(when, Order.id).limit(MAX_ROWS)).all()
|
||
users = {u.id: u for u in db.scalars(select(User))}
|
||
pos = dict(db.execute(select(Position.id, Position.name)).all())
|
||
eq = {e.id: e for e in db.scalars(select(Equipment))}
|
||
loc = dict(db.execute(select(Location.id, Location.name)).all())
|
||
cl = dict(db.execute(select(Checklist.id, Checklist.name)).all())
|
||
rows = []
|
||
for o in orders:
|
||
ex = users.get(o.executor_id)
|
||
pid = o.executor_position_id or (ex.position_id if ex else None)
|
||
e = eq.get(o.equipment_id)
|
||
rv = users.get(o.reviewed_by_id)
|
||
rows.append({
|
||
"order_id": o.id,
|
||
"finished_at": o.completed_at or o.submitted_at,
|
||
"equipment": e.name if e else "—",
|
||
"inventory_no": e.inventory_no if e else "",
|
||
"location": loc.get(o.location_id) or "",
|
||
"checklist": cl.get(o.checklist_id) or o.title,
|
||
"title": o.title,
|
||
"executor": (ex.full_name or ex.username) if ex else "—",
|
||
"position": pos.get(pid) or "—",
|
||
"status": o.status,
|
||
"reviewer": (rv.full_name or rv.username) if rv else "",
|
||
"deviations": sum(1 for i in o.items if i.out_of_range),
|
||
"items_done": sum(1 for i in o.items if i.value not in (None, "")),
|
||
"items_total": len(o.items),
|
||
})
|
||
return rows
|
||
|
||
|
||
def describe_filters(db: Session, f: Filters) -> list[str]:
|
||
out = []
|
||
if f.equipment_id and (e := db.get(Equipment, f.equipment_id)):
|
||
out.append(f"машина: {e.name}")
|
||
if f.position_id and (p := db.get(Position, f.position_id)):
|
||
out.append(f"должность: {p.name}")
|
||
if f.user_id and (u := db.get(User, f.user_id)):
|
||
out.append(f"исполнитель: {u.full_name or u.username}")
|
||
if f.include_review:
|
||
out.append("включая наряды на проверке")
|
||
return out
|
||
|
||
|
||
def safe_cell(v):
|
||
"""Защита CSV от формул Excel: значения, начинающиеся с = + - @, экранируются апострофом."""
|
||
if isinstance(v, str) and v[:1] in ("=", "+", "-", "@", "\t", "\r"):
|
||
return "'" + v
|
||
return v
|
||
|
||
|
||
# ---------- CSV ----------
|
||
def to_csv(rows: list[dict]) -> bytes:
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=";")
|
||
w.writerow(["№ наряда", "Когда", "Машина", "Инв. №", "Место", "Чек-лист", "Выполнил", "Должность", "Статус", "Отклонений"])
|
||
for r in rows:
|
||
w.writerow([safe_cell(x) for x in (r["order_id"], fmt_dt(r["finished_at"]), r["equipment"], r["inventory_no"], r["location"],
|
||
r["checklist"], r["executor"], r["position"], STATUS_RU.get(r["status"], r["status"]), r["deviations"])])
|
||
return ("" + buf.getvalue()).encode("utf-8") # BOM — чтобы Excel понял кодировку
|
||
|
||
|
||
# ---------- PDF ----------
|
||
class Report(FPDF):
|
||
def footer(self):
|
||
self.set_y(-12)
|
||
self.set_font("DejaVu", "", 8)
|
||
self.set_text_color(120)
|
||
self.cell(0, 6, f"Стр. {self.page_no()} / {{nb}}", align=Align.R)
|
||
|
||
|
||
def to_pdf(title: str, f: Filters, filters_text: list[str], rows: list[dict], 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, title, new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("DejaVu", "", 9)
|
||
pdf.set_text_color(70)
|
||
pdf.cell(0, 5, f"Период: {fmt_dt(f.start)} — {fmt_dt(f.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}. Нарядов: {len(rows)}", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(3)
|
||
pdf.set_text_color(0)
|
||
pdf.set_font("DejaVu", "", 8.5)
|
||
widths = (16, 30, 52, 55, 45, 46, 18, 15)
|
||
if rows:
|
||
with pdf.table(col_widths=widths, line_height=5, text_align=("RIGHT", "LEFT", "LEFT", "LEFT", "LEFT", "LEFT", "LEFT", "CENTER"),
|
||
headings_style=FontFace(emphasis="BOLD", fill_color=(230, 233, 240)),
|
||
padding=1.2, repeat_headings=1) as t:
|
||
h = t.row()
|
||
for c in ("№", "Когда", "Машина", "Чек-лист", "Выполнил", "Должность", "Статус", "Откл."):
|
||
h.cell(c)
|
||
for r in rows:
|
||
row = t.row()
|
||
row.cell(str(r["order_id"]))
|
||
row.cell(fmt_dt(r["finished_at"]))
|
||
row.cell(r["equipment"] + (f" ({r['inventory_no']})" if r["inventory_no"] else ""))
|
||
row.cell(r["checklist"])
|
||
row.cell(r["executor"])
|
||
row.cell(r["position"])
|
||
row.cell("Выполн." if r["status"] == "completed" else "Проверка")
|
||
row.cell(str(r["deviations"]) if r["deviations"] else "—")
|
||
else:
|
||
pdf.cell(0, 8, "За выбранный период выполненных нарядов нет.", new_x="LMARGIN", new_y="NEXT")
|
||
return bytes(pdf.output())
|
||
|
||
|
||
# ======================= «Текущее состояние»: обслуживаемые узлы машин =======================
|
||
@dataclass
|
||
class StateFilters:
|
||
equipment_id: int | None = None
|
||
location_id: int | None = None
|
||
state: str = "" # "" — все; due | soon | attention (пора + скоро) | ok
|
||
include_inactive: bool = False
|
||
|
||
|
||
def _fmt_d(iso: str | None) -> str:
|
||
return datetime.fromisoformat(iso).strftime("%d.%m.%Y") if iso else "—"
|
||
|
||
|
||
def equipment_state(db: Session, f: StateFilters) -> list[dict]:
|
||
"""По строке на каждый обслуживаемый узел: правило, когда обслуживался, остаток, прогноз и состояние."""
|
||
q = select(ServiceUnit, Equipment).join(Equipment, Equipment.id == ServiceUnit.equipment_id)
|
||
if not f.include_inactive:
|
||
q = q.where(ServiceUnit.is_active.is_(True), Equipment.is_active.is_(True))
|
||
if f.equipment_id:
|
||
q = q.where(Equipment.id == f.equipment_id)
|
||
if f.location_id:
|
||
q = q.where(Equipment.location_id == f.location_id)
|
||
loc = dict(db.execute(select(Location.id, Location.name)).all())
|
||
rates, rows = {}, []
|
||
for u, e in db.execute(q.order_by(Equipment.name, ServiceUnit.name)).all():
|
||
if e.id not in rates:
|
||
rates[e.id] = svc.daily_rate(db, e.id)
|
||
ev = svc.evaluate(u, e.meter_hours or 0, rates[e.id])
|
||
rows.append({
|
||
"equipment_id": e.id, "equipment": e.name, "inventory_no": e.inventory_no, "location": loc.get(e.location_id) or "",
|
||
"unit_id": u.id, "unit": u.name, "rule": ev["rule"], "state": ev["state"], "state_ru": svc.STATE_RU[ev["state"]],
|
||
"percent": ev["percent"], "meter_hours": e.meter_hours or 0, "last_service_at": config.iso(u.last_service_at),
|
||
"last_service_hours": u.last_service_hours, "used_hours": ev["used_hours"], "remaining_hours": ev["remaining_hours"],
|
||
"due_date": ev["due_date"], "remaining_days": ev["remaining_days"], "forecast_date": ev["forecast_date"],
|
||
"daily_rate": round(rates[e.id], 1) if rates[e.id] else None, "is_active": u.is_active,
|
||
})
|
||
rows.sort(key=lambda r: (svc.state_rank(r["state"]), -r["percent"], r["equipment"], r["unit"]))
|
||
return rows
|
||
|
||
|
||
def filter_state(rows: list[dict], state: str) -> list[dict]:
|
||
if state == "attention":
|
||
return [r for r in rows if r["state"] in ("due", "soon")]
|
||
return [r for r in rows if r["state"] == state] if state else rows
|
||
|
||
|
||
def remaining_text(r: dict) -> str:
|
||
bits = []
|
||
if r["remaining_hours"] is not None:
|
||
v = r["remaining_hours"]
|
||
bits.append(f"{svc.num(v)} м/ч" if v > 0 else f"просрочено {svc.num(-v)} м/ч")
|
||
if r["remaining_days"] is not None:
|
||
v = r["remaining_days"]
|
||
bits.append(f"{v} дн." if v > 0 else f"просрочено {-v} дн." if v < 0 else "сегодня")
|
||
return "; ".join(bits) or "—"
|
||
|
||
|
||
def state_csv(rows: list[dict]) -> bytes:
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=";")
|
||
w.writerow(["Машина", "Инв. №", "Место", "Узел", "Правило", "Состояние", "Выработано, %", "Наработка машины, м/ч",
|
||
"Обслужен", "Наработка при обслуживании, м/ч", "Остаток", "Срок (календарь)", "Прогноз даты"])
|
||
for r in rows:
|
||
w.writerow([safe_cell(x) for x in (
|
||
r["equipment"], r["inventory_no"], r["location"], r["unit"], r["rule"], r["state_ru"], r["percent"],
|
||
svc.num(r["meter_hours"]), _fmt_d(r["last_service_at"]), svc.num(r["last_service_hours"]),
|
||
remaining_text(r), _fmt_d(r["due_date"]), _fmt_d(r["forecast_date"]))])
|
||
return ("" + buf.getvalue()).encode("utf-8")
|
||
|
||
|
||
STATE_FILL = {"due": (254, 226, 226), "soon": (254, 243, 199), "ok": (220, 252, 231), "nodata": (235, 235, 235)}
|
||
|
||
|
||
def state_pdf(f: StateFilters, filters_text: list[str], rows: list[dict], 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)
|
||
if filters_text:
|
||
pdf.cell(0, 5, "Фильтры: " + "; ".join(filters_text), new_x="LMARGIN", new_y="NEXT")
|
||
cnt = {k: sum(1 for r in rows if r["state"] == k) for k in ("due", "soon", "ok")}
|
||
pdf.cell(0, 5, f"Сформировано: {fmt_dt(config.now())}, {who}. Узлов: {len(rows)} (требуется: {cnt['due']}, скоро: {cnt['soon']}, в норме: {cnt['ok']})",
|
||
new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(3)
|
||
pdf.set_text_color(0)
|
||
pdf.set_font("DejaVu", "", 8.5)
|
||
widths = (50, 38, 44, 22, 28, 28, 35, 32)
|
||
if rows:
|
||
with pdf.table(col_widths=widths, line_height=5, text_align=("LEFT",) * 8,
|
||
headings_style=FontFace(emphasis="BOLD", fill_color=(230, 233, 240)), padding=1.2, repeat_headings=1) as t:
|
||
h = t.row()
|
||
for c in ("Машина", "Узел", "Правило", "Состояние", "Обслужен", "Наработка (м/ч)", "Остаток", "Срок / прогноз"):
|
||
h.cell(c)
|
||
for r in rows:
|
||
row = t.row()
|
||
row.cell(r["equipment"] + (f" ({r['inventory_no']})" if r["inventory_no"] else ""))
|
||
row.cell(r["unit"])
|
||
row.cell(r["rule"])
|
||
row.cell(f"{r['state_ru']} · {r['percent']}%", style=FontFace(fill_color=STATE_FILL[r["state"]]))
|
||
row.cell(_fmt_d(r["last_service_at"]) if r["last_service_at"] else "—")
|
||
row.cell(f"сейчас {svc.num(r['meter_hours'])}; было {svc.num(r['last_service_hours'])}")
|
||
row.cell(remaining_text(r))
|
||
row.cell("; ".join(x for x in (("срок " + _fmt_d(r["due_date"])) if r["due_date"] else "",
|
||
("прогноз " + _fmt_d(r["forecast_date"])) if r["forecast_date"] else "") if x) or "—")
|
||
else:
|
||
pdf.cell(0, 8, "Обслуживаемых узлов по выбранным условиям нет.", new_x="LMARGIN", new_y="NEXT")
|
||
return bytes(pdf.output())
|