#!/usr/bin/env python3 """plan_fact.py — Контролёр план-факта. Полная реализация.""" import argparse, csv, json, sys, re, hashlib from pathlib import Path from collections import defaultdict, OrderedDict REQUIRED = ["row_id", "period", "branch", "plan_rub", "actual_rub", "currency"] OPTIONAL = ["operation_date", "document_no", "category", "description"] ALL_FIELDS = REQUIRED + OPTIONAL def sha256(p): h = hashlib.sha256() with open(p, "rb") as f: for blk in iter(lambda: f.read(4096), b""): h.update(blk) return h.hexdigest() def stop(out, period, reason, details): out.mkdir(parents=True, exist_ok=True) with open(out / f"Протокол_{period}.json", "w", encoding="utf-8") as f: json.dump({"status": "STOP", "reason": reason, "details": details}, f, ensure_ascii=False, indent=2) sys.exit(2) def diff_fields(r1, r2): """Сравнить две записи по ВСЕМ 10 полям (сырые значения), вернуть различия.""" diffs = {} for k in ALL_FIELDS: v1, v2 = r1.get(k, ""), r2.get(k, "") if v1 != v2: diffs[k] = {"first": v1, "second": v2} return diffs def read_csv(path, period, allowed_dupes, out): """Прочитать CSV, проверить поля. Вернуть (rows_norm, sha256, count, branches, dupes).""" basename = path.name try: file_sha = sha256(path) except (OSError, IOError) as e: stop(out, period, "file_read_error", {"file": basename, "error": str(e)}) rows_raw = OrderedDict() # row_id -> raw row dict (no strip) rows_norm = OrderedDict() # row_id -> normalized row dict (stripped) first_seen = {} # row_id -> csv line number dupes_found = [] count = 0 branches = set() try: with open(path, encoding="utf-8-sig") as f: reader = csv.DictReader(f) hdr = reader.fieldnames or [] # --- уникальность заголовков --- if len(hdr) != len(set(hdr)): dup_hdrs = [h for h in hdr if hdr.count(h) > 1] stop(out, period, "duplicate_headers", {"file": basename, "duplicate_headers": sorted(set(dup_hdrs))}) # --- неизвестные колонки --- known = set(ALL_FIELDS) unknown = [h for h in hdr if h not in known] if unknown: stop(out, period, "unknown_columns", {"file": basename, "unknown_columns": unknown}) # --- недостающие обязательные заголовки --- missing = [h for h in REQUIRED if h not in hdr] if missing: stop(out, period, "missing_headers", {"file": basename, "missing": missing}) for lineno, row in enumerate(reader, start=2): count += 1 # --- проверить, что row содержит все обязательные поля --- for h in REQUIRED: if h not in row or row[h] is None: stop(out, period, "missing_field_in_row", {"file": basename, "line": lineno, "field": h}) rid_raw = row["row_id"] rid = rid_raw.strip() # --- лишние значения в строке (ключ None у DictReader) --- if None in row: stop(out, period, "extra_values_in_row", {"file": basename, "line": lineno, "hint": "row has more values than headers"}) # --- непустые обязательные поля (после strip) --- for h in REQUIRED: v = row.get(h, "") if v is None or v.strip() == "": stop(out, period, "empty_field", {"file": basename, "line": lineno, "field": h, "row_id": rid}) # --- непустые присутствующие optional-поля (отсутствие колонки — нормально) --- for h in OPTIONAL: if h in reader.fieldnames: v = row.get(h, "") if v is None or v.strip() == "": stop(out, period, "empty_optional_field", {"file": basename, "line": lineno, "field": h, "row_id": rid}) # --- период --- if row["period"].strip() != period: stop(out, period, "wrong_period", {"file": basename, "line": lineno, "row_id": rid, "expected": period, "got": row["period"].strip()}) # --- валюта --- if row["currency"].strip() != "RUB": stop(out, period, "wrong_currency", {"file": basename, "line": lineno, "row_id": rid, "got": row["currency"].strip()}) # --- целые суммы --- for money_field in ("plan_rub", "actual_rub"): v = row[money_field].strip() if not re.fullmatch(r"-?\d+", v): stop(out, period, "non_integer", {"file": basename, "line": lineno, "row_id": rid, "field": money_field, "value": v}) # --- сохранить сырые и нормализованные --- raw_row = {k: row.get(k, "") for k in ALL_FIELDS} norm_row = {k: row.get(k, "").strip() for k in ALL_FIELDS} br = norm_row["branch"] branches.add(br) # --- повтор row_id --- if rid in rows_raw: # Сравниваем сырые записи (без strip) diffs = diff_fields(rows_raw[rid], raw_row) if diffs: stop(out, period, "conflict", {"file": basename, "row_id": rid, "first_line": first_seen[rid], "second_line": lineno, "differences": diffs}) else: # Точный повтор (сырые значения совпадают) key = f"{basename}:{rid}" if key not in allowed_dupes: stop(out, period, "exact_duplicate_no_permission", {"file": basename, "row_id": rid, "first_line": first_seen[rid], "second_line": lineno, "hint": f"Use --allow-exact-duplicate {basename}:{rid}"}) dupes_found.append({"row_id": rid, "first_line": first_seen[rid], "second_line": lineno, "resolution": "allowed_one_copy"}) continue rows_raw[rid] = raw_row rows_norm[rid] = norm_row first_seen[rid] = lineno except (OSError, IOError) as e: stop(out, period, "file_read_error", {"file": basename, "error": str(e)}) except csv.Error as e: stop(out, period, "csv_parse_error", {"file": basename, "error": str(e)}) return list(rows_norm.values()), file_sha, count, branches, dupes_found def main(): ap = argparse.ArgumentParser(description="Контролёр план-факта") ap.add_argument("--period", required=True, help="YYYY-MM") ap.add_argument("--sales", required=True, help="Path to sales.csv") ap.add_argument("--cogs", required=True, help="Path to cogs.csv") ap.add_argument("--opex", required=True, help="Path to opex.csv") ap.add_argument("--out", required=True, help="Output directory (must be empty or new)") ap.add_argument("--allow-exact-duplicate", action="append", default=[], help="BASENAME:ROW_ID — allow one copy of exact duplicate") args = ap.parse_args() # --- период: YYYY-MM, месяц 01..12 --- m = re.fullmatch(r"(\d{4})-(\d{2})", args.period) if not m: print(f"Invalid period format: {args.period}", file=sys.stderr) sys.exit(2) month_num = int(m.group(2)) if month_num < 1 or month_num > 12: print(f"Invalid month in period: {args.period}", file=sys.stderr) sys.exit(2) out = Path(args.out) # --- выходная папка: пустая или не существует --- if out.exists(): if not out.is_dir(): print(f"Output path exists but is not a directory: {out}", file=sys.stderr) sys.exit(2) if any(out.iterdir()): print(f"STOP: output directory {out} is not empty. " f"Use a new empty directory for each run.", file=sys.stderr) sys.exit(2) # Создаём папку только после всех проверок allowed = set(args.allow_exact_duplicate) inputs = OrderedDict([("sales", Path(args.sales)), ("cogs", Path(args.cogs)), ("opex", Path(args.opex))]) file_data = {} file_meta = {} branch_sets = {} for role, path in inputs.items(): rows, fsha, cnt, brs, dupes = read_csv(path, args.period, allowed, out) file_data[role] = rows file_meta[role] = {"path": str(path), "sha256": fsha, "rows": cnt, "duplicates": dupes} branch_sets[role] = brs # --- сверить наборы филиалов трёх CSV --- ref_branches = branch_sets["sales"] for role in ("cogs", "opex"): if branch_sets[role] != ref_branches: stop(out, args.period, "branch_mismatch", {"file1": "sales.csv", "branches1": sorted(ref_branches), "file2": f"{role}.csv", "branches2": sorted(branch_sets[role])}) branch_list = sorted(ref_branches) # --- Агрегация по филиалам --- agg = OrderedDict() for br in branch_list: agg[br] = {"sp": 0, "sa": 0, "cp": 0, "ca": 0, "op": 0, "oa": 0} for row in file_data["sales"]: b = row["branch"] agg[b]["sp"] += int(row["plan_rub"]) agg[b]["sa"] += int(row["actual_rub"]) for row in file_data["cogs"]: b = row["branch"] agg[b]["cp"] += int(row["plan_rub"]) agg[b]["ca"] += int(row["actual_rub"]) for row in file_data["opex"]: b = row["branch"] agg[b]["op"] += int(row["plan_rub"]) agg[b]["oa"] += int(row["actual_rub"]) # --- Построение строк таблицы --- COLS = ["Категория", "sales_p", "sales_a", "cogs_p", "cogs_a", "opex_p", "opex_a", "res_p", "res_a", "d_sales", "d_cogs", "d_opex", "d_res"] table = [] totals = {"sp": 0, "sa": 0, "cp": 0, "ca": 0, "op": 0, "oa": 0} bridge_checks = [] for br in branch_list: s = agg[br] rp = s["sp"] - s["cp"] - s["op"] ra = s["sa"] - s["ca"] - s["oa"] ds = s["sa"] - s["sp"] dc = s["ca"] - s["cp"] do_ = s["oa"] - s["op"] dr = ra - rp bridge = ds - dc - do_ bridge_ok = (bridge == dr) bridge_checks.append({ "branch": br, "bridge_check": "d_sales - d_cogs - d_opex == d_res", "bridge_value": bridge, "d_res": dr, "pass": bridge_ok }) table.append([br, s["sp"], s["sa"], s["cp"], s["ca"], s["op"], s["oa"], rp, ra, ds, dc, do_, dr]) for k in totals: totals[k] += s[k] # --- ИТОГО --- t = totals trp = t["sp"] - t["cp"] - t["op"] tra = t["sa"] - t["ca"] - t["oa"] tds = t["sa"] - t["sp"] tdc = t["ca"] - t["cp"] tdo = t["oa"] - t["op"] tdr = tra - trp tbridge = tds - tdc - tdo t_bridge_ok = (tbridge == tdr) bridge_checks.append({ "branch": "ИТОГО", "bridge_check": "d_sales - d_cogs - d_opex == d_res", "bridge_value": tbridge, "d_res": tdr, "pass": t_bridge_ok }) # --- Проверка: сумма филиалов == ИТОГО --- total_row_vals = { "sales_p": t["sp"], "sales_a": t["sa"], "cogs_p": t["cp"], "cogs_a": t["ca"], "opex_p": t["op"], "opex_a": t["oa"], "res_p": trp, "res_a": tra, "d_sales": tds, "d_cogs": tdc, "d_opex": tdo, "d_res": tdr } branch_sums = {col: 0 for col in COLS[1:]} for r in table: for i, col in enumerate(COLS[1:], start=1): branch_sums[col] += r[i] sum_checks = [] for col in COLS[1:]: ok = branch_sums[col] == total_row_vals[col] sum_checks.append({ "column": col, "sum_branches": branch_sums[col], "total": total_row_vals[col], "pass": ok }) # --- Проверить ВСЕ pass до записи файлов --- all_pass = True failed_checks = [] for bc in bridge_checks: if not bc["pass"]: all_pass = False failed_checks.append({"type": "bridge", "branch": bc["branch"], "expected": bc["d_res"], "got": bc["bridge_value"]}) for sc in sum_checks: if not sc["pass"]: all_pass = False failed_checks.append({"type": "sum_check", "column": sc["column"], "sum_branches": sc["sum_branches"], "total": sc["total"]}) if not all_pass: out.mkdir(parents=True, exist_ok=True) with open(out / f"Протокол_{args.period}.json", "w", encoding="utf-8") as f: json.dump({"status": "STOP", "reason": "check_failure", "details": {"failed_checks": failed_checks}}, f, ensure_ascii=False, indent=2) sys.exit(2) # --- Запись CSV (только после всех проверок) --- out.mkdir(parents=True, exist_ok=True) csv_path = out / f"Сводная_таблица_{args.period}.csv" with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(COLS) table.append(["ИТОГО", t["sp"], t["sa"], t["cp"], t["ca"], t["op"], t["oa"], trp, tra, tds, tdc, tdo, tdr]) for r in table: w.writerow(r) # --- Запись протокола --- proto = { "status": "OK", "period": args.period, "inputs": file_meta, "branches": branch_list, "bridge_checks": bridge_checks, "sum_checks": sum_checks } proto_path = out / f"Протокол_{args.period}.json" with open(proto_path, "w", encoding="utf-8") as f: json.dump(proto, f, ensure_ascii=False, indent=2) sys.exit(0) if __name__ == "__main__": main()