581 lines
28 KiB
Python
581 lines
28 KiB
Python
"""
|
||
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()
|