#!/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 api_request(url: str, token: str, path: str, method: str, payload: dict | None, allow_self_signed: bool) -> dict: body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8") request = urllib.request.Request( url.rstrip("/") + path, data=body, method=method, headers={ "authorization": f"Bearer {token}", **({"content-type": "application/json"} if body is not None else {}), }, ) 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 post(url: str, token: str, payload: dict, allow_self_signed: bool) -> dict: return api_request(url, token, "/api/v1/staff/imports/owner-workbook", "POST", payload, allow_self_signed) 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") profession_response = api_request( args.server, args.token, "/api/v1/staff/professions", "GET", None, args.allow_self_signed) existing_professions = { item["name"] for item in profession_response.get("professions", []) } for profession_name in sorted(set(PROFESSIONS.values()) - existing_professions): code = "xlsx-" + hashlib.sha256( profession_name.casefold().encode("utf-8")).hexdigest()[:12] api_request(args.server, args.token, "/api/v1/staff/professions", "POST", {"code": code, "name": profession_name}, args.allow_self_signed) print(f"Created profession: {profession_name}") 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())