274 lines
11 KiB
Python
274 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""Read the owner's workbook and import raw 2026 facts into Krona Staff.
|
||
|
||
The workbook is never modified. Personal data stays in memory and is sent only to
|
||
the explicitly supplied HTTPS Staff endpoint.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import ssl
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from collections import defaultdict
|
||
from datetime import datetime
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from pathlib import Path
|
||
|
||
import openpyxl
|
||
|
||
FINANCE_COLUMNS = {
|
||
3: "income_cash",
|
||
4: "income_card",
|
||
5: "income_transfer",
|
||
6: "income_sber_market",
|
||
7: "income_yandex",
|
||
8: "income_certificate",
|
||
10: "income_recycling",
|
||
12: "expense_products",
|
||
13: "expense_supplies",
|
||
14: "expense_taxi",
|
||
15: "expense_communications",
|
||
16: "expense_advertising",
|
||
17: "expense_owner_payroll",
|
||
18: "expense_taxes",
|
||
19: "expense_rent",
|
||
20: "expense_utilities",
|
||
21: "expense_other",
|
||
22: "expense_gifts",
|
||
}
|
||
UNIT_SHEETS = {"Горский": ("gorsky", "Горский"), "Красный": ("krasny", "Красный")}
|
||
POINT_CODES = {"Г": "gorsky", "К": "krasny", "Ф": "krasny"}
|
||
PROFESSIONS = {
|
||
"Повар": "Повар",
|
||
"Доп Повар": "Дополнительный повар",
|
||
"Стажер": "Стажёр",
|
||
"Админ": "Администратор",
|
||
}
|
||
|
||
|
||
def money_minor(value: object) -> int | None:
|
||
if not isinstance(value, (int, float, Decimal)):
|
||
return None
|
||
return int((Decimal(str(value)) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
||
|
||
|
||
def normalized_name(value: object) -> str:
|
||
return " ".join(str(value or "").split())
|
||
|
||
|
||
def stable_ref(prefix: str, value: str) -> str:
|
||
return prefix + hashlib.sha256(value.encode("utf-8")).hexdigest()[:24]
|
||
|
||
|
||
def extract(path: Path) -> tuple[dict[str, list[dict]], dict[str, object]]:
|
||
workbook = openpyxl.load_workbook(path, data_only=True, read_only=False)
|
||
units = [{"code": code, "name": name, "external_ref": f"xlsx-unit:{code}"}
|
||
for code, name in UNIT_SHEETS.values()]
|
||
finance: list[dict] = []
|
||
monthly = defaultdict(lambda: {"income_minor": 0, "expense_minor": 0})
|
||
|
||
for sheet_name, (unit_code, _) in UNIT_SHEETS.items():
|
||
sheet = workbook[sheet_name]
|
||
for row in range(2, sheet.max_row + 1):
|
||
business_date = sheet.cell(row, 1).value
|
||
if not isinstance(business_date, datetime) or business_date.year != 2026:
|
||
continue
|
||
date_text = business_date.date().isoformat()
|
||
for column, metric_key in FINANCE_COLUMNS.items():
|
||
value_minor = money_minor(sheet.cell(row, column).value)
|
||
if value_minor is None or value_minor == 0:
|
||
continue
|
||
finance.append({
|
||
"org_unit_code": unit_code,
|
||
"business_date": date_text,
|
||
"metric_key": metric_key,
|
||
"value_minor": value_minor,
|
||
"currency_code": "RUB",
|
||
})
|
||
bucket = monthly[date_text[:7]]
|
||
bucket["income_minor" if metric_key.startswith("income_") else "expense_minor"] += value_minor
|
||
|
||
graph = workbook["График"]
|
||
date_columns = {
|
||
column: graph.cell(1, column).value.date().isoformat()
|
||
for column in range(4, graph.max_column + 1)
|
||
if isinstance(graph.cell(1, column).value, datetime)
|
||
and graph.cell(1, column).value.year == 2026
|
||
}
|
||
employee_blocks: list[tuple[int, str, str]] = []
|
||
professions_by_name: dict[str, set[str]] = defaultdict(set)
|
||
for row in range(2, graph.max_row):
|
||
if graph.cell(row, 3).value != "Часов":
|
||
continue
|
||
name = normalized_name(graph.cell(row, 2).value)
|
||
profession = PROFESSIONS.get(normalized_name(graph.cell(row - 1, 1).value))
|
||
if not name or not profession:
|
||
continue
|
||
employee_blocks.append((row, name, profession))
|
||
professions_by_name[name].add(profession)
|
||
|
||
names = sorted(professions_by_name)
|
||
employees = []
|
||
refs: dict[str, str] = {}
|
||
for name in names:
|
||
identity_hash = hashlib.sha256(name.casefold().encode("utf-8")).hexdigest()[:16]
|
||
reference = f"xlsx:staff:{identity_hash}"
|
||
refs[name] = reference
|
||
professions = professions_by_name[name]
|
||
priority = ("Администратор", "Повар", "Дополнительный повар", "Стажёр")
|
||
primary = next(role for role in priority if role in professions)
|
||
employees.append({
|
||
"external_ref": reference,
|
||
"personnel_number": f"XLSX-{identity_hash.upper()}",
|
||
"display_name": name,
|
||
"position": primary,
|
||
})
|
||
|
||
schedule_totals: dict[tuple[str, str, str], int] = defaultdict(int)
|
||
schedule_locations: dict[tuple[str, str, str], set[str]] = defaultdict(set)
|
||
payroll_totals: dict[tuple[str, str, str], int] = defaultdict(int)
|
||
payout_totals: dict[tuple[str, str, str], int] = defaultdict(int)
|
||
inferred_points = 0
|
||
for row, name, _profession in employee_blocks:
|
||
for column, date_text in date_columns.items():
|
||
hours = graph.cell(row, column).value
|
||
if not isinstance(hours, (int, float)) or hours <= 0:
|
||
continue
|
||
point_code = normalized_name(graph.cell(row - 1, column).value)
|
||
unit_code = POINT_CODES.get(point_code)
|
||
if unit_code is None:
|
||
inferred_points += 1
|
||
lowered = name.casefold()
|
||
unit_code = "krasny" if ("крас" in lowered or "кр." in lowered) else "gorsky"
|
||
identity = (refs[name], date_text, unit_code)
|
||
schedule_totals[identity] += int(
|
||
Decimal(str(hours * 60)).quantize(Decimal("1")))
|
||
schedule_locations[identity].add(point_code or "inferred")
|
||
salary_minor = money_minor(graph.cell(row + 1, column).value)
|
||
if salary_minor:
|
||
payroll_totals[identity] += salary_minor
|
||
payout_minor = money_minor(graph.cell(row + 2, column).value)
|
||
if payout_minor:
|
||
payout_totals[identity] += abs(payout_minor)
|
||
|
||
schedule = [{
|
||
"external_key": stable_ref("xlsx:shift:", "|".join(identity)),
|
||
"employee_external_ref": identity[0],
|
||
"org_unit_code": identity[2],
|
||
"business_date": identity[1],
|
||
"planned_minutes": minutes,
|
||
"location_code": ",".join(sorted(schedule_locations[identity])),
|
||
} for identity, minutes in sorted(schedule_totals.items())]
|
||
payroll = [{
|
||
"external_key": stable_ref("xlsx:payroll:accrual:", "|".join(identity)),
|
||
"employee_external_ref": identity[0],
|
||
"business_date": identity[1],
|
||
"kind": "accrual",
|
||
"amount_minor": amount,
|
||
"note": "Импорт начисления из листа «График»",
|
||
} for identity, amount in sorted(payroll_totals.items())]
|
||
payroll.extend({
|
||
"external_key": stable_ref("xlsx:payroll:payout:", "|".join(identity)),
|
||
"employee_external_ref": identity[0],
|
||
"business_date": identity[1],
|
||
"kind": "payout",
|
||
"amount_minor": amount,
|
||
"note": "Импорт выплаты из строки «Долг» листа «График»",
|
||
} for identity, amount in sorted(payout_totals.items()))
|
||
|
||
summary = workbook["Сводная"]
|
||
for unit_code, row in (("gorsky", 31), ("krasny", 43)):
|
||
for column in range(3, 15):
|
||
period_end = summary.cell(2, column).value
|
||
amount = money_minor(summary.cell(row, column).value)
|
||
if not isinstance(period_end, datetime) or period_end.year != 2026 or not amount:
|
||
continue
|
||
business_date = period_end.date().isoformat()
|
||
finance.append({
|
||
"org_unit_code": unit_code,
|
||
"business_date": business_date,
|
||
"metric_key": "expense_employee_payroll",
|
||
"value_minor": amount,
|
||
"currency_code": "RUB",
|
||
})
|
||
monthly[business_date[:7]]["expense_minor"] += amount
|
||
|
||
report = {
|
||
"year": 2026,
|
||
"units": len(units),
|
||
"employees": len(employees),
|
||
"financial_entries": len(finance),
|
||
"schedule_items": len(schedule),
|
||
"payroll_entries": len(payroll),
|
||
"inferred_point_codes": inferred_points,
|
||
"monthly": dict(sorted(monthly.items())),
|
||
"totals": {
|
||
"income_minor": sum(item["income_minor"] for item in monthly.values()),
|
||
"expense_minor": sum(item["expense_minor"] for item in monthly.values()),
|
||
},
|
||
}
|
||
return {
|
||
"units": units,
|
||
"employees": employees,
|
||
"financial_entries": finance,
|
||
"schedule_items": schedule,
|
||
"payroll_entries": payroll,
|
||
}, report
|
||
|
||
|
||
def chunks(values: list[dict], size: int = 500):
|
||
for offset in range(0, len(values), size):
|
||
yield values[offset:offset + size]
|
||
|
||
|
||
def post(url: str, token: str, payload: dict, allow_self_signed: bool) -> dict:
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
request = urllib.request.Request(
|
||
url.rstrip("/") + "/api/v1/staff/imports/owner-workbook",
|
||
data=body,
|
||
method="POST",
|
||
headers={"authorization": f"Bearer {token}", "content-type": "application/json"},
|
||
)
|
||
context = ssl._create_unverified_context() if allow_self_signed else ssl.create_default_context()
|
||
with urllib.request.urlopen(request, context=context, timeout=90) as response:
|
||
return json.loads(response.read().decode("utf-8"))
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("workbook", type=Path)
|
||
parser.add_argument("--server", default=os.getenv("KRONA_STAFF_SERVER", ""))
|
||
parser.add_argument("--token", default=os.getenv("KRONA_STAFF_TOKEN", ""))
|
||
parser.add_argument("--allow-self-signed", action="store_true")
|
||
parser.add_argument("--apply", action="store_true", help="Send data; otherwise only validate and report")
|
||
args = parser.parse_args()
|
||
|
||
payload, report = extract(args.workbook)
|
||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||
if not args.apply:
|
||
return 0
|
||
if not args.server.lower().startswith("https://"):
|
||
raise SystemExit("--server must be an https:// URL")
|
||
if len(args.token) < 32:
|
||
raise SystemExit("Provide --token or KRONA_STAFF_TOKEN")
|
||
|
||
source = f"{args.workbook.name}:2026"
|
||
batches = [
|
||
{"units": payload["units"], "employees": payload["employees"]},
|
||
*({"financial_entries": part} for part in chunks(payload["financial_entries"])),
|
||
*({"schedule_items": part} for part in chunks(payload["schedule_items"])),
|
||
*({"payroll_entries": part} for part in chunks(payload["payroll_entries"])),
|
||
]
|
||
for number, batch in enumerate(batches, start=1):
|
||
result = post(args.server, args.token, {"source_name": source, **batch}, args.allow_self_signed)
|
||
print(f"Imported batch {number}/{len(batches)}: {result}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|