Phase 6: Add tree-style progress visualization

- 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 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-01-19 22:44:41 +01:00
parent 290030f5e8
commit 5315e4f346

View File

@@ -40,6 +40,43 @@ logging.basicConfig(
logger = logging.getLogger('outline_import') 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: class OutlineImporter:
"""Import documents into Outline with hierarchy preservation.""" """Import documents into Outline with hierarchy preservation."""
@@ -181,14 +218,16 @@ class OutlineImporter:
Returns: Returns:
True if API is accessible and authenticated True if API is accessible and authenticated
""" """
logger.info("Checking API connectivity...") print("Checking API connectivity...", end=" ")
result = self._api_request("/api/auth.info") result = self._api_request("/api/auth.info")
if result and "data" in result: if result and "data" in result:
user = result["data"].get("user", {}) user = result["data"].get("user", {})
team = result["data"].get("team", {}) team = result["data"].get("team", {})
logger.info(f"Authenticated as: {user.get('name', 'Unknown')} ({user.get('email', 'N/A')})") print("")
logger.info(f"Team: {team.get('name', 'Unknown')}") logger.debug(f"Authenticated as: {user.get('name', 'Unknown')} ({user.get('email', 'N/A')})")
logger.debug(f"Team: {team.get('name', 'Unknown')}")
return True return True
print("")
logger.error("Health check failed: Unable to verify authentication") logger.error("Health check failed: Unable to verify authentication")
return False return False
@@ -453,7 +492,7 @@ class OutlineImporter:
parent_document_id: Optional[str] = None parent_document_id: Optional[str] = None
) -> Tuple[int, int, int]: ) -> Tuple[int, int, int]:
""" """
Import a single collection. Import a single collection with tree-style output.
Args: Args:
collection_dir: Path to collection directory collection_dir: Path to collection directory
@@ -492,108 +531,144 @@ class OutlineImporter:
# Check if collection exists # Check if collection exists
if collection_name in self.existing_collections: if collection_name in self.existing_collections:
if self.force: if self.force:
logger.info(f" Deleting existing collection \"{collection_name}\"...") print(f" Deleting existing collection \"{collection_name}\"...")
if not self.dry_run: if not self.dry_run:
self._delete_collection(self.existing_collections[collection_name]) self._delete_collection(self.existing_collections[collection_name])
del self.existing_collections[collection_name] del self.existing_collections[collection_name]
else: else:
logger.info(f" Collection exists, skipping...") print(f" Collection exists, skipping...")
self.stats["collections_skipped"] += 1 self.stats["collections_skipped"] += 1
return (0, doc_count, 0) return (0, doc_count, 0)
# Create collection # Create collection
logger.info(f" Creating collection...") if self.dry_run:
collection_id = self._create_collection(collection_name) print(f" [DRY RUN] Would create collection \"{collection_name}\"")
if not collection_id: collection_id = "dry-run-collection-id"
self.stats["collections_errors"] += 1 else:
self.errors.append({ print(f" Creating collection...", end=" ")
"type": "collection", collection_id = self._create_collection(collection_name)
"name": collection_name, if not collection_id:
"error": "Failed to create collection" print("✗ failed")
}) self.stats["collections_errors"] += 1
return (0, 0, 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 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) doc_tree = self.build_document_tree(documents)
import_order = self.flatten_for_import(doc_tree)
# Import documents # Import documents with tree visualization
created = 0 created = 0
skipped = 0 skipped = 0
errors = 0 errors = 0
for doc_meta in import_order: def import_tree_recursive(
old_id = doc_meta["id"] docs: List[Dict],
title = doc_meta["title"] prefix: str = " ",
filename = doc_meta["filename"] coll_id: str = None,
old_parent_id = doc_meta.get("parent_id") default_parent_id: str = None
) -> Tuple[int, int, int]:
"""Recursively import documents with tree-style output."""
nonlocal created, skipped, errors
# Resolve parent ID for i, doc in enumerate(docs):
new_parent_id = parent_document_id # Default for single mode is_last = (i == len(docs) - 1)
if old_parent_id: connector = TreePrinter.ELBOW if is_last else TreePrinter.TEE
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")
# Read content old_id = doc["id"]
content = self.read_document_content(collection_dir, filename) title = doc["title"]
if content is None: filename = doc["filename"]
self._print_doc_status(title, "error", "file not found") old_parent_id = doc.get("parent_id")
errors += 1 children = doc.get("_children", []) or doc.get("children", [])
self.stats["documents_errors"] += 1
self.errors.append({
"type": "document",
"title": title,
"collection": collection_name,
"error": "File not found"
})
continue
# Create document # Resolve parent ID
new_id = self._create_document( new_parent_id = default_parent_id
collection_id, if old_parent_id:
title, new_parent_id = self.id_map.get(old_parent_id, default_parent_id)
content,
parent_document_id=new_parent_id
)
if new_id: # Read content
self.id_map[old_id] = new_id content = self.read_document_content(collection_dir, filename)
self._print_doc_status(title, "created") if content is None:
created += 1 line = TreePrinter.format_line(title, "error", "file not found", prefix + connector)
self.stats["documents_created"] += 1 print(line)
else: errors += 1
self._print_doc_status(title, "error", "API error") self.stats["documents_errors"] += 1
errors += 1 self.errors.append({
self.stats["documents_errors"] += 1 "type": "document",
self.errors.append({ "title": title,
"type": "document", "collection": collection_name,
"title": title, "error": "File not found"
"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
# 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) 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: def import_all(self) -> None:
"""Import all collections from source directory.""" """Import all collections from source directory."""
start_time = time.time() start_time = time.time()
@@ -602,9 +677,9 @@ class OutlineImporter:
mode_str = "Single collection" if self.single_mode else "Collection per folder" mode_str = "Single collection" if self.single_mode else "Collection per folder"
dry_run_str = " (DRY RUN)" if self.dry_run else "" dry_run_str = " (DRY RUN)" if self.dry_run else ""
print("=" * 60) print("════════════════════════════════════════════════════════════")
print(f" OUTLINE IMPORT{dry_run_str}") print(f" OUTLINE IMPORT{dry_run_str}")
print("=" * 60) print("════════════════════════════════════════════════════════════")
print() print()
print(f"Source: {self.source_dir}/") print(f"Source: {self.source_dir}/")
print(f"Target: {self.base_url}") print(f"Target: {self.base_url}")
@@ -687,9 +762,12 @@ class OutlineImporter:
# Print summary # Print summary
duration = time.time() - start_time duration = time.time() - start_time
print() print()
print("=" * 60) print("════════════════════════════════════════════════════════════")
print("SUMMARY") if self.dry_run:
print("=" * 60) print("DRY RUN SUMMARY")
else:
print("SUMMARY")
print("════════════════════════════════════════════════════════════")
print(f" Collections: {self.stats['collections_created']} created, " print(f" Collections: {self.stats['collections_created']} created, "
f"{self.stats['collections_skipped']} skipped, " f"{self.stats['collections_skipped']} skipped, "
f"{self.stats['collections_errors']} errors") f"{self.stats['collections_errors']} errors")
@@ -697,11 +775,11 @@ class OutlineImporter:
f"{self.stats['documents_skipped']} skipped, " f"{self.stats['documents_skipped']} skipped, "
f"{self.stats['documents_errors']} errors") f"{self.stats['documents_errors']} errors")
print(f" Duration: {duration:.1f} seconds") print(f" Duration: {duration:.1f} seconds")
print("=" * 60) print("════════════════════════════════════════════════════════════")
if self.errors: if self.errors:
print() 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: def load_settings(settings_file: str = "settings.json") -> Dict: