""" generate.py — Оркестратор генерации ~100 тестовых файлов. Схема Опуса: 10 контрактов × (договор + спека v1 + спека v2 + 1-3 допника) + ошибочные, форматы, ZIP, дубликаты, мусор. Вывод: out/ с подпапками + manifest.json (ground truth). """ import os, json, random, shutil, zipfile import pools import templates import corrupt OUT = os.path.join(os.path.dirname(__file__), "out") MANIFEST = {} def ensure_dir(path): os.makedirs(path, exist_ok=True) def save_doc(doc, path, comment=""): doc.save(path) print(f" ✓ {os.path.basename(path)}") # ═══════════════════════════════════════════════════════════════ # Генерация нормальных контрактов # ═══════════════════════════════════════════════════════════════ def generate_valid(): """10 контрактов: договор + спека v1 + спека v2 + 1-3 допника.""" out = os.path.join(OUT, "valid") ensure_dir(out) for ci, company in enumerate(pools.COMPANIES[:10]): num = pools.contract_number(ci) date = pools.random_date(2025, 12, 7) prefix = company["prefix"] # ── Договор ── ctx_contract = { "company": company, "number": num, "date": date, "company_idx": ci, "comment": ( f"Базовый договор контракта {prefix}-{num}. " f"К нему привязаны спецификации и допсоглашения через НОМЕР={num}." ) } fname = f"договор-{prefix}-{num}.docx" fpath = os.path.join(out, fname) save_doc(templates.build_contract(ctx_contract), fpath) MANIFEST[fname] = { "type": "contract", "company": prefix, "number": num, "date": date, "status": "valid", "groups": [{"contract_number": num, "counterparty": company["prefix"]}], } # ── Спецификация v1 ── svcs_v1, tot_v1, tot_str_v1 = pools.pick_services(random.randint(5, 9)) # Присвоить даты из пула for s in svcs_v1: s["date_start"] = pools.random_date(2026, 3, 3) ctx_spec_v1 = { "company": company, "number": num, "date": date, "contract_date": date, "version": 1, "services": svcs_v1, "total": tot_v1, "total_str": tot_str_v1, "label": f"Абонентские услуги с {date}", "comment": ( f"Спецификация v1 (базовая) к договору {num}. " f"{len(svcs_v1)} услуг, итого {tot_str_v1} руб." ) } fname_v1 = f"спецификация-{prefix}-{num}.docx" fpath_v1 = os.path.join(out, fname_v1) save_doc(templates.build_spec(ctx_spec_v1), fpath_v1) MANIFEST[fname_v1] = { "type": "spec", "company": prefix, "number": num, "version": 1, "date": date, "status": "valid", "services": [{"name": s["name"], "price": s["price"], "qty": s["qty"], "sum": s["sum"]} for s in svcs_v1], } # ── Спецификация v2 (ревизия с diff-ами) ── svcs_v2, changes = _make_v2(svcs_v1) tot_v2 = round(sum(s["sum"] for s in svcs_v2), 2) tot_str_v2 = pools.format_price(tot_v2) # Определить diff diffs = _compute_diff(svcs_v1, svcs_v2) # Добавить changes в описание if changes: diffs["description"] = ", ".join(changes) ctx_spec_v2 = { "company": company, "number": num, "date": date, "contract_date": date, "version": 2, "services": svcs_v2, "total": tot_v2, "total_str": tot_str_v2, "label": f"Абонентские услуги с {pools.random_date(2026, 4, 3)}", "comment": ( f"Спецификация v2 (ревизия) к договору {num}. " f"Изменения относительно v1: {diffs['description']}. " f"{len(svcs_v2)} услуг, итого {tot_str_v2} руб." ) } fname_v2 = f"спецификация-{prefix}-{num}_v2.docx" fpath_v2 = os.path.join(out, fname_v2) save_doc(templates.build_spec(ctx_spec_v2), fpath_v2) MANIFEST[fname_v2] = { "type": "spec", "company": prefix, "number": num, "version": 2, "date": date, "status": "valid", "services": [{"name": s["name"], "price": s["price"], "qty": s["qty"], "sum": s["sum"]} for s in svcs_v2], "diff": diffs, } # ── Допники (1-3) ── n_add = random.randint(2, 4) for ai in range(1, n_add + 1): inst, itot, itot_s = pools.pick_services(random.randint(1, 3)) mon, mtot, mtot_s = pools.pick_services(random.randint(2, 5)) adate = pools.random_date(2026, 1, 6) ctx_add = { "company": company, "number": num, "date": adate, "contract_date": date, "addendum_num": ai, "services_install": inst, "install_total": itot, "install_total_str": itot_s, "services_monthly": mon, "monthly_total": mtot, "monthly_total_str": mtot_s, "comment": ( f"Допсоглашение №{ai} к договору {num}. " f"Инсталляц.: {len(inst)} стр., {itot_s} руб. " f"Абонентск.: {len(mon)} стр., {mtot_s} руб." ) } fname_add = f"допник-{ai}-{prefix}-{num}.docx" fpath_add = os.path.join(out, fname_add) save_doc(templates.build_addendum(ctx_add), fpath_add) MANIFEST[fname_add] = { "type": "addendum", "company": prefix, "number": num, "addendum_num": ai, "date": adate, "status": "valid", } def _make_v2(svcs_v1): """ Создать v2 из v1 с преднамеренными diff-ами: - Изменить цену у 1-2 услуг (±15-30%) - Изменить объём у 1 услуги - Удалить 1 услугу (если > 3) - Добавить 1 новую услугу """ import copy svcs = copy.deepcopy(svcs_v1) changes = [] # Изменить цену if len(svcs) >= 1: idx = random.randint(0, len(svcs) - 1) old_price = svcs[idx]["price"] new_price = round(old_price * random.uniform(0.70, 1.30), 2) svcs[idx]["price"] = new_price svcs[idx]["price_str"] = pools.format_price(new_price) svcs[idx]["sum"] = round(new_price * svcs[idx]["qty"], 2) svcs[idx]["sum_str"] = pools.format_price(svcs[idx]["sum"]) changes.append(f"цена {svcs[idx]['name'][:40]} {old_price}→{new_price}") # Изменить объём if len(svcs) >= 2: idx = random.randint(0, len(svcs) - 1) old_qty = svcs[idx]["qty"] new_qty = old_qty + random.choice([-2, -1, 1, 2, 3]) if new_qty >= 1: svcs[idx]["qty"] = new_qty svcs[idx]["sum"] = round(svcs[idx]["price"] * new_qty, 2) svcs[idx]["sum_str"] = pools.format_price(svcs[idx]["sum"]) changes.append(f"объём {svcs[idx]['name'][:40]} {old_qty}→{new_qty}") # Удалить услугу if len(svcs) > 3: idx = random.randint(0, len(svcs) - 1) removed = svcs.pop(idx) changes.append(f"удалена: {removed['name'][:40]}") # Добавить новую услугу new_svc_pool = [s for s in pools.SERVICES if not any(s[0] == x["name"] for x in svcs)] if new_svc_pool: name, base_price, min_q, max_q, cat = random.choice(new_svc_pool) price = pools.vary_price(base_price) qty = pools.vary_qty(min_q, max_q) s = round(price * qty, 2) svcs.append({ "num": len(svcs) + 1, "name": name, "price": price, "price_str": pools.format_price(price), "qty": qty, "sum": s, "sum_str": pools.format_price(s), "cat": cat, "date_start": pools.random_date(2026, 4, 3), }) changes.append(f"добавлена: {name[:40]}") # Пересчитать номера for i, s in enumerate(svcs, 1): s["num"] = i return svcs, changes def _compute_diff(v1, v2): """Вычислить разницу между v1 и v2 для manifest.json.""" names_v1 = {s["name"] for s in v1} names_v2 = {s["name"] for s in v2} added = [s["name"][:60] for s in v2 if s["name"] not in names_v1] removed = [s["name"][:60] for s in v1 if s["name"] not in names_v2] changed = [] for s1 in v1: for s2 in v2: if s1["name"] == s2["name"]: if s1["price"] != s2["price"]: changed.append(f"цена {s1['name'][:40]}: {s1['price']}→{s2['price']}") if s1["qty"] != s2["qty"]: changed.append(f"объём {s1['name'][:40]}: {s1['qty']}→{s2['qty']}") desc_parts = [] if added: desc_parts.append(f"+{len(added)} услуг") if removed: desc_parts.append(f"-{len(removed)} услуг") if changed: desc_parts.append(f"~{len(changed)} изменений") description = ", ".join(desc_parts) if desc_parts else "без изменений" return { "added": added, "removed": removed, "changed": changed, "description": description, } # ═══════════════════════════════════════════════════════════════ # Ошибочные кейсы # ═══════════════════════════════════════════════════════════════ def generate_errors(): out = os.path.join(OUT, "errors") ensure_dir(out) ctx_tmp = { "company": pools.COMPANIES[0], "number": "ERR01", "date": "01.06.2026", "company_idx": 0, "comment": "Ошибочный кейс." } # Битый XML tmp = os.path.join(out, "_tmp_valid.docx") templates.build_contract(ctx_tmp).save(tmp) corrupt.corrupt_xml(tmp, os.path.join(out, "corrupt_xml.docx")) MANIFEST["corrupt_xml.docx"] = {"type": "error", "error": "corrupt_xml", "status": "expected_fail"} # Пустой corrupt.make_empty_docx(os.path.join(out, "empty.docx")) MANIFEST["empty.docx"] = {"type": "error", "error": "empty_docx", "status": "expected_fail"} corrupt.make_empty_file(os.path.join(out, "empty_zero.docx")) MANIFEST["empty_zero.docx"] = {"type": "error", "error": "zero_bytes", "status": "expected_fail"} # Только текст corrupt.make_text_only_docx(os.path.join(out, "text_only.docx")) MANIFEST["text_only.docx"] = {"type": "error", "error": "text_only", "status": "expected_parse_ok_no_tables"} # Только таблицы corrupt.make_table_only_docx(os.path.join(out, "table_only.docx")) MANIFEST["table_only.docx"] = {"type": "error", "error": "table_only", "status": "expected_parse_ok"} # Без номера corrupt.make_no_number_docx(os.path.join(out, "no_number.docx"), templates.build_contract, ctx_tmp) MANIFEST["no_number.docx"] = {"type": "error", "error": "no_number", "status": "expected_classify_fail"} # Нестандартный заголовок corrupt.make_nonstandard_title_docx(os.path.join(out, "nonstandard_title.docx")) MANIFEST["nonstandard_title.docx"] = {"type": "error", "error": "nonstandard_title", "status": "expected_classify_ok"} # Гигант corrupt.make_giant_docx(os.path.join(out, "giant_500.docx"), 500) MANIFEST["giant_500.docx"] = {"type": "error", "error": "giant", "status": "expected_slow"} # Мусор (11 шт.) trash_types = ["Счёт-фактура", "Письмо-запрос", "Коммерческое предложение", "Акт выполненных работ", "Накладная", "Платёжное поручение", "Счёт на оплату", "Договор аренды офиса", "Служебная записка", "Прайс-лист", "Гарантийное письмо"] for i, tt in enumerate(trash_types): fname = f"trash_{i+1}.docx" corrupt.make_trash_docx(os.path.join(out, fname), tt) MANIFEST[fname] = {"type": "trash", "label": tt, "status": "expected_classify_other"} # Сирота — спецификация с номером, для которого нет договора ctx_orphan = { "company": pools.COMPANIES[0], "number": "ORPHAN", "date": "01.06.2026", "contract_date": "01.01.2026", "version": 1, "services": pools.pick_services(3)[0], "total": 0, "total_str": "0", "label": "Абонентские услуги", "comment": "Спецификация-сирота: номер ORPHAN не принадлежит ни одному договору." } ctx_orphan["total"] = round(sum(s["sum"] for s in ctx_orphan["services"]), 2) ctx_orphan["total_str"] = pools.format_price(ctx_orphan["total"]) save_doc(templates.build_spec(ctx_orphan), os.path.join(out, "orphan_spec.docx")) MANIFEST["orphan_spec.docx"] = {"type": "error", "error": "orphan", "status": "expected_group_unresolved"} print(f" Сгенерировано ошибочных: 21 + 2 orphan") # Добавим ещё сирот и edge-кейсов for vi in range(3): onum = f"GHOST{vi}" osvcs, _, _ = pools.pick_services(2) ctx_o = {"company": pools.COMPANIES[vi], "number": onum, "date": pools.random_date(), "contract_date": pools.random_date(), "version": 1, "services": osvcs, "total": sum(s["sum"] for s in osvcs), "total_str": pools.format_price(sum(s["sum"] for s in osvcs)), "label": "Абонентские услуги", "comment": f"Спецификация-сирота {onum}: нет родительского договора."} ctx_o["total"] = round(sum(s["sum"] for s in osvcs), 2) ctx_o["total_str"] = pools.format_price(ctx_o["total"]) fn = f"orphan_spec_{vi+1}.docx" save_doc(templates.build_spec(ctx_o), os.path.join(out, fn)) MANIFEST[fn] = {"type": "error", "error": "orphan", "number": onum, "status": "expected_group_unresolved"} # Договор без таблицы реквизитов (edge case для парсера) doc_no_table = templates.build_contract(ctx_tmp) # Удаляем таблицы for t in doc_no_table.tables: t._element.getparent().remove(t._element) save_doc(doc_no_table, os.path.join(out, "contract_no_rekv.docx")) MANIFEST["contract_no_rekv.docx"] = {"type": "error", "error": "no_rekv_table", "status": "expected_parse_ok"} print(f" ✓ доп. ошибки: +5") # ── Кириллический контракт (ХХХ002) ── out_v = os.path.join(OUT, "valid") kc = pools.COMPANIES[-1] # ХХХ002 knum = "03700_K" kdate = "15.03.2026" ctx_k = {"company": kc, "number": knum, "date": kdate, "company_idx": 99, "comment": f"Кириллический контракт {kc['prefix']}-{knum}. Edge-case: кириллица в названии компании."} fk = f"договор-{kc['prefix']}-{knum}.docx" save_doc(templates.build_contract(ctx_k), os.path.join(out_v, fk)) MANIFEST[fk] = {"type": "contract", "company": kc["prefix"], "number": knum, "date": kdate, "status": "valid"} # Спека к кириллическому ksvcs, ktot, ktot_s = pools.pick_services(6) for s in ksvcs: s["date_start"] = pools.random_date(2026, 3, 3) ctx_ks = {"company": kc, "number": knum, "date": kdate, "contract_date": kdate, "version": 1, "services": ksvcs, "total": ktot, "total_str": ktot_s, "label": f"Абонентские услуги", "comment": f"Спецификация кириллического контракта {knum}."} fks = f"спецификация-{kc['prefix']}-{knum}.docx" save_doc(templates.build_spec(ctx_ks), os.path.join(out_v, fks)) MANIFEST[fks] = {"type": "spec", "company": kc["prefix"], "number": knum, "version": 1, "status": "valid"} print(f" ✓ кириллический контракт: +2 файла") # ═══════════════════════════════════════════════════════════════ # .doc и .pdf форматы # ═══════════════════════════════════════════════════════════════ def generate_formats(): out = os.path.join(OUT, "formats") ensure_dir(out) # Проверить наличие LibreOffice import subprocess as sp has_lo = False try: sp.run(["libreoffice", "--version"], capture_output=True, timeout=5) has_lo = True except (FileNotFoundError, sp.TimeoutExpired): pass if not has_lo: print(" ⚠ LibreOffice не найден — .doc/.pdf пропущены") return # Взять один валидный контракт как основу ci = 0; c = pools.COMPANIES[ci]; num = pools.contract_number(ci); date = pools.random_date() ctx = {"company": c, "number": num, "date": date, "company_idx": ci, "comment": "Конвертирован в .doc / .pdf для теста."} docx_path = os.path.join(out, "_base.docx") templates.build_contract(ctx).save(docx_path) # .doc doc_path = os.path.join(out, f"договор-{c['prefix']}-{num}.doc") ok = corrupt.convert_to_doc(docx_path, doc_path) if ok: print(f" ✓ .doc конвертирован: {os.path.basename(doc_path)}") MANIFEST[os.path.basename(doc_path)] = {"type": "contract", "format": "doc", "status": "valid"} else: print(f" ⚠ .doc конвертация не удалась (нет LibreOffice?)") # Нормальный PDF pdf_path = os.path.join(out, f"договор-{c['prefix']}-{num}.pdf") corrupt.make_broken_pdf(docx_path, pdf_path) # эта функция делает битый — используем её, но с полным файлом # На самом деле make_broken_pdf обрезает. Сделаем нормальный отдельно. # Перезапишем полным — make_broken_pdf уже обрезал, сделаем заново с 100% import tempfile, subprocess as sp tmp = tempfile.mkdtemp() shutil.copy(docx_path, os.path.join(tmp, "t.docx")) sp.run(["libreoffice", "--headless", "--convert-to", "pdf", "--outdir", tmp, os.path.join(tmp, "t.docx")], timeout=30, capture_output=True) src = os.path.join(tmp, "t.pdf") if os.path.exists(src): shutil.move(src, pdf_path) print(f" ✓ .pdf: {os.path.basename(pdf_path)}") MANIFEST[os.path.basename(pdf_path)] = {"type": "contract", "format": "pdf", "status": "valid"} shutil.rmtree(tmp, ignore_errors=True) # Битый PDF broken_pdf = os.path.join(out, f"broken.pdf") corrupt.make_broken_pdf(docx_path, broken_pdf) if os.path.exists(broken_pdf): print(f" ✓ битый .pdf: {os.path.basename(broken_pdf)}") MANIFEST[os.path.basename(broken_pdf)] = {"type": "error", "error": "broken_pdf", "status": "expected_fail"} os.remove(docx_path) # убрать временный # ═══════════════════════════════════════════════════════════════ # ZIP-архивы # ═══════════════════════════════════════════════════════════════ def generate_zips(): out = os.path.join(OUT, "zips") ensure_dir(out) valid_dir = os.path.join(OUT, "valid") if not os.path.exists(valid_dir): print(" ⚠ valid/ не найден, ZIP-ы не сгенерированы") return # Набор файлов одного контракта files_for_zip = [f for f in os.listdir(valid_dir) if f.startswith("договор-XXX001")] if files_for_zip: zip_path = os.path.join(out, "contract_XXX001.zip") with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf: for f in files_for_zip: zf.write(os.path.join(valid_dir, f), f) # ZIP: все файлы контракта XXX005 files5 = [f for f in os.listdir(valid_dir) if "XXX005" in f] if files5: zp = os.path.join(out, "contract_XXX005.zip") with zipfile.ZipFile(zp, 'w', zipfile.ZIP_DEFLATED) as zf: for f in files5: zf.write(os.path.join(valid_dir, f), f) print(f" ✓ ZIP XXX005: {os.path.basename(zp)} ({len(files5)} файлов)") MANIFEST[os.path.basename(zp)] = {"type": "zip", "files": files5, "status": "valid"} # ZIP: только спецификации (без договора) specs = [f for f in os.listdir(valid_dir) if f.startswith("спецификация")][:4] if specs: zp2 = os.path.join(out, "only_specs.zip") with zipfile.ZipFile(zp2, 'w', zipfile.ZIP_DEFLATED) as zf: for f in specs: zf.write(os.path.join(valid_dir, f), f) print(f" ✓ ZIP только спеки: {os.path.basename(zp2)} ({len(specs)} файлов)") MANIFEST[os.path.basename(zp2)] = {"type": "zip", "specs_only": True, "status": "valid"} # ZIP с дубликатами имён внутри if len(files_for_zip) >= 1: zip_path2 = os.path.join(out, "duplicates_inside.zip") with zipfile.ZipFile(zip_path2, 'w', zipfile.ZIP_DEFLATED) as zf: zf.write(os.path.join(valid_dir, files_for_zip[0]), files_for_zip[0]) zf.write(os.path.join(valid_dir, files_for_zip[0]), files_for_zip[0]) print(f" ✓ ZIP с дубликатами: {os.path.basename(zip_path2)}") MANIFEST[os.path.basename(zip_path2)] = {"type": "zip", "duplicates": True, "status": "valid"} # ZIP с битым файлом внутри err_dir = os.path.join(OUT, "errors") if os.path.exists(err_dir) and files_for_zip: zip_path3 = os.path.join(out, "mixed_with_error.zip") with zipfile.ZipFile(zip_path3, 'w', zipfile.ZIP_DEFLATED) as zf: zf.write(os.path.join(valid_dir, files_for_zip[0]), files_for_zip[0]) zf.write(os.path.join(err_dir, "corrupt_xml.docx"), "corrupt_xml.docx") print(f" ✓ ZIP с битым: {os.path.basename(zip_path3)}") MANIFEST[os.path.basename(zip_path3)] = {"type": "zip", "has_error": True, "status": "valid"} print(f" ✓ ZIP-ы: 5 шт.") # ═══════════════════════════════════════════════════════════════ # Дубликаты # ═══════════════════════════════════════════════════════════════ def generate_duplicates(): out = os.path.join(OUT, "duplicates") ensure_dir(out) valid_dir = os.path.join(OUT, "valid") if not os.path.exists(valid_dir): return files = [f for f in os.listdir(valid_dir) if f.endswith('.docx')] if len(files) >= 3: # Точная копия (другое имя) × 3 for i in range(min(3, len(files))): src = os.path.join(valid_dir, files[i]) dst = os.path.join(out, f"copy_{i+1}_" + files[i]) shutil.copy(src, dst) MANIFEST[os.path.basename(dst)] = {"type": "duplicate", "original": files[i], "kind": "content_copy"} # Одинаковое имя, разный контент × 3 for i in range(min(3, len(files)-1)): shutil.copy(os.path.join(valid_dir, files[i+3]), os.path.join(out, files[i])) MANIFEST[files[i]] = {"type": "duplicate", "original": files[i], "kind": "name_collision"} # Рассинхрон: имя файла ≠ номер в тексте if len(files) >= 2: src = os.path.join(valid_dir, files[0]) dst = os.path.join(out, "mismatch_" + files[1]) shutil.copy(src, dst) MANIFEST[os.path.basename(dst)] = {"type": "error", "error": "name_content_mismatch", "status": "expected_classify_different"} shutil.copy(os.path.join(valid_dir, files[1]), os.path.join(out, "mismatch_" + files[0])) MANIFEST[os.path.basename("mismatch_" + files[0])] = {"type": "error", "error": "name_content_mismatch", "status": "expected_classify_different"} print(f" ✓ Дубликаты + рассинхрон: 8 шт.") # ═══════════════════════════════════════════════════════════════ # Main # ═══════════════════════════════════════════════════════════════ def main(): print("=" * 60) print("Генератор тестовых документов (схема Опуса)") print("=" * 60) print("\n── Валидные контракты ──") generate_valid() print("\n── Ошибочные кейсы ──") generate_errors() print("\n── Форматы (.doc/.pdf) ──") generate_formats() print("\n── ZIP-архивы ──") generate_zips() print("\n── Дубликаты ──") generate_duplicates() # Сохранить manifest manifest_path = os.path.join(OUT, "manifest.json") with open(manifest_path, 'w', encoding='utf-8') as f: json.dump(MANIFEST, f, ensure_ascii=False, indent=2) # Сгенерировать human-readable README.md readme_path = os.path.join(OUT, "README.md") lines = ["# Тестовые файлы — описание\n"] cats = {"contract": "📄 Договоры", "spec": "📋 Спецификации", "addendum": "📎 Допсоглашения", "error": "❌ Ошибочные", "trash": "🗑 Мусор", "zip": "📦 ZIP-архивы", "duplicate": "📋 Дубликаты"} by_cat = {} for fname, info in MANIFEST.items(): cat = info.get("type", "?") by_cat.setdefault(cat, []).append((fname, info)) for cat, title in cats.items(): items = by_cat.get(cat, []) if not items: continue lines.append(f"\n## {title} ({len(items)} шт.)\n") for fname, info in sorted(items): company = info.get("company", "") number = info.get("number", "") ver = f" v{info['version']}" if info.get("version") else "" err = info.get("error", "") label = info.get("label", "") extra = f" — {err}" if err else (f" — {label}" if label else "") lines.append(f"- `{fname}` {company} {number}{ver}{extra}") lines.append(f"\n\nВсего: **{len(MANIFEST)}** файлов") with open(readme_path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines)) total = len(MANIFEST) print(f"\n{'=' * 60}") print(f"Всего файлов в manifest.json: {total}") print(f"Вывод: {OUT}/") print(f"Манифест: {manifest_path}") print(f"{'=' * 60}") if __name__ == "__main__": main()