""" differ.py — Сравнение строк спецификаций между версиями договора. Чистая функция: diff(rows_old, rows_new) → список изменений. Не зависит от БД, LLM, Flask. Только от данных. change_type: added — новая строка (была в new, нет в old) deleted — удалённая строка (была в old, нет в new) changed — изменённая строка (совпали по row_num, но значения разные) unchanged — без изменений """ # Поля для сравнения (все кроме row_num) _COMPARE_FIELDS = ["name", "price", "qty", "sum", "date_start"] def diff(rows_old: list, rows_new: list) -> dict: """ Сравнить два списка строк спецификации. Args: rows_old: строки из предыдущего допника rows_new: строки из нового допника Returns: {"changes": [...], "ambiguous": [...]} changes: [{row_num, change_type, old_values, new_values}, ...] ambiguous: строки которые не удалось сопоставить """ # Индексация по row_num old_by_num = {r["row_num"]: r for r in rows_old} new_by_num = {r["row_num"]: r for r in rows_new} all_nums = sorted(set(list(old_by_num.keys()) + list(new_by_num.keys()))) changes = [] ambiguous = [] for num in all_nums: in_old = num in old_by_num in_new = num in new_by_num if in_old and not in_new: # Строка удалена changes.append({ "row_num": num, "change_type": "deleted", "old_values": _row_values(old_by_num[num]), "new_values": None, }) elif not in_old and in_new: # Новая строка changes.append({ "row_num": num, "change_type": "added", "old_values": None, "new_values": _row_values(new_by_num[num]), }) else: # Строка есть в обоих — сравниваем значения old_vals = _row_values(old_by_num[num]) new_vals = _row_values(new_by_num[num]) if old_vals == new_vals: changes.append({ "row_num": num, "change_type": "unchanged", "old_values": old_vals, "new_values": new_vals, }) else: # Частичное изменение — показываем что именно поменялось changed_fields = {} for field in _COMPARE_FIELDS: if old_vals.get(field) != new_vals.get(field): changed_fields[field] = { "old": old_vals.get(field), "new": new_vals.get(field), } changes.append({ "row_num": num, "change_type": "changed", "old_values": old_vals, "new_values": new_vals, "changed_fields": changed_fields, }) return { "changes": changes, "ambiguous": ambiguous, "summary": { "added": sum(1 for c in changes if c["change_type"] == "added"), "deleted": sum(1 for c in changes if c["change_type"] == "deleted"), "changed": sum(1 for c in changes if c["change_type"] == "changed"), "unchanged": sum(1 for c in changes if c["change_type"] == "unchanged"), }, } def _row_values(row: dict) -> dict: """Извлечь только значимые поля из строки спецификации.""" return {f: row.get(f) for f in _COMPARE_FIELDS}