20389568923

Committed 20 Dec 2025 05:06AM UTC coverage: 87.375% (+5.6%) from 81.75%

Build # 20389568923

Build Type

Pull #7

github

Committed by

speedyk-005

Commit Message

refactor: optimize core chunking and validation logic

- Simplified document chunker upfront validation to on-demand processing
- Optimized code chunker error message formatting to use single limit formatting call
- Replaced regex with efficient string operations for decorator/function detection
- Enhanced visualizer with async file operations and proper dependencies
- Modernized JavaScript patterns: replaced deprecated functions and optional chaining
- Improved frontend accessibility with proper form label associations
- Fixed CSS validation errors and removed duplicate selectors
- Enhanced type hints and error handling throughout codebase

These changes significantly reduce cognitive complexity, improve performance, and enhance maintainability while preserving all existing functionality.

Pull Request Pull Request #7: Merge develop branch to main

Coverage Stats

478 of 552 new or added lines in 17 files covered. (86.59%)

1308 of 1497 relevant lines covered (87.37%)

4.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.08

/src/chunklet/visualizer/visualizer.py

import os
import json
import tempfile
import traceback
import mimetypes
from typing import Callable
import aiofiles

try:
    import uvicorn
    from charset_normalizer import detect
    from fastapi import FastAPI, UploadFile, File, Form, HTTPException
    from fastapi.responses import HTMLResponse
    from fastapi.staticfiles import StaticFiles
except ImportError:
    # Lambda placeholders prevent "None is not callable" errors when imports fail
    # This allows the module to be imported without dependencies, with proper error handling later
    uvicorn = None
    detect = None
    FastAPI = None
    UploadFile = None
    File = lambda x: x  # noqa: E731
    Form = lambda x: x  # noqa: E731
    HTTPException = None
    HTMLResponse = lambda x: x  # noqa: E731
    StaticFiles = None

from chunklet.document_chunker import DocumentChunker
from chunklet.code_chunker import CodeChunker
from chunklet.common.validation import validate_input


