472 lines
22 KiB
Python
472 lines
22 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):
|
||
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(1, 3)
|
||
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"}
|
||
|
||
# Мусор (3 шт.)
|
||
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" Сгенерировано ошибочных: 13")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# .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)
|
||
print(f" ✓ ZIP: {os.path.basename(zip_path)} ({len(files_for_zip)} файлов)")
|
||
MANIFEST[os.path.basename(zip_path)] = {"type": "zip", "files": files_for_zip, "status": "valid"}
|
||
|
||
# ZIP с дубликатами имён внутри
|
||
if len(files_for_zip) >= 2:
|
||
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"}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# Дубликаты
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
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:
|
||
# Точная копия (другое имя)
|
||
src = os.path.join(valid_dir, files[0])
|
||
shutil.copy(src, os.path.join(out, "copy_of_" + files[0]))
|
||
MANIFEST["copy_of_" + files[0]] = {"type": "duplicate", "original": files[0], "kind": "content_copy"}
|
||
|
||
# Одинаковое имя, разный контент
|
||
shutil.copy(os.path.join(valid_dir, files[1]), os.path.join(out, files[0]))
|
||
MANIFEST[files[0]] = {"type": "duplicate", "original": files[0], "kind": "name_collision"}
|
||
|
||
print(f" ✓ Дубликаты: 2 шт.")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# 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)
|
||
|
||
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()
|