Main migration entry point.
(self, url_filter: Optional[str] = None)
| 308 | print(f"\n Total: {len(docs)} docs across {len(path_counts)} directories") |
| 309 | |
| 310 | def migrate(self, url_filter: Optional[str] = None) -> None: |
| 311 | """Main migration entry point.""" |
| 312 | mode = "[DRY RUN] " if self.dry_run else "" |
| 313 | print(f"\n{mode}Starting migration from {self.site_url}") |
| 314 | print(f"Output directory: {self.output_dir}") |
| 315 | if url_filter: |
| 316 | print(f"URL filter: {url_filter}") |
| 317 | print() |
| 318 | |
| 319 | docs = self.get_all_docs() |
| 320 | |
| 321 | if url_filter: |
| 322 | docs = [d for d in docs if url_filter.lower() in d.get("link", "").lower()] |
| 323 | print(f"After filter: {len(docs)} docs match '{url_filter}'") |
| 324 | |
| 325 | print(f"\nTotal docs to process: {len(docs)}") |
| 326 | |
| 327 | path_entries: Dict[str, List[dict]] = {} |
| 328 | |
| 329 | for i, doc in enumerate(docs, 1): |
| 330 | self.stats["total"] += 1 |
| 331 | title = self.clean_title(doc.get("title", {}).get("rendered", "")) |
| 332 | link = doc.get("link", "") |
| 333 | output_path, _ = self.detect_output_path(link) |
| 334 | print(f"[{i:3d}/{len(docs)}] {title[:65]}") |
| 335 | |
| 336 | try: |
| 337 | zh_file, fm = self.save_document(doc) |
| 338 | if fm is None: |
| 339 | self.stats["skipped"] += 1 |
| 340 | print(f" SKIP (empty content)") |
| 341 | continue |
| 342 | |
| 343 | if output_path not in path_entries: |
| 344 | path_entries[output_path] = [] |
| 345 | path_entries[output_path].append({ |
| 346 | "id": fm["id"], |
| 347 | "title": fm["title"], |
| 348 | "status": fm["status"], |
| 349 | "source_url": fm["source_url"], |
| 350 | "source_url_verified": fm["source_url_verified"], |
| 351 | "effective_date": fm.get("effective_date", ""), |
| 352 | }) |
| 353 | |
| 354 | self.stats["success"] += 1 |
| 355 | if self.dry_run: |
| 356 | print(f" -> [DRY] {output_path}/{doc.get('slug', '')}.zh.md") |
| 357 | else: |
| 358 | print(f" -> {zh_file.relative_to(self.output_dir)}") |
| 359 | except Exception as e: |
| 360 | self.stats["error"] += 1 |
| 361 | print(f" ERROR: {e}") |
| 362 | |
| 363 | time.sleep(self.delay) |
| 364 | |
| 365 | if not self.dry_run: |
| 366 | print("\nUpdating _index.json files...") |
| 367 | for path, entries in path_entries.items(): |
no test coverage detected