diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..b1c1577 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,31 @@ +name: Tests + +on: + push: + pull_request: + workflow_dispatch: + +jobs: + pytest: + name: Pytest (${{ matrix.python-version }}) + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: "pip" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install -e . pytest + + - name: Run tests + run: python -m pytest tests diff --git a/deepdoc/common/misc_utils.py b/deepdoc/common/misc_utils.py index 521fa7b..cf3efd2 100644 --- a/deepdoc/common/misc_utils.py +++ b/deepdoc/common/misc_utils.py @@ -15,6 +15,7 @@ # import logging +import os logger = logging.getLogger(__name__) @@ -31,3 +32,13 @@ def pip_install_torch(): return True except ImportError: return False + + +def parse_bool(value: str | None, default: bool = False) -> bool: + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def offline_mode_or_from_env(offline: bool | None = None) -> bool: + return offline if offline is not None else parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False) diff --git a/deepdoc/common/model_store.py b/deepdoc/common/model_store.py index 8ccfecf..e8f6589 100644 --- a/deepdoc/common/model_store.py +++ b/deepdoc/common/model_store.py @@ -22,17 +22,14 @@ import logging import os from dataclasses import dataclass +from importlib import resources from pathlib import Path +from ..common.misc_utils import offline_mode_or_from_env GLOBAL_MODELSCOPE_REPO_ENV = "DEEPDOC_MODELSCOPE_REPO" GLOBAL_MODELSCOPE_REVISION_ENV = "DEEPDOC_MODELSCOPE_REVISION" - - -def _parse_bool(value: str | None, default: bool = False) -> bool: - if value is None: - return default - return value.strip().lower() in {"1", "true", "yes", "on"} +TOKENIZER_MODEL_DIR_ENV = "DEEPDOC_TOKENIZER_MODEL_DIR" def _normalize_provider(provider: str | None) -> str: @@ -56,6 +53,18 @@ def _model_home_path(model_home: str | None) -> Path: return Path.home().joinpath(".cache", "deepdoc") +def _resolve_tokenizer_dict_path() -> Path: + configured_dir = os.getenv(TOKENIZER_MODEL_DIR_ENV) + if configured_dir: + dictionary = Path(configured_dir).expanduser().resolve().joinpath("huqie.txt") + else: + dictionary = Path(str(resources.files("deepdoc").joinpath("dict", "huqie.txt"))).resolve() + + if not dictionary.exists(): + raise FileNotFoundError("Tokenizer dictionary not found: {}. Set {} to a directory containing huqie.txt.".format(dictionary, TOKENIZER_MODEL_DIR_ENV)) + return dictionary + + @dataclass(frozen=True) class BundleSpec: name: str @@ -243,7 +252,7 @@ def resolve_bundle_dir( spec = BUNDLES[bundle] provider_name = _normalize_provider(provider) - offline_mode = offline if offline is not None else _parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False) + offline_mode = offline_mode_or_from_env(offline) explicit_local = os.getenv(spec.local_dir_env) if explicit_local: @@ -356,5 +365,5 @@ def resolve_tokenizer_dict_prefix( provider: str | None = None, offline: bool | None = None, ) -> str: - bundle_dir = Path(resolve_bundle_dir("tokenizer", model_home=model_home, provider=provider, offline=offline)) - return str(bundle_dir.joinpath("huqie")) + del model_home, provider, offline + return str(_resolve_tokenizer_dict_path().with_suffix("")) diff --git a/deepdoc/config.py b/deepdoc/config.py index c6aa82e..1ba540f 100644 --- a/deepdoc/config.py +++ b/deepdoc/config.py @@ -11,16 +11,11 @@ resolve_xgb_model_dir, validate_bundle_dir, ) +from .common.misc_utils import offline_mode_or_from_env ProviderType = Literal["local", "modelscope", "auto"] -def _parse_bool(value: str | None, default: bool = False) -> bool: - if value is None: - return default - return value.strip().lower() in {"1", "true", "yes", "on"} - - def _normalize_provider(provider: str) -> ProviderType: normalized = provider.strip().lower() aliases = { @@ -31,9 +26,7 @@ def _normalize_provider(provider: str) -> ProviderType: } normalized = aliases.get(normalized, normalized) if normalized not in {"local", "modelscope", "auto"}: - raise ValueError( - "Unsupported model provider '{}'. Use one of: local, modelscope, auto.".format(provider) - ) + raise ValueError("Unsupported model provider '{}'. Use one of: local, modelscope, auto.".format(provider)) return normalized # type: ignore[return-value] @@ -64,13 +57,10 @@ def resolve_dict_path(self) -> str: if dictionary.is_dir(): dictionary = dictionary.joinpath("huqie.txt") if dictionary.suffix != ".txt": - raise ValueError( - "TokenizerConfig.dict_path must point to a '.txt' dictionary file, got: {}".format(dictionary) - ) + raise ValueError("TokenizerConfig.dict_path must point to a '.txt' dictionary file, got: {}".format(dictionary)) _require_file( dictionary, - "Tokenizer dictionary not found: {}. Provide a valid TokenizerConfig.dict_path." - .format(dictionary), + "Tokenizer dictionary not found: {}. Provide a valid TokenizerConfig.dict_path.".format(dictionary), ) return str(dictionary) @@ -86,7 +76,7 @@ def from_env(cls) -> "TokenizerConfig": return cls( dict_path=dict_path, - offline=_parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False), + offline=offline_mode_or_from_env(None), nltk_data_dir=os.getenv("DEEPDOC_NLTK_DATA_DIR"), ) @@ -107,10 +97,7 @@ def _resolve_bundle_dir(self, bundle: str, explicit_dir: str | None) -> str: candidate = Path(explicit_dir).expanduser().resolve() exists, missing = validate_bundle_dir(bundle, candidate) if not exists: - raise FileNotFoundError( - "Missing required files for '{}' bundle in {}: {}" - .format(bundle, candidate, ", ".join(missing)) - ) + raise FileNotFoundError("Missing required files for '{}' bundle in {}: {}".format(bundle, candidate, ", ".join(missing))) return str(candidate) model_provider = self.normalized_provider() diff --git a/deepdoc/depend/nltk_manager.py b/deepdoc/depend/nltk_manager.py index 9b2886b..08e1af5 100644 --- a/deepdoc/depend/nltk_manager.py +++ b/deepdoc/depend/nltk_manager.py @@ -2,19 +2,13 @@ import os import threading from pathlib import Path +from ..common.misc_utils import offline_mode_or_from_env logger = logging.getLogger(__name__) _RESOURCE_SPECS: tuple[tuple[str, tuple[str, ...]], ...] = ( - ( - "punkt", - ( - "tokenizers/punkt", - "tokenizers/punkt.zip", - "tokenizers/punkt_tab", - "tokenizers/punkt_tab.zip", - ), - ), + ("punkt", ("tokenizers/punkt", "tokenizers/punkt.zip")), + ("punkt_tab", ("tokenizers/punkt_tab", "tokenizers/punkt_tab.zip")), ("wordnet", ("corpora/wordnet", "corpora/wordnet.zip")), ( "averaged_perceptron_tagger", @@ -31,12 +25,6 @@ _ensured_keys: set[tuple[str, bool]] = set() -def _parse_bool(value: str | None, default: bool = False) -> bool: - if value is None: - return default - return value.strip().lower() in {"1", "true", "yes", "on"} - - def _resolve_nltk_data_dir(data_dir: str | None) -> Path | None: # Resolution precedence: # 1) explicit arg @@ -88,7 +76,7 @@ def ensure_nltk_data( import nltk resolved_dir = _resolve_nltk_data_dir(data_dir) - offline_mode = offline if offline is not None else _parse_bool(os.getenv("DEEPDOC_OFFLINE"), default=False) + offline_mode = offline_mode_or_from_env(offline) auto_download_mode = not offline_mode _ensure_search_path(nltk, resolved_dir) @@ -117,10 +105,7 @@ def ensure_nltk_data( if missing_packages: searched_paths = ", ".join(nltk.data.path) raise RuntimeError( - "Missing required NLTK packages: {}. Searched paths: {}. " - "Set DEEPDOC_NLTK_DATA_DIR to a local NLTK data path, or disable offline mode by setting " - "DEEPDOC_OFFLINE=0." - .format( + "Missing required NLTK packages: {}. Searched paths: {}. Set DEEPDOC_NLTK_DATA_DIR to a local NLTK data path, or disable offline mode by setting DEEPDOC_OFFLINE=0.".format( ", ".join(missing_packages), searched_paths, ) diff --git a/pyproject.toml b/pyproject.toml index c8bdd65..66136a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,6 +89,11 @@ extend-select = ["ASYNC", "ASYNC1"] ignore = ["E402"] [tool.pytest.ini_options] +addopts = [ + "--ignore=tests/test_data", + "--ignore=tests/test_results", +] +testpaths = ["tests"] markers = [ "p1: high priority test cases", "p2: medium priority test cases", diff --git a/tests/test_model_store.py b/tests/test_model_store.py index 9b91f78..4e16733 100644 --- a/tests/test_model_store.py +++ b/tests/test_model_store.py @@ -22,10 +22,6 @@ def _create_combined_repo_layout(root: Path) -> None: for name in ms.BUNDLES["xgb"].required_files: _touch(root / "xgb" / name) - # Tokenizer bundle - for name in ms.BUNDLES["tokenizer"].required_files: - _touch(root / "tokenizer" / name) - class TestModelStoreSharedRepo(unittest.TestCase): def setUp(self) -> None: @@ -68,12 +64,10 @@ def snapshot_download( with patch.object(ms, "_import_modelscope_snapshot_download", return_value=snapshot_download): vision_dir = Path(ms.resolve_bundle_dir("vision", model_home=tmp, provider="modelscope", offline=False)) xgb_dir = Path(ms.resolve_bundle_dir("xgb", model_home=tmp, provider="modelscope", offline=False)) - tok_dir = Path(ms.resolve_bundle_dir("tokenizer", model_home=tmp, provider="modelscope", offline=False)) expected_root = (Path(tmp) / "modelscope" / "Xorbits__deepdoc" / "v1").resolve() self.assertEqual(vision_dir.resolve(), (expected_root / "vision").resolve()) self.assertEqual(xgb_dir.resolve(), (expected_root / "xgb").resolve()) - self.assertEqual(tok_dir.resolve(), (expected_root / "tokenizer").resolve()) self.assertGreaterEqual(len(calls), 1) for call in calls: @@ -134,8 +128,22 @@ def snapshot_download(*args, **kwargs) -> str: # pragma: no cover with patch.object(ms, "_import_modelscope_snapshot_download", return_value=snapshot_download): vision_dir = Path(ms.resolve_bundle_dir("vision", model_home=tmp, provider="auto", offline=False)) xgb_dir = Path(ms.resolve_bundle_dir("xgb", model_home=tmp, provider="auto", offline=False)) - tok_dir = Path(ms.resolve_bundle_dir("tokenizer", model_home=tmp, provider="auto", offline=False)) self.assertEqual(vision_dir.resolve(), (expected_root / "vision").resolve()) self.assertEqual(xgb_dir.resolve(), (expected_root / "xgb").resolve()) - self.assertEqual(tok_dir.resolve(), (expected_root / "tokenizer").resolve()) + + def test_resolve_tokenizer_dict_prefix_uses_packaged_dict_by_default(self) -> None: + prefix = Path(ms.resolve_tokenizer_dict_prefix()) + + self.assertEqual(prefix.name, "huqie") + self.assertTrue(prefix.with_suffix(".txt").exists()) + + def test_resolve_tokenizer_dict_prefix_uses_env_dir_when_set(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tokenizer_dir = Path(tmp) / "tokenizer" + _touch(tokenizer_dir / "huqie.txt") + os.environ[ms.TOKENIZER_MODEL_DIR_ENV] = str(tokenizer_dir) + + prefix = Path(ms.resolve_tokenizer_dict_prefix()) + + self.assertEqual(prefix.resolve(), (tokenizer_dir / "huqie").resolve()) diff --git a/tests/test_parser.py b/tests/test_parser.py index 17d9b18..d5e1b34 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -1,59 +1,89 @@ import os +import shutil import sys -import unittest import tempfile -import shutil +import unittest +import warnings from pathlib import Path -from unittest.mock import patch, MagicMock -from io import BytesIO + +from docx import Document +from pptx import Presentation +from pypdf import PdfWriter # 添加 deepdoc 路径到 sys.path -sys.path.append(os.path.join(os.path.dirname(__file__), '..')) +sys.path.append(os.path.join(os.path.dirname(__file__), "..")) # 导入所有解析器（使用正确的类名） -from parser import ( - TxtParser, MarkdownParser, JsonParser, HtmlParser, - ExcelParser, PptParser, DocxParser, PdfParser -) -from parser.figure_parser import VisionFigureParser - -# 导入视觉模型 -from ..depend.simple_cv_model import create_vision_model +from deepdoc import TxtParser, MarkdownParser, JsonParser, HtmlParser, ExcelParser, PptParser, DocxParser, PdfParser class TestParsers(unittest.TestCase): """测试所有解析器的功能""" - + def setUp(self): """测试前的准备工作""" # 创建测试数据目录 self.test_data_dir = Path(__file__).parent / "test_data" self.test_data_dir.mkdir(exist_ok=True) - + # 创建测试结果目录 self.test_results_dir = Path(__file__).parent / "test_results" self.test_results_dir.mkdir(exist_ok=True) - - # 创建视觉模型实例（用于图片解析） - self.vision_model = create_vision_model("qwen") - # 使用配置文件 - # self.vision_model = create_vision_model("deepdoc_config.yaml") - + # 创建临时文件 self.temp_dir = tempfile.mkdtemp() - + def tearDown(self): """测试后的清理工作""" # 清理临时文件 shutil.rmtree(self.temp_dir, ignore_errors=True) - - def create_test_file(self, content, filename, encoding='utf-8'): + + def create_test_file(self, content, filename, encoding="utf-8"): """创建测试文件""" file_path = self.test_data_dir / filename - with open(file_path, 'w', encoding=encoding) as f: + with open(file_path, "w", encoding=encoding) as f: f.write(content) - return file_path - + return os.fspath(file_path) + + def create_test_docx_file(self, filename): + """创建测试Word文件""" + file_path = Path(self.temp_dir) / filename + doc = Document() + doc.add_heading("测试Word标题", level=1) + doc.add_paragraph("这是Word正文内容。") + table = doc.add_table(rows=2, cols=2) + table.cell(0, 0).text = "列1" + table.cell(0, 1).text = "列2" + table.cell(1, 0).text = "数据1" + table.cell(1, 1).text = "数据2" + doc.save(file_path) + return os.fspath(file_path) + + def create_test_pptx_file(self, filename): + """创建测试PPT文件""" + file_path = Path(self.temp_dir) / filename + presentation = Presentation() + + slide = presentation.slides.add_slide(presentation.slide_layouts[1]) + slide.shapes.title.text = "测试PPT标题" + slide.placeholders[1].text = "第一页内容" + + slide = presentation.slides.add_slide(presentation.slide_layouts[1]) + slide.shapes.title.text = "第二页标题" + slide.placeholders[1].text = "第二页内容" + + presentation.save(file_path) + return os.fspath(file_path) + + def create_test_pdf_file(self, filename): + """创建测试PDF文件""" + file_path = Path(self.temp_dir) / filename + writer = PdfWriter() + writer.add_blank_page(width=612, height=792) + with open(file_path, "wb") as f: + writer.write(f) + return os.fspath(file_path) + def test_txt_parser(self): """测试文本解析器""" # 创建测试文本文件 @@ -62,25 +92,25 @@ def test_txt_parser(self): 测试中文和English混合内容。 包含特殊字符：!@#$%^&*() 包含数字：1234567890""" - + test_file = self.create_test_file(test_content, "test.txt") - + # 测试解析 parser = TxtParser() result = parser(test_file) - + # 验证结果 self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + # 检查内容 all_text = " ".join([chunk[0] for chunk in result]) self.assertIn("测试文档", all_text) self.assertIn("中文和English", all_text) - + print(f"✅ TXT解析成功: {len(result)} 个chunk") - + def test_markdown_parser(self): """测试Markdown解析器""" test_content = """# 测试标题 @@ -105,27 +135,27 @@ def test_markdown_parser(self): | 行1 | 数据1 | 数据2 | | 行2 | 数据3 | 数据4 | """ - + test_file = self.create_test_file(test_content, "test.md") - + # 测试解析 parser = MarkdownParser() - + # 读取文件内容 - with open(test_file, 'r', encoding='utf-8') as f: + with open(test_file, "r", encoding="utf-8") as f: content = f.read() - + # 测试表格提取 text, tables = parser.extract_tables_and_remainder(content) - + # 验证结果 self.assertIsNotNone(text) self.assertIsInstance(tables, list) self.assertIn("测试标题", text) self.assertIn("粗体", text) - + print(f"✅ Markdown解析成功: {len(tables)} 个表格") - + def test_json_parser(self): """测试JSON解析器""" test_content = """{ @@ -141,29 +171,29 @@ def test_json_parser(self): "active": true, "score": 95.5 }""" - + test_file = self.create_test_file(test_content, "test.json") - + # 读取文件为二进制 - with open(test_file, 'rb') as f: + with open(test_file, "rb") as f: binary_content = f.read() - + # 测试解析 parser = JsonParser() result = parser(binary_content) - + # 验证结果 self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + # 检查内容 all_text = " ".join(result) self.assertIn("测试用户", all_text) self.assertIn("北京", all_text) - + print(f"✅ JSON解析成功: {len(result)} 个chunk") - + def test_html_parser(self): """测试HTML解析器""" test_content = """ @@ -199,25 +229,25 @@ def test_html_parser(self): """ - + test_file = self.create_test_file(test_content, "test.html") - + # 测试解析 parser = HtmlParser() result = parser(test_file) - + # 验证结果 self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + # 检查内容 all_text = " ".join(result) - self.assertIn("测试页面", all_text) self.assertIn("主标题", all_text) - + self.assertIn("这是div中的内容。", all_text) + print(f"✅ HTML解析成功: {len(result)} 行") - + def test_excel_parser(self): """测试Excel解析器""" # 创建简单的CSV文件作为Excel测试 @@ -226,120 +256,166 @@ def test_excel_parser(self): 李四,30,上海,设计师 王五,28,广州,产品经理 赵六,35,深圳,销售经理""" - + test_file = self.create_test_file(test_content, "test.csv") - + # 读取文件为二进制 - with open(test_file, 'rb') as f: + with open(test_file, "rb") as f: binary_content = f.read() - + # 测试解析 parser = ExcelParser() result = parser(binary_content) - + # 验证结果 self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + # 检查内容 all_text = " ".join(result) self.assertIn("张三", all_text) self.assertIn("工程师", all_text) - + print(f"✅ Excel/CSV解析成功: {len(result)} 行") - + def test_excel_parser_html_output(self): """测试Excel解析器的HTML输出""" test_content = """姓名,年龄,城市,职业 张三,25,北京,工程师 李四,30,上海,设计师""" - + test_file = self.create_test_file(test_content, "test_html.csv") - + # 读取文件为二进制 - with open(test_file, 'rb') as f: + with open(test_file, "rb") as f: binary_content = f.read() - + # 测试HTML输出 parser = ExcelParser() result = parser.html(binary_content) - + # 验证结果 self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + # 检查HTML内容 html_content = result[0] self.assertIn("", html_content) self.assertIn("
", html_content) self.assertIn("张三", html_content) - + print(f"✅ Excel HTML输出成功: {len(result)} 个表格") - - @unittest.skip("需要真实的PPT文件") + def test_ppt_parser(self): """测试PPT解析器""" - # 这个测试需要真实的PPT文件 - pass - - @unittest.skip("需要真实的Word文件") + test_file = self.create_test_pptx_file("test.pptx") + + parser = PptParser() + result = parser(test_file, from_page=0, to_page=10) + + self.assertIsNotNone(result) + self.assertIsInstance(result, list) + self.assertGreaterEqual(len(result), 2) + + all_text = " ".join(result) + self.assertIn("测试PPT标题", all_text) + self.assertIn("第一页内容", all_text) + self.assertIn("第二页内容", all_text) + + print(f"✅ PPT解析成功: {len(result)} 页") + def test_docx_parser(self): """测试Word解析器""" - # 这个测试需要真实的Word文件 - pass - - @unittest.skip("需要真实的PDF文件") + test_file = self.create_test_docx_file("test.docx") + + parser = DocxParser() + secs, tbls = parser(test_file) + + self.assertIsInstance(secs, list) + self.assertIsInstance(tbls, list) + self.assertGreater(len(secs), 0) + + all_text = " ".join(text for text, _style in secs if text) + self.assertIn("测试Word标题", all_text) + self.assertIn("这是Word正文内容。", all_text) + + all_tables = " ".join(line for table in tbls for line in table) + self.assertIn("列1: 数据1", all_tables) + self.assertIn("列2: 数据2", all_tables) + + print(f"✅ DOCX解析成功: {len(secs)} 个段落, {len(tbls)} 个表格") + def test_pdf_parser(self): """测试PDF解析器""" - # 这个测试需要真实的PDF文件 - pass - + test_file = self.create_test_pdf_file("test.pdf") + + parser = PdfParser() + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + text_content, tbls = parser(test_file, need_image=False) + except Exception as exc: + msg = str(exc).lower() + asset_markers = ("model", "onnx", "xgb", "nltk", "wordnet", "huqie", "no such file", "not found") + if any(marker in msg for marker in asset_markers): + self.skipTest(f"PdfParser runtime assets unavailable: {exc}") + raise + + self.assertIsInstance(text_content, str) + self.assertIsInstance(tbls, list) + self.assertEqual(parser.total_page, 1) + + print(f"✅ PDF解析成功: {len(text_content)} 个文本字符, {len(tbls)} 个表格/图片块") + @unittest.skip("需要真实的图片文件") def test_figure_parser(self): """测试图片解析器""" # 这个测试需要真实的图片文件 pass - + def test_parser_error_handling(self): """测试解析器错误处理""" # 测试不存在的文件 non_existent_file = self.test_data_dir / "non_existent.txt" - + parser = TxtParser() - with self.assertRaises(Exception): - parser(non_existent_file) - + with self.assertRaises(FileNotFoundError): + parser(os.fspath(non_existent_file)) + print("✅ 错误处理测试通过") - + def test_encoding_handling(self): """测试编码处理""" # 创建GBK编码的文件 test_content = "这是GBK编码的中文内容" - test_file = self.create_test_file(test_content, "test_gbk.txt", encoding='gbk') - + test_file = self.create_test_file(test_content, "test_gbk.txt", encoding="gbk") + parser = TxtParser() - result = parser(test_file) - + with open(test_file, "rb") as f: + binary_content = f.read() + + result = parser(test_file, binary_content) + self.assertIsNotNone(result) self.assertIsInstance(result, list) - + # 检查内容 all_text = " ".join([chunk[0] for chunk in result]) self.assertIn("GBK编码", all_text) - + print("✅ 编码处理测试通过") class TestParserIntegration(unittest.TestCase): """测试解析器集成功能""" - + def setUp(self): """测试前的准备工作""" self.test_data_dir = Path(__file__).parent / "test_data" self.test_data_dir.mkdir(exist_ok=True) - + def test_multiple_parsers(self): """测试多个解析器同时工作""" # 创建不同类型的测试文件 @@ -348,63 +424,57 @@ def test_multiple_parsers(self): "test2.md": "# 标题\n这是Markdown内容", "test3.json": '{"name": "测试", "value": 123}', "test4.html": "

标题

内容

", - "test5.csv": "姓名,年龄\n张三,25\n李四,30" - } - - parsers = { - "txt": TxtParser(), - "md": MarkdownParser(), - "json": JsonParser(), - "html": HtmlParser(), - "excel": ExcelParser() + "test5.csv": "姓名,年龄\n张三,25\n李四,30", } - + + parsers = {"txt": TxtParser(), "md": MarkdownParser(), "json": JsonParser(), "html": HtmlParser(), "excel": ExcelParser()} + results = {} - + for filename, content in files.items(): file_path = self.test_data_dir / filename - with open(file_path, 'w', encoding='utf-8') as f: + with open(file_path, "w", encoding="utf-8") as f: f.write(content) - + # 根据文件扩展名选择解析器 - ext = filename.split('.')[-1] - if ext == 'txt': - parser = parsers['txt'] - result = parser(file_path) - elif ext == 'md': - parser = parsers['md'] - with open(file_path, 'r', encoding='utf-8') as f: + ext = filename.split(".")[-1] + if ext == "txt": + parser = parsers["txt"] + result = parser(os.fspath(file_path)) + elif ext == "md": + parser = parsers["md"] + with open(file_path, "r", encoding="utf-8") as f: content = f.read() text, tables = parser.extract_tables_and_remainder(content) result = [text] # 简化处理 - elif ext == 'json': - parser = parsers['json'] - with open(file_path, 'rb') as f: + elif ext == "json": + parser = parsers["json"] + with open(file_path, "rb") as f: binary_content = f.read() result = parser(binary_content) - elif ext == 'html': - parser = parsers['html'] - result = parser(file_path) - elif ext == 'csv': - parser = parsers['excel'] - with open(file_path, 'rb') as f: + elif ext == "html": + parser = parsers["html"] + result = parser(os.fspath(file_path)) + elif ext == "csv": + parser = parsers["excel"] + with open(file_path, "rb") as f: binary_content = f.read() result = parser(binary_content) else: continue - + results[filename] = result - + # 验证所有解析都成功 self.assertEqual(len(results), 5) for filename, result in results.items(): self.assertIsNotNone(result) self.assertIsInstance(result, list) self.assertGreater(len(result), 0) - + print("✅ 多解析器集成测试通过") -if __name__ == '__main__': +if __name__ == "__main__": # 运行所有测试 - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/tests/test_real_files.py b/tests/test_real_files.py index c92e99a..bd841ea 100644 --- a/tests/test_real_files.py +++ b/tests/test_real_files.py @@ -7,47 +7,32 @@ from datetime import datetime # 清除代理设置 -os.environ['HTTP_PROXY'] = '' -os.environ['HTTPS_PROXY'] = '' -os.environ['http_proxy'] = '' -os.environ['https_proxy'] = '' -os.environ['NO_PROXY'] = '*' -os.environ['no_proxy'] = '*' +os.environ["HTTP_PROXY"] = "" +os.environ["HTTPS_PROXY"] = "" +os.environ["http_proxy"] = "" +os.environ["https_proxy"] = "" +os.environ["NO_PROXY"] = "*" +os.environ["no_proxy"] = "*" # 添加 deepdoc 路径到 sys.path -sys.path.append(os.path.join(os.path.dirname(__file__), '..')) +sys.path.append(os.path.join(os.path.dirname(__file__), "..")) # 导入所有解析器 -from parser import ( - TxtParser, MarkdownParser, JsonParser, HtmlParser, - ExcelParser, PptParser, DocxParser, PdfParser -) -from parser.figure_parser import VisionFigureParser - -# 导入视觉模型 -from ..depend.simple_cv_model import create_vision_model +from deepdoc import TxtParser, MarkdownParser, JsonParser, HtmlParser, ExcelParser, PptParser, DocxParser, PdfParser class RealFileParserTest: """真实文件解析测试类""" - + def __init__(self): """初始化测试环境""" # 创建测试数据目录 self.test_data_dir = Path(__file__).parent / "test_data" - + # 创建测试结果目录 self.test_results_dir = Path(__file__).parent / "test_results" self.test_results_dir.mkdir(exist_ok=True) - - # 创建视觉模型实例 - try: - self.vision_model = create_vision_model("qwen") - print("✅ 视觉模型创建成功") - except Exception as e: - print(f"⚠️ 视觉模型创建失败: {e}") - self.vision_model = None - + # 初始化解析器 self.parsers = { "txt": TxtParser(), @@ -59,83 +44,72 @@ def __init__(self): "docx": DocxParser(), "ppt": PptParser(), } - + # 测试文件映射 self.test_files = { "pdf": "test.pdf", - "docx": "test.docx", + "docx": "test.docx", "ppt": "test.ppt", "xlsx": "test.xlsx", "jpg": "test.jpg", } - + # 测试结果存储 self.results = {} - + def log(self, message): """日志输出""" timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] {message}") - + def save_result(self, file_type, result, parse_time): """保存解析结果""" result_file = self.test_results_dir / f"{file_type}_result.txt" - - with open(result_file, 'w', encoding='utf-8') as f: + + with open(result_file, "w", encoding="utf-8") as f: f.write(f"文件类型: {file_type}\n") f.write(f"解析时间: {parse_time:.2f}秒\n") f.write(f"结果类型: {type(result)}\n") f.write(f"结果长度: {len(str(result))}字符\n") f.write("-" * 50 + "\n") f.write(str(result)) - + # 保存到结果字典 - self.results[file_type] = { - "success": True, - "parse_time": parse_time, - "result_length": len(str(result)), - "result_type": str(type(result)), - "result_file": str(result_file) - } - + self.results[file_type] = {"success": True, "parse_time": parse_time, "result_length": len(str(result)), "result_type": str(type(result)), "result_file": str(result_file)} + def save_error(self, file_type, error, parse_time): """保存错误信息""" error_file = self.test_results_dir / f"{file_type}_error.txt" - - with open(error_file, 'w', encoding='utf-8') as f: + + with open(error_file, "w", encoding="utf-8") as f: f.write(f"文件类型: {file_type}\n") f.write(f"解析时间: {parse_time:.2f}秒\n") f.write(f"错误类型: {type(error).__name__}\n") f.write(f"错误信息: {str(error)}\n") f.write("-" * 50 + "\n") f.write(traceback.format_exc()) - + # 保存到结果字典 - self.results[file_type] = { - "success": False, - "parse_time": parse_time, - "error": str(error), - "error_file": str(error_file) - } - + self.results[file_type] = {"success": False, "parse_time": parse_time, "error": str(error), "error_file": str(error_file)} + def test_pdf_parser(self): """测试PDF解析器""" self.log("开始测试PDF解析器...") - + file_path = self.test_data_dir / self.test_files["pdf"] if not file_path.exists(): self.log(f"❌ PDF文件不存在: {file_path}") return - + start_time = time.time() try: parser = self.parsers["pdf"] # 传递字符串路径而不是 Path 对象 result = parser(str(file_path)) - + parse_time = time.time() - start_time self.log(f"✅ PDF解析成功，耗时: {parse_time:.2f}秒") - + # 显示结果预览 if isinstance(result, list): self.log(f" 解析出 {len(result)} 个chunk") @@ -145,9 +119,9 @@ def test_pdf_parser(self): else: preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) self.log(f" 结果预览: {preview}") - + self.save_result("pdf", result, parse_time) - + except Exception as e: parse_time = time.time() - start_time self.log(f"❌ PDF解析失败: {e}") @@ -156,21 +130,21 @@ def test_pdf_parser(self): def test_docx_parser(self): """测试Word解析器""" self.log("开始测试Word解析器...") - + file_path = self.test_data_dir / self.test_files["docx"] if not file_path.exists(): self.log(f"❌ Word文件不存在: {file_path}") return - + start_time = time.time() try: parser = self.parsers["docx"] # 传递字符串路径而不是 Path 对象 result = parser(str(file_path)) - + parse_time = time.time() - start_time self.log(f"✅ Word解析成功，耗时: {parse_time:.2f}秒") - + # 显示结果预览 if isinstance(result, list): self.log(f" 解析出 {len(result)} 个chunk") @@ -180,9 +154,9 @@ def test_docx_parser(self): else: preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) self.log(f" 结果预览: {preview}") - + self.save_result("docx", result, parse_time) - + except Exception as e: parse_time = time.time() - start_time self.log(f"❌ Word解析失败: {e}") @@ -191,21 +165,21 @@ def test_docx_parser(self): def test_ppt_parser(self): """测试PPT解析器""" self.log("开始测试PPT解析器...") - + file_path = self.test_data_dir / self.test_files["ppt"] if not file_path.exists(): self.log(f"❌ PPT文件不存在: {file_path}") return - + start_time = time.time() try: parser = self.parsers["ppt"] # 添加必需的参数 result = parser(str(file_path), from_page=0, to_page=100000) - + parse_time = time.time() - start_time self.log(f"✅ PPT解析成功，耗时: {parse_time:.2f}秒") - + # 显示结果预览 if isinstance(result, list): self.log(f" 解析出 {len(result)} 个chunk") @@ -215,72 +189,36 @@ def test_ppt_parser(self): else: preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) self.log(f" 结果预览: {preview}") - + self.save_result("ppt", result, parse_time) - + except Exception as e: parse_time = time.time() - start_time self.log(f"❌ PPT解析失败: {e}") self.save_error("ppt", e, parse_time) - - def test_image_parser(self): - """测试图片解析器""" - self.log("开始测试图片解析器...") - - if not self.vision_model: - self.log("❌ 视觉模型不可用，跳过图片解析测试") - return - - file_path = self.test_data_dir / self.test_files["jpg"] - if not file_path.exists(): - self.log(f"❌ 图片文件不存在: {file_path}") - return - - start_time = time.time() - try: - # 读取图片文件 - with open(file_path, 'rb') as f: - image_data = f.read() - - # 使用视觉模型解析 - result = self.vision_model.describe_with_prompt(image_data) - - parse_time = time.time() - start_time - self.log(f"✅ 图片解析成功，耗时: {parse_time:.2f}秒") - - # 显示结果预览 - preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) - self.log(f" 结果预览: {preview}") - - self.save_result("image", result, parse_time) - - except Exception as e: - parse_time = time.time() - start_time - self.log(f"❌ 图片解析失败: {e}") - self.save_error("image", e, parse_time) - + def test_excel_parser_real(self): """测试Excel解析器（真实文件）""" self.log("开始测试Excel解析器（真实文件）...") - + file_path = self.test_data_dir / self.test_files["xlsx"] if not file_path.exists(): self.log(f"❌ Excel文件不存在: {file_path}") return - + start_time = time.time() try: parser = self.parsers["excel"] - + # 读取文件为二进制 - with open(file_path, 'rb') as f: + with open(file_path, "rb") as f: binary_content = f.read() - + result = parser(binary_content) - + parse_time = time.time() - start_time self.log(f"✅ Excel解析成功，耗时: {parse_time:.2f}秒") - + # 显示结果预览 if isinstance(result, list): self.log(f" 解析出 {len(result)} 行数据") @@ -290,32 +228,32 @@ def test_excel_parser_real(self): else: preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result) self.log(f" 结果预览: {preview}") - + self.save_result("excel", result, parse_time) - + except Exception as e: parse_time = time.time() - start_time self.log(f"❌ Excel解析失败: {e}") self.save_error("excel", e, parse_time) - + def generate_summary_report(self): """生成测试总结报告""" self.log("生成测试总结报告...") - + summary_file = self.test_results_dir / "summary.txt" - - with open(summary_file, 'w', encoding='utf-8') as f: + + with open(summary_file, "w", encoding="utf-8") as f: f.write("DeepDoc 真实文件解析测试报告\n") f.write("=" * 50 + "\n") f.write(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"测试文件数量: {len(self.test_files)}\n\n") - + success_count = sum(1 for r in self.results.values() if r.get("success", False)) f.write(f"成功解析: {success_count}/{len(self.results)}\n\n") - + f.write("详细结果:\n") f.write("-" * 30 + "\n") - + for file_type, result in self.results.items(): f.write(f"\n{file_type.upper()}:\n") if result.get("success", False): @@ -329,35 +267,34 @@ def generate_summary_report(self): f.write(f" 解析时间: {result.get('parse_time', 0):.2f}秒\n") f.write(f" 错误信息: {result.get('error', 'Unknown')}\n") f.write(f" 错误文件: {result.get('error_file', 'N/A')}\n") - + # 同时保存JSON格式的详细结果 json_file = self.test_results_dir / "results.json" - with open(json_file, 'w', encoding='utf-8') as f: + with open(json_file, "w", encoding="utf-8") as f: json.dump(self.results, f, indent=2, ensure_ascii=False) - + self.log(f"✅ 测试报告已保存到: {summary_file}") self.log(f"✅ 详细结果已保存到: {json_file}") - + def run_all_tests(self): """运行所有测试""" self.log("开始DeepDoc真实文件解析测试") self.log("=" * 50) - + # 运行各个解析器测试 self.test_pdf_parser() self.test_docx_parser() self.test_ppt_parser() - self.test_image_parser() self.test_excel_parser_real() - + # 生成总结报告 self.generate_summary_report() - + self.log("=" * 50) self.log("测试完成！") -if __name__ == '__main__': +if __name__ == "__main__": # 运行测试 tester = RealFileParserTest() - tester.run_all_tests() \ No newline at end of file + tester.run_all_tests() diff --git a/tests/test_simple_cv_model.py b/tests/test_simple_cv_model.py deleted file mode 100644 index 09541bd..0000000 --- a/tests/test_simple_cv_model.py +++ /dev/null @@ -1,384 +0,0 @@ -import os -import sys -import unittest -from unittest.mock import patch, MagicMock -from io import BytesIO -from PIL import Image, ImageDraw, ImageFont -import base64 - -# 清除代理设置 -os.environ['HTTP_PROXY'] = '' -os.environ['HTTPS_PROXY'] = '' -os.environ['http_proxy'] = '' -os.environ['https_proxy'] = '' -os.environ['NO_PROXY'] = '*' -os.environ['no_proxy'] = '*' - -# 添加 deepdoc 路径到 sys.path -sys.path.append(os.path.join(os.path.dirname(__file__), '..')) - -from ..depend.simple_cv_model import QWenCV, VisionModelFactory, create_vision_model - - -class TestQWenCV(unittest.TestCase): - """测试 QWenCV 模型类""" - - def setUp(self): - """测试前的准备工作""" - # 测试用的API密钥（实际使用时需要替换为真实的） - self.test_api_key = "sk-ef14e92efd674970b810f15fa095e4c4" - self.test_model_name = "qwen-vl-max" - - # 创建测试图片 - self.test_image = self.create_test_image() - - # 创建 QWenCV 实例 - self.qwen_cv = QWenCV( - key=self.test_api_key, - model_name=self.test_model_name, - lang="Chinese" - ) - - def create_test_image(self): - """创建测试用的图片""" - # 创建一个简单的测试图片 - img = Image.new('RGB', (100, 100), color='white') - img_bytes = BytesIO() - img.save(img_bytes, format='JPEG') - img_bytes.seek(0) - return img_bytes - - def test_init(self): - """测试 QWenCV 初始化""" - self.assertEqual(self.qwen_cv.key, self.test_api_key) - self.assertEqual(self.qwen_cv.model_name, self.test_model_name) - self.assertEqual(self.qwen_cv.lang, "Chinese") - - def test_image2base64(self): - """测试图片转base64功能""" - # 测试 BytesIO 格式 - b64 = self.qwen_cv.image2base64(self.test_image) - self.assertIsInstance(b64, str) - self.assertTrue(len(b64) > 0) - - # 测试 bytes 格式 - img_bytes = self.test_image.getvalue() - b64_bytes = self.qwen_cv.image2base64(img_bytes) - self.assertEqual(b64, b64_bytes) - - # 测试 PIL Image 格式 - img = Image.open(self.test_image) - b64_pil = self.qwen_cv.image2base64(img) - self.assertEqual(b64, b64_pil) - - def test_prompt_generation(self): - """测试提示词生成""" - # 测试默认提示词 - prompt = self.qwen_cv.prompt(self.test_image.getvalue()) - self.assertIsInstance(prompt, list) - self.assertEqual(len(prompt), 1) - self.assertEqual(prompt[0]["role"], "user") - self.assertIn("content", prompt[0]) - - # 测试自定义提示词 - custom_prompt = self.qwen_cv.vision_llm_prompt( - self.test_image.getvalue(), - "请描述这张图片" - ) - self.assertIsInstance(custom_prompt, list) - self.assertIn("请描述这张图片", str(custom_prompt)) - - def test_temp_image_saving(self): - """测试临时图片保存功能""" - img_bytes = self.test_image.getvalue() - temp_path = self.qwen_cv._save_temp_image(img_bytes) - - # 检查文件是否存在 - self.assertTrue(os.path.exists(temp_path)) - self.assertTrue(temp_path.endswith('.jpg')) - - # 清理临时文件 - if os.path.exists(temp_path): - os.remove(temp_path) - - @patch('dashscope.MultiModalConversation.call') - def test_describe_success(self, mock_call): - """测试成功的图片描述""" - # 模拟成功的API响应 - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.output.choices = [{ - "message": { - "content": [{"text": "这是一张白色背景的图片"}] - } - }] - mock_response.usage.output_tokens = 10 - mock_call.return_value = mock_response - - # 测试描述功能 - result, tokens = self.qwen_cv.describe(self.test_image) - - self.assertEqual(result, "这是一张白色背景的图片") - self.assertEqual(tokens, 10) - mock_call.assert_called_once() - - @patch('dashscope.MultiModalConversation.call') - def test_describe_with_prompt_success(self, mock_call): - """测试带自定义提示词的图片描述""" - # 模拟成功的API响应 - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.output.choices = [{ - "message": { - "content": [{"text": "自定义描述结果"}] - } - }] - mock_response.usage.output_tokens = 15 - mock_call.return_value = mock_response - - # 测试带提示词的描述功能 - result, tokens = self.qwen_cv.describe_with_prompt( - self.test_image, - "请详细描述这张图片的内容" - ) - - self.assertEqual(result, "自定义描述结果") - self.assertEqual(tokens, 15) - mock_call.assert_called_once() - - @patch('dashscope.MultiModalConversation.call') - def test_describe_failure(self, mock_call): - """测试API调用失败的情况""" - # 模拟失败的API响应 - mock_response = MagicMock() - mock_response.status_code = 400 - mock_response.message = "API调用失败" - mock_call.return_value = mock_response - - # 测试失败情况 - result, tokens = self.qwen_cv.describe(self.test_image) - - self.assertEqual(result, "API调用失败") - self.assertEqual(tokens, 0) - - def test_factory_creation(self): - """测试通过工厂创建 QWenCV 实例""" - config = { - "provider": "qwen", - "model_name": "qwen-vl-max", - "api_key": self.test_api_key, - "lang": "Chinese" - } - - model = VisionModelFactory.create_model(config) - self.assertIsInstance(model, QWenCV) - self.assertEqual(model.key, self.test_api_key) - self.assertEqual(model.model_name, "qwen-vl-max") - - def test_create_vision_model_function(self): - """测试便捷函数""" - config = { - "provider": "qwen", - "model_name": "qwen-vl-max", - "api_key": self.test_api_key - } - - model = create_vision_model(config) - self.assertIsInstance(model, QWenCV) - - def test_language_settings(self): - """测试不同语言设置""" - # 测试中文设置 - qwen_chinese = QWenCV(self.test_api_key, self.test_model_name, lang="Chinese") - self.assertEqual(qwen_chinese.lang, "Chinese") - - # 测试英文设置 - qwen_english = QWenCV(self.test_api_key, self.test_model_name, lang="English") - self.assertEqual(qwen_english.lang, "English") - - def test_invalid_provider(self): - """测试无效的提供商""" - config = { - "provider": "invalid_provider", - "api_key": self.test_api_key - } - - with self.assertRaises(ValueError): - VisionModelFactory.create_model(config) - - def tearDown(self): - """测试后的清理工作""" - # 清理可能创建的临时文件 - tmp_dir = os.path.join(os.getcwd(), "tmp") - if os.path.exists(tmp_dir): - for file in os.listdir(tmp_dir): - if file.endswith('.jpg'): - os.remove(os.path.join(tmp_dir, file)) - - -class TestQWenCVIntegration(unittest.TestCase): - """集成测试：测试真实的API调用（需要真实的API密钥）""" - - def setUp(self): - """设置真实的API密钥进行测试""" - # 直接使用你的API密钥 - self.real_api_key = "sk-ef14e92efd674970b810f15fa095e4c4" - - self.qwen_cv = QWenCV( - key=self.real_api_key, - model_name="qwen-vl-max", - lang="Chinese" - ) - - def create_test_image(self): - """创建更真实的测试图片""" - # 创建一个更复杂的测试图片 - img = Image.new('RGB', (400, 300), color='white') - - # 添加一些内容到图片中 - draw = ImageDraw.Draw(img) - - # 添加文字 - try: - # 尝试使用系统字体 - font = ImageFont.truetype("/System/Library/Fonts/Arial.ttf", 20) - except: - # 如果找不到字体，使用默认字体 - font = ImageFont.load_default() - - draw.text((50, 50), "Test Document", fill='black', font=font) - draw.text((50, 100), "This is a test image for OCR", fill='black', font=font) - draw.text((50, 150), "包含中文测试", fill='black', font=font) - - # 添加一些图形 - draw.rectangle([(50, 200), (200, 250)], outline='black', width=2) - draw.text((60, 210), "Test Box", fill='black', font=font) - - img_bytes = BytesIO() - img.save(img_bytes, format='JPEG', quality=95) - img_bytes.seek(0) - return img_bytes - - def load_real_test_image(self, image_path): - """加载真实的测试图片""" - if os.path.exists(image_path): - with open(image_path, 'rb') as f: - image_data = f.read() - return BytesIO(image_data) - else: - print(f"警告: 找不到测试图片 {image_path}，使用生成的测试图片") - return self.create_test_image() - - def test_real_api_call_with_generated_image(self): - """测试真实的API调用 - 使用生成的测试图片""" - test_image = self.create_test_image() - - try: - result, tokens = self.qwen_cv.describe(test_image) - - # 检查是否是错误响应 - if "error" in result.lower() or "input format error" in result.lower(): - self.fail(f"API返回错误: {result}") - - # 验证返回结果 - self.assertIsInstance(result, str) - self.assertGreater(len(result), 0) - self.assertIsInstance(tokens, int) - self.assertGreaterEqual(tokens, 0) - - print(f"✅ 生成图片API调用成功") - print(f"�� 结果: {result[:200]}...") - print(f"🔢 Token使用量: {tokens}") - - except Exception as e: - self.fail(f"API调用失败: {str(e)}") - - def test_real_api_call_with_real_images(self): - """测试真实的API调用 - 使用真实图片文件""" - # 定义可能的测试图片路径 - test_image_paths = [ - "test_images/document.png", - "test_images/chart.jpg", - "test_images/photo.png", - "test_images/sample.pdf", # 如果有PDF转图片的话 - "../test_images/document.png", # 上级目录 - "../../test_images/document.png" # 上上级目录 - ] - - found_images = [] - for image_path in test_image_paths: - if os.path.exists(image_path): - found_images.append(image_path) - - if not found_images: - print("⚠️ 没有找到真实测试图片，跳过此测试") - self.skipTest("没有找到真实测试图片文件") - - # 测试每个找到的图片 - for image_path in found_images: - print(f"\n🖼️ 测试图片: {image_path}") - test_image = self.load_real_test_image(image_path) - - try: - result, tokens = self.qwen_cv.describe(test_image) - - # 检查是否是错误响应 - if "error" in result.lower() or "input format error" in result.lower(): - print(f"❌ API返回错误: {result}") - continue - - # 验证返回结果 - self.assertIsInstance(result, str) - self.assertGreater(len(result), 0) - self.assertIsInstance(tokens, int) - self.assertGreaterEqual(tokens, 0) - - print(f"✅ 真实图片API调用成功") - print(f"�� 结果: {result[:200]}...") - print(f"🔢 Token使用量: {tokens}") - - except Exception as e: - print(f"❌ 图片 {image_path} 测试失败: {str(e)}") - # 不直接失败，继续测试其他图片 - - def test_real_api_call_with_custom_prompt(self): - """测试带自定义提示词的真实API调用""" - test_image = self.create_test_image() - - try: - custom_prompt = "请详细描述这张图片中的文字内容和布局结构" - result, tokens = self.qwen_cv.describe_with_prompt(test_image, custom_prompt) - - # 检查是否是错误响应 - if "error" in result.lower() or "input format error" in result.lower(): - self.fail(f"API返回错误: {result}") - - # 验证返回结果 - self.assertIsInstance(result, str) - self.assertGreater(len(result), 0) - self.assertIsInstance(tokens, int) - self.assertGreaterEqual(tokens, 0) - - print(f"✅ 自定义提示词API调用成功") - print(f"�� 结果: {result[:200]}...") - print(f"🔢 Token使用量: {tokens}") - - except Exception as e: - self.fail(f"API调用失败: {str(e)}") - - def tearDown(self): - """测试后的清理工作""" - # 清理可能创建的临时文件 - tmp_dir = os.path.join(os.getcwd(), "tmp") - if os.path.exists(tmp_dir): - for file in os.listdir(tmp_dir): - if file.endswith('.jpg'): - try: - os.remove(os.path.join(tmp_dir, file)) - except: - pass - - -if __name__ == "__main__": - # 运行测试 - unittest.main(verbosity=2) \ No newline at end of file