class Visualizer:
    """A FastAPI-based web interface for visualizing document and code chunks.

    This server allows users to upload text or code files, processes them with
    Chunklet's `DocumentChunker` or `CodeChunker`, and returns the chunked
    data along with statistics. A minimal frontend interface is served at the
    root endpoint.

    Attributes:
        host (str): Host IP to bind the FastAPI server.
        port (int): Port number to run the server on.
        document_chunker (DocumentChunker): Chunklet document chunker instance.
        code_chunker (CodeChunker): Chunklet code chunker instance.
        app (FastAPI): FastAPI application instance.
    """

    @validate_input
    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 8000,
        token_counter: Callable[[str], int] | None = None,
    ):
        """Initializes the Visualizer server and configures chunkers.

        Args:
            host (str): Host IP to run the server. Defaults to "127.0.0.1".
            port (int): Port number to run the server. Defaults to 8000.
            token_counter (Optional[Callable[[str], int]]): Function to count tokens
                in text/code. Required for chunkers if used with `max_tokens`.
        """
        if FastAPI is None:
            raise ImportError(
                "The 'fastapi' library is not installed. "
                "Please install it with 'pip install fastapi>=0.115.12' or install the visualization extras "
                "with 'pip install 'chunklet-py[visualization]''"
            )

        self.host = host
        self.port = port
        self._token_counter = token_counter

        self.app = FastAPI()

        # Initialize chunkers
        self.document_chunker = DocumentChunker(token_counter=token_counter)
        self.code_chunker = CodeChunker(token_counter=token_counter)

        base_dir = os.path.dirname(os.path.abspath(__file__))
        static_dir = os.path.join(base_dir, "static")

        self.app.mount("/static", StaticFiles(directory=static_dir), name="static")

        # API endpoints
        self.app.get("/api/token_counter_status")(self._get_token_counter_status)
        self.app.get("/health")(self._get_health_check)
        self.app.get("/")(self._get_index)
        self.app.post("/api/chunk")(self._chunk_file)

    # Instance endpoint methods
    def _get_token_counter_status(self):
        return {"token_counter_available": self.token_counter is not None}

    def _get_health_check(self):
        """Health check endpoint for testing."""
        return {"status": "healthy"}

    async def _get_index(self):
        """Serves the main HTML interface for the visualizer.

        Returns:
            HTMLResponse: The content of index.html if exists, else a default heading.
        """
        base_dir = os.path.dirname(os.path.abspath(__file__))
        static_dir = os.path.join(base_dir, "static")
        index_path = os.path.join(static_dir, "index.html")
        if os.path.exists(index_path):
            async with aiofiles.open(index_path, "r", encoding="utf-8") as f:
                content = await f.read()
                return HTMLResponse(content=content)
        return HTMLResponse(content="<h1>Text Chunk Visualizer</h1>")

    @validate_input
    async def _chunk_file(
        self,
        file: UploadFile = File(...),
        mode: str = Form("document"),
        params: str = Form("{}"),
    ) -> dict:
        """Processes an uploaded file and returns chunked output.

        Args:
            self: The Visualizer instance.
            file (UploadFile): File uploaded by the client.
            mode (str): Determines which chunker to use ("document" or "code").
            params (str): JSON string containing chunking parameters.

        Returns:
            dict: Contains original text, chunked content, and statistics.

        Raises:
            HTTPException: If chunking fails.
        """
        # Parse params JSON and filter out None values
        try:
            chunker_params = json.loads(params)
            chunker_params = {k: v for k, v in chunker_params.items() if v is not None}
        except (json.JSONDecodeError, TypeError):
            raise HTTPException(400, f"Invalid chunking parameters JSON: {params}")

        print(f"Processing: {file.filename} in {mode} mode")

        # Use Python mimetypes instead of browser content_type
        mimetype, _ = mimetypes.guess_type(file.filename or "")
        if not mimetype or not mimetype.startswith("text/"):
            raise HTTPException(400, "Only text files are supported.")

        if detect is None:
            raise HTTPException(
                400,
                "charset-normalizer library is not available. Please install visualization dependencies."
                "with 'pip install 'chunklet-py[visualization]''",
            )

        # Saved as txt file since they are all plaintext anyway
        with tempfile.NamedTemporaryFile(mode="wb", suffix=".txt", delete=False) as tmp:
            content = await file.read()
            tmp.write(content)
            tmp_path = tmp.name

        encoding = detect(content).get("encoding", "utf-8")
        text = content.decode(encoding, errors="ignore")
        chunker = self.code_chunker if mode == "code" else self.document_chunker

        try:
            chunks = [
                dict(chunk) for chunk in chunker.chunk(tmp_path, **chunker_params)
            ]

            return {
                "text": text,
                "chunks": chunks,
                "stats": {
                    "text_length": len(text),
                    "chunk_count": len(chunks),
                    "mode": mode,
                },
            }

        except Exception as e:
            traceback.print_exc()
            raise HTTPException(
                500,
                f"Chunking failed. Please check the server terminal for specific error details. ({str(e)})",
            )
        finally:
            # Always cleanup temp file
            try:
                os.unlink(tmp_path)
            except OSError:
                pass

    @property
    def token_counter(self):
        """Get the current token counter function."""
        return self._token_counter

    @token_counter.setter
    @validate_input
    def token_counter(self, value):
        """Set the token counter and update both chunkers."""
        self._token_counter = value
        self.document_chunker.token_counter = value
        self.code_chunker.token_counter = value

    def serve(self):
        """Starts the FastAPI server and prints the server URL."""
        if uvicorn is None:
            raise ImportError(
                "The 'uvicorn' library is not installed. "
                "Please install it with 'pip install uvicorn>=0.34.0' or install the visualization extras "
                "with 'pip install 'chunklet-py[visualization]''"
            )

        print(" =" * 20)
        print("\nTEXT CHUNK VISUALIZER")
        print("= " * 20)
        print(f"URL: http://{self.host}:{self.port}")

        uvicorn.run(self.app, host=self.host, port=self.port)


if __name__ == "__main__":  # pragma: no cover
    visualizer = Visualizer()
    visualizer.serve()

