| 56 | |
| 57 | |
| 58 | class PostsMigrator: |
| 59 | def __init__(self, site_url: str, output_dir: str, delay: float = 0.3, |
| 60 | dry_run: bool = False): |
| 61 | self.site_url = site_url.rstrip("/") |
| 62 | self.output_dir = Path(output_dir) |
| 63 | self.delay = delay |
| 64 | self.dry_run = dry_run |
| 65 | self.session = requests.Session() |
| 66 | self.session.headers.update({ |
| 67 | "User-Agent": "docmcp-knowledge-migrator/1.0" |
| 68 | }) |
| 69 | self.h2t = html2text.HTML2Text() |
| 70 | self.h2t.ignore_links = False |
| 71 | self.h2t.ignore_images = False |
| 72 | self.h2t.body_width = 0 |
| 73 | self.h2t.protect_links = True |
| 74 | self.h2t.unicode_snob = True |
| 75 | self.stats: Dict[str, int] = { |
| 76 | "total": 0, "success": 0, "skipped": 0, "error": 0 |
| 77 | } |
| 78 | self._category_cache: Dict[int, str] = {} |
| 79 | |
| 80 | def api_get(self, endpoint: str, params: Optional[dict] = None) -> Optional[object]: |
| 81 | url = f"{self.site_url}/wp-json/wp/v2/{endpoint}" |
| 82 | try: |
| 83 | resp = self.session.get(url, params=params, timeout=30) |
| 84 | if resp.status_code == 400: |
| 85 | return None |
| 86 | resp.raise_for_status() |
| 87 | return resp.json() |
| 88 | except requests.RequestException as e: |
| 89 | print(f" ERROR fetching {url}: {e}") |
| 90 | return None |
| 91 | |
| 92 | def get_all_posts(self) -> List[dict]: |
| 93 | """Fetch all published posts with pagination.""" |
| 94 | all_posts = [] |
| 95 | page = 1 |
| 96 | params: dict = { |
| 97 | "per_page": 100, |
| 98 | "status": "publish", |
| 99 | "_fields": "id,title,slug,link,content,excerpt,date,modified,categories,tags", |
| 100 | } |
| 101 | while True: |
| 102 | params["page"] = page |
| 103 | posts = self.api_get("posts", params) |
| 104 | if not posts or not isinstance(posts, list): |
| 105 | break |
| 106 | all_posts.extend(posts) |
| 107 | print(f" Fetched page {page}: {len(posts)} posts (total: {len(all_posts)})") |
| 108 | if len(posts) < 100: |
| 109 | break |
| 110 | page += 1 |
| 111 | time.sleep(self.delay) |
| 112 | return all_posts |
| 113 | |
| 114 | def get_category_slug(self, cat_id: int) -> str: |
| 115 | """Get category slug by ID (cached).""" |