368 lines
13 KiB
Python
368 lines
13 KiB
Python
"""Генерация Excel для учётных документов."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import zipfile
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import xlrd
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Alignment, Border, Font, Side
|
|
from openpyxl.utils import get_column_letter
|
|
from flask import Flask
|
|
|
|
from app import db
|
|
from app.models import Ingredient, LoadingReport, LoadingReportComponent, Recipe
|
|
from app.services.feed_accounting.sp20_report_builder import Sp20Report, build_journal_rows, build_sp20_report
|
|
from app.services.sklad_metrics import list_stock_balance_items
|
|
|
|
|
|
def _template_path(app: Flask) -> Path:
|
|
p = Path(app.root_path).parent / "static" / "templates" / "forma-sp-20.xls"
|
|
if not p.is_file():
|
|
p = Path(app.static_folder or "") / "templates" / "forma-sp-20.xls"
|
|
return p
|
|
|
|
|
|
def _header_font():
|
|
return Font(bold=True)
|
|
|
|
|
|
def build_stock_balances_xlsx(app: Flask, date_from: str, date_to: str) -> bytes:
|
|
items = list_stock_balance_items(
|
|
db,
|
|
LoadingReport,
|
|
LoadingReportComponent,
|
|
Ingredient,
|
|
Recipe,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
)
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Остатки"
|
|
ws.append(["Учёт кормов — остатки на складе"])
|
|
ws.append([f"Период: {date_from} — {date_to}"])
|
|
ws.append([])
|
|
headers = [
|
|
"Компонент",
|
|
"Всего, кг",
|
|
"Поступление, кг",
|
|
"Израсходовано, кг",
|
|
"Осталось, кг",
|
|
"План кг/сут",
|
|
"Осталось дней",
|
|
]
|
|
ws.append(headers)
|
|
for h in range(1, len(headers) + 1):
|
|
ws.cell(row=4, column=h).font = _header_font()
|
|
for item in items:
|
|
ws.append(
|
|
[
|
|
item["name"],
|
|
item["total_kg"],
|
|
item["inflow_kg"],
|
|
item["consumed_total_kg"],
|
|
item["remaining_kg"],
|
|
item.get("planned_consumption_per_day_kg") or "",
|
|
item.get("days_left_plan") if item.get("days_left_plan") is not None else "",
|
|
]
|
|
)
|
|
for col in range(1, 8):
|
|
ws.column_dimensions[get_column_letter(col)].width = 16
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
_XLRD_LINE_STYLE = {
|
|
0: None,
|
|
1: "thin",
|
|
2: "medium",
|
|
3: "dashed",
|
|
4: "dotted",
|
|
5: "thick",
|
|
6: "double",
|
|
7: "hair",
|
|
8: "mediumDashed",
|
|
9: "dashDot",
|
|
10: "mediumDashDot",
|
|
11: "dashDotDot",
|
|
12: "mediumDashDotDot",
|
|
13: "slantDashDot",
|
|
}
|
|
|
|
_XLRD_HOR_ALIGN = {
|
|
0: "general",
|
|
1: "left",
|
|
2: "center",
|
|
3: "right",
|
|
4: "fill",
|
|
5: "justify",
|
|
6: "centerContinuous",
|
|
}
|
|
|
|
_XLRD_VERT_ALIGN = {
|
|
0: "bottom",
|
|
1: "top",
|
|
2: "center",
|
|
3: "justify",
|
|
}
|
|
|
|
|
|
def _xlrd_side(line_style: int) -> Side:
|
|
style = _XLRD_LINE_STYLE.get(line_style)
|
|
if not style:
|
|
return Side()
|
|
return Side(style=style)
|
|
|
|
|
|
def _xlrd_border(xf_border) -> Border:
|
|
return Border(
|
|
left=_xlrd_side(xf_border.left_line_style),
|
|
right=_xlrd_side(xf_border.right_line_style),
|
|
top=_xlrd_side(xf_border.top_line_style),
|
|
bottom=_xlrd_side(xf_border.bottom_line_style),
|
|
)
|
|
|
|
|
|
def _xlrd_font(xls_book, xf) -> Font:
|
|
f = xls_book.font_list[xf.font_index]
|
|
kwargs: dict = {}
|
|
if f.bold:
|
|
kwargs["bold"] = True
|
|
if f.italic:
|
|
kwargs["italic"] = True
|
|
if f.underline_type:
|
|
kwargs["underline"] = "single"
|
|
if f.height:
|
|
kwargs["size"] = max(6.0, f.height / 20.0)
|
|
if f.name:
|
|
kwargs["name"] = f.name
|
|
return Font(**kwargs) if kwargs else Font()
|
|
|
|
|
|
def _xlrd_alignment(xf) -> Alignment | None:
|
|
a = xf.alignment
|
|
kwargs: dict = {}
|
|
if a.hor_align in _XLRD_HOR_ALIGN:
|
|
kwargs["horizontal"] = _XLRD_HOR_ALIGN[a.hor_align]
|
|
if a.vert_align in _XLRD_VERT_ALIGN:
|
|
kwargs["vertical"] = _XLRD_VERT_ALIGN[a.vert_align]
|
|
if a.text_wrapped:
|
|
kwargs["wrap_text"] = True
|
|
if a.rotation:
|
|
kwargs["text_rotation"] = a.rotation
|
|
return Alignment(**kwargs) if kwargs else None
|
|
|
|
|
|
def _copy_cell_styles(xls_sheet, xls_book, ws) -> None:
|
|
"""Границы, шрифты, выравнивание — без них форма не похожа на бланк."""
|
|
for r in range(xls_sheet.nrows):
|
|
for c in range(xls_sheet.ncols):
|
|
xf_idx = xls_sheet.cell_xf_index(r, c)
|
|
xf = xls_book.xf_list[xf_idx]
|
|
cell = ws.cell(row=r + 1, column=c + 1)
|
|
cell.border = _xlrd_border(xf.border)
|
|
font = _xlrd_font(xls_book, xf)
|
|
if font != Font():
|
|
cell.font = font
|
|
alignment = _xlrd_alignment(xf)
|
|
if alignment is not None:
|
|
cell.alignment = alignment
|
|
|
|
|
|
def _copy_row_heights(xls_sheet, ws) -> None:
|
|
for row_idx, info in xls_sheet.rowinfo_map.items():
|
|
if info.height:
|
|
ws.row_dimensions[row_idx + 1].height = info.height / 20.0
|
|
|
|
|
|
def _copy_merged_cells(xls_sheet, ws) -> None:
|
|
"""Перенос объединённых ячеек из xlrd в openpyxl (без этого форма «разъезжается»)."""
|
|
for rlo, rhi, clo, chi in xls_sheet.merged_cells:
|
|
ws.merge_cells(
|
|
start_row=rlo + 1,
|
|
end_row=rhi,
|
|
start_column=clo + 1,
|
|
end_column=chi,
|
|
)
|
|
|
|
|
|
def _copy_column_widths(xls_sheet, ws) -> None:
|
|
for col_idx, colinfo in xls_sheet.colinfo_map.items():
|
|
letter = get_column_letter(col_idx + 1)
|
|
# xlrd: ширина в 1/256 символа; openpyxl — приблизительно в символах
|
|
ws.column_dimensions[letter].width = max(2.0, colinfo.width / 256.0)
|
|
|
|
|
|
def _copy_xls_sheet_to_ws(xls_sheet, xls_book, ws) -> None:
|
|
for r in range(xls_sheet.nrows):
|
|
for c in range(xls_sheet.ncols):
|
|
val = xls_sheet.cell_value(r, c)
|
|
if val != "" and val is not None:
|
|
ws.cell(row=r + 1, column=c + 1, value=val)
|
|
_copy_cell_styles(xls_sheet, xls_book, ws)
|
|
_copy_merged_cells(xls_sheet, ws)
|
|
_copy_column_widths(xls_sheet, ws)
|
|
_copy_row_heights(xls_sheet, ws)
|
|
|
|
|
|
def _fill_sp20_data_rows(
|
|
ws,
|
|
lines,
|
|
year: int,
|
|
month: int,
|
|
*,
|
|
start_row: int,
|
|
end_row: int,
|
|
) -> int:
|
|
"""Заполняет строки таблицы (дата, головы, корм, кг). Возвращает индекс следующей строки."""
|
|
row_idx = start_row
|
|
for line in lines:
|
|
if row_idx > end_row:
|
|
break
|
|
ws.cell(row=row_idx, column=1, value=f"{line.day:02d}.{month:02d}.{year}")
|
|
ws.cell(row=row_idx, column=3, value=line.heads or "")
|
|
ws.cell(row=row_idx, column=7, value=line.feed_name)
|
|
ws.cell(row=row_idx, column=10, value=line.quantity_kg)
|
|
row_idx += 1
|
|
return row_idx
|
|
|
|
|
|
def _fill_sp20_sheet1(ws, report: Sp20Report) -> None:
|
|
org = report.org
|
|
ws.cell(row=8, column=4, value=org.get("organization_name") or "")
|
|
ws.cell(row=8, column=27, value=org.get("okpo") or "")
|
|
ws.cell(row=9, column=5, value=org.get("department") or "")
|
|
ws.cell(row=10, column=2, value=org.get("farm_name") or "")
|
|
ws.cell(row=11, column=2, value=org.get("brigade") or "")
|
|
ws.cell(row=7, column=27, value=date(report.year, report.month, 1).strftime("%d.%m.%Y"))
|
|
ws.cell(row=14, column=1, value=report.animal_type)
|
|
ws.cell(row=13, column=13, value=org.get("responsible_person") or org.get("zootechnician") or "")
|
|
if report.lines:
|
|
norm = report.lines[0].norm_per_head
|
|
if norm is not None:
|
|
ws.cell(row=15, column=7, value=norm)
|
|
_fill_sp20_data_rows(ws, report.lines, report.year, report.month, start_row=20, end_row=31)
|
|
|
|
|
|
def _fill_sp20_sheet2(ws, report: Sp20Report) -> None:
|
|
"""Подписи и продолжение таблицы на стр. 2 формы."""
|
|
org = report.org
|
|
remaining = list(report.lines[12:]) if len(report.lines) > 12 else []
|
|
if remaining:
|
|
_fill_sp20_data_rows(ws, remaining, report.year, report.month, start_row=5, end_row=15)
|
|
ws.cell(row=23, column=9, value=org.get("zootechnician") or "")
|
|
ws.cell(row=23, column=26, value=org.get("brigade") or org.get("warehouse_keeper") or "")
|
|
ws.cell(row=26, column=18, value=org.get("responsible_person") or "")
|
|
|
|
|
|
def _add_groups_sheet(wb: Workbook, report: Sp20Report) -> None:
|
|
ws = wb.create_sheet("Группы выгрузки")
|
|
days = report.days_in_month
|
|
header = ["Группа", "Рецепт", "Тип", "Значение"] + [str(d) for d in range(1, days + 1)] + ["Итого, кг"]
|
|
ws.append(header)
|
|
for cell in ws[1]:
|
|
cell.font = _header_font()
|
|
for g in report.groups:
|
|
row = [
|
|
g.get("group_name"),
|
|
g.get("recipe_name"),
|
|
g.get("distribution_type"),
|
|
g.get("distribution_value"),
|
|
]
|
|
daily = g.get("daily") or {}
|
|
total = 0.0
|
|
for d in range(1, days + 1):
|
|
v = float(daily.get(d, 0) or 0)
|
|
row.append(round(v, 2) if v else "")
|
|
total += v
|
|
row.append(round(total, 2))
|
|
ws.append(row)
|
|
|
|
|
|
def _fill_journal_sheet(ws, journal_rows, org, year, month) -> None:
|
|
ws.append(["Журнал учёта расхода кормов"])
|
|
ws.append([org.get("organization_name") or ""])
|
|
ws.append([f"Период: {month:02d}.{year}"])
|
|
ws.append([])
|
|
headers = ["Вид корма", "Ед.изм.", "Расход за месяц, кг", "С начала года, кг"]
|
|
ws.append(headers)
|
|
header_row = ws.max_row
|
|
for cell in ws[header_row]:
|
|
cell.font = _header_font()
|
|
for row in journal_rows:
|
|
ws.append([row["name"], row["unit"], row["month_kg"], row["ytd_kg"]])
|
|
|
|
|
|
def build_groups_xlsx(app: Flask, year: int, month: int) -> bytes:
|
|
report = build_sp20_report(app, year, month)
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
_add_groups_sheet(wb, report)
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def build_sp20_xlsx(app: Flask, year: int, month: int, *, include_journal: bool = False) -> bytes:
|
|
"""Только официальные листы стр1 + стр2 (как forma-sp-20). Журнал и группы — отдельные файлы."""
|
|
_ = include_journal # legacy query param, больше не добавляет листы в СП-20
|
|
report = build_sp20_report(app, year, month)
|
|
template = _template_path(app)
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
if template.is_file():
|
|
book = xlrd.open_workbook(str(template), formatting_info=True)
|
|
for si, name in enumerate(book.sheet_names()[:2]):
|
|
xsh = book.sheet_by_index(si)
|
|
ws = wb.create_sheet(title=name[:31])
|
|
_copy_xls_sheet_to_ws(xsh, book, ws)
|
|
_fill_sp20_sheet1(wb.worksheets[0], report)
|
|
if len(wb.worksheets) > 1:
|
|
_fill_sp20_sheet2(wb.worksheets[1], report)
|
|
else:
|
|
ws = wb.create_sheet("СП-20")
|
|
ws.append(["Ведомость учёта расхода кормов (форма СП-20)"])
|
|
_fill_sp20_sheet1(ws, report)
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def build_journal_xlsx(app: Flask, year: int, month: int) -> bytes:
|
|
from app.services.org_settings import get_org_settings
|
|
|
|
org = get_org_settings(app)
|
|
journal = build_journal_rows(app, year, month)
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "Журнал расхода"
|
|
_fill_journal_sheet(ws, journal, org, year, month)
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def build_documents_zip(app: Flask, year: int, month: int, date_from: str, date_to: str) -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(
|
|
f"WESP_ostatki_{date_from}_{date_to}.xlsx",
|
|
build_stock_balances_xlsx(app, date_from, date_to),
|
|
)
|
|
zf.writestr(f"WESP_SP20_{year}-{month:02d}.xlsx", build_sp20_xlsx(app, year, month))
|
|
zf.writestr(f"WESP_gruppy_{year}-{month:02d}.xlsx", build_groups_xlsx(app, year, month))
|
|
zf.writestr(f"WESP_zhurnal_{year}-{month:02d}.xlsx", build_journal_xlsx(app, year, month))
|
|
try:
|
|
from app.services.feed_accounting.pdf_feed_accounting import build_journal_pdf, build_sp20_pdf
|
|
|
|
zf.writestr(f"WESP_SP20_{year}-{month:02d}.pdf", build_sp20_pdf(app, year, month))
|
|
zf.writestr(f"WESP_zhurnal_{year}-{month:02d}.pdf", build_journal_pdf(app, year, month))
|
|
except Exception:
|
|
pass
|
|
return buf.getvalue()
|