1	import os	5✔
2	import json	5✔
3	import tempfile	5✔
4	import traceback	5✔
5	import mimetypes	5✔
6	from typing import Callable	5✔
7	import aiofiles	5✔
8
9	try:	5✔
10	import uvicorn	5✔
11	from charset_normalizer import detect	5✔
12	from fastapi import FastAPI, UploadFile, File, Form, HTTPException	5✔
13	from fastapi.responses import HTMLResponse	5✔
14	from fastapi.staticfiles import StaticFiles	5✔
NEW 15	except ImportError:	×
16	# Lambda placeholders prevent "None is not callable" errors when imports fail
17	# This allows the module to be imported without dependencies, with proper error handling later
NEW 18	uvicorn = None	×
NEW 19	detect = None	×
NEW 20	FastAPI = None	×
NEW 21	UploadFile = None	×
NEW 22	File = lambda x: x # noqa: E731	×
NEW 23	Form = lambda x: x # noqa: E731	×
NEW 24	HTTPException = None	×
NEW 25	HTMLResponse = lambda x: x # noqa: E731	×
NEW 26	StaticFiles = None	×
27
28	from chunklet.document_chunker import DocumentChunker	5✔
29	from chunklet.code_chunker import CodeChunker	5✔
30	from chunklet.common.validation import validate_input	5✔
31
32
33	class Visualizer:	5✔
34	"""A FastAPI-based web interface for visualizing document and code chunks.
35
36	This server allows users to upload text or code files, processes them with
37	Chunklet's `DocumentChunker` or `CodeChunker`, and returns the chunked
38	data along with statistics. A minimal frontend interface is served at the
39	root endpoint.
40
41	Attributes:
42	host (str): Host IP to bind the FastAPI server.
43	port (int): Port number to run the server on.
44	document_chunker (DocumentChunker): Chunklet document chunker instance.
45	code_chunker (CodeChunker): Chunklet code chunker instance.
46	app (FastAPI): FastAPI application instance.
47	"""
48
49	@validate_input	5✔
50	def __init__(	5✔
51	self,
52	host: str = "127.0.0.1",
53	port: int = 8000,
54	token_counter: Callable[[str], int] \| None = None,
55	):
56	"""Initializes the Visualizer server and configures chunkers.
57
58	Args:
59	host (str): Host IP to run the server. Defaults to "127.0.0.1".
60	port (int): Port number to run the server. Defaults to 8000.
61	token_counter (Optional[Callable[[str], int]]): Function to count tokens
62	in text/code. Required for chunkers if used with `max_tokens`.
63	"""
64	if FastAPI is None:	5✔
NEW 65	raise ImportError(	×
66	"The 'fastapi' library is not installed. "
67	"Please install it with 'pip install fastapi>=0.115.12' or install the visualization extras "
68	"with 'pip install 'chunklet-py[visualization]''"
69	)
70
71	self.host = host	5✔
72	self.port = port	5✔
73	self._token_counter = token_counter	5✔
74
75	self.app = FastAPI()	5✔
76
77	# Initialize chunkers
78	self.document_chunker = DocumentChunker(token_counter=token_counter)	5✔
79	self.code_chunker = CodeChunker(token_counter=token_counter)	5✔
80
81	base_dir = os.path.dirname(os.path.abspath(__file__))	5✔
82	static_dir = os.path.join(base_dir, "static")	5✔
83
84	self.app.mount("/static", StaticFiles(directory=static_dir), name="static")	5✔
85
86	# API endpoints
87	self.app.get("/api/token_counter_status")(self._get_token_counter_status)	5✔
88	self.app.get("/health")(self._get_health_check)	5✔
89	self.app.get("/")(self._get_index)	5✔
90	self.app.post("/api/chunk")(self._chunk_file)	5✔
91
92	# Instance endpoint methods
93	def _get_token_counter_status(self):	5✔
94	return {"token_counter_available": self.token_counter is not None}	5✔
95
96	def _get_health_check(self):	5✔
97	"""Health check endpoint for testing."""
98	return {"status": "healthy"}	5✔
99
100	async def _get_index(self):	5✔
101	"""Serves the main HTML interface for the visualizer.
102
103	Returns:
104	HTMLResponse: The content of index.html if exists, else a default heading.
105	"""
NEW 106	base_dir = os.path.dirname(os.path.abspath(__file__))	×
NEW 107	static_dir = os.path.join(base_dir, "static")	×
NEW 108	index_path = os.path.join(static_dir, "index.html")	×
NEW 109	if os.path.exists(index_path):	×
NEW 110	async with aiofiles.open(index_path, "r", encoding="utf-8") as f:	×
NEW 111	content = await f.read()	×
NEW 112	return HTMLResponse(content=content)	×
NEW 113	return HTMLResponse(content="<h1>Text Chunk Visualizer</h1>")	×
114
115	@validate_input	5✔
116	async def _chunk_file(	5✔
117	self,
118	file: UploadFile = File(...),
119	mode: str = Form("document"),
120	params: str = Form("{}"),
121	) -> dict:
122	"""Processes an uploaded file and returns chunked output.
123
124	Args:
125	self: The Visualizer instance.
126	file (UploadFile): File uploaded by the client.
127	mode (str): Determines which chunker to use ("document" or "code").
128	params (str): JSON string containing chunking parameters.
129
130	Returns:
131	dict: Contains original text, chunked content, and statistics.
132
133	Raises:
134	HTTPException: If chunking fails.
135	"""
136	# Parse params JSON and filter out None values
137	try:	5✔
138	chunker_params = json.loads(params)	5✔
139	chunker_params = {k: v for k, v in chunker_params.items() if v is not None}	5✔
NEW 140	except (json.JSONDecodeError, TypeError):	×
NEW 141	raise HTTPException(400, f"Invalid chunking parameters JSON: {params}")	×
142
143	print(f"Processing: {file.filename} in {mode} mode")	5✔
144
145	# Use Python mimetypes instead of browser content_type
146	mimetype, _ = mimetypes.guess_type(file.filename or "")	5✔
147	if not mimetype or not mimetype.startswith("text/"):	5✔
148	raise HTTPException(400, "Only text files are supported.")	5✔
149
150	if detect is None:	5✔
NEW 151	raise HTTPException(	×
152	400,
153	"charset-normalizer library is not available. Please install visualization dependencies."
154	"with 'pip install 'chunklet-py[visualization]''",
155	)
156
157	# Saved as txt file since they are all plaintext anyway
158	with tempfile.NamedTemporaryFile(mode="wb", suffix=".txt", delete=False) as tmp:	5✔
159	content = await file.read()	5✔
160	tmp.write(content)	5✔
161	tmp_path = tmp.name	5✔
162
163	encoding = detect(content).get("encoding", "utf-8")	5✔
164	text = content.decode(encoding, errors="ignore")	5✔
165	chunker = self.code_chunker if mode == "code" else self.document_chunker	5✔
166
167	try:	5✔
168	chunks = [	5✔
169	dict(chunk) for chunk in chunker.chunk(tmp_path, **chunker_params)
170	]
171
172	return {	5✔
173	"text": text,
174	"chunks": chunks,
175	"stats": {
176	"text_length": len(text),
177	"chunk_count": len(chunks),
178	"mode": mode,
179	},
180	}
181
NEW 182	except Exception as e:	×
NEW 183	traceback.print_exc()	×
NEW 184	raise HTTPException(	×
185	500,
186	f"Chunking failed. Please check the server terminal for specific error details. ({str(e)})",
187	)
188	finally:
189	# Always cleanup temp file
190	try:	5✔
191	os.unlink(tmp_path)	5✔
NEW 192	except OSError:	×
NEW 193	pass	×
194
195	@property	5✔
196	def token_counter(self):	5✔
197	"""Get the current token counter function."""
198	return self._token_counter	5✔
199
200	@token_counter.setter	5✔
201	@validate_input	5✔
202	def token_counter(self, value):	5✔
203	"""Set the token counter and update both chunkers."""
204	self._token_counter = value	5✔
205	self.document_chunker.token_counter = value	5✔
206	self.code_chunker.token_counter = value	5✔
207
208	def serve(self):	5✔
209	"""Starts the FastAPI server and prints the server URL."""
210	if uvicorn is None:	5✔
NEW 211	raise ImportError(	×
212	"The 'uvicorn' library is not installed. "
213	"Please install it with 'pip install uvicorn>=0.34.0' or install the visualization extras "
214	"with 'pip install 'chunklet-py[visualization]''"
215	)
216
217	print(" =" * 20)	5✔
218	print("\nTEXT CHUNK VISUALIZER")	5✔
219	print("= " * 20)	5✔
220	print(f"URL: http://{self.host}:{self.port}")	5✔
221
222	uvicorn.run(self.app, host=self.host, port=self.port)	5✔
223
224
225	if __name__ == "__main__": # pragma: no cover
226	visualizer = Visualizer()
227	visualizer.serve()

speedyk-005 / chunklet-py / 20389568923

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous