proposa-ai/scripts/tests/test_template_text_reading.py

import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from docx import Document
from docx.enum.text import WD_BREAK
from docx.oxml import OxmlElement
from models import BidOutline, Chapter, CompanyInfo, TemplateSection, TemplateStructure, TenderAnalysis
from step3_outlining.template_parser import (
    _read_template_body_texts, get_template_text_for_chapter, get_template_text_for_section,
)
from step4_writing import _write_single_chapter
from step4_writing.content_injector import _ContentInjector


class TemplateTextReadingTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.path = Path(self.directory.name) / 'template.docx'
        doc = Document()
        doc.add_paragraph('第四章 服务方案')
        doc.add_paragraph('')
        doc.add_table(rows=1, cols=1).cell(0, 0).text = '表内文本不计入正文索引'
        p = doc.add_paragraph('正文\t制表\n换行')
        p.add_run().add_break(WD_BREAK.PAGE)
        link = OxmlElement('w:hyperlink')
        run = OxmlElement('w:r')
        text = OxmlElement('w:t')
        text.text = '链接'
        run.append(text)
        run.append(OxmlElement('w:noBreakHyphen'))
        link.append(run)
        p._p.append(link)
        doc.add_paragraph('第五章 其他方案')
        self.expected = [p.text for p in doc.paragraphs]
        doc.save(self.path)
        self.structure = TemplateStructure(file_path=str(self.path), sections=[
            TemplateSection(name='第四章 服务方案', start_para=0, end_para=2),
            TemplateSection(name='第五章 其他方案', start_para=3, end_para=99),
        ])

    def test_top_level_indices_and_visible_text_match_python_docx(self):
        self.assertEqual(_read_template_body_texts(str(self.path)), self.expected)

    def test_parallel_section_and_title_reads_do_not_use_native_parser(self):
        expected = '\n'.join(t.strip() for t in self.expected[:3] if t.strip())
        def read(index):
            if index % 2:
                return get_template_text_for_section(self.structure, '第四章 服务方案')
            return get_template_text_for_chapter(self.structure, '第四章 服务方案', use_llm=False)
        with patch('step3_outlining.template_parser.DocxDocument', side_effect=AssertionError(
            '并发纯文本取文不得进入python-docx原生解析器'
        )), ThreadPoolExecutor(max_workers=5) as pool:
            self.assertEqual(list(pool.map(read, range(25))), [expected] * 25)
        self.assertEqual(get_template_text_for_section(self.structure, '第五章 其他方案'), self.expected[3])
        self.assertEqual(get_template_text_for_section(self.structure, '不存在'), '')

    def test_step4_chapter_generation_uses_streamed_template_text(self):
        chapter = Chapter(id='4', title='服务方案', template_section='第四章 服务方案')
        outline = BidOutline(project_name='测试项目', chapters=[chapter])
        with patch('step3_outlining.template_parser.DocxDocument', side_effect=TypeError(
            "'CT_P' object is not callable"
        )):
            result = _write_single_chapter(
                chapter=chapter, writer=SimpleNamespace(), injector=_ContentInjector({}, []),
                analysis=TenderAnalysis(project_name='测试项目'), company_info=CompanyInfo(),
                template_structure=self.structure, cfg=SimpleNamespace(enforce_word_limit=False),
                outline=outline, placeholder_map={},
            )
        self.assertEqual(result.generated_content, '\n'.join(t for t in self.expected[:3] if t))
        self.assertTrue(result.preserve_template_layout)