From 5315e4f346d6083eb22deebfd4dd5473e8f2cf94 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Jan 2026 22:44:41 +0100 Subject: [PATCH] Phase 6: Add tree-style progress visualization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TreePrinter class for formatted output - Tree characters (├──, └──, │) for hierarchy - Status indicators (✓ created, ○ skipped, ✗ error) - Box-drawing characters for header/summary - Consistent output format matching spec Co-Authored-By: Claude Opus 4.5 --- outline_import.py | 256 ++++++++++++++++++++++++++++++---------------- 1 file changed, 167 insertions(+), 89 deletions(-) diff --git a/outline_import.py b/outline_import.py index d8ac2d2..ea92860 100644 --- a/outline_import.py +++ b/outline_import.py @@ -40,6 +40,43 @@ logging.basicConfig( logger = logging.getLogger('outline_import') +class TreePrinter: + """Utility for printing tree-style output.""" + + PIPE = "│ " + ELBOW = "└── " + TEE = "├── " + BLANK = " " + + @staticmethod + def format_line(title: str, status: str, message: str = None, prefix: str = "") -> str: + """Format a tree line with status indicator.""" + # Status symbols and labels + if status == "created": + symbol = "✓" + label = "created" + elif status == "skipped": + symbol = "○" + label = "skipped" + elif status == "dry_run": + symbol = "○" + label = "(dry run)" + else: + symbol = "✗" + label = message or "error" + + # Truncate title if too long + max_title_len = 40 + if len(title) > max_title_len: + display_title = title[:max_title_len - 3] + "..." + else: + display_title = title + + # Format: prefix + title + padding + symbol + label + filename = f"{display_title}.md" + return f"{prefix}{filename:<45} {symbol} {label}" + + class OutlineImporter: """Import documents into Outline with hierarchy preservation.""" @@ -181,14 +218,16 @@ class OutlineImporter: Returns: True if API is accessible and authenticated """ - logger.info("Checking API connectivity...") + print("Checking API connectivity...", end=" ") result = self._api_request("/api/auth.info") if result and "data" in result: user = result["data"].get("user", {}) team = result["data"].get("team", {}) - logger.info(f"Authenticated as: {user.get('name', 'Unknown')} ({user.get('email', 'N/A')})") - logger.info(f"Team: {team.get('name', 'Unknown')}") + print("✓") + logger.debug(f"Authenticated as: {user.get('name', 'Unknown')} ({user.get('email', 'N/A')})") + logger.debug(f"Team: {team.get('name', 'Unknown')}") return True + print("✗") logger.error("Health check failed: Unable to verify authentication") return False @@ -453,7 +492,7 @@ class OutlineImporter: parent_document_id: Optional[str] = None ) -> Tuple[int, int, int]: """ - Import a single collection. + Import a single collection with tree-style output. Args: collection_dir: Path to collection directory @@ -492,108 +531,144 @@ class OutlineImporter: # Check if collection exists if collection_name in self.existing_collections: if self.force: - logger.info(f" Deleting existing collection \"{collection_name}\"...") + print(f" Deleting existing collection \"{collection_name}\"...") if not self.dry_run: self._delete_collection(self.existing_collections[collection_name]) del self.existing_collections[collection_name] else: - logger.info(f" Collection exists, skipping...") + print(f" Collection exists, skipping...") self.stats["collections_skipped"] += 1 return (0, doc_count, 0) # Create collection - logger.info(f" Creating collection...") - collection_id = self._create_collection(collection_name) - if not collection_id: - self.stats["collections_errors"] += 1 - self.errors.append({ - "type": "collection", - "name": collection_name, - "error": "Failed to create collection" - }) - return (0, 0, 1) + if self.dry_run: + print(f" [DRY RUN] Would create collection \"{collection_name}\"") + collection_id = "dry-run-collection-id" + else: + print(f" Creating collection...", end=" ") + collection_id = self._create_collection(collection_name) + if not collection_id: + print("✗ failed") + self.stats["collections_errors"] += 1 + self.errors.append({ + "type": "collection", + "name": collection_name, + "error": "Failed to create collection" + }) + return (0, 0, 1) + print(f"✓ (id: {collection_id[:8]}...)") - if not self.dry_run: - logger.info(f" ✓ (id: {collection_id[:8]}...)") self.stats["collections_created"] += 1 - # Build document tree and flatten for import + # Build document tree for tree-style import doc_tree = self.build_document_tree(documents) - import_order = self.flatten_for_import(doc_tree) - # Import documents + # Import documents with tree visualization created = 0 skipped = 0 errors = 0 - for doc_meta in import_order: - old_id = doc_meta["id"] - title = doc_meta["title"] - filename = doc_meta["filename"] - old_parent_id = doc_meta.get("parent_id") + def import_tree_recursive( + docs: List[Dict], + prefix: str = " ", + coll_id: str = None, + default_parent_id: str = None + ) -> Tuple[int, int, int]: + """Recursively import documents with tree-style output.""" + nonlocal created, skipped, errors - # Resolve parent ID - new_parent_id = parent_document_id # Default for single mode - if old_parent_id: - new_parent_id = self.id_map.get(old_parent_id) - if not new_parent_id and not self.dry_run: - logger.warning(f"Parent not found for {title}, creating as root-level") + for i, doc in enumerate(docs): + is_last = (i == len(docs) - 1) + connector = TreePrinter.ELBOW if is_last else TreePrinter.TEE - # Read content - content = self.read_document_content(collection_dir, filename) - if content is None: - self._print_doc_status(title, "error", "file not found") - errors += 1 - self.stats["documents_errors"] += 1 - self.errors.append({ - "type": "document", - "title": title, - "collection": collection_name, - "error": "File not found" - }) - continue + old_id = doc["id"] + title = doc["title"] + filename = doc["filename"] + old_parent_id = doc.get("parent_id") + children = doc.get("_children", []) or doc.get("children", []) - # Create document - new_id = self._create_document( - collection_id, - title, - content, - parent_document_id=new_parent_id - ) + # Resolve parent ID + new_parent_id = default_parent_id + if old_parent_id: + new_parent_id = self.id_map.get(old_parent_id, default_parent_id) - if new_id: - self.id_map[old_id] = new_id - self._print_doc_status(title, "created") - created += 1 - self.stats["documents_created"] += 1 - else: - self._print_doc_status(title, "error", "API error") - errors += 1 - self.stats["documents_errors"] += 1 - self.errors.append({ - "type": "document", - "title": title, - "collection": collection_name, - "error": "API error during creation" - }) + # Read content + content = self.read_document_content(collection_dir, filename) + if content is None: + line = TreePrinter.format_line(title, "error", "file not found", prefix + connector) + print(line) + errors += 1 + self.stats["documents_errors"] += 1 + self.errors.append({ + "type": "document", + "title": title, + "collection": collection_name, + "error": "File not found" + }) + # Skip children if parent failed + if children: + child_prefix = prefix + (TreePrinter.BLANK if is_last else TreePrinter.PIPE) + print(f"{child_prefix}└── (children skipped due to parent failure)") + continue + + # Create document + if self.dry_run: + line = TreePrinter.format_line(title, "dry_run", prefix=prefix + connector) + print(line) + self.id_map[old_id] = f"dry-run-{old_id}" + created += 1 + self.stats["documents_created"] += 1 + else: + new_id = self._create_document( + coll_id, + title, + content, + parent_document_id=new_parent_id + ) + + if new_id: + self.id_map[old_id] = new_id + line = TreePrinter.format_line(title, "created", prefix=prefix + connector) + print(line) + created += 1 + self.stats["documents_created"] += 1 + else: + line = TreePrinter.format_line(title, "error", "API error", prefix + connector) + print(line) + errors += 1 + self.stats["documents_errors"] += 1 + self.errors.append({ + "type": "document", + "title": title, + "collection": collection_name, + "error": "API error during creation" + }) + # Skip children if parent failed + if children: + child_prefix = prefix + (TreePrinter.BLANK if is_last else TreePrinter.PIPE) + print(f"{child_prefix}└── (children skipped due to parent failure)") + continue + + # Process children recursively + if children: + child_prefix = prefix + (TreePrinter.BLANK if is_last else TreePrinter.PIPE) + import_tree_recursive( + children, + prefix=child_prefix, + coll_id=coll_id, + default_parent_id=self.id_map.get(old_id, default_parent_id) + ) + + # Start recursive import + import_tree_recursive( + doc_tree, + prefix=" ", + coll_id=collection_id, + default_parent_id=parent_document_id + ) return (created, skipped, errors) - def _print_doc_status(self, title: str, status: str, message: str = None): - """Print document import status.""" - if status == "created": - symbol = "✓" - label = "created" - elif status == "skipped": - symbol = "○" - label = "skipped" - else: - symbol = "✗" - label = message or "error" - - # This will be enhanced in Phase 6 with tree formatting - logger.info(f" {symbol} {title[:50]:<50} {label}") - def import_all(self) -> None: """Import all collections from source directory.""" start_time = time.time() @@ -602,9 +677,9 @@ class OutlineImporter: mode_str = "Single collection" if self.single_mode else "Collection per folder" dry_run_str = " (DRY RUN)" if self.dry_run else "" - print("=" * 60) + print("════════════════════════════════════════════════════════════") print(f" OUTLINE IMPORT{dry_run_str}") - print("=" * 60) + print("════════════════════════════════════════════════════════════") print() print(f"Source: {self.source_dir}/") print(f"Target: {self.base_url}") @@ -687,9 +762,12 @@ class OutlineImporter: # Print summary duration = time.time() - start_time print() - print("=" * 60) - print("SUMMARY") - print("=" * 60) + print("════════════════════════════════════════════════════════════") + if self.dry_run: + print("DRY RUN SUMMARY") + else: + print("SUMMARY") + print("════════════════════════════════════════════════════════════") print(f" Collections: {self.stats['collections_created']} created, " f"{self.stats['collections_skipped']} skipped, " f"{self.stats['collections_errors']} errors") @@ -697,11 +775,11 @@ class OutlineImporter: f"{self.stats['documents_skipped']} skipped, " f"{self.stats['documents_errors']} errors") print(f" Duration: {duration:.1f} seconds") - print("=" * 60) + print("════════════════════════════════════════════════════════════") if self.errors: print() - logger.warning(f"Encountered {len(self.errors)} errors during import") + print(f"Encountered {len(self.errors)} errors during import:") def load_settings(settings_file: str = "settings.json") -> Dict: