Coverage for rbt_mcp_server/document

1"""

2DocumentService - Core CRUD logic for document operations.

4@REQ: REQ-rbt-mcp-tool

5@BP: BP-rbt-mcp-tool

6@TASK: TASK-003-DocumentService

8Provides document loading, saving, reading, and updating operations.

9Integrates PathResolver, DocumentCache, and MarkdownConverter.

10"""

12import os

13import copy

14from pathlib import Path

15from typing import Dict, Any, Optional

16import sys

18# Add converter module to path

19converter_path = os.path.join(os.path.dirname(__file__), '..', 'converter')

20if converter_path not in sys.path:

21 sys.path.insert(0, converter_path)

23from converter import MarkdownConverter

24from .path_resolver import PathResolver

25from .cache import DocumentCache

26from .models import PathInfo

27from .errors import ToolError

30class DocumentService:

31 """

32 Core service for document CRUD operations.

34 Features:

35 - Load documents (with cache support)

36 - Save documents (as .new.md)

37 - Get outline (lightweight structure without blocks)

38 - Read/update sections

39 - Clear cache

41 @REQ: REQ-rbt-mcp-tool

42 @BP: BP-rbt-mcp-tool

43 @TASK: TASK-003-DocumentService

44 """

46 def __init__(self, root_dir: str):

47 """

48 Initialize DocumentService.

50 Args:

51 root_dir: Root directory for all documents

53 @REQ: REQ-rbt-mcp-tool

54 @BP: BP-rbt-mcp-tool

55 @TASK: TASK-003-DocumentService

56 """

57 self.root_dir = root_dir

58 self.path_resolver = PathResolver(root_dir)

59 self.cache = DocumentCache(max_size=10, ttl_seconds=300)

60 self.converter = MarkdownConverter()

62 # Start cache cleanup thread

63 self.cache.start()

65 def load_document(self, path_info: PathInfo) -> Dict[str, Any]:

66 """

67 Load document JSON from file or cache.

69 Priority: Cache -> File system

70 Automatically caches loaded documents.

72 Args:

73 path_info: PathInfo with resolved file path

75 Returns:

76 Document JSON data (deep copy)

78 Raises:

79 FileNotFoundError: If file doesn't exist

80 Exception: If conversion fails

82 @REQ: REQ-rbt-mcp-tool

83 @BP: BP-rbt-mcp-tool

84 @TASK: TASK-003-DocumentService

85 """

86 file_path = path_info.file_path

88 # Try to get from cache first

89 cached_data = self.cache.get(file_path)

90 if cached_data is not None:

91 # Return deep copy to prevent modification of cached data

92 return copy.deepcopy(cached_data)

94 # Cache miss - load from file

95 if not os.path.exists(file_path):

96 raise FileNotFoundError(f"Document not found: {file_path}")

98 # Read file content

99 with open(file_path, 'r', encoding='utf-8') as f:

100 md_content = f.read()

101

102 # Convert to JSON

103 json_data = self.converter.to_json(md_content)

104

105 # Store in cache

106 self.cache.put(file_path, json_data)

107

108 # Return deep copy

109 return copy.deepcopy(json_data)

110

111 def save_document(self, path_info: PathInfo, json_data: Dict[str, Any]) -> None:

112 """

113 Save document as .new.md file.

114

115 Atomic operation: Write to .tmp first, then rename to .new.md

116 Updates cache after successful save.

117

118 Args:

119 path_info: PathInfo with file path information

120 json_data: Document JSON data to save

121

122 Raises:

123 Exception: If conversion or file write fails

124

125 @REQ: REQ-rbt-mcp-tool

126 @BP: BP-rbt-mcp-tool

127 @TASK: TASK-003-DocumentService

128 """

129 # Convert JSON to Markdown

130 md_content = self.converter.to_md(json_data)

131

132 # Determine target path (.new.md)

133 original_path = path_info.file_path

134 if original_path.endswith(".new.md"):

135 target_path = original_path

136 elif original_path.endswith(".md"):

137 target_path = original_path.replace(".md", ".new.md")

138 else:

139 target_path = original_path + ".new.md"

140

141 # Write to temporary file first (atomic operation)

142 tmp_path = target_path + ".tmp"

143 try:

144 with open(tmp_path, 'w', encoding='utf-8') as f:

145 f.write(md_content)

146

147 # Rename to final path (atomic on POSIX systems)

148 os.rename(tmp_path, target_path)

149

150 finally:

151 # Clean up tmp file if it still exists

152 if os.path.exists(tmp_path):

153 os.remove(tmp_path)

154

155 # Update cache

156 self.cache.put(target_path, json_data)

157

158 def get_outline(self, path_info: PathInfo) -> Dict[str, Any]:

159 """

160 Get document outline (structure without blocks).

161

162 Returns lightweight JSON with:

163 - metadata

164 - info

165 - title

166 - sections tree (without blocks)

167

168 Args:

169 path_info: PathInfo with file path

170

171 Returns:

172 Outline JSON (deep copy)

173

174 @REQ: REQ-rbt-mcp-tool

175 @BP: BP-rbt-mcp-tool

176 @TASK: TASK-003-DocumentService

177 """

178 # Load full document

179 json_data = self.load_document(path_info)

180

181 # Deep copy to avoid modifying cache

182 outline = copy.deepcopy(json_data)

183

184 # Remove all blocks recursively

185 def remove_blocks(sections):

186 """Recursively remove blocks from sections."""

187 for section in sections:

188 if "blocks" in section:

189 del section["blocks"]

190 if "sections" in section:

191 remove_blocks(section["sections"])

192

193 if "sections" in outline:

194 remove_blocks(outline["sections"])

195

196 return outline

197

198 def read_section(self, path_info: PathInfo, section_id: str) -> Dict[str, Any]:

199 """

200 Read specific section by ID.

201

202 Args:

203 path_info: PathInfo with file path

204 section_id: Section ID to read

205

206 Returns:

207 Section data with blocks (deep copy)

208

209 Raises:

210 ToolError: If section not found

211

212 @REQ: REQ-rbt-mcp-tool

213 @BP: BP-rbt-mcp-tool

214 @TASK: TASK-003-DocumentService

215 """

216 # Load document

217 json_data = self.load_document(path_info)

218

219 # Search for section

220 def find_section(sections, target_id):

221 """Recursively search for section by ID."""

222 for section in sections:

223 if section.get("id") == target_id:

224 return section

225 if "sections" in section:

226 result = find_section(section["sections"], target_id)

227 if result:

228 return result

229 return None

230

231 section = None

232 if "sections" in json_data:

233 section = find_section(json_data["sections"], section_id)

234

235 if section is None:

236 raise ToolError(

237 "SECTION_NOT_FOUND",

238 f"Section '{section_id}' not found in document"

239 )

240

241 # Return deep copy

242 return copy.deepcopy(section)

243

244 def update_section_summary(

245 self,

246 path_info: PathInfo,

247 section_id: str,

248 new_summary: str

249 ) -> None:

250 """

251 Update section summary.

252

253 Modifies the section's summary field and saves as .new.md

254

255 Args:

256 path_info: PathInfo with file path

257 section_id: Section ID to update

258 new_summary: New summary text

259

260 Raises:

261 ToolError: If section not found

262

263 @REQ: REQ-rbt-mcp-tool

264 @BP: BP-rbt-mcp-tool

265 @TASK: TASK-003-DocumentService

266 """

267 # Load document

268 json_data = self.load_document(path_info)

269

270 # Search and update section

271 def update_section(sections, target_id, summary):

272 """Recursively search and update section summary."""

273 for section in sections:

274 if section.get("id") == target_id:

275 section["summary"] = summary

276 return True

277 if "sections" in section:

278 if update_section(section["sections"], target_id, summary):

279 return True

280 return False

281

282 updated = False

283 if "sections" in json_data:

284 updated = update_section(json_data["sections"], section_id, new_summary)

285

286 if not updated:

287 raise ToolError(

288 "SECTION_NOT_FOUND",

289 f"Section '{section_id}' not found in document"

290 )

291

292 # Save document

293 self.save_document(path_info, json_data)

294

295 def update_block(

296 self,

297 path_info: PathInfo,

298 block_id: str,

299 content: Optional[str] = None,

300 title: Optional[str] = None,

301 items: Optional[list] = None,

302 language: Optional[str] = None,

303 header: Optional[list] = None,

304 rows: Optional[list] = None

305 ) -> None:

306 """

307 Update block content based on block type.

308

309 Supports updating different block types with appropriate parameters:

310 - paragraph: content

311 - code: content, language (optional)

312 - list: title (optional), items

313 - table: header, rows

314

315 Args:

316 path_info: PathInfo with file path

317 block_id: Block ID to update

318 content: New content (for paragraph/code blocks)

319 title: New title (for list blocks)

320 items: New items list (for list blocks)

321 language: Code language (for code blocks)

322 header: Table header row (for table blocks)

323 rows: Table data rows (for table blocks)

324

325 Raises:

326 ToolError: If block not found

327

328 @REQ: REQ-rbt-mcp-tool

329 @BP: BP-rbt-mcp-tool

330 @TASK: TASK-010-UpdateBlockTool

331 """

332 # Load document

333 json_data = self.load_document(path_info)

334

335 # Search for block and update it

336 def find_and_update_block(sections, target_id):

337 """Recursively search for block and update it."""

338 for section in sections:

339 # Search in this section's blocks

340 if "blocks" in section:

341 for block in section["blocks"]:

342 if block.get("id") == target_id:

343 # Found the block - update based on type

344 block_type = block.get("type")

345

346 if block_type == "paragraph":

347 if content is not None:

348 block["content"] = content

349

350 elif block_type == "code":

351 if content is not None:

352 block["content"] = content

353 if language is not None:

354 block["language"] = language

355

356 elif block_type == "list":

357 if title is not None:

358 block["title"] = title

359 if items is not None:

360 block["items"] = items

361

362 elif block_type == "table":

363 if header is not None:

364 block["header"] = header

365 if rows is not None:

366 block["rows"] = rows

367

368 return True

369

370 # Search in sub-sections

371 if "sections" in section:

372 if find_and_update_block(section["sections"], target_id):

373 return True

374 return False

375

376 updated = False

377 if "sections" in json_data:

378 updated = find_and_update_block(json_data["sections"], block_id)

379

380 if not updated:

381 raise ToolError(

382 "BLOCK_NOT_FOUND",

383 f"Block '{block_id}' not found in document"

384 )

385

386 # Save document

387 self.save_document(path_info, json_data)

388

389 def delete_block(

390 self,

391 path_info: PathInfo,

392 block_id: str

393 ) -> None:

394 """

395 Delete block from document.

396

397 Removes the specified block from its containing section.

398 Does not affect other blocks in the section.

399

400 Args:

401 path_info: PathInfo with file path

402 block_id: Block ID to delete

403

404 Raises:

405 ToolError: If block not found

406

407 @REQ: REQ-rbt-mcp-tool

408 @BP: BP-rbt-mcp-tool

409 @TASK: TASK-011-DeleteBlockTool

410 """

411 # Load document

412 json_data = self.load_document(path_info)

413

414 # Search for block and delete it from all sections

415 def find_and_delete_block(sections, target_id):

416 """Recursively search for block and delete it."""

417 for section in sections:

418 # Search in this section's blocks

419 if "blocks" in section:

420 for i, block in enumerate(section["blocks"]):

421 if block.get("id") == target_id:

422 # Found the block - delete it

423 del section["blocks"][i]

424 return True

425

426 # Search in sub-sections

427 if "sections" in section:

428 if find_and_delete_block(section["sections"], target_id):

429 return True

430 return False

431

432 deleted = False

433 if "sections" in json_data:

434 deleted = find_and_delete_block(json_data["sections"], block_id)

435

436 if not deleted:

437 raise ToolError(

438 "BLOCK_NOT_FOUND",

439 f"Block '{block_id}' not found in document"

440 )

441

442 # Save document

443 self.save_document(path_info, json_data)

444

445 def append_list_item(

446 self,

447 path_info: PathInfo,

448 block_id: str,

449 item: str

450 ) -> None:

451 """

452 Append item to list block.

453

454 Only works with list type blocks. Appends new item to the end of

455 the items array.

456

457 Args:

458 path_info: PathInfo with file path

459 block_id: Block ID (must be a list type block)

460 item: Item text to append

461

462 Raises:

463 ToolError: If block not found or block is not a list type

464

465 @REQ: REQ-rbt-mcp-tool

466 @BP: BP-rbt-mcp-tool

467 @TASK: TASK-012-AppendListItemTool

468 """

469 # Load document

470 json_data = self.load_document(path_info)

471

472 # Search for block in all sections

473 def find_and_append_to_block(sections, target_id, new_item):

474 """Recursively search for block and append item."""

475 for section in sections:

476 # Search in this section's blocks

477 if "blocks" in section:

478 for block in section["blocks"]:

479 if block.get("id") == target_id:

480 # Found the block - verify it's a list type

481 block_type = block.get("type")

482 if block_type != "list":

483 raise ToolError(

484 "INVALID_BLOCK_TYPE",

485 f"Block '{target_id}' is type '{block_type}', not 'list'. "

486 f"append_list_item only works with list blocks."

487 )

488

489 # Append item to the list

490 if "items" not in block:

491 block["items"] = []

492 block["items"].append(new_item)

493 return True

494

495 # Search in sub-sections

496 if "sections" in section:

497 if find_and_append_to_block(section["sections"], target_id, new_item):

498 return True

499 return False

500

501 found = False

502 if "sections" in json_data:

503 found = find_and_append_to_block(json_data["sections"], block_id, item)

504

505 if not found:

506 raise ToolError(

507 "BLOCK_NOT_FOUND",

508 f"Block '{block_id}' not found in document"

509 )

510

511 # Save document

512 self.save_document(path_info, json_data)

513

514 def clear_cache(self, file_path: Optional[str] = None) -> None:

515 """

516 Clear document cache.

517

518 Args:

519 file_path: If provided, clear only this file; if None, clear all

520

521 @REQ: REQ-rbt-mcp-tool

522 @BP: BP-rbt-mcp-tool

523 @TASK: TASK-003-DocumentService

524 """

525 self.cache.clear(file_path)

526

527 def __del__(self):

528 """Stop cache cleanup thread on destruction."""

529 if hasattr(self, 'cache'):

530 self.cache.stop()

Coverage for rbt_mcp_server/document_service.py: 42%

179 statements