112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
"""Производственный календарь — порт themes/merakomis/day/model.php."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
|
|
from app.labor import _column_lookup, _prefixed_col, _resolve_table
|
|
from app.labor_calendar_logic import (
|
|
format_day_context,
|
|
get_work_hours_in_date,
|
|
)
|
|
from app.main import _quote_ident, _table_columns
|
|
from app.merakomis_schema import DAY_TABLE, DAY_TYPE_NO_WORK
|
|
|
|
|
|
def _load_day_row(cur, db: str, day: date) -> dict | None:
|
|
table = _resolve_table(cur, db, DAY_TABLE)
|
|
cols = _table_columns(cur, db, table)
|
|
lut = _column_lookup(cols)
|
|
date_col = _prefixed_col(lut, DAY_TABLE, "date")
|
|
if not date_col:
|
|
return None
|
|
type_col = _prefixed_col(lut, DAY_TABLE, "type")
|
|
text_col = _prefixed_col(lut, DAY_TABLE, "text")
|
|
hours_col = _prefixed_col(lut, DAY_TABLE, "hours")
|
|
tq = _quote_ident(table)
|
|
sel = [f"{_quote_ident(date_col)} AS d"]
|
|
if type_col:
|
|
sel.append(f"{_quote_ident(type_col)} AS type")
|
|
if text_col:
|
|
sel.append(f"{_quote_ident(text_col)} AS text")
|
|
if hours_col:
|
|
sel.append(f"{_quote_ident(hours_col)} AS hours")
|
|
cur.execute(
|
|
f"SELECT {', '.join(sel)} FROM {tq} WHERE {_quote_ident(date_col)} = %s LIMIT 1",
|
|
(day.isoformat(),),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
|
|
def get_work_hours_by_date(cur, db: str, day: date) -> float:
|
|
row = _load_day_row(cur, db, day)
|
|
if row and row.get("hours") is not None:
|
|
return float(row["hours"])
|
|
return get_work_hours_in_date(day)
|
|
|
|
|
|
def is_no_work_day(cur, db: str, day: date) -> bool:
|
|
row = _load_day_row(cur, db, day)
|
|
if row and int(row.get("type") or 0) == DAY_TYPE_NO_WORK:
|
|
return True
|
|
return day.weekday() >= 5
|
|
|
|
|
|
def format_day_row(row: dict) -> dict:
|
|
d = row.get("d") or row.get("date")
|
|
if isinstance(d, date):
|
|
ds = d.isoformat()
|
|
else:
|
|
ds = str(d)[:10]
|
|
return {
|
|
"date": ds,
|
|
"type": int(row.get("type") or 0),
|
|
"text": (row.get("text") or "").strip(),
|
|
}
|
|
|
|
|
|
def get_by_range_formatted(cur, db: str, begin: date, end: date) -> dict[str, dict]:
|
|
table = _resolve_table(cur, db, DAY_TABLE)
|
|
cols = _table_columns(cur, db, table)
|
|
lut = _column_lookup(cols)
|
|
date_col = _prefixed_col(lut, DAY_TABLE, "date")
|
|
type_col = _prefixed_col(lut, DAY_TABLE, "type")
|
|
text_col = _prefixed_col(lut, DAY_TABLE, "text")
|
|
hours_col = _prefixed_col(lut, DAY_TABLE, "hours")
|
|
if not date_col:
|
|
return {
|
|
day.isoformat(): format_day_context(None, day)
|
|
for day in iter_dates(begin, end)
|
|
}
|
|
tq = _quote_ident(table)
|
|
parts = [f"{_quote_ident(date_col)} AS d"]
|
|
if type_col:
|
|
parts.append(f"{_quote_ident(type_col)} AS type")
|
|
if text_col:
|
|
parts.append(f"{_quote_ident(text_col)} AS text")
|
|
if hours_col:
|
|
parts.append(f"{_quote_ident(hours_col)} AS hours")
|
|
cur.execute(
|
|
f"""
|
|
SELECT {', '.join(parts)} FROM {tq}
|
|
WHERE {_quote_ident(date_col)} BETWEEN %s AND %s
|
|
""",
|
|
(begin.isoformat(), end.isoformat()),
|
|
)
|
|
rows = {}
|
|
for row in cur.fetchall():
|
|
value = row.get("d") or row.get("date")
|
|
key = value.isoformat() if isinstance(value, date) else str(value)[:10]
|
|
rows[key] = row
|
|
return {
|
|
day.isoformat(): format_day_context(rows.get(day.isoformat()), day)
|
|
for day in iter_dates(begin, end)
|
|
}
|
|
|
|
|
|
def iter_dates(begin: date, end: date):
|
|
d = begin
|
|
while d <= end:
|
|
yield d
|
|
d += timedelta(days=1)
|