189 lines
7.9 KiB
Python
189 lines
7.9 KiB
Python
"""
|
|
corrupt.py — Порча docx/pdf, конвертация в .doc.
|
|
|
|
Схема Опуса: 17 типов ошибочных кейсов.
|
|
"""
|
|
|
|
import zipfile, io, os, shutil, tempfile, subprocess
|
|
|
|
def corrupt_xml(docx_path, out_path):
|
|
"""
|
|
Битый XML внутри docx: обрезать тег в word/document.xml.
|
|
Парсер должен вернуть ✗, pipeline не падает.
|
|
"""
|
|
with zipfile.ZipFile(docx_path, 'r') as zin:
|
|
data = {name: zin.read(name) for name in zin.namelist()}
|
|
|
|
# Найти document.xml и обрезать последние 200 байт
|
|
doc_key = 'word/document.xml'
|
|
if doc_key in data:
|
|
xml = data[doc_key]
|
|
# Обрезать закрывающие теги
|
|
cut = max(len(xml) - 500, len(xml) // 2)
|
|
data[doc_key] = xml[:cut]
|
|
|
|
with zipfile.ZipFile(out_path, 'w', zipfile.ZIP_DEFLATED) as zout:
|
|
for name, content in data.items():
|
|
zout.writestr(name, content)
|
|
|
|
def make_empty_docx(out_path):
|
|
"""Пустой docx — без текста, без таблиц."""
|
|
from docx import Document
|
|
doc = Document()
|
|
doc.save(out_path)
|
|
|
|
def make_empty_file(out_path):
|
|
"""Файл нулевого размера."""
|
|
with open(out_path, 'wb') as f:
|
|
pass
|
|
|
|
def make_text_only_docx(out_path, title="Соглашение об услугах"):
|
|
"""Docx только с текстом, без таблиц. Крайний случай для парсера."""
|
|
from docx import Document
|
|
doc = Document()
|
|
p = doc.add_paragraph()
|
|
p.add_run(title).bold = True
|
|
doc.add_paragraph("Настоящий документ является соглашением об оказании услуг.")
|
|
doc.add_paragraph("Стороны: Заказчик и Исполнитель.")
|
|
doc.add_paragraph("Стоимость услуг: 100 000 рублей.")
|
|
doc.add_paragraph("Дата: 01.01.2026")
|
|
doc.save(out_path)
|
|
|
|
def make_table_only_docx(out_path):
|
|
"""Docx только с таблицей, без параграфов. Крайний случай для парсера."""
|
|
from docx import Document
|
|
doc = Document()
|
|
table = doc.add_table(rows=3, cols=6)
|
|
table.style = 'Table Grid'
|
|
headers = ["№", "Наименование", "Цена", "Объем", "Сумма", "Дата"]
|
|
for i, h in enumerate(headers):
|
|
table.rows[0].cells[i].text = h
|
|
table.rows[1].cells[0].text = "1"
|
|
table.rows[1].cells[1].text = "Тестовая услуга"
|
|
table.rows[1].cells[2].text = "1000,00"
|
|
table.rows[1].cells[3].text = "2"
|
|
table.rows[1].cells[4].text = "2000,00"
|
|
table.rows[1].cells[5].text = "01.06.2026"
|
|
doc.save(out_path)
|
|
|
|
def make_no_number_docx(out_path, template_builder, ctx):
|
|
"""
|
|
Документ БЕЗ номера договора в тексте.
|
|
classify должен вернуть ошибку или 'other'.
|
|
"""
|
|
doc = template_builder(ctx)
|
|
# Удаляем параграф с номером (второй параграф после заголовка)
|
|
# Простой подход: заменяем текст номера на пробел
|
|
for p in doc.paragraphs:
|
|
if ctx["number"] in p.text:
|
|
p.text = p.text.replace(f"№ {ctx['number']}", "№ ")
|
|
break
|
|
doc.save(out_path)
|
|
|
|
def make_nonstandard_title_docx(out_path):
|
|
"""
|
|
Документ с нестандартным заголовком — classify должен классифицировать
|
|
по содержимому, а не по заголовку.
|
|
"""
|
|
from docx import Document
|
|
doc = Document()
|
|
p = doc.add_paragraph()
|
|
p.add_run("Соглашение о предоставлении технологических услуг").bold = True
|
|
doc.add_paragraph("№ 99999")
|
|
doc.add_paragraph('ООО "Нестандарт" и ООО "НУБЕС" заключили настоящее соглашение.')
|
|
doc.add_paragraph("Предмет: оказание услуг ЦОД.")
|
|
# Таблица услуг
|
|
table = doc.add_table(rows=2, cols=6)
|
|
table.style = 'Table Grid'
|
|
for i, h in enumerate(["№", "Наименование", "Цена", "Объем", "Сумма", "Дата"]):
|
|
table.rows[0].cells[i].text = h
|
|
table.rows[1].cells[0].text = "1"
|
|
table.rows[1].cells[1].text = "Аренда стойко-места"
|
|
table.rows[1].cells[2].text = "200 000,00"
|
|
table.rows[1].cells[3].text = "2"
|
|
table.rows[1].cells[4].text = "400 000,00"
|
|
table.rows[1].cells[5].text = "01.06.2026"
|
|
doc.save(out_path)
|
|
|
|
def make_giant_docx(out_path, rows=500):
|
|
"""Документ с гигантской таблицей — тест производительности."""
|
|
from docx import Document
|
|
doc = Document()
|
|
p = doc.add_paragraph()
|
|
p.add_run("Договор № GIANT").bold = True
|
|
doc.add_paragraph("Спецификация с большим количеством позиций.")
|
|
table = doc.add_table(rows=1, cols=6)
|
|
table.style = 'Table Grid'
|
|
for i, h in enumerate(["№", "Наименование", "Цена", "Объем", "Сумма", "Дата"]):
|
|
table.rows[0].cells[i].text = h
|
|
for r in range(rows):
|
|
row = table.add_row()
|
|
row.cells[0].text = str(r + 1)
|
|
row.cells[1].text = f"Услуга {r+1}"
|
|
row.cells[2].text = f"{1000 + r * 10},00"
|
|
row.cells[3].text = "1"
|
|
row.cells[4].text = f"{1000 + r * 10},00"
|
|
row.cells[5].text = "01.06.2026"
|
|
doc.save(out_path)
|
|
|
|
def make_trash_docx(out_path, label="Счёт-фактура"):
|
|
"""Мусорный документ — не договор. Негативная классификация."""
|
|
from docx import Document
|
|
doc = Document()
|
|
p = doc.add_paragraph()
|
|
p.add_run(label).bold = True
|
|
doc.add_paragraph("№ СФ-2026-001 от 15.06.2026")
|
|
doc.add_paragraph("Поставщик: ООО Рога и Копыта")
|
|
doc.add_paragraph("Покупатель: ЗАО XXX001")
|
|
doc.add_paragraph("Сумма: 150 000,00 руб.")
|
|
doc.save(out_path)
|
|
|
|
def convert_to_doc(docx_path, out_path):
|
|
"""Конвертировать docx → doc через LibreOffice."""
|
|
out_dir = os.path.dirname(out_path) or "."
|
|
out_name = os.path.basename(out_path)
|
|
tmp = tempfile.mkdtemp()
|
|
shutil.copy(docx_path, os.path.join(tmp, "temp.docx"))
|
|
try:
|
|
subprocess.run(
|
|
["libreoffice", "--headless", "--convert-to", "doc", "--outdir", tmp,
|
|
os.path.join(tmp, "temp.docx")],
|
|
timeout=30, capture_output=True
|
|
)
|
|
src = os.path.join(tmp, "temp.doc")
|
|
if os.path.exists(src):
|
|
shutil.move(src, out_path)
|
|
return True
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
return False
|
|
|
|
def make_broken_pdf(docx_path, out_path):
|
|
"""Сгенерировать PDF из docx и обрезать байты — битый PDF."""
|
|
# Сначала генерируем нормальный PDF через LibreOffice
|
|
out_dir = os.path.dirname(out_path) or "."
|
|
tmp = tempfile.mkdtemp()
|
|
shutil.copy(docx_path, os.path.join(tmp, "temp.docx"))
|
|
try:
|
|
subprocess.run(
|
|
["libreoffice", "--headless", "--convert-to", "pdf", "--outdir", tmp,
|
|
os.path.join(tmp, "temp.docx")],
|
|
timeout=30, capture_output=True
|
|
)
|
|
src = os.path.join(tmp, "temp.pdf")
|
|
if os.path.exists(src):
|
|
with open(src, 'rb') as f:
|
|
data = f.read()
|
|
# Обрезать последние 30% байт
|
|
cut = int(len(data) * 0.7)
|
|
with open(out_path, 'wb') as f:
|
|
f.write(data[:cut])
|
|
return True
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
return False
|