#!/usr/bin/env python3 """Read-only checker for the M03 educational CSV bundle. Python 3.8+, standard library. This is not an importer or a general validator of SelSup/Skirbo exports. The sample has order/reserve links and separate receipt/adjustment/document links. It does not contain or validate order-to-shipment links. Run: python m03-check-export.py /path/to/downloads Run fault example: python m03-check-export.py /path/to/downloads --demo-broken """ import argparse import csv import hashlib import json import sys from collections import defaultdict from datetime import datetime from decimal import Decimal, InvalidOperation from pathlib import Path def parse_iso(value): """Accept UTC Z without requiring Python 3.11's expanded fromisoformat.""" return datetime.fromisoformat(value[:-1] + '+00:00' if value.endswith('Z') else value) def check(folder, demo_broken=False): errors = [] notes = [] def error(code, message): errors.append(f"{code}: {message}") def local_file(name): candidate = (folder / name).resolve() if candidate.parent != folder: raise ValueError(f"Путь должен указывать на файл выбранной папки: {name}") return candidate try: manifest = json.loads(local_file('m03-example-manifest.json').read_text(encoding='utf-8')) except (OSError, ValueError) as exc: return [f"MANIFEST: {exc}"], notes if manifest.get('schema_version') != 'm03-sample-v1': return ['SCHEMA: поддерживается только учебная схема m03-sample-v1'], notes required = {'owners','products','variants','warehouses','cells','orders','order-lines', 'reserves','stock','movements','documents','history','attachments'} if set(manifest.get('files', {})) != required: return ['FILES: состав учебного манифеста отличается от ожидаемых 13 сущностей'], notes data = {} for name, spec in manifest['files'].items(): try: f = local_file(spec['filename']) raw = f.read_bytes() if hashlib.sha256(raw).hexdigest() != spec['sha256']: error('HASH', f"{f.name}: байты отличаются от зафиксированного экспорта") with f.open(encoding='utf-8-sig', newline='') as handle: reader = csv.DictReader(handle) if reader.fieldnames != spec['columns']: error('COLUMNS', f"{f.name}: другой порядок или состав колонок") data[name] = list(reader) if len(data[name]) != spec['rows']: error('COUNT', f"{f.name}: {len(data[name])}, ожидалось {spec['rows']}") if any(None in row or None in row.values() for row in data[name]): error('CSV', f"{f.name}: неодинаковое количество полей") except (OSError, ValueError, KeyError, csv.Error) as exc: error('READ', f"{name}: {exc}") if errors: return errors, notes if demo_broken: # In-memory mutations only: no file is changed. data['order-lines'][0]['variant_id'] = 'V999' data['variants'][0]['sku'] = '123' notes.append('УЧЕБНАЯ ПОРЧА В ПАМЯТИ: ссылка L001 → V999 и sku V001: 000123 → 123. Файлы не изменены.') ids = {} primary = {'owners':'owner_id','products':'product_id','variants':'variant_id', 'warehouses':'warehouse_id','cells':'cell_id','orders':'order_id', 'order-lines':'line_id','reserves':'reserve_id','stock':'stock_id', 'movements':'movement_id','documents':'document_id','history':'history_id', 'attachments':'attachment_id'} for name, rows in data.items(): key = primary[name] ids[name] = {} for row in rows: val = row[key] if not val or val in ids[name]: error('KEY', f"{name}: пустой или повторный {key}={val!r}") ids[name][val] = row links = {'owner_id':'owners','product_id':'products','variant_id':'variants', 'warehouse_id':'warehouses','cell_id':'cells','order_id':'orders', 'line_id':'order-lines','document_id':'documents'} for name, rows in data.items(): for row in rows: label = f"{name}/{row[primary[name]]}" for field, target in links.items(): if field == primary[name] or field not in row: continue if row[field] not in ids[target]: error('BROKEN_LINK', f"{label}: {field}={row[field]!r} не найден в {target}") owner = row.get('owner_id') for foreign in ('product_id','variant_id','document_id'): if owner and foreign in row and foreign != primary[name]: target = ids[links[foreign]].get(row[foreign], {}) if target and target.get('owner_id') != owner: error('OWNER', f"{label}: владелец расходится с {foreign}") if 'cell_id' in row and 'warehouse_id' in row: cell = ids['cells'].get(row['cell_id'], {}) if cell and cell['warehouse_id'] != row['warehouse_id']: error('WAREHOUSE', f"{label}: ячейка относится к другому складу") for field in ('created_at','occurred_at','snapshot_at'): if field in row: try: dt = parse_iso(row[field]) if dt.tzinfo is None: raise ValueError('нет часового пояса') if dt > parse_iso(manifest['snapshot_at']): raise ValueError('событие позже момента среза') except ValueError as exc: error('TIME', f"{label}/{field}: {exc}") for point in manifest['control_values']: row = ids[point['file']].get(point['id'], {}) got = row.get(point['field']) if got != point['expected']: error('IDENTIFIER_CHANGED', f"{point['file']}/{point['id']}/{point['field']}: {got!r}, ожидалось {point['expected']!r}") for row in data['order-lines']: order = ids['orders'].get(row['order_id'], {}) variant = ids['variants'].get(row['variant_id'], {}) if order and variant and order['owner_id'] != variant['owner_id']: error('OWNER', f"строка {row['line_id']}: товар другого владельца") def quantity(row, field, label, allow_negative=False): try: q = Decimal(row[field]) if not q.is_finite() or q != q.to_integral_value() or (not allow_negative and q < 0): raise InvalidOperation() return q except (InvalidOperation, KeyError): error('QUANTITY', f"{label}: {field} должен быть целым числом штук") return None def stock_key(row): return tuple(row[x] for x in ('owner_id','variant_id','warehouse_id','cell_id')) reserved = defaultdict(Decimal) reserved_by_line = defaultdict(Decimal) for r in data['reserves']: line = ids['order-lines'].get(r['line_id'], {}) order = ids['orders'].get(line.get('order_id'), {}) if line and line['variant_id'] != r['variant_id']: error('RESERVE_LINK', f"{r['reserve_id']}: вариант отличается от строки заказа") if order and (order['owner_id'] != r['owner_id'] or order['status'] != 'OPEN'): error('RESERVE_ORDER', f"{r['reserve_id']}: другой владелец или заказ не OPEN") q = quantity(r,'qty',r['reserve_id']) if r['status'] != 'ACTIVE': error('RESERVE_STATUS', f"{r['reserve_id']}: учебный набор содержит только ACTIVE") if q is not None: reserved[stock_key(r)] += q reserved_by_line[r['line_id']] += q for line in data['order-lines']: line_qty = quantity(line,'qty',line['line_id']) rqty = reserved_by_line[line['line_id']] if line_qty is not None and rqty > line_qty: error('OVER_RESERVE', f"{line['line_id']}: резерв больше количества заказа") movements = defaultdict(Decimal) for m in data['movements']: q = quantity(m,'qty_delta',m['movement_id'],allow_negative=True) if q is not None: movements[stock_key(m)] += q totals = defaultdict(Decimal) seen_stock = set() for s in data['stock']: k = stock_key(s) if k in seen_stock: error('STOCK_KEY', f"{s['stock_id']}: повтор владельца/варианта/склада/ячейки") seen_stock.add(k) qs = {f:quantity(s,f,s['stock_id']) for f in ('on_hand','reserved','available')} if any(q is None for q in qs.values()): continue if s['unit'] != 'pcs': error('UNIT', f"{s['stock_id']}: пример поддерживает только pcs") if qs['on_hand'] - qs['reserved'] != qs['available']: error('AVAILABLE', f"{s['stock_id']}: on_hand − reserved != available") if reserved[k] != qs['reserved']: error('RESERVE_TOTAL', f"{s['stock_id']}: сумма связанных резервов не совпала") if Decimal(manifest['opening_balances']) + movements[k] != qs['on_hand']: error('MOVEMENT_TOTAL', f"{s['stock_id']}: начальный остаток + движения не совпали") for f,q in qs.items(): totals[f]+=q if (set(reserved)|set(movements))-seen_stock: error('UNLISTED_STOCK', 'Резерв или движение относится к отсутствующей строке остатка') for key, expected in manifest['expected_totals'].items(): if totals[key] != Decimal(expected): error('CONTROL_TOTAL', f"{key}: {totals[key]}, ожидалось {expected}") for h in data['history']: if h['entity'] not in ids or h['entity_id'] not in ids[h['entity']]: error('HISTORY_LINK', f"{h['history_id']}: объект истории отсутствует") for a in data['attachments']: try: raw=local_file(a['filename']).read_bytes() if len(raw)!=int(a['bytes']) or hashlib.sha256(raw).hexdigest()!=a['sha256']: error('ATTACHMENT', f"{a['attachment_id']}: размер или SHA-256 не совпали") except (OSError,ValueError) as exc: error('ATTACHMENT', f"{a['attachment_id']}: {exc}") notes.append(f"Прочитано {len(data)} CSV, {sum(map(len,data.values()))} записей; физических единиц {totals['on_hand']}, в резерве {totals['reserved']}, доступно {totals['available']}.") notes.append('Проверка охватывает связи заказов с резервами и отдельно движений приёмки/корректировки с документами. Связи заказа с отгрузкой в наборе нет.') notes.append('Проверка относится к учебному набору. Права реальной WMS, полнота её API/истории и импорт не проверяются.') return errors, notes if __name__ == '__main__': parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('folder',type=Path) parser.add_argument('--demo-broken',action='store_true',help='Смоделировать две ошибки в памяти, без записи файлов') args=parser.parse_args() try: errors,notes=check(args.folder.resolve(),args.demo_broken) except (KeyError, TypeError, ValueError, InvalidOperation) as exc: errors,notes=[f'FORMAT: структура не соответствует учебной схеме: {exc}'],[] print('\n'.join(notes)) if errors: print(f'НАЙДЕНЫ ОШИБКИ: {len(errors)}') print('\n'.join(errors)) sys.exit(1) print('Проверки учебных связей, количеств и файлов пройдены